MCP Integration

┌────────────────────────────────────────────────────────────┐
│  Your Python application (SDK Host)                        │
│                                                            │
│   ┌──────────────────────────┐                             │
│   │ create_sdk_mcp_server(...) │ ← In-Process tools        │
│   │  + @tool(...)             │     defined inline, no extra process │
│   └──────────────────────────┘                             │
│                  │                                         │
│                  ▼                                         │
│   ┌──────────────────────────┐                             │
│   │  query({mcp_servers})    │── stdio ─▶ qodercli child   │
│   │  / QoderSDKClient(...)   │                             │
│   └──────────────────────────┘                             │
│                                          │                 │
│                                          ├── stdio ──▶ MCP server (process)
│                                          ├── sse   ──▶ MCP server (HTTP/SSE)
│                                          └── http  ──▶ MCP server (Streamable HTTP)
└────────────────────────────────────────────────────────────┘

import asyncio
from typing import Annotated

from qoder_agent_sdk import (
    QoderAgentOptions,
    create_sdk_mcp_server,
    query,
    tool,
)


@tool("greet", "Greet someone.", {"name": Annotated[str, "Recipient name"]})
async def greet(args):
    return {"content": [{"type": "text", "text": f"Hello, {args['name']}!"}]}


server = create_sdk_mcp_server(name="my_tools", tools=[greet])


async def main():
    options = QoderAgentOptions(
        mcp_servers={"my_tools": server},
        allowed_tools=["mcp__my_tools__greet"],
    )
    async for msg in query(prompt="Use the greet tool to greet Alice", options=options):
        print(msg)


asyncio.run(main())

def tool(
    name: str,
    description: str,
    input_schema: type | dict[str, Any],
    annotations: ToolAnnotations | None = None,
) -> Callable[[Handler], SdkMcpTool[Any]]: ...


def create_sdk_mcp_server(
    name: str,
    version: str = "1.0.0",
    tools: list[SdkMcpTool[Any]] | None = None,
) -> McpSdkServerConfig: ...

# Success return
{"content": [{"type": "text", "text": "result"}]}

# Business failure: use is_error instead of throwing an exception
{"content": [{"type": "text", "text": "error description"}], "is_error": True}

@tool(
    "query_db",
    "Read-only SQL query.",
    {"sql": Annotated[str, "SQL query statement"]},
    annotations=ToolAnnotations(readOnlyHint=True),
)
async def query_db(args):
    sql = args["sql"]
    if not sql.lstrip().upper().startswith("SELECT"):
        return {
            "is_error": True,
            "content": [{"type": "text", "text": "Only SELECT statements are allowed"}],
        }
    rows = await db.query(sql)
    return {"content": [{"type": "text", "text": json.dumps(rows)}]}

@tool("watch", "Watch a counter", {"max": int})
async def watch(args, extra):
    for i in range(args["max"]):
        if extra.signal.is_set():
            return {"content": [{"type": "text", "text": f"aborted at {i}"}]}
        await asyncio.sleep(0.01)
    return {"content": [{"type": "text", "text": "done"}]}

class McpStdioServerConfig(TypedDict):
    type: NotRequired[Literal["stdio"]]    # optional; stdio is the default
    command: str                           # executable command
    args: NotRequired[list[str]]           # command arguments
    env: NotRequired[dict[str, str]]       # environment variables
    tools: NotRequired[list[McpServerToolPolicy]]

options = QoderAgentOptions(
    mcp_servers={
        "fs": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/project"],
        },
        "gh": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-github"],
            "env": {"GITHUB_TOKEN": os.environ["GITHUB_TOKEN"]},
        },
    },
)

class McpSSEServerConfig(TypedDict):
    type: Literal["sse"]
    url: str
    headers: NotRequired[dict[str, str]]
    tools: NotRequired[list[McpServerToolPolicy]]


class McpHttpServerConfig(TypedDict):
    type: Literal["http"]                   # Streamable HTTP
    url: str
    headers: NotRequired[dict[str, str]]
    tools: NotRequired[list[McpServerToolPolicy]]

options = QoderAgentOptions(
    mcp_servers={
        "analytics": {
            "type": "http",
            "url": "https://analytics.example.com/mcp",
            "headers": {"Authorization": f"Bearer {os.environ['ANALYTICS_TOKEN']}"},
        },
    },
)

mcp__<server_name>__<tool_name>

options = QoderAgentOptions(
    mcp_servers={"my_tools": server},
    tools=[
        "Read", "Grep",                  # built-in tools you still want
        "mcp__my_tools__greet",
        "mcp__my_tools__search_docs",
    ],
)

options = QoderAgentOptions(
    mcp_servers={"my_tools": server},
    allowed_tools=[
        "mcp__my_tools__greet",          # pre-approved, no prompt
        "mcp__my_tools__search_docs",
    ],
)

options = QoderAgentOptions(
    mcp_servers={
        "keep": {"command": "..."},
        "drop": {"command": "..."},
    },
    allowed_mcp_server_names=["keep"],   # 'drop' still appears in status but does not connect
    strict_mcp_config=True,              # do not load MCP servers from settings.json / .mcp.json
)

async with QoderSDKClient(options) as client:
    status = await client.get_mcp_status()
    # Returns McpStatusResponse: {"mcpServers": [McpServerStatus, ...]}
    # Each McpServerStatus contains:
    #   name, status, serverInfo?, error?, config?, scope?, tools?

    for server in status["mcpServers"]:
        print(f"{server['name']}: {server['status']}")
        if server["status"] == "connected":
            print("  tools:", [t["name"] for t in server.get("tools", [])])

async def on_status(msg):
    print(f"{msg['server_name']} -> {msg['status']}")
    if msg.get("error"):
        print("  error:", msg["error"])


options = QoderAgentOptions(
    mcp_servers={...},
    on_mcp_status_change=on_status,
)

options = QoderAgentOptions(
    control_request_timeout_ms=20_000,   # default 60_000; pass 0 to disable
)

async def handle_oauth(request: McpOAuthRequest) -> McpOAuthResolution | None:
    # Open request['auth_url'] in an Electron BrowserWindow / system browser
    callback_url = await open_browser_and_wait_for_callback(request["auth_url"])
    return {"callbackUrl": callback_url}


options = QoderAgentOptions(
    mcp_servers={"analytics": {"type": "http", "url": "https://analytics.example.com/mcp"}},
    on_mcp_oauth_required=handle_oauth,
    control_request_timeout_ms=120_000,   # user authorization may take a while
)

async with QoderSDKClient(options) as client:
    status = await client.get_mcp_status()
    for server in status["mcpServers"]:
        if server["status"] != "needs-auth":
            continue
        result = await client.mcp_authenticate(server["name"])
        if result.get("requiresUserAction"):
            await open_in_browser(result["authUrl"])
            callback_url = await wait_for_user_paste_callback()
            await client.mcp_submit_oauth_callback_url(server["name"], callback_url)
        # Silent path (cached client + valid refresh token): requiresUserAction=False
        # No UI needed; the server transitions directly to connected

    # The tools list is now stable; safe to send user messages
    await client.query("first user message")

from qoder_agent_sdk import ElicitationRequest, ElicitationResult, QoderAgentOptions


async def on_elicitation(req: ElicitationRequest) -> ElicitationResult:
    # Form mode: req["requestedSchema"] is a JSON Schema; the returned content must match.
    if req.get("mode") == "form":
        return {"action": "accept", "content": {"token": "xxx"}}
    # Return decline / cancel for any case you cannot handle; the CLI relays the result back to the MCP server.
    return {"action": "decline"}


options = QoderAgentOptions(
    mcp_servers={"my_server": {"type": "http", "url": "..."}},
    on_elicitation=on_elicitation,
)

from qoder_agent_sdk import HookMatcher, QoderAgentOptions


async def on_elicit(input, tool_use_id, context):
    print(
        "elicit from",
        input["mcp_server_name"],
        "mode=",
        input["mode"],
        "schema=",
        input.get("requested_schema"),
    )
    return {"continue_": True}


options = QoderAgentOptions(
    mcp_servers={"my_server": {"type": "http", "url": "..."}},
    hooks={
        "Elicitation": [HookMatcher(hooks=[on_elicit])],
    },
)

from qoder_agent_sdk import (
    # Factories
    create_sdk_mcp_server,
    tool,
    SdkMcpTool,
    # Configs
    McpServerConfig,
    McpStdioServerConfig,
    McpSSEServerConfig,
    McpHttpServerConfig,
    McpSdkServerConfig,
    McpServerToolPolicy,
    # Status
    McpServerStatus,
    McpServerStatusConfig,
    McpServerConnectionStatus,
    McpServerInfo,
    McpToolInfo,
    McpToolAnnotations,
    McpStatusResponse,
    McpStatusChangeMessage,
    # Runtime changes
    McpSetServersResult,
    # OAuth
    OAuthToken,
    McpOAuthRequest,
    McpOAuthResolution,
    McpOAuthTokenResolution,
    McpOAuthCallbackUrlResolution,
    McpOAuthCodeResolution,
    OnMcpOAuthRequired,
    OnMcpStatusChange,
)

import asyncio
import json
import os
from typing import Annotated

from mcp.types import ToolAnnotations

from qoder_agent_sdk import (
    AssistantMessage,
    McpOAuthRequest,
    McpOAuthResolution,
    QoderAgentOptions,
    QoderSDKClient,
    ResultMessage,
    TextBlock,
    create_sdk_mcp_server,
    tool,
)


# 1. Define application tools
@tool(
    "get_user_orders",
    "Query a user's orders, optionally filtered by status.",
    {
        "user_id": Annotated[str, "User UUID"],
        "status": Annotated[str, "Filter by order status: pending/paid/shipped/cancelled"],
    },
    annotations=ToolAnnotations(readOnlyHint=True),
)
async def get_user_orders(args):
    try:
        orders = await db.get_orders(args["user_id"], args.get("status"))
        return {"content": [{"type": "text", "text": json.dumps(orders)}]}
    except Exception as e:
        return {
            "is_error": True,
            "content": [{"type": "text", "text": f"Query failed: {e}"}],
        }


# 2. Assemble the server
crm = create_sdk_mcp_server(name="crm", tools=[get_user_orders])


# 3. Prepare the inbound OAuth callback (called by the CLI when a remote server requires auth)
async def handle_oauth(request: McpOAuthRequest) -> McpOAuthResolution | None:
    callback_url = await open_browser_and_wait_for_callback(request["auth_url"])
    return {"callbackUrl": callback_url}


# 4. Start the client; complete authentication before the first message
async def main():
    options = QoderAgentOptions(
        mcp_servers={
            "crm": crm,
            "analytics": {"type": "http", "url": "https://analytics.example.com/mcp"},
        },
        allowed_tools=["mcp__crm__get_user_orders"],
        on_mcp_oauth_required=handle_oauth,
        control_request_timeout_ms=120_000,
    )

    async with QoderSDKClient(options) as client:
        # Outbound fallback: even when the CLI did not actively go inbound, pull status and drive auth ourselves
        status = await client.get_mcp_status()
        for server in status["mcpServers"]:
            if server["status"] != "needs-auth":
                continue
            result = await client.mcp_authenticate(server["name"])
            if result.get("requiresUserAction"):
                callback_url = await open_browser_and_wait_for_callback(result["authUrl"])
                await client.mcp_submit_oauth_callback_url(server["name"], callback_url)

        # 5. Consume messages (the tools list is now stable; the prefix cache will be established correctly)
        await client.query("Find user-123's recent paid orders")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)
            elif isinstance(msg, ResultMessage):
                if msg.subtype == "success":
                    print("done, cost=", msg.total_cost_usd)
                else:
                    print("failed:", msg.subtype)


asyncio.run(main())