Factories

Registered callables that produce a fresh Agent, Team, or Workflow per request. AgentFactory, TeamFactory, and WorkflowFactory are thin subclasses of BaseFactory[T] and share the same constructor; they differ only in the component type produced.

Import

from agno.agent import AgentFactory
from agno.team import TeamFactory
from agno.workflow import WorkflowFactory
from agno.factory import (
    BaseFactory,
    RequestContext,
    TrustedContext,
    FactoryError,
    FactoryValidationError,
    FactoryPermissionError,
    FactoryContextRequired,
)

Parameters

Shared constructor signature for AgentFactory, TeamFactory, and WorkflowFactory:

Parameter	Type	Default	Description
id	str	required	Stable handle used in API URLs (e.g. POST /agents/{id}/runs). The produced component’s id is enforced to match this. Empty string raises ValueError.
db	Union[BaseDb, AsyncBaseDb]	required	Database for session storage. Used as the fallback db if the produced component’s db attribute is falsy. None raises ValueError.
factory	Union[Callable[[RequestContext], T], Callable[[RequestContext], Awaitable[T]]]	required	Callable invoked on every run-scoped request. Returns a fresh component of type T. Both sync and async callables are accepted.
name	Optional[str]	None	Human-readable name for UI discovery.
description	Optional[str]	None	Description for UI discovery.
input_schema	Optional[Type[BaseModel]]	None	Pydantic model describing the expected shape of factory_input in the run request. Used for validation and OpenAPI schema generation.

After the factory returns, AgentOS enforces the component’s id to match the factory’s id, fills the component’s db from the factory if unset, sets store_events=True so per-step run events persist on the run output, and runs the component-specific initialize_*().

RequestContext

AgentOS passes a RequestContext as the sole argument to every factory call.

Field	Type	Source	Description
user_id	Optional[str]	request.state.user_id (set by middleware), with form field as fallback	Caller identity. JWT claims reach this field only via JWTMiddleware setting request.state.user_id.
session_id	Optional[str]	request.state.session_id, with form field as fallback	Conversation/session handle. Same precedence rules as user_id.
input	Optional[BaseModel | dict]	factory_input form field or query parameter	Validated against the factory’s input_schema if set. Otherwise the parsed JSON object or None.
trusted	TrustedContext	request.state.claims / request.state.scopes	Populated only by verified middleware. Use for authorization.
trusted.claims	Mapping[str, Any]	request.state.claims	Token claims set by verified middleware (e.g. JWTMiddleware).
trusted.scopes	frozenset[str]	request.state.scopes	RBAC scopes set by verified middleware.
request	Any	Raw FastAPI Request	Escape hatch for anything not plumbed through the typed fields. Typed as Any in the source to avoid forcing a FastAPI import.

factory_input by endpoint

Endpoint	Source
POST /{agents,teams,workflows}/{id}/runs	Form field
POST /workflows/{id}/runs/{run_id}/continue	Form field
GET /workflows/{id}/runs/{run_id}	Query parameter
GET /workflows/{id}/runs	Query parameter
WS /workflows/{id}/ws	JSON field in the message body
POST /agents/{id}/runs/{run_id}/continue and POST /teams/{id}/runs/{run_id}/continue	Not read. The factory re-runs with ctx.input=None.
POST .../cancel	Not read. The factory is not invoked.

Factory setting	Effect on ctx.input
input_schema=SomeModel	ctx.input is a validated SomeModel instance. Invalid input returns 400 before the factory runs.
No input_schema	ctx.input is the parsed dict or None.

Identity precedence

user_id and session_id follow the same precedence as the prototype path. Verified middleware wins over the form field. On POST /runs, if both are set and disagree, the form value is overridden and a warning is logged.

1. request.state.user_id / request.state.session_id set by verified middleware (e.g. JWTMiddleware lifting the sub claim). Wins if present.
2. Explicit form field on the run request. Used only when no verified value is set.
3. None.

Exceptions

All factory exceptions inherit from FactoryError. The router maps them to HTTP status codes:

Exception	HTTP	When
FactoryValidationError	400	factory_input failed validation against input_schema. Raised automatically by AgentOS.
FactoryPermissionError	403	Factory decided the caller is not authorized. Raise from inside the factory.
FactoryError	500	Base for all factory errors. A bad return type, sync invocation of an async factory, or any uncaught exception inside the factory surfaces as 500.
FactoryContextRequired	internal	Raised by the get_*_by_id helpers when a factory is matched without a RequestContext. Not normally surfaced: HTTP routers raise their own 400 before this fires, and the workflow WebSocket replay path catches it and falls back to event-buffer / DB lookup.

Usage

from agno.agent import Agent, AgentFactory
from agno.db.postgres import PostgresDb
from agno.factory import RequestContext
from agno.models.openai import OpenAIResponses
from agno.os import AgentOS

db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai")


def build_agent(ctx: RequestContext) -> Agent:
    return Agent(model=OpenAIResponses(id="gpt-5.4"), db=db)


agent_factory = AgentFactory(id="tenant-agent", db=db, factory=build_agent)

agent_os = AgentOS(agents=[agent_factory])
app = agent_os.get_app()

Mixing Factories and Prototypes

agent_os = AgentOS(
    agents=[
        Agent(id="support-agent", model=OpenAIResponses(id="gpt-5.4"), db=db),
        AgentFactory(id="tenant-agent", db=db, factory=build_agent),
    ],
)

Sending factory_input

curl -X POST http://localhost:7777/agents/tenant-agent/runs \
    -F 'message=Hello' \
    -F 'factory_input={"persona": "analyst"}' \
    -F 'stream=false'

See Also

• Factories overview - Conceptual overview
• AgentFactory - Build an Agent per request
• TeamFactory - Build a Team per request
• WorkflowFactory - Build a Workflow per request
• Factory examples - Runnable examples mirroring the cookbook