Multi-agent patterns.

import asyncio

from matilda_agent_sdk import Agent, Runner


async def main():
    runner = Runner()

    researcher = Agent(
        name="researcher",
        purpose="analysis",
        instructions="Produce a structured list of key facts for a blog post. Bullet points only.",
    )

    writer = Agent(
        name="writer",
        purpose="general",
        instructions="Given research notes, write an engaging blog post draft under 400 words.",
    )

    editor = Agent(
        name="editor",
        purpose="general",
        instructions="Polish the draft for clarity, grammar, and flow. Return the full revised post.",
    )

    topic = "Why developers are adopting AI coding assistants"

    # Stage 1 → 2 → 3
    research = await runner.run(researcher, f"Research this topic: {topic}")
    draft = await runner.run(writer, research.final_output)
    edited = await runner.run(editor, draft.final_output)

    print(edited.final_output)

    # Total token usage across the pipeline
    total_tokens = sum(
        r.usage.output_tokens for r in (research, draft, edited) if r.usage
    )
    print(f"Total output tokens: {total_tokens}")


asyncio.run(main())

import asyncio

from matilda_agent_sdk import Agent, Runner


async def main():
    runner = Runner()

    security_reviewer = Agent(
        name="security-reviewer",
        purpose="analysis",
        instructions="Review code for vulnerabilities. Report only security issues.",
    )

    performance_reviewer = Agent(
        name="performance-reviewer",
        purpose="analysis",
        instructions="Review code for efficiency. Report only performance issues.",
    )

    synthesiser = Agent(
        name="synthesiser",
        purpose="analysis",
        instructions="Given reviews from multiple reviewers, produce a prioritised action list.",
    )

    code = "def get_user_data(user_id, db): ..."

    # Fan-out: two reviewers analyse concurrently
    security, performance = await asyncio.gather(
        runner.run(security_reviewer, f"Review this code:\n```python\n{code}\n```"),
        runner.run(performance_reviewer, f"Review this code:\n```python\n{code}\n```"),
    )

    # Fan-in: synthesiser merges the reviews
    combined_input = "\n".join(
        [
            "## Security Review", security.final_output,
            "## Performance Review", performance.final_output,
        ]
    )

    synthesis = await runner.run(synthesiser, combined_input)
    print(synthesis.final_output)


asyncio.run(main())

import asyncio

from matilda_agent_sdk import Agent, Runner, ToolResult

runner = Runner()

billing_specialist = Agent(
    name="billing-specialist",
    instructions="You are a billing support specialist.",
)

technical_specialist = Agent(
    name="technical-specialist",
    instructions="You are a technical support specialist. Include code examples when relevant.",
)

triage_agent = Agent(
    name="triage",
    purpose="general",
    instructions="You are a customer support triage specialist.",
)

triage_tools = [
    {
        "name": "ask_billing_specialist",
        "description": "Route billing questions to the billing specialist.",
        "parameters": {"type": "object", "properties": {"question": {"type": "string"}}, "required": ["question"]},
    },
    {
        "name": "ask_technical_specialist",
        "description": "Route technical questions to the technical specialist.",
        "parameters": {"type": "object", "properties": {"question": {"type": "string"}}, "required": ["question"]},
    },
]


async def ask_billing_specialist(args, ctx):
    result = await runner.run(billing_specialist, f"Answer: {args['question']}")
    return ToolResult(content=result.final_output)


async def ask_technical_specialist(args, ctx):
    result = await runner.run(technical_specialist, f"Answer: {args['question']}")
    return ToolResult(content=result.final_output)


tool_handlers = {
    "ask_billing_specialist": ask_billing_specialist,
    "ask_technical_specialist": ask_technical_specialist,
}


async def main():
    triage_prompt = "\n".join(
        [
            "A customer asked:",
            '"I\'m getting a 401 Unauthorized error when calling the /api/chat endpoint."',
            "You MUST forward this to a specialist by calling a tool.",
            "After the specialist responds, relay their answer.",
        ]
    )

    await runner.run(
        triage_agent,
        triage_prompt,
        tool_handlers=tool_handlers,
        client_tools=triage_tools,
        max_tool_roundtrips=6,
        callbacks={
            "on_tool_call": lambda name, args: print(f"Triage chose: {name}"),
            "on_tool_result": lambda name, result, is_error: print("Specialist responded."),
            "on_token": lambda delta: print(delta, end="", flush=True),
        },
    )


asyncio.run(main())