Workflow Examples

Build complex multi-agent workflows with orchestration patterns

Overview

Fast Agent provides powerful workflow patterns for building sophisticated multi-agent systems. These patterns are inspired by Anthropic’s Building Effective Agents guide.

Agents as Tools (Simple)

The simplest orchestration pattern: use agents as tools for other agents.

examples/workflows/agents_as_tools_simple.py

import asyncio
from fast_agent import FastAgent

fast = FastAgent("Agents-as-Tools simple demo")

@fast.agent(
    name="NY-Project-Manager",
    instruction="Return NY time + timezone, plus a one-line project status.",
    servers=["time"],
)
@fast.agent(
    name="London-Project-Manager",
    instruction="Return London time + timezone, plus a one-line news update.",
    servers=["time"],
)
@fast.agent(
    name="PMO-orchestrator",
    instruction=(
        "Get reports. Always use one tool call per project/news. "  # parallelization
        "Responsibilities: NY projects: [OpenAI, Fast-Agent, Anthropic]. London news: [Economics, Art, Culture]. "  # routing
        "Aggregate results and add a one-line PMO summary."
    ),
    default=True,
    agents=["NY-Project-Manager", "London-Project-Manager"],  # orchestrator-workers
)
async def main() -> None:
    async with fast.run() as agent:
        await agent("Get PMO report. Projects: all. News: Art, Culture")

if __name__ == "__main__":
    asyncio.run(main())

The agents parameter makes other agents available as tools. The orchestrator can call them just like any other tool.

Markdown-Based Configuration

You can also define agents using markdown files:

PMO-orchestrator.md

---
type: agent
name: PMO-orchestrator
default: true
agents:
- NY-Project-Manager
- London-Project-Manager
---
Get reports. Always use one tool call per project/news. Responsibilities: NY projects: [OpenAI, Fast-Agent, Anthropic]. London news: [Economics, Art, Culture]. Aggregate results and add a one-line PMO summary.

NY-Project-Manager.md

---
type: agent
name: NY-Project-Manager
servers:
- time
---
Return NY time + timezone, plus a one-line project status.

Run with:

fast-agent go

Router Workflow

Route requests to specialized agents or handle them directly:

examples/workflows/router.py

import asyncio
from fast_agent import FastAgent

fast = FastAgent("Router Workflow")

SAMPLE_REQUESTS = [
    "Download and summarize https://llmindset.co.uk/posts/2024/12/mcp-build-notes/",
    "Analyze the quality of the Python codebase in the current working directory",
    "What are the key principles of effective beekeeping?",
]

@fast.agent(
    name="fetcher",
    model="haiku",
    instruction="You are an agent, with a tool enabling you to fetch URLs.",
    servers=["fetch"],
)
@fast.agent(
    name="code_expert",
    model="haiku",
    instruction="""You are an expert in code analysis and software engineering.
    When asked about code, architecture, or development practices,
    you provide thorough and practical insights.""",
    servers=["filesystem"],
)
@fast.agent(
    name="general_assistant",
    model="haiku",
    instruction="""You are a knowledgeable assistant that provides clear,
    well-reasoned responses about general topics, concepts, and principles.""",
)
@fast.router(
    name="route",
    model="sonnet",
    default=True,
    agents=["code_expert", "general_assistant", "fetcher"],
)
async def main() -> None:
    async with fast.run() as agent:
        await agent.interactive(agent_name="route")
        for request in SAMPLE_REQUESTS:
            await agent.route(request)
        await agent.interactive()

if __name__ == "__main__":
    asyncio.run(main())

The router can either delegate to specialized agents or use its own tools directly, providing flexible request handling.

Parallel Workflow (Fan-Out/Fan-In)

Process work in parallel and aggregate results:

examples/workflows/parallel.py

import asyncio
from pathlib import Path
from fast_agent import FastAgent
from fast_agent.core.prompt import Prompt

fast = FastAgent("Parallel Workflow")

@fast.agent(
    name="proofreader",
    instruction="""Review the short story for grammar, spelling, and punctuation errors.
    Identify any awkward phrasing or structural issues that could improve clarity. 
    Provide detailed feedback on corrections.""",
)
@fast.agent(
    name="fact_checker",
    instruction="""Verify the factual consistency within the story. Identify any contradictions,
    logical inconsistencies, or inaccuracies in the plot, character actions, or setting. 
    Highlight potential issues with reasoning or coherence.""",
)
@fast.agent(
    name="style_enforcer",
    instruction="""Analyze the story for adherence to style guidelines.
    Evaluate the narrative flow, clarity of expression, and tone. Suggest improvements to 
    enhance storytelling, readability, and engagement.""",
    model="sonnet",
)
@fast.agent(
    name="grader",
    instruction="""Compile the feedback from the Proofreader, Fact Checker, and Style Enforcer
    into a structured report. Summarize key issues and categorize them by type. 
    Provide actionable recommendations for improving the story, 
    and give an overall grade based on the feedback.""",
)
@fast.parallel(
    fan_out=["proofreader", "fact_checker", "style_enforcer"],
    fan_in="grader",
    name="parallel",
)
async def main() -> None:
    async with fast.run() as agent:
        await agent.parallel.send(
            Prompt.user("Student short story submission", Path("short_story.txt"))
        )

if __name__ == "__main__":
    asyncio.run(main())

The fan_out agents run in parallel, and their results are aggregated by the fan_in agent.

Orchestrator with Iterative Planning

Use an iterative planner to coordinate complex workflows:

examples/workflows/orchestrator.py

import asyncio
from fast_agent import FastAgent

fast = FastAgent("Orchestrator-Workers")

@fast.agent(
    "author",
    instruction="""You are to role play a poorly skilled writer, 
    who makes frequent grammar, punctuation and spelling errors. You enjoy
    writing short stories, but the narrative doesn't always make sense""",
    servers=["filesystem"],
)
@fast.agent(
    name="finder",
    instruction="""You are an agent with access to the filesystem, 
            as well as the ability to fetch URLs. Your job is to identify 
            the closest match to a user's request, make the appropriate tool calls, 
            and return the URI and CONTENTS of the closest match.""",
    servers=["fetch", "filesystem"],
    model="gpt-4.1",
)
@fast.agent(
    name="writer",
    instruction="""You are an agent that can write to the filesystem.
            You are tasked with taking the user's input, addressing it, and 
            writing the result to disk in the appropriate location.""",
    servers=["filesystem"],
)
@fast.agent(
    name="proofreader",
    instruction="""Review the short story for grammar, spelling, and punctuation errors.
            Identify any awkward phrasing or structural issues that could improve clarity. 
            Provide detailed feedback on corrections.""",
    servers=["fetch"],
    model="gpt-4.1",
)
@fast.iterative_planner(
    name="orchestrate",
    agents=["finder", "writer", "proofreader"],
    model="sonnet",
    plan_iterations=5,
)
async def main() -> None:
    async with fast.run() as agent:
        task = """Load the student's short story from short_story.md, 
        and generate a report with feedback across proofreading, 
        factuality/logical consistency and style adherence. Use the style rules from 
        https://apastyle.apa.org/learn/quick-guide-on-formatting and 
        https://apastyle.apa.org/learn/quick-guide-on-references.
        Write the graded report to graded_report.md in the same directory as short_story.md"""
        
        await agent.orchestrate(task)

if __name__ == "__main__":
    asyncio.run(main())

MAKER: K-Voting for Reliability

Achieve high reliability through statistical consensus voting:

examples/workflows/maker.py

import asyncio
from typing import Any, cast
from fast_agent import FastAgent

fast = FastAgent("MAKER Example")

@fast.agent(
    name="classifier",
    model="claude-3-haiku-20240307", 
    instruction="""You are a customer support intent classifier.
Classify the customer message into exactly one of: COMPLAINT, QUESTION, REQUEST, FEEDBACK.
Respond with ONLY the single word classification, nothing else.

Examples:
- "This product is broken!" → COMPLAINT
- "How do I reset my password?" → QUESTION
- "Please cancel my subscription" → REQUEST
- "Just wanted to say I love the new feature" → FEEDBACK""",
)
@fast.maker(
    name="reliable_classifier",
    worker="classifier",
    k=3,  # Require 3-vote margin for consensus
    max_samples=10,
    match_strategy="normalized",
    red_flag_max_length=20,  # Discard verbose responses
)
async def main():
    async with fast.run() as agent:
        test_cases = [
            "I've been waiting for 3 days now.",
            "Can someone explain how this works?",
            "This isn't what I expected.",
            "I'd like to speak to a manager.",
        ]
        
        for text in test_cases:
            result = await agent.reliable_classifier.send(text)
            stats = cast("Any", agent.reliable_classifier).last_result
            print(f"{text} → {result} (votes: {stats.votes})")

if __name__ == "__main__":
    asyncio.run(main())

When to use MAKER: Long chains of simple steps where errors compound (ETL pipelines, code migration, document processing). Trade compute for accuracy.

When NOT to use MAKER: Single classifications, creative tasks, complex reasoning, or when occasional errors are acceptable.

Chaining Pattern

Chain agents sequentially, passing outputs as inputs:

import asyncio
from fast_agent import FastAgent

fast = FastAgent("Chaining Example")

@fast.agent(
    name="url_fetcher",
    instruction="Fetch the content from the provided URL.",
    servers=["fetch"],
)
@fast.agent(
    name="post_writer",
    instruction="Write a social media post based on the provided content.",
)
@fast.agent(
    name="social_media",
    instruction="Polish and optimize the post for maximum engagement.",
)
async def main():
    async with fast.run() as agent:
        # Manual chaining
        content = await agent.url_fetcher("https://example.com/article")
        draft = await agent.post_writer(content)
        final = await agent.social_media(draft)
        print(final)

if __name__ == "__main__":
    asyncio.run(main())

Next Steps

MCP Examples

Learn about MCP server integration

Advanced Patterns

Explore hooks, RAG, and production patterns

Workflows Guide

Deep dive into workflow patterns

Agent Configuration

Configure agents with YAML and markdown