Agent Invoke Registry

Scope the agent invocation registry per tenant or per embedded server — race-free

The agent invoke registry maps an agent id to a template agent, so a request handler can resolve and invoke the right agent by id.

python

from praisonaiagents import Agent
from praisonai.api.agent_invoke import register_agent, get_agent

agent = Agent(
    name="assistant",
    instructions="You are a helpful assistant."
)
register_agent("assistant", agent)          # default (module-global) registry
# Later, in a request handler:
resolved = get_agent("assistant")

Diagram

Quick Start

Default (single-tenant)

Use the module-level functions — no registry= kwarg needed. This is what almost every user wants.

python

from praisonaiagents import Agent
from praisonai.api.agent_invoke import register_agent, get_agent

agent = Agent(
    name="assistant",
    instructions="You are a helpful assistant."
)
register_agent("assistant", agent)
resolved = get_agent("assistant")   # -> agent

Scoped (multi-tenant / multi-embedding)

Construct an AgentRegistry() per tenant and pass it via the registry= kwarg. Two AgentOS gateways in the same interpreter can now hold isolated agent sets.

python

from praisonaiagents import Agent
from praisonai.api.agent_invoke import AgentRegistry, register_agent, get_agent

agent = Agent(name="assistant", instructions="You are a helpful assistant.")

tenant_a = AgentRegistry()
tenant_b = AgentRegistry()

register_agent("assistant", agent, registry=tenant_a)
get_agent("assistant", registry=tenant_a)   # -> agent
get_agent("assistant", registry=tenant_b)   # -> None (isolated)

Bind a per-tenant registry to a mounted FastAPI app

Passing registry= to register_agent(...) scopes the free functions, but the mounted router’s route handlers resolve their registry through Depends(_registry_dependency). Override that dependency so /agents, /agents/{id}, and /agents/{id}/invoke serve your registry instead of the process-global one.

python

from fastapi import FastAPI
from praisonaiagents import Agent
from praisonai.api.agent_invoke import (
    AgentRegistry, register_agent, router, _registry_dependency,
)

# Per-tenant registry — isolated from the module-global one
tenant_registry = AgentRegistry()
register_agent("assistant", Agent(name="assistant",
    instructions="You are a helpful assistant."),
    registry=tenant_registry)

app = FastAPI()
app.include_router(router)

# This one line makes every route handler resolve through tenant_registry
# instead of the process-global _default_registry.
app.dependency_overrides[_registry_dependency] = lambda: tenant_registry

The override is per FastAPI app instance, so two AgentOS gateways in the same interpreter now serve their own agent set through /agents, /agents/{id}, and /agents/{id}/invoke — not a shared list.

How It Works

Every invocation resolves through resolve_session_agent(agent_id, session_id, *, registry=None), which clones the registered template per request so concurrent callers never share mutable conversation state. The invoke handler passes in whatever Depends(_registry_dependency) resolved to, so a per-app override reaches the resolver.

Diagram

unregister pops the id atomically under a lock (dict.pop), so a concurrent unregister can’t race between the membership check and the delete — the previous split check-then-delete returned uncaught 500s on concurrent deletes, which is the reason the class exists.

Per-request clones are closed for you. The invoke handler wraps every clone in try/finally and closes it (preferring aclose(), falling back to sync close() in a worker thread) after every request. Only real clones are closed; when the resolved agent is the shared registered template (plain mocks / non-isolatable objects), it is left untouched. Deriving ownership from the resolved object itself — rather than a second, independently-timed registry lookup — is race-free even when the registry entry is concurrently replaced. See Agent Lifecycle Cleanup.

Configuration Options

AgentRegistry methods:

Method	Signature	Description
`register`	`register(agent_id: str, agent: Any) -> None`	Add an agent template under `agent_id`.
`unregister`	`unregister(agent_id: str) -> bool`	Atomic remove; returns `True` if it existed. Race-free under concurrent calls.
`get`	`get(agent_id: str) -> Optional[Any]`	Look up an agent template by id.
`list`	`list() -> list[str]`	List all registered agent ids.

Module-level functions (backward-compatible):

Function	New signature	Notes
`register_agent`	`register_agent(agent_id, agent, *, registry=None)`	Uses module-global when `registry` is omitted.
`unregister_agent`	`unregister_agent(agent_id, *, registry=None)`	Same.
`get_agent`	`get_agent(agent_id, *, registry=None)`	Same.
`list_registered_agents`	`list_registered_agents(*, registry=None)`	Same.

Best Practices

Prefer the default registry unless you run multiple tenants

The module-global is already thread-safe, and the registry= kwarg is the escape hatch, not the default path. Use plain register_agent(agent_id, agent) / get_agent(agent_id) for single-tenant setups.

Give each embedded server its own AgentRegistry

Two AgentOS gateways sharing the module global will step on each other’s agent ids. Construct one AgentRegistry() per gateway and thread it through your register_agent(...) calls.

python

from praisonai.api.agent_invoke import AgentRegistry, register_agent

gateway_a_registry = AgentRegistry()
gateway_b_registry = AgentRegistry()

register_agent("assistant", agent, registry=gateway_a_registry)

When embedding the router, override _registry_dependency

Just calling register_agent(..., registry=my_reg) isn’t enough — the router’s route handlers resolve their registry via Depends(_registry_dependency). To serve your registry through the API, add:

python

app.dependency_overrides[_registry_dependency] = lambda: my_reg

_registry_dependency is an override key, not a helper you call from user code — its name starts with _ on purpose.

Registry holds templates, not live conversations

Every invocation goes through resolve_session_agent, which clones the template per request via clone_for_channel. Do not stash mutable state on the registered agent expecting it to survive across calls.

Don't stash live resources on the registered template

The registry holds a template; every invoke resolves a per-request clone via clone_for_channel(). Live resources (open sockets, connection pools, subprocesses) put on the template survive across requests but are also not released by the invoke wrapper’s per-request cleanup — the wrapper deliberately skips the shared template. Attach resources to the clone through Agent‘s standard fields so aclose() on the clone cleans them up.

Registering Agents

Agents are registered in-process with register_agent(...) — this is the only supported path.

python

from praisonaiagents import Agent
from praisonai.api.agent_invoke import register_agent

agent = Agent(name="assistant", instructions="Help the user")
register_agent("assistant", agent)

POST /api/v1/agents/{agent_id}/register is not a registration surface. It returns 501 Not Implemented:

json

{
  "detail": "Dynamic HTTP registration of 'assistant' is not supported; register agents in-process via praisonai.api.agent_invoke.register_agent(...)"
}

AgentOS Chat Session Isolation

Session isolation cloning path

Serve Agents Auth

Surface where this registry is invoked

Agent Lifecycle Cleanup

How the per-request clone releases its resources on teardown

n8n API

Per-request clone cleanup in the HTTP invoke endpoint

Agent Invoke Registry

Quick Start

How It Works

Configuration Options

Best Practices

Registering Agents

Related