Workflows

Workflows let you compose multiple agents into a single higher-level capability (e.g. chaining steps, routing, or adding reliability via voting). They can be used alongside MCP servers defined in fastagent.config.yaml.

Workflows and MCP Servers

To generate examples use fast-agent quickstart workflow.

Agents can use MCP Servers defined in fastagent.config.yaml:

fastagent.config.yaml

# Example of a STDIO sever named "fetch"
mcp:
  servers:
    fetch:
      command: "uvx"
      args: ["mcp-server-fetch"]

social.py

@fast.agent(
    "url_fetcher",
    "Given a URL, provide a complete and comprehensive summary",
    servers=["fetch"],  # Name of an MCP Server defined in fastagent.config.yaml
)
@fast.agent(
    "social_media",
    """
    Write a 280 character social media post for any given text.
    Respond only with the post, never use hashtags.
    """,
)
@fast.chain(
    name="post_writer",
    sequence=["url_fetcher", "social_media"],
)
async def main():
    async with fast.run() as agent:
        await agent.post_writer.send("http://fast-agent.ai")

Saved as social.py you can run the workflow from the command line with:

uv run social.py --agent post_writer --message "<url>"

Add the --quiet switch to disable progress and message display and return only the final response.

Read more about running fast-agent agents here

Workflow Types

fast-agent has built-in support for common agentic workflow patterns (including those referenced in Anthropic's Building Effective Agents).

Chain

The chain workflow offers a declarative approach to calling Agents in sequence.

@fast.chain(
  name="post_writer",
  sequence=["url_fetcher", "social_media"],
)

async with fast.run() as agent:
  await agent.interactive(agent="post_writer")

Chains can be incorporated in other workflows, or contain other workflow elements (including other Chains). You can set an instruction to describe its capabilities to other workflow steps if needed.

Parallel

The parallel workflow sends the same message to multiple agents simultaneously (fan_out), then optionally uses a fan_in agent to process the combined content.

@fast.agent("translate_fr", "Translate the text to French")
@fast.agent("translate_de", "Translate the text to German")
@fast.agent("translate_es", "Translate the text to Spanish")

@fast.parallel(
  name="translate",
  fan_out=["translate_fr", "translate_de", "translate_es"],
)

If you don't specify a fan_in agent, parallel returns the combined agent results verbatim.

Evaluator-Optimizer

Evaluator-Optimizers combine 2 agents: one to generate content (the generator), and the other to judge that content and provide actionable feedback (the evaluator). Messages are sent to the generator first, then the pair run in a loop until either the evaluator is satisfied with the quality, or the maximum number of refinements is reached. The final result from the generator is returned.

@fast.evaluator_optimizer(
  name="researcher",
  generator="web_searcher",
  evaluator="quality_assurance",
  min_rating="EXCELLENT",
  max_refinements=3,
)

async with fast.run() as agent:
  await agent.researcher.send("produce a report on how to make the perfect espresso")

Router

Routers use an LLM to assess a message and route it to the most appropriate agent. The routing prompt is automatically generated based on the agent instructions and available servers.

@fast.router(
  name="route",
  agents=["agent1", "agent2", "agent3"],
)

NB - If only one agent is supplied to the router, it forwards directly.

Orchestrator

Given a complex task, the Orchestrator uses an LLM to generate a plan to divide the task amongst the available Agents. Plans can either be built once at the beginning (plan_type="full") or iteratively (plan_type="iterative").

@fast.orchestrator(
  name="orchestrate",
  agents=["task1", "task2", "task3"],
)

Iterative Planner

The iterative_planner workflow is a specialized orchestrator for long-running plans that are refined over multiple iterations.

@fast.iterative_planner(
  name="planner",
  agents=["task1", "task2", "task3"],
)

MAKER

MAKER (“Massively decomposed Agentic processes with K-voting Error Reduction”) wraps a worker agent and samples it repeatedly until a response achieves a k-vote margin over all alternatives (“first-to-ahead-by-k” voting). This is useful for long chains of simple steps where rare errors would otherwise compound.

Reference: Solving a Million-Step LLM Task with Zero Errors
Credit: Lucid Programmer (PR author)

@fast.agent(
  name="classifier",
  instruction="Reply with only: A, B, or C.",
)
@fast.maker(
  name="reliable_classifier",
  worker="classifier",
  k=3,
  max_samples=25,
  match_strategy="normalized",
  red_flag_max_length=16,
)
async def main():
  async with fast.run() as agent:
    await agent.reliable_classifier.send("Classify: ...")

Workflow Reference