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.

---

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.

---

Related

AgentOS Chat Session Isolation

Session isolation cloning path

Serve Agents Auth

Surface where this registry is invoked