Claude Agent SDK

  Claude Agent SDK (query)               Naive
  ────────────────────────               ─────
  agent loop ─────── tool call ─────────▶ loop
    │  model picks an MCP tool
    ▼
  mcp_servers["naive"] (SSE + 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 claude-agent-sdk httpx
npm install -g @anthropic-ai/claude-code

export NAIVE_API_KEY=nv_sk_live_...
export ANTHROPIC_API_KEY=sk-ant-...

from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    ResultMessage,
    ToolPermissionContext,
    PermissionResultAllow,
    PermissionResultDeny,
)

async def gate(tool_name: str, input: dict, context: ToolPermissionContext):
    # Hold spend for out-of-band review; allow everything else.
    if tool_name == "mcp__naive__naive_cards_create":
        # ...surface tool_name / input to a human in your UI, then decide...
        return PermissionResultDeny(message="Card issuance requires human approval.")
    return PermissionResultAllow()

import asyncio

async def stream(text: str):
    yield {
        "type": "user",
        "message": {"role": "user", "content": text},
        "parent_tool_use_id": None,
        "session_id": "alice",
    }

async def main():
    session = mint_session(ALICE_USER_ID)
    options = ClaudeAgentOptions(
        model="claude-sonnet-4-5",
        mcp_servers={
            "naive": {
                "type": "sse",
                "url": session["mcp"]["url"],
                "headers": session["mcp"]["headers"],
            }
        },
        allowed_tools=["mcp__naive__*"],
        can_use_tool=gate,  # in-run gate
    )

    async for message in query(
        prompt=stream("Issue a $50 virtual card called 'Ads budget'."),
        options=options,
    ):
        if isinstance(message, ResultMessage):
            print(message.result)

asyncio.run(main())

{
  "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 run_for_user(user_id: str, task: str) -> str | None:
    session = mint_session(user_id)  # fresh, per-user session

    options = ClaudeAgentOptions(
        model="claude-sonnet-4-5",
        system_prompt="Use the Naive tools to act on this user's real account.",
        mcp_servers={
            "naive": {
                "type": "sse",
                "url": session["mcp"]["url"],
                "headers": session["mcp"]["headers"],
            }
        },
        allowed_tools=["mcp__naive__*"],
    )

    result = None
    async for message in query(prompt=task, options=options):
        if isinstance(message, ResultMessage) and message.subtype == "success":
            result = message.result
    return result

# Same agent code, different tenant — isolated identity, spend, and approvals each time.
# await run_for_user("<user_id>", "Connect GitHub and set up a $50 ads card.")