Building a Simple Multi-Agent Workflow in Python: Router + Specialist Agents

Most AI assistants fail for a boring reason: one prompt is asked to do everything. It answers billing questions, debugs API errors, and handles sales inquiries with the same instructions. The result is a bloated system prompt and answers that are inconsistent.

The fix is a pattern used in production agentic AI systems: a router agent that classifies each request, and specialist agents that each do one job well.

In this tutorial you'll learn how to build a multi-agent AI workflow in Python with a router and specialist agents. By the end you'll have about 120 lines of runnable code with structured routing, a fallback path, and a specialist that uses a tool.

Table of Contents

• What is a multi-agent workflow?
• What is a router agent?
• Prerequisites
• Architecture overview
• Step 1: Set up the project
• Step 2: Build the router agent
• Step 3: Build the specialist agents
• Step 4: Wire the orchestrator
• Step 5: Run it
• Design rules that keep it reliable
• How do you extend a multi-agent workflow?
• Conclusion

What is a multi-agent workflow?

A multi-agent workflow splits one large task across several focused AI agents. Each agent has its own instructions, and sometimes its own tools. A coordinating layer decides which agent handles which piece of work.

Compared with a single do-everything prompt, this gives you:

• Focus: each agent has a short, specific prompt, so it stays on task.
• Testability: you can evaluate the router and each specialist separately.
• Control: you can restrict tools and data per agent.
• Maintainability: adding a capability means adding an agent, not rewriting a prompt.

What is a router agent?

A router agent is a lightweight classifier. It reads the incoming request and outputs one decision: which specialist should handle this? It does not answer the question itself.

Key idea: the router's only job is to pick a route. Keeping it narrow makes it fast, cheap, and easy to test.

Prerequisites

Before you start, make sure you have:

• [ ] Python 3.10 or newer
• [ ] An API key for an LLM provider (this tutorial uses the Anthropic SDK, but the pattern is provider-agnostic)
• [ ] Basic familiarity with Python functions and dataclasses
• [ ] A terminal and a virtual environment tool (venv works fine)

Architecture overview

Here is the flow we're building:

                  ┌──────────────────┐
  User message ─▶ │   Router Agent   │
                  └────────┬─────────┘
                           │ route = billing | technical | sales | general
        ┌──────────────┬───┴──────────┬──────────────┐
        ▼              ▼              ▼              ▼
  ┌──────────┐  ┌────────────┐  ┌──────────┐  ┌──────────┐
  │ Billing  │  │ Technical  │  │  Sales   │  │ General  │
  │ (+ tool) │  │  Support   │  │  Agent   │  │ fallback │
  └────┬─────┘  └─────┬──────┘  └────┬─────┘  └────┬─────┘
       └──────────────┴──────┬───────┴─────────────┘
                             ▼
                      Final response

Step 1: Set up the project

Create a folder and install the one dependency:

mkdir multi-agent-router && cd multi-agent-router
python -m venv .venv && source .venv/bin/activate
pip install anthropic
export ANTHROPIC_API_KEY="your-key-here"

Now create agents.py and add the shared setup. Everything goes through one small ask() helper, so swapping providers later means changing a single function.

# agents.py
import json
import os
import re
import time
from dataclasses import dataclass, field
from typing import Callable, Optional

from anthropic import Anthropic

client = Anthropic()  # reads ANTHROPIC_API_KEY from the environment
MODEL = os.getenv("MODEL_NAME", "claude-sonnet-5")


def ask(system: str, user: str, max_tokens: int = 500) -> str:
    """Single choke point for every LLM call in the workflow."""
    response = client.messages.create(
        model=MODEL,
        max_tokens=max_tokens,
        system=system,
        messages=[{"role": "user", "content": user}],
    )
    return response.content[0].text

Step 2: Build the router agent

The router returns JSON so the rest of your code can trust its output. Two details matter here: the list of valid routes is explicit, and there is a safe fallback if the model returns something unexpected.

FENCE = "`" * 3  # avoids typing triple backticks inside a Markdown code block


def route(message: str) -> tuple[str, str]:
    text = ask(ROUTER_PROMPT, message, max_tokens=120).strip()
    # Models sometimes wrap JSON in code fences; strip them defensively.
    text = text.removeprefix(FENCE + "json").removeprefix(FENCE).removesuffix(FENCE).strip()
    try:
        data = json.loads(text)
        if data.get("route") in VALID_ROUTES:
            return data["route"], data.get("reason", "")
    except json.JSONDecodeError:
        pass
    return "general", "Router output was invalid; used fallback route."

> **Why the fallback matters:** in a multi-agent system, an unhandled routing failure breaks the entire request. A default route keeps the user experience intact and gives you a log line to investigate.

Step 3: Build the specialist agents

Each specialist is a small dataclass: a name, a system prompt, and an optional context builder. The context builder is where a specialist gets its tool. In this case, the billing agent looks up invoice data before the model answers.

@dataclass
class Agent:
    name: str
    system_prompt: str
    build_context: Optional[Callable[[str], str]] = None

    def run(self, message: str) -> str:
        context = self.build_context(message) if self.build_context else ""
        user_input = f"{message}\n\n{context}".strip()
        return ask(self.system_prompt, user_input, max_tokens=600)


# --- A tiny "tool": replace with a real database or API call ---
INVOICES = {
    "INV-1001": {"amount": "$49.00", "status": "paid", "date": "2026-08-30"},
    "INV-1002": {"amount": "$149.00", "status": "overdue", "date": "2026-09-05"},
}


def billing_context(message: str) -> str:
    match = re.search(r"INV-\d+", message.upper())
    if not match:
        return ""
    invoice = INVOICES.get(match.group())
    if not invoice:
        return f"[Tool result] No invoice found with ID {match.group()}."
    return f"[Tool result] {match.group()}: {json.dumps(invoice)}"


SPECIALISTS = {
    "billing": Agent(
        name="Billing Agent",
        system_prompt=(
            "You are a billing support specialist. Answer using the tool result "
            "when one is provided. Never invent invoice details. If data is "
            "missing, say so and ask for the invoice ID."
        ),
        build_context=billing_context,
    ),
    "technical": Agent(
        name="Technical Support Agent",
        system_prompt=(
            "You are a technical support engineer. Give concise, step-by-step "
            "troubleshooting. Ask for error messages or logs if none are given."
        ),
    ),
    "sales": Agent(
        name="Sales Agent",
        system_prompt=(
            "You are a helpful sales assistant. Explain plans and next steps "
            "clearly, avoid pressure, and offer to book a demo when appropriate."
        ),
    ),
    "general": Agent(
        name="General Assistant",
        system_prompt=(
            "You are a friendly general assistant. Answer briefly, and if the "
            "question is outside your scope, say what you can help with."
        ),
    ),
}

Privacy note: only pass each specialist the data it needs. Here the billing agent sees invoice fields, and nothing else sees them. Scoping context per agent is one of the biggest practical advantages of this pattern.

Step 4: Wire the orchestrator

The orchestrator is a plain function: route, dispatch, and record a trace. You don't need a framework for this.

def handle(message: str) -> dict:
    start = time.perf_counter()

    chosen_route, reason = route(message)
    agent = SPECIALISTS[chosen_route]
    answer = agent.run(message)

    return {
        "route": chosen_route,
        "agent": agent.name,
        "router_reason": reason,
        "answer": answer,
        "latency_s": round(time.perf_counter() - start, 2),
    }


if __name__ == "__main__":
    samples = [
        "Why is invoice INV-1002 still showing as overdue?",
        "I keep getting a 401 error when calling your webhook endpoint.",
        "Do you offer a reseller plan for agencies?",
        "What are your support hours?",
    ]
    for text in samples:
        result = handle(text)
        print(f"\n> {text}")
        print(f"  route={result['route']} ({result['router_reason']})")
        print(f"  agent={result['agent']} | {result['latency_s']}s")
        print(f"  {result['answer']}")

Step 5: Run it

python agents.py

You should see each message routed to the right specialist, with a trace line showing the route, reason, and latency:

> Why is invoice INV-1002 still showing as overdue?
  route=billing (Question about an existing invoice status.)
  agent=Billing Agent | 2.31s
  Invoice INV-1002 for $149.00, dated 2026-09-05, is currently marked overdue...

Your output will differ, since the model wording varies. What matters is that the route is correct for each message.

Design rules that keep it reliable

Working code is the easy part. These rules are what make a router-and-specialist system dependable in production.

1. Keep the router narrow

Give it a short prompt, a closed list of routes, and no ability to answer. If routing accuracy drops, add examples to the router prompt instead of making it smarter in other ways.

2. Always validate and fall back

Never trust model output blindly. Check the route against an allow-list and default to a safe agent, as route() does above.

3. Give each specialist one job and minimal context

Narrow prompts reduce hallucination. Scoped data reduces risk. If a specialist needs a new skill, that's a signal to create another specialist.

4. Log the route, reason, and latency for every request

Routing errors are the most common failure in this pattern, and they're invisible without traces. The reason field costs a few tokens and saves hours of debugging.

5. Test the router separately from the specialists

Build a small table of messages and expected routes, and run it after every prompt change:

TEST_CASES = [
    ("Where is my refund?", "billing"),
    ("The SDK throws a timeout on init", "technical"),
    ("Can I get a demo?", "sales"),
    ("Who founded your company?", "general"),
]

def test_router():
    for message, expected in TEST_CASES:
        got, _ = route(message)
        assert got == expected, f"{message!r}: expected {expected}, got {got}"

How do you extend a multi-agent workflow?

Once the basic router works, common next steps include:

• Add a new specialist: one new Agent entry and one new line in the router prompt.
• Use real tools: replace billing_context with a database query or API call, or move to native tool calling.
• Run agents in parallel: for requests that touch multiple domains, dispatch several specialists and merge the answers.
• Add a review step: a lightweight "checker" agent that validates the specialist's answer before it reaches the user.
• Add human handoff: route low-confidence or sensitive requests to a person.

Prefer JavaScript? The structure is identical. You need one ask() function, one JSON-returning router, and a map of specialists. Only the syntax changes.

Conclusion

You now know how to build a multi-agent AI workflow in Python with a router and specialist agents. Key takeaways:

• A router agent decides who should answer, and a specialist agent answers.
• Structured output plus a fallback route keeps the system predictable.
• Scoped prompts and scoped data make each agent more accurate and safer.
• Traces and router tests turn a demo into something you can trust.

Start with two or three specialists, measure routing accuracy, and grow from there.

At Botsailors, we build multi-agent and agentic AI systems for omnichannel automation. If you'd like a follow-up on parallel agents, tool calling, or evaluation, let us know in the comments. What would you build with a router and specialists?