Workflow orchestrator
Chain a planner, a researcher, and a writer into a single multi-step workflow using plain Python async orchestration.
What this builds
A workflow orchestrator runs several specialized agents in sequence, feeding each one's output into the next. Here we wire up three agents — a planner that turns a brief into an outline, a researcher that gathers facts for each section, and a writer that produces the final draft — and a run_workflow function that drives them with asyncio.
Each agent has its own instructions, and one returns structured output so the next stage can consume it as typed data instead of free text. The orchestration itself is ordinary Python: there is no special "team" object, just three Agent instances and an async function that calls arun on each in turn.
This example shows agents, structured output, and async orchestration.
AgentRoute does not yet ship a built-in Team/Company primitive or agent-to-agent (A2A) calls. You orchestrate multiple agents with plain Python today, exactly as shown below. Native multi-agent orchestration and A2A are tracked on the platform roadmap.
The orchestrator
The planner returns a typed Plan so the researcher can iterate over its sections. The researcher and writer return plain text. The run_workflow function awaits each stage and passes the result forward.
import asyncio
from pydantic import BaseModel, Field
from agentroute import Agent
class Section(BaseModel):
title: str
goal: str = Field(description="What this section should accomplish")
class Plan(BaseModel):
sections: list[Section]
planner = Agent(
name="planner",
model="claude-sonnet-4",
instructions=(
"You are an editorial planner. Turn the user's brief into a tight "
"outline of 3-5 sections. Each section has a title and a one-line goal."
),
output=Plan,
)
researcher = Agent(
name="researcher",
model="claude-sonnet-4",
instructions=(
"You are a researcher. Given a section title and goal, produce 3-5 "
"concise, factual bullet points a writer can build on. No fluff."
),
)
writer = Agent(
name="writer",
model="claude-sonnet-4",
instructions=(
"You are a writer. Given an outline and research notes, write a clear, "
"well-structured article in Markdown. Use the notes; do not invent facts."
),
)
async def run_workflow(brief: str) -> str:
# 1. Plan: brief -> typed outline.
plan_result = await planner.arun(brief)
plan: Plan = plan_result.output
# 2. Research: one call per section, run concurrently.
research = await asyncio.gather(
*(
researcher.arun(f"Section: {s.title}\nGoal: {s.goal}")
for s in plan.sections
)
)
# 3. Write: stitch the outline + notes into one prompt for the writer.
notes = "\n\n".join(
f"## {s.title}\n{str(r)}"
for s, r in zip(plan.sections, research)
)
draft_result = await writer.arun(
f"Brief:\n{brief}\n\nOutline and research notes:\n{notes}"
)
# Aggregate usage across every agent call so you can see total cost.
total_cost = (
plan_result.usage.total_cost_usd
+ sum(r.usage.total_cost_usd for r in research)
+ draft_result.usage.total_cost_usd
)
print(f"Workflow cost: ${total_cost:.4f}")
return str(draft_result)
if __name__ == "__main__":
brief = "Write a short explainer on why vector databases matter for RAG."
article = asyncio.run(run_workflow(brief))
print(article)The planner's result.output is a real Plan instance because we passed output=Plan, so plan.sections is a typed list[Section] — no parsing. The research stage uses asyncio.gather to fan out one researcher call per section in parallel, then the writer runs once on the combined notes. Each Result carries its own Usage, so summing total_cost_usd gives you the workflow's total spend.
Run it
export AGENTROUTE_API_KEY="sk-or-..."
python orchestrator.pyOne OpenRouter key (AGENTROUTE_API_KEY or OPENROUTER_API_KEY) covers every model="claude-sonnet-4" call across all three agents.