workflows/factory.py returns an AugmentedLLM that can be treated like any other LLM in the framework—compose it, expose it as a tool, or wrap it with additional logic.
Patterns at a glance
| Pattern | Reach for it when… | Factory helper(s) | Highlights | Runnable example |
|---|---|---|---|---|
| Parallel (Map-Reduce) | You need multiple specialists to look at the same request concurrently | create_parallel_llm(...) | Fan-out/fan-in via FanOut + FanIn, accepts agents and plain callables | workflow_parallel |
| Router | Requests must be dispatched to the best skill, server, or function | create_router_llm(...), create_router_embedding(...) | Confidence-scored results, route_to_{agent,server,function} helpers, optional embedding routing | workflow_router |
| Intent Classifier | You need lightweight intent buckets before routing or automation | create_intent_classifier_llm(...), create_intent_classifier_embedding(...) | Returns structured IntentClassificationResult with entities and metadata | workflow_intent_classifier |
| Planner (Orchestrator) | A goal requires multi-step planning and coordination across agents | create_orchestrator(...) | Switch between full and iterative planning, override planner/synthesizer roles | workflow_orchestrator_worker |
| Deep Research | Long-horizon investigations with budgets, memory, and policy checks | create_deep_orchestrator(...) | Knowledge extraction, policy engine, Temporal-friendly execution | workflow_deep_orchestrator |
| Evaluator-Optimizer | You want an automated reviewer to approve or iterate on drafts | create_evaluator_optimizer_llm(...) | QualityRating thresholds, detailed feedback loop, refinement_history | workflow_evaluator_optimizer |
| Build Your Own | You need a bespoke pattern stitched from the primitives above | Mix helpers, native agents, and @app.tool decorators | Compose routers, parallel fan-outs, evaluators, or custom callables | See all workflows + create_swarm(...) |
Before you start
- Model your specialists as
AgentSpecor instantiateAgent/AugmentedLLMobjects up front. The factory helpers accept any combination. - Run everything inside
async with app.run() as running_app:so the sharedContextis initialised (server registry, executor, tracing, secrets). - Tune behaviour with
RequestParams(temperature, max tokens, strict schema mode) and provider-specific options (provider="anthropic", Azure/OpenAI models, etc.). - Expose the returned AugmentedLLM directly (
await llm.generate_str(...)) or wrap it with@app.tool/@app.async_toolto make it callable over MCP.
Composable building blocks
- Patterns are just AugmentedLLMs, so you can nest them—e.g. route to an orchestrator, run parallel fan-outs inside a planner step, or wrap the output of any pattern with an evaluator-optimizer loop.
- Mix LLM-powered steps with deterministic functions. Routers accept plain Python callables; parallel workflows blend
AgentSpecwith helpers likefan_out_functions. - Share state via the
Context: reuse secrets, telemetry, the executor, and the token counter across nested patterns without additional wiring.
Observability and control
- Every pattern reports token usage through the global
TokenCounter. Callawait llm.get_token_node()to inspect fan-out costs, planner iterations, or evaluation loops. - Adjust concurrency and retries centrally in
mcp_agent.config.yaml(executor.max_concurrent_activities, retry policy) instead of per-pattern plumbing. - Enable tracing (
otel.enabled: true) to see spans for planner steps, router decisions, evaluator iterations, and MCP tool calls in Jaeger or any OTLP backend.