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.

request.state.user_id / request.state.session_id set by verified middleware (e.g. JWTMiddleware lifting the sub claim). Wins if present.
Explicit form field on the run request. Used only when no verified value is set.
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

Install uv pip install 'agno[os]' openai 'psycopg[binary]' and export OPENAI_API_KEY. Start the local Postgres database on port 5532 with database, user, and password ai.

Save this as factory_server.py:

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

Start the server before sending requests:

uv run uvicorn factory_server:app --host 0.0.0.0 --port 7777

Mixing Factories and Prototypes

Replace the agent_os declaration above with this composition fragment, then call get_app() on the new instance:

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`

This sends a JSON object as the form field. The minimal factory above ignores ctx.input; read it inside build_agent to apply the persona or other request configuration.

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