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.

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")

Quick Start

1. Default (single-tenant)

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

   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
   
2. 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.

   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)
   
3. 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.

   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.

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.

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:

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.

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:

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

Previously this endpoint returned 200 with a placeholder payload that implied success, but nothing was registered. Client code that relied on 200 OK to confirm registration should call register_agent(...) in the server process instead.

---

Related

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