Router Workflow

import asyncio
from fast_agent import FastAgent

fast = FastAgent("Router Workflow")

@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:
        # Router intelligently selects the appropriate agent
        await agent.route.send("Analyze the code quality of main.py")
        await agent.route.send("What are the principles of effective beekeeping?")
        await agent.route.send("Fetch and summarize https://example.com")

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

class RoutingResponse(BaseModel):
    agent: str              # Name of selected agent
    confidence: str         # high, medium, or low
    reasoning: str | None   # Explanation of routing decision

{
  "agent": "code_expert",
  "confidence": "high",
  "reasoning": "Request specifically asks for code analysis which is the code_expert's specialty"
}

ROUTING_AGENT_INSTRUCTION = """
Select from the following agents to handle the request:
<fastagent:agents>
[
  {"name": "code_expert", "description": "Expert in code analysis..."},
  {"name": "general_assistant", "description": "Knowledgeable assistant..."},
  {"name": "fetcher", "description": "Agent that can fetch URLs..."}
]
</fastagent:agents>

You must respond with the 'name' of one of the agents listed above.
"""

# Request: "Download and summarize https://example.com"
# Router can either:
# 1. Use 'fetch' tool directly (if it has the server)
# 2. Delegate to 'fetcher' agent

@fast.router(
    name="smart_router",
    model="sonnet",
    agents=["specialist1", "specialist2"],
    servers=["fetch"],  # Router has tools too!
)

@fast.agent(
    "billing_specialist",
    instruction="Handle all billing, payments, and invoice questions",
    servers=["database"],
)
@fast.agent(
    "technical_support",
    instruction="Resolve technical issues, bugs, and product questions",
    servers=["jira", "filesystem"],
)
@fast.agent(
    "account_manager",
    instruction="Handle account changes, upgrades, and general inquiries",
    servers=["database"],
)
@fast.router(
    name="customer_service",
    model="sonnet",
    agents=["billing_specialist", "technical_support", "account_manager"],
)
async def main() -> None:
    async with fast.run() as agent:
        await agent.customer_service.prompt()  # Interactive customer service

@fast.agent(
    "content_reviewer",
    instruction="Review content for policy compliance and safety",
)
@fast.agent(
    "spam_detector",
    instruction="Identify and classify spam, scams, and low-quality content",
)
@fast.agent(
    "sensitivity_checker",
    instruction="Check for sensitive topics requiring special handling",
)
@fast.router(
    name="moderation_router",
    model="sonnet",
    agents=["content_reviewer", "spam_detector", "sensitivity_checker"],
)
async def main() -> None:
    async with fast.run() as agent:
        result = await agent.moderation_router.send("Check this user submission...")

@fast.agent(
    "code_reviewer",
    instruction="Review code for quality, style, and best practices",
    servers=["filesystem", "github"],
)
@fast.agent(
    "test_generator",
    instruction="Generate comprehensive unit tests for code",
    servers=["filesystem"],
)
@fast.agent(
    "doc_writer",
    instruction="Write clear technical documentation",
    servers=["filesystem"],
)
@fast.agent(
    "refactorer",
    instruction="Suggest and implement code refactoring improvements",
    servers=["filesystem"],
)
@fast.router(
    name="dev_assistant",
    model="sonnet",
    agents=["code_reviewer", "test_generator", "doc_writer", "refactorer"],
)
async def main() -> None:
    async with fast.run() as agent:
        await agent.dev_assistant.prompt()  # Interactive development assistant

CUSTOM_ROUTING = """
You are a specialized router for medical queries.

Prioritization rules:
1. Emergency queries ALWAYS go to emergency_handler
2. Prescription questions go to pharmacist
3. General health goes to general_practitioner
4. If uncertain, route to general_practitioner

Available agents:
{context}
"""

@fast.router(
    name="medical_router",
    agents=["emergency_handler", "pharmacist", "general_practitioner"],
    routing_instruction=CUSTOM_ROUTING,
)

import logging
logging.basicConfig(level=logging.INFO)

# You'll see:
# INFO: Routing structured request to agent: code_expert (confidence: high)
# INFO: Reasoning: Request specifically mentions code analysis