Google ADK

# Control plane: a reusable policy template.
curl -X POST https://api.usenaive.ai/v1/account-kits \
  -H "Authorization: Bearer $NAIVE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Pro",
    "primitives_config": {
      "cards": { "enabled": true, "requiresApproval": true, "defaults": { "spending_limit_cents": 50000 } },
      "vault": { "enabled": true }
    },
    "connections_config": { "mode": "allowlist", "toolkits": ["github", "gmail", "stripe"] }
  }'
# → { "id": "<kit_id>", ... }

# Provision one of your end-users and assign the kit.
curl -X POST https://api.usenaive.ai/v1/users \
  -H "Authorization: Bearer $NAIVE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{ "external_id": "user_123", "email": "alice@acme.com", "account_kit_id": "<kit_id>" }'
# → { "id": "<user_id>", ... }

import os
import httpx

NAIVE_API = "https://api.usenaive.ai"
AUTH = {"Authorization": f"Bearer {os.environ['NAIVE_API_KEY']}"}

ALICE_USER_ID = "<user_id>"  # from the control-plane step

def mint_session(user_id: str, ttl_ms: int = 15 * 60 * 1000) -> dict:
    r = httpx.post(
        f"{NAIVE_API}/v1/users/{user_id}/sessions",
        headers=AUTH,
        json={"ttl_ms": ttl_ms},
    )
    r.raise_for_status()
    return r.json()  # { id, expires_at, mcp: { url, headers, expires_at } }

import asyncio
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.mcp_tool import McpToolset, SseConnectionParams
from google.genai import types

async def main():
    session = mint_session(ALICE_USER_ID)

    naive_tools = McpToolset(
        connection_params=SseConnectionParams(
            url=session["mcp"]["url"],          # scoped, per-user endpoint
            headers=session["mcp"]["headers"],  # scoped bearer — never in the URL
        ),
    )

    agent = LlmAgent(
        model="gemini-2.5-flash",
        name="ops_agent",
        instruction="You are Alice's operations agent. Use Naive tools to act on her real account.",
        tools=[naive_tools],
    )

    svc = InMemorySessionService()
    run = await svc.create_session(app_name="naive_ops", user_id="alice")
    runner = Runner(app_name="naive_ops", agent=agent, session_service=svc)

    message = types.Content(
        role="user",
        parts=[types.Part(text=(
            "Connect my GitHub, then issue a $50 virtual card called "
            "'Ads budget' for our marketing spend."
        ))],
    )

    try:
        async for event in runner.run_async(
            user_id="alice", session_id=run.id, new_message=message
        ):
            if event.is_final_response():
                print(event.content.parts[0].text)
    finally:
        await naive_tools.close()

asyncio.run(main())