DSPy

  DSPy (ReAct reasoning loop)              Naive
  ─────────────────────────────           ─────
  LM reasons
    │  tool call (naive_cards_create)
    ▼  dspy.Tool (async) → session.call_tool
  ClientSession ── sse_client ──────────────▶ /mcp/sse/sess_…   (per-user session)
    (Authorization: Bearer nv_sess_…)          │  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 dspy mcp httpx

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

GATED = {"naive_cards_create"}

def gate(tool: dspy.Tool, confirm) -> dspy.Tool:
    """Wrap a dspy.Tool so a human confirms before it runs. Schema is preserved."""
    inner = tool.func

    async def guarded(*args, **kwargs):
        if not await confirm(tool.name, kwargs):
            return {"status": "rejected_by_human", "tool": tool.name}
        return await inner(*args, **kwargs)

    return dspy.Tool(
        func=guarded,
        name=tool.name,
        desc=tool.desc,
        args=tool.args,
        arg_types=tool.arg_types,
        arg_desc=tool.arg_desc,
    )

# ...inside the ClientSession block, before building the agent:
naive_tools = [
    gate(t, confirm) if t.name in GATED else t
    for t in (dspy.Tool.from_mcp_tool(session, x) for x in listed.tools)
]

{
  "status": "pending_approval",
  "approval_id": "65589c8b-e033-4a65-b16c-379211c94429",
  "action_type": "cards.create",
  "primitive": "cards",
  "title": "Issue virtual card \"Ads budget\""
}

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

async def handle(user_id: str, request: str) -> str:
    mcp = mint_session(user_id)["mcp"]  # fresh, per-user session

    async with sse_client(url=mcp["url"], headers=mcp["headers"]) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            listed = await session.list_tools()

            # Optional: only expose a subset of Naive's tools to this agent.
            allowed = {"naive_connections_connect", "naive_cards_create"}
            naive_tools = [
                dspy.Tool.from_mcp_tool(session, t)
                for t in listed.tools
                if t.name in allowed
            ]

            agent = dspy.ReAct(OpsAgent, tools=naive_tools)
            result = await agent.acall(request=request)
            return result.summary

# Same agent definition, different tenant — isolated identity, spend, and approvals each time.