Coordinator (~AutoPattern)

graph = TransitionGraph(
    initial_speaker=user.agent_id,                     # the user kicks off
    transitions=[],                                    # no per-specialist rules
    default_target=AgentTarget(coordinator.agent_id),  # every non-handoff turn → coordinator
    max_turns=14,
)

from typing import Literal
from autogen.beta import ToolResult
from autogen.beta.network import Finish, Handoff

async def handoff(to: Literal["researcher", "critic", "cost_analyst", "operator"], reason: str = "") -> ToolResult:
    """Hand off the conversation to a participant by name."""
    return ToolResult(Handoff(target=to, reason=reason), final=True)

async def finish(summary: str = "") -> ToolResult:
    """End the conversation cleanly with a brief summary."""
    return ToolResult(Finish(summary=summary), final=True)

# ... then register them on the coordinator agent:
coordinator_agent.tool(handoff)
coordinator_agent.tool(finish)

researcher_resume = Resume(
    summary="Evidence-based analyst — cites studies, benchmarks, and data.",
    claimed_capabilities=["research", "literature-review"],
)
critic_resume = Resume(
    summary="Skeptical risk analyst — surfaces failure modes and overlooked costs.",
    claimed_capabilities=["risk-analysis", "review"],
)
cost_resume = Resume(
    summary="Cost analyst — models total cost of ownership and budget impact.",
    claimed_capabilities=["cost-modelling", "budgeting"],
)
operator_resume = Resume(
    summary="Operations engineer — speaks to day-2 ops, on-call load, and maintenance.",
    claimed_capabilities=["operations", "reliability"],
)

roster = {
    "researcher": researcher_resume,
    "critic": critic_resume,
    "cost_analyst": cost_resume,
    "operator": operator_resume,
}
roster_block = "\n".join(f"- `{name}`: {r.summary}" for name, r in roster.items())
# coordinator prompt embeds f"{roster_block}"; the same Resume objects
# are passed to register() so the prompt and the hub registry agree.

"""Cookbook — Coordinator (manager + specialists) pattern.

One coordinator at the centre directs several specialists via a single
generic ``handoff(to)`` tool. Specialists reply; the channel default
routes every non-handoff turn back to the coordinator. The coordinator
ends the channel with a ``finish(summary)`` tool.

Both routing tools return ``ToolResult(..., final=True)`` so each
coordinator round produces exactly one routing decision.
"""

import asyncio
from typing import Literal

from dotenv import load_dotenv

from autogen.beta import Agent, ToolResult
from autogen.beta.config import AnthropicConfig
from autogen.beta.knowledge import MemoryKnowledgeStore
from autogen.beta.network import (
    EV_CHANNEL_CLOSED,
    EV_PACKET,
    EV_TEXT,
    WORKFLOW_TYPE,
    AgentTarget,
    Finish,
    Handoff,
    Hub,
    HubClient,
    LocalLink,
    Passport,
    Resume,
    TransitionGraph,
)
from autogen.beta.testing import TestConfig

load_dotenv()

async def handoff(to: Literal["researcher", "critic", "cost_analyst", "operator"], reason: str = "") -> ToolResult:
    """Hand off the conversation to a specialist by name."""
    return ToolResult(Handoff(target=to, reason=reason), final=True)

async def finish(summary: str = "") -> ToolResult:
    """End the conversation with a summary recommendation."""
    return ToolResult(Finish(summary=summary), final=True)

async def main() -> None:
    config = AnthropicConfig(
        model="claude-sonnet-4-6",
        extra_body={"tool_choice": {"type": "auto", "disable_parallel_tool_use": True}},
    )

    hub_obj = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0)
    link = LocalLink(hub_obj)

    user_hc = HubClient(link, hub=hub_obj)
    coord_hc = HubClient(link, hub=hub_obj)
    researcher_hc = HubClient(link, hub=hub_obj)
    critic_hc = HubClient(link, hub=hub_obj)
    cost_hc = HubClient(link, hub=hub_obj)
    operator_hc = HubClient(link, hub=hub_obj)

    # Each Resume is built once and used twice: folded into the
    # coordinator's system message and passed to register().
    researcher_resume = Resume(
        summary="Evidence-based analyst — cites studies, benchmarks, and data.",
        claimed_capabilities=["research", "literature-review"],
    )
    critic_resume = Resume(
        summary="Skeptical risk analyst — surfaces failure modes and overlooked costs.",
        claimed_capabilities=["risk-analysis", "review"],
    )
    cost_resume = Resume(
        summary="Cost analyst — models total cost of ownership and budget impact.",
        claimed_capabilities=["cost-modelling", "budgeting"],
    )
    operator_resume = Resume(
        summary="Operations engineer — speaks to day-2 ops, on-call load, and maintenance.",
        claimed_capabilities=["operations", "reliability"],
    )

    roster = {
        "researcher": researcher_resume,
        "critic": critic_resume,
        "cost_analyst": cost_resume,
        "operator": operator_resume,
    }
    roster_block = "\n".join(f"- `{name}`: {r.summary}" for name, r in roster.items())

    user_agent = Agent("user", config=TestConfig())

    coord_agent = Agent(
        "coordinator",
        prompt=(
            "You coordinate a discussion between these specialists:\n"
            f"{roster_block}\n"
            "\n"
            "Each turn, do exactly ONE of:\n"
            "- Call `handoff(to=<name>, reason=...)` to consult a "
            "specialist who has not yet weighed in.\n"
            "- Call `finish(summary=...)` once ALL specialists have "
            "contributed, summarising a recommendation."
        ),
        config=config,
    )
    coord_agent.tool(handoff)
    coord_agent.tool(finish)

    researcher_agent = Agent(
        "researcher",
        prompt=(
            "You are a researcher. When consulted, give a concise, "
            "evidence-based input — cite relevant studies, benchmarks, "
            "or data. 2-3 sentences. You have no tools; just reply."
        ),
        config=config,
    )
    critic_agent = Agent(
        "critic",
        prompt=(
            "You are a critic. When consulted, give a skeptical, "
            "risk-focused take — what could go wrong, what's being "
            "overlooked. 2-3 sentences. You have no tools; just reply."
        ),
        config=config,
    )
    cost_agent = Agent(
        "cost_analyst",
        prompt=(
            "You are a cost analyst. When consulted, give a concise "
            "take on total cost of ownership and budget impact — "
            "name the cost drivers. 2-3 sentences. You have no tools; "
            "just reply."
        ),
        config=config,
    )
    operator_agent = Agent(
        "operator",
        prompt=(
            "You are an operations engineer. When consulted, speak to "
            "day-2 operations — on-call load, maintenance, and "
            "reliability. 2-3 sentences. You have no tools; just reply."
        ),
        config=config,
    )

    user = await user_hc.register(user_agent, Passport(name="user"), Resume(), attach_plugin=False)
    coordinator = await coord_hc.register(
        coord_agent, Passport(name="coordinator"), Resume(), attach_plugin=False
    )
    researcher = await researcher_hc.register(
        researcher_agent, Passport(name="researcher"), researcher_resume, attach_plugin=False
    )
    critic = await critic_hc.register(critic_agent, Passport(name="critic"), critic_resume, attach_plugin=False)
    cost_analyst = await cost_hc.register(
        cost_agent, Passport(name="cost_analyst"), cost_resume, attach_plugin=False
    )
    operator = await operator_hc.register(
        operator_agent, Passport(name="operator"), operator_resume, attach_plugin=False
    )

    graph = TransitionGraph(
        initial_speaker=user.agent_id,
        transitions=[],
        default_target=AgentTarget(coordinator.agent_id),
        max_turns=14,
    )

    channel = await user.open(
        type=WORKFLOW_TYPE,
        target=[
            coordinator.agent_id,
            researcher.agent_id,
            critic.agent_id,
            cost_analyst.agent_id,
            operator.agent_id,
        ],
        knobs={"graph": graph.to_dict()},
    )
    print(f"channel: {channel.channel_id}\n")

    name_by_id = {
        user.agent_id: "user",
        coordinator.agent_id: "coordinator",
        researcher.agent_id: "researcher",
        critic.agent_id: "critic",
        cost_analyst.agent_id: "cost_analyst",
        operator.agent_id: "operator",
    }

    await channel.send("Should we adopt Kubernetes for an 8-person startup?")

    close_env = await user.wait_for_channel_event(
        channel_id=channel.channel_id,
        predicate=lambda e: e.event_type == EV_CHANNEL_CLOSED,
        timeout=180.0,
    )

    for env in await hub_obj.read_wal(channel.channel_id):
        speaker = name_by_id.get(env.sender_id, env.sender_id[:8])
        if env.event_type == EV_TEXT:
            print(f"{speaker:>13}: {env.event_data['text']}")
        elif env.event_type == EV_PACKET:
            routing = env.event_data.get("routing", {}) or {}
            kind = routing.get("kind")
            if kind == "handoff":
                target = name_by_id.get(routing.get("target", ""), "?")
                print(f"{speaker:>13}: [handoff → {target}] {routing.get('reason', '')}")
            elif kind == "finish":
                print(f"{speaker:>13}: [finish] {routing.get('summary', '')}")
            else:
                body = env.event_data.get("body", "")
                if body:
                    print(f"{speaker:>13}: {body}")

    print(f"\nclosed: reason={close_env.event_data.get('reason')!r}")

    await user_hc.close()
    await coord_hc.close()
    await researcher_hc.close()
    await critic_hc.close()
    await cost_hc.close()
    await operator_hc.close()
    await hub_obj.close()

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