AutoGen

  Microsoft AutoGen (team.run_stream)    Naive
  ───────────────────────────────       ─────
  AssistantAgent ── tool call ──▶ loop
        │  model picks an MCP tool
        ▼
  McpWorkbench (SSE + scoped bearer) ─▶  /mcp/sse/sess_…   (per-user session)
                                          │  AccountKit-gated, scoped to one user
                                          ▼
                                    connect GitHub · issue a $50 card · run a capability
                                          │
                                    sensitive? → 202 pending_approval (human-in-the-loop)

pip install -U "autogen-agentchat" "autogen-ext[openai,mcp]" httpx

export NAIVE_API_KEY=nv_sk_live_...
export OPENAI_API_KEY=sk-...

{
  "status": "pending_approval",
  "approval_id": "65589c8b-e033-4a65-b16c-379211c94429",
  "action": "cards.create",
  "primitive": "cards",
  "title": "Issue virtual card \"Ads budget\"",
  "message": "This action requires human approval before it executes."
}

def list_pending(user_id: str) -> list[dict]:
    r = httpx.get(
        f"{NAIVE_API}/v1/users/{user_id}/approvals",
        headers=AUTH,
        params={"status": "pending"},
    )
    r.raise_for_status()
    return r.json()["approvals"]

def approve(user_id: str, approval_id: str) -> dict:
    r = httpx.post(
        f"{NAIVE_API}/v1/users/{user_id}/approvals/{approval_id}/approve",
        headers=AUTH,
    )
    r.raise_for_status()
    return r.json()

# Find what the agent queued for Alice, then approve (or deny) it.
for a in list_pending(ALICE_USER_ID):
    # ...show a["title"] / a["action_type"] to a human in your UI...
    approve(ALICE_USER_ID, a["id"])  # API replays cards.create → real card

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.mcp import McpWorkbench, SseServerParams

async def run_team(user_id: str, task: str):
    session = mint_session(user_id)  # fresh, per-user session
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    naive_mcp = SseServerParams(
        url=session["mcp"]["url"],
        headers=session["mcp"]["headers"],
        sse_read_timeout=300,
    )

    async with McpWorkbench(server_params=naive_mcp) as workbench:
        planner = AssistantAgent(
            name="planner",
            model_client=model_client,
            system_message=(
                "Break the request into concrete steps for the operator. "
                "When the work is done, reply with the single word DONE."
            ),
        )
        operator = AssistantAgent(
            name="operator",
            model_client=model_client,
            workbench=workbench,            # only the operator can transact
            reflect_on_tool_use=True,
            system_message="Use Naive tools to act on this user's real account.",
        )

        team = RoundRobinGroupChat(
            [planner, operator],
            termination_condition=TextMentionTermination("DONE"),
        )
        await Console(team.run_stream(task=task))

    await model_client.close()

asyncio.run(run_team(
    "<user_id>",
    "Connect GitHub, then issue a $50 'Ads budget' card for marketing.",
))