> ## 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. # WorkflowFactory > Build a Workflow per request whose steps, agents, and model depend on the caller. v2.6.0 A `WorkflowFactory` produces a fresh `Workflow` for each request. Register it in `AgentOS(workflows=[...])` alongside any prototype workflows. ```python basic_workflow_factory.py theme={null} from agno.agent import Agent from agno.db.postgres import PostgresDb from agno.factory import RequestContext from agno.models.openai import OpenAIResponses from agno.os import AgentOS from agno.workflow import Step, Workflow, WorkflowFactory db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai") def build_content_pipeline(ctx: RequestContext) -> Workflow: user_id = ctx.user_id or "anonymous" drafter = Agent( name="Drafter", model=OpenAIResponses(id="gpt-5.4"), instructions=( f"You are a content drafter for tenant {user_id}. " "Write a first draft based on the topic. Keep it focused and concise." ), ) editor = Agent( name="Editor", model=OpenAIResponses(id="gpt-5.4"), instructions=( f"You are an editor for tenant {user_id}. " "Review the draft for clarity, grammar, and structure. Output the final version." ), ) return Workflow( name="Content Pipeline", description="Draft then edit content", db=db, steps=[ Step(name="draft", description="Write the first draft", agent=drafter), Step(name="edit", description="Edit and finalize", agent=editor), ], ) content_pipeline_factory = WorkflowFactory( id="content-pipeline", db=db, factory=build_content_pipeline, name="Content Pipeline", description="Builds a draft-then-edit content workflow per tenant.", ) agent_os = AgentOS(workflows=[content_pipeline_factory]) app = agent_os.get_app() if __name__ == "__main__": agent_os.serve(app="basic_workflow_factory:app", port=7777, reload=True) ``` Run the workflow like any other workflow: ```bash theme={null} curl -X POST http://localhost:7777/workflows/content-pipeline/runs \ -F 'message=Write a blog post about sustainable energy' \ -F 'user_id=tenant_42' \ -F 'stream=false' ``` See the [Factories reference](/reference/agent-os/factories) for the full constructor signature and parameter list. ## Step Shape That Varies by Tier The step graph can change per request. A common pattern: add or drop steps based on a tier read from `ctx.trusted.claims`. ```python theme={null} def build_article_pipeline(ctx: RequestContext) -> Workflow: tier = ctx.trusted.claims.get("tier", "free") model_id = TIER_MODELS.get(tier, "gpt-4.1-mini") steps = [] if tier == "enterprise": steps.append(Step(name="research", agent=researcher(model_id))) steps.append(Step(name="draft", agent=drafter(model_id))) steps.append(Step(name="edit", agent=editor(model_id))) return Workflow(name="Article Pipeline", db=db, steps=steps) ``` The `tier` claim is read from verified middleware, so the `tier` cannot be changed by the request body. See [Authorization From Verified Context](/agent-os/factories/agent-factory#authorization-from-verified-context) for the trust model and [JWT Role Factory](/examples/agent-os/factories/jwt-role-factory) for an end-to-end JWT example. ## Async Factories Use an async callable when you need to fetch context from a database, an HTTP service, or any other awaitable resource. ```python async_workflow_factory.py theme={null} async def build_pipeline(ctx: RequestContext) -> Workflow: config = await fetch_pipeline_config(ctx.user_id) steps = [Step(name=s.name, agent=build_agent(s)) for s in config.steps] return Workflow(name="Pipeline", db=db, steps=steps) pipeline_factory = WorkflowFactory(id="pipeline", db=db, factory=build_pipeline) ``` ## With an Input Schema Declare a Pydantic model on the factory to validate client-supplied `factory_input` before the factory runs. ```python workflow_input_schema.py theme={null} from pydantic import BaseModel class PipelineConfig(BaseModel): include_research: bool = False def build_pipeline(ctx: RequestContext) -> Workflow: cfg: PipelineConfig = ctx.input steps = [] if cfg.include_research: steps.append(Step(name="research", agent=researcher)) steps.extend([ Step(name="draft", agent=drafter), Step(name="edit", agent=editor), ]) return Workflow(name="Pipeline", db=db, steps=steps) pipeline_factory = WorkflowFactory( id="pipeline", db=db, factory=build_pipeline, input_schema=PipelineConfig, ) ``` ## Error Handling Raise `FactoryPermissionError` from inside the factory to reject unauthorized callers with HTTP 403. AgentOS raises `FactoryValidationError` (400) automatically when `factory_input` fails `input_schema` validation. ```python theme={null} from agno.factory import FactoryPermissionError def build_pipeline(ctx: RequestContext) -> Workflow: if "workflows:run" not in ctx.trusted.scopes: raise FactoryPermissionError("Missing 'workflows:run' scope") ... ``` See the [Factories reference](/reference/agent-os/factories) for the full exception hierarchy and the post-resolve behavior. ## Factory Re-invocation on Read Endpoints `GET /workflows/{id}/runs/{run_id}` and `GET /workflows/{id}/runs` re-invoke the factory because the workflow has to be reconstructed to read its session state. Pass `factory_input` as a query parameter to drive the rebuild: ```bash theme={null} curl "http://localhost:7777/workflows/content-pipeline/runs/RUN_ID?session_id=SESSION_ID&factory_input=%7B%22tier%22%3A%22enterprise%22%7D" ``` ## Developer Resources * [Factories overview](/agent-os/factories/overview) * [RequestContext fields](/reference/agent-os/factories#requestcontext) * [Factories reference](/reference/agent-os/factories) * [Factory examples](/examples/agent-os/factories/overview)