Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.agno.com/llms.txt

Use this file to discover all available pages before exploring further.

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:
ParameterTypeDefaultDescription
idstrrequiredStable 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.
dbUnion[BaseDb, AsyncBaseDb]requiredDatabase for session storage. Used as the fallback db if the produced component’s db attribute is falsy. None raises ValueError.
factoryUnion[Callable[[RequestContext], T], Callable[[RequestContext], Awaitable[T]]]requiredCallable invoked on every run-scoped request. Returns a fresh component of type T. Both sync and async callables are accepted.
nameOptional[str]NoneHuman-readable name for UI discovery.
descriptionOptional[str]NoneDescription for UI discovery.
input_schemaOptional[Type[BaseModel]]NonePydantic 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.
FieldTypeSourceDescription
user_idOptional[str]request.state.user_id (set by middleware), with form field as fallbackCaller identity. JWT claims reach this field only via JWTMiddleware setting request.state.user_id.
session_idOptional[str]request.state.session_id, with form field as fallbackConversation/session handle. Same precedence rules as user_id.
inputOptional[BaseModel | dict]factory_input form field or query parameterValidated against the factory’s input_schema if set. Otherwise the parsed JSON object or None.
trustedTrustedContextrequest.state.claims / request.state.scopesPopulated only by verified middleware. Use for authorization.
trusted.claimsMapping[str, Any]request.state.claimsToken claims set by verified middleware (e.g. JWTMiddleware).
trusted.scopesfrozenset[str]request.state.scopesRBAC scopes set by verified middleware.
requestAnyRaw FastAPI RequestEscape 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

EndpointSource
POST /{agents,teams,workflows}/{id}/runsForm field
POST /workflows/{id}/runs/{run_id}/continueForm field
GET /workflows/{id}/runs/{run_id}Query parameter
GET /workflows/{id}/runsQuery parameter
WS /workflows/{id}/wsJSON field in the message body
POST /agents/{id}/runs/{run_id}/continue and POST /teams/{id}/runs/{run_id}/continueNot read. The factory re-runs with ctx.input=None.
POST .../cancelNot read. The factory is not invoked.
Factory settingEffect on ctx.input
input_schema=SomeModelctx.input is a validated SomeModel instance. Invalid input returns 400 before the factory runs.
No input_schemactx.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:
ExceptionHTTPWhen
FactoryValidationError400factory_input failed validation against input_schema. Raised automatically by AgentOS.
FactoryPermissionError403Factory decided the caller is not authorized. Raise from inside the factory.
FactoryError500Base for all factory errors. A bad return type, sync invocation of an async factory, or any uncaught exception inside the factory surfaces as 500.
FactoryContextRequiredinternalRaised 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