MCP Integration

MCP (Model Context Protocol) is an open protocol for AI Agents to invoke external tools. The Python SDK has built-in MCP client capabilities — the host application only needs to describe “which MCP servers exist,” and the SDK automatically handles connection, tool discovery, message routing, OAuth, and state synchronization.

Architecture Overview

┌────────────────────────────────────────────────────────────┐
│  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)
└────────────────────────────────────────────────────────────┘

• In-Process: The tool is a Python async function running in your own process. The product of create_sdk_mcp_server communicates with the CLI via the SDK’s control channel without spawning a child process.
• External: You declare a child process or remote URL in the configuration; the CLI handles connection, discovery, and invocation.

---

Three Integration Methods

Method	Config type	Process Boundary	Use Case
In-Process	'sdk' (created by create_sdk_mcp_server)	Same process	Custom application tools that need direct access to host state
Stdio	'stdio' (can be omitted)	Child process	Existing MCP toolkits (@modelcontextprotocol/server-*)
SSE / HTTP	'sse' / 'http'	Remote	Remote services, SaaS tools, services requiring OAuth

All three methods can be mixed — register multiple servers of different types in the same query() / QoderSDKClient.

💡 mcp_servers can also accept a str / pathlib.Path pointing to a JSON configuration file path; the SDK will pass it through to the CLI as --mcp-config <path>.

---

In-Process Server (Recommended)

In-process tools are the most straightforward extension method: define a regular async function, declare a schema with the decorator, and it becomes callable by the Agent. For detailed @tool() / schema / handler behavior, see tools.md; this section only covers parts related to MCP server assembly.

30-Second Getting Started

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())

@tool() / create_sdk_mcp_server() Full Signatures

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: ...

Parameter	Description
name (tool)	Tool name; the fully-qualified name will be mcp__<server>__<name>
description	Description for the model, determining when the AI invokes it — clearly state What/When
input_schema	Three forms: simple dict / TypedDict / full JSON Schema dict; see Tools Reference - input_schema
annotations	MCP tool annotations; see table below
name (server)	Server name (determines tool prefix mcp__<name>__)
version	Defaults to '1.0.0'
tools	List of SdkMcpTool

The return value McpSdkServerConfig looks like {"type": "sdk", "name": ..., "instance": ...} and can be placed directly into options.mcp_servers.

Annotations Actually Consumed

The three fields below are consumed by the SDK and returned to the host via get_mcp_status().mcpServers[i].tools[i].annotations:

Field	What it does	Host-side reads as
readOnlyHint	Declares the tool is read-only. Read-only tools may run concurrently (they don’t block each other in the same batch); the TUI renders a [read-only] badge in tool details	annotations.readOnly
destructiveHint	Declares the tool performs destructive operations. The TUI renders a [destructive] badge in tool details	annotations.destructive
openWorldHint	Declares the tool interacts with the outside world (e.g., web search, third-party API calls). The TUI renders an [open-world] badge in tool details	annotations.openWorld

Note that host-side field names drop the Hint suffix: readOnlyHint → annotations.readOnly, and so on. The annotations object only contains fields that were explicitly set.

⚠️ These fields do NOT affect auto-mode permission decisions. The CLI treats server-declared annotations as unverifiable advisory metadata (servers can freely under- or over-declare) and intentionally keeps them out of the permission pipeline — admitting them would launder authority for the server’s self-assessment. To hard-block specific tools, use the allowed_tools allowlist or hooks — annotations are for host-side identification (get_mcp_status) and TUI display only.

idempotentHint and title are not currently consumed by the SDK — passing them won’t error, but they neither affect CLI behavior nor appear in get_mcp_status(). If your application needs this information, maintain the mapping yourself on the host side.

💡 About maxResultSizeChars: The Python SDK writes anthropic/maxResultSizeChars into the tool’s _meta via ToolAnnotations(maxResultSizeChars=...), and the CLI uses this to relax the default 50K return length limit. This field is a Python-side incremental capability (TS exposes it via the same-named annotation; wire format is identical).

Handler Return Value

# 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}

For business failures, use is_error: True instead of throwing an exception. The full content type description, along with several behavior differences between Python and TS (resource_link degrades to text, top-level _meta is not passed through, binary embedded resources are skipped), is in Tools Reference - CallToolResult.

@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)}]}

Handler Cancellation Signal

The handler can optionally accept a second parameter ToolInvocationContext to cooperatively exit when the CLI cancels the current call via extra.signal:

@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"}]}

⚠️ Do not reuse the same server config across multiple query() calls: Each query binds an independent transport. Reusing the same config has no side effects, but you also won’t get “cross-query shared state” — for shared state, place it in module scope outside the handler closure.

---

Stdio Server

Communicates with MCP servers via a child process’s stdin/stdout. The @modelcontextprotocol/server-* packages on NPM are all stdio implementations.

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"]},
        },
    },
)

When command is unreachable or fails to start, it does not bring down the entire query — that server’s status stays non-'connected', and other servers are unaffected.

---

SSE / HTTP Server

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']}"},
        },
    },
)

Likewise, an unreachable remote URL won’t hang the query; the server status stays non-'connected', and other servers are unaffected. For remote services requiring OAuth, see OAuth Authentication.

---

Tool Naming and Allowlists

The CLI uniformly prefixes MCP tools when exposing them to the model:

mcp__<server_name>__<tool_name>

For example, server name my_tools with tool name greet gives the model the tool name mcp__my_tools__greet. Server names may contain hyphens and other special characters (my-tools → mcp__my-tools__<tool>).

tools: Restrict Which Tools the Model Can See

Use tools when you want the model to only see a subset of tools. The CLI adds every built-in tool not in the list to the disallow set — effectively a “visibility allowlist”:

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",
    ],
)

⚠️ Omitting tools means everything is exposed: all built-in tools plus every tool from connected MCP servers reach the model. For production, list them explicitly to tighten scope.

allowed_tools: Pre-approval (Not a Visibility Allowlist)

allowed_tools adds listed tools to the “always-allow” rule set — calls skip the permission prompt, but unlisted tools are not hidden. Use it to whitelist low-risk MCP tools for unattended use:

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

Omitting allowed_tools just means no pre-approval rules — the model still sees and can call every tool; write operations simply route through the regular permission_mode approval flow. See Permissions docs for full semantics.

allowed_mcp_server_names: Process-Server Allowlist

Only filters process-based (stdio/sse/http) servers; does not affect in-process servers. Combined with strict_mcp_config=True, it can prevent the CLI from loading additional local configurations:

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
)

⚠️ Omitting allowed_mcp_server_names means all process servers connect; to tighten, list them explicitly. In-process servers are never filtered by this field.

---

Runtime Management (QoderSDKClient)

query() is a single-shot iterator and cannot change servers or auth mid-stream. For runtime management of MCP, use QoderSDKClient, which exposes status queries, OAuth, server add/remove, reconnect / toggle, etc., as public methods.

⚠️ Caching Principle: MCP server config / auth state changes rebuild the tools list, which breaks the prompt prefix cache mid-session. The SDK provides methods for “querying status + completing auth before the first message”; the server set itself should be configured once at startup via options.mcp_servers, creating a new QoderSDKClient when necessary.

Querying Status

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", [])])

💡 The MCP handshake occurs after the CLI completes initialize but before the first user message. QoderSDKClient.connect() already waits until initialize returns; handshake IO may take a few hundred milliseconds, so poll get_mcp_status() until connected if needed.

Subscribing to Status Changes

Pick one of two ways:

Method 1 (recommended): Attach an on_mcp_status_change callback on options; it is called every time status changes.

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,
)

Method 2: Consume the message stream and filter system/mcp_status_change. The callback and the message stream share the same payload; the callback simply removes the need for filtering.

Runtime Add/Remove Server / Reconnect / Toggle

Method	Purpose
client.set_mcp_servers(servers)	Replace the current MCP configuration with the full desired mapping; returns {added, removed, errors}
client.reconnect_mcp_server(name)	Reconnect a specific server, typically to recover from a 'failed' state
client.toggle_mcp_server(name, enabled)	Enable / disable a server; disabling disconnects it and removes its tools

⚠️ All three methods trigger a tools-list rebuild and therefore break the prompt prefix cache. In production, prefer to configure mcp_servers fully at startup; reserve these APIs for debugging and local development.

Controlling Request Timeout

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

After timeout, the SDK automatically writes a control_cancel_request and rejects the current Future.

---

OAuth Authentication

Remote MCP servers (HTTP/SSE) often require OAuth. The CLI has a complete built-in OAuth 2.0 + PKCE + Dynamic Client Registration (RFC 7591) implementation. The Python SDK exposes both inbound (CLI proactively asks the host to complete OAuth) and outbound (host actively triggers OAuth) paths; choose based on your host shape.

⚠️ Caching Principle: After OAuth completes, the CLI reconnects to the server and rediscovers tools, which inevitably breaks the prompt prefix cache mid-session. Complete authentication before sending the first user message so the tools list stabilizes before the conversation starts.

💡 This section only covers CLI-driven OAuth: The CLI performs metadata discovery, PKCE, token exchange, and token persistence itself. There is another server-driven auth path — where the server uses MCP elicitation/create to have the client redirect to a URL for authorization. The two paths are independent and won’t trigger simultaneously. See Elicitation: Server Requests User Input.

Inbound: on_mcp_oauth_required Callback

When the CLI detects during handshake that a server requires OAuth, it pushes an McpOAuthRequest to the SDK via control_request, and the SDK invokes the host’s on_mcp_oauth_required callback. The host returns one of the following resolutions:

Return type	Meaning
OAuthToken or {"token": OAuthToken}	The host runs the entire OAuth flow itself and injects the token directly into the CLI
{"callbackUrl": "..."}	The host returns the full callback URL (including code / state); the CLI parses it and exchanges for a token
{"code": "...", "state": "..."}	The host extracts the code itself and returns it to the CLI
None	Reject; the CLI marks that server as failed

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
)

Outbound: Host Actively Drives Authentication

When the host’s own UI has a “Sign in” entry, it can actively invoke:

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")

Method	Purpose	When to Call
client.mcp_authenticate(name, redirect_uri=None)	Initiate OAuth; returns {authUrl?, requiresUserAction}; on silent renewal, requiresUserAction=False and no UI is needed	Before the first user message
client.mcp_submit_oauth_callback_url(name, callback_url)	Submit the complete callback URL (with code/state)	Before the first user message
client.inject_mcp_token(name, token)	The host runs the entire OAuth flow itself and injects the OAuthToken into the CLI	Before the first user message
client.mcp_clear_auth(name)	Delete the OAuth credentials stored by the CLI — equivalent to “sign out”	Any time; the next tool call will trigger re-authentication

redirect_uri is optional, overriding the default OAuth callback target (Electron custom protocol, enterprise intranet callback addresses, etc.).

The CLI stores tokens in the system Keychain by default (macOS / Linux Secret Service), falling back to ~/.qoder/mcp-oauth-tokens.json (0o600 permissions + cross-process locking).

---

Elicitation: Server Requests User Input

MCP elicitation/create is a server → client request used to have the client display an interaction to the user (form mode collects structured input; url mode asks the user to visit a URL to complete an action).

✅ The Python SDK is now aligned with the TS SDK: QoderAgentOptions.on_elicitation accepts an async callback that returns ElicitationResult, with the same signature as the TS version. When the callback is not set, the SDK still defaults to answering {"action": "cancel"}. The Elicitation / ElicitationResult hook events still fire in parallel as a read-only observation channel.

⚠️ The current CLI does not advertise the elicitation.url capability. A server’s elicit({mode: 'url'}) is rejected directly by the CLI (MCP error -32602: Client does not support URL-mode elicitation requests), so URL-mode elicit will not reach the SDK, and the system/elicitation_complete notification will not fire on the current CLI either. Once the CLI enables the URL capability, this path will automatically become operational.

Responding to elicit with on_elicitation

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,
)

• Field names follow the TS SDK’s camelCase (serverName / elicitationId / requestedSchema / displayName); the CLI’s snake_case payload is converted automatically by the SDK.
• Returning None is equivalent to {"action": "cancel"}, making it convenient for the host to bail out from a fallback path.
• You can also return a mcp.types.ElicitResult Pydantic model (the SDK calls model_dump).

Observing elicitation (hook channel)

After on_elicitation lands, the Elicitation / ElicitationResult hooks still fire in parallel — they are a read-only observation channel and do not make decisions.

Hook Event	Timing	Payload TypedDict
Elicitation	When server request arrives	ElicitationHookInput — mcp_server_name / message / mode / elicitation_id? / requested_schema? / url? / title?
ElicitationResult	After SDK / host completes its response	ElicitationResultHookInput — mcp_server_name / action / mode / elicitation_id? / content?

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])],
    },
)

Boundary with the OAuth Path

• CLI-driven OAuth (mcp_authenticate / inject_mcp_token / on_mcp_oauth_required): Token stored in qodercli Keychain; driven when get_mcp_status() shows needs-auth; does NOT trigger the Elicitation hook.
• Server-driven elicit: Token stays internal to the server; get_mcp_status() does not show needs-auth; decisions are made via the on_elicitation callback (the SDK auto-cancels when not registered).

The two paths are not mutually exclusive but don’t overlap: the same server typically uses only one.

---

Options Reference

Field	Type	Default	Description
mcp_servers	dict[str, McpServerConfig] | str | Path	{}	Server name → config; or a JSON config file path
allowed_mcp_server_names	list[str]	[]	Process-based server allowlist (does not affect in-process); empty list means all are open
strict_mcp_config	bool	False	Prevent the CLI from loading additional MCP from user config files
tools	list[str] | ToolsPreset | None	None	Model-visible tool allowlist; omitting means every built-in + MCP tool is visible
allowed_tools	list[str]	[]	Pre-approval list (skip permission prompts; does not control visibility); empty list means no pre-approval rules
disallowed_tools	list[str]	[]	Explicit deny list; takes precedence over allow
control_request_timeout_ms	int	60_000	Control request timeout (including mcp series), 0 to disable
on_mcp_oauth_required	OnMcpOAuthRequired | None	None	Triggered when the CLI detects that a server requires OAuth
on_mcp_status_change	OnMcpStatusChange | None	None	Triggered on every server status change; equivalent to filtering the system/mcp_status_change stream
on_elicitation	OnElicitation | None	None	Host decision when a server requests user input via MCP elicitation/create; the SDK auto-cancels if not set
hooks['Elicitation']	list[HookMatcher]	–	Read-only observation hook when a server requests user input (decisions go through on_elicitation)

Methods on QoderSDKClient

Method	Description	When to Call
get_mcp_status()	Get current status of all MCP servers	Any time
set_mcp_servers(servers)	Replace the entire MCP server configuration; returns {added, removed, errors}	Any time (breaks the prefix cache)
reconnect_mcp_server(name)	Reconnect a specific server	Any time
toggle_mcp_server(name, enabled)	Enable / disable a server	Any time
mcp_authenticate(name, redirect_uri=None)	Actively initiate OAuth	Before the first user message
mcp_submit_oauth_callback_url(name, callback_url)	Submit OAuth callback URL	Before the first user message
inject_mcp_token(name, token)	Inject a token after running the full OAuth flow on the host	Before the first user message
mcp_clear_auth(name)	Delete stored OAuth credentials	Any time

---

Type Reference

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,
)

McpServerStatus.status enum (McpServerConnectionStatus):

Value	Meaning
'pending'	Registered, connection not yet started
'connecting'	Handshaking
'connected'	Connected, tools are callable
'failed'	Connection failed (check the error field)
'needs-auth'	Requires OAuth, proceed with auth flow
'disabled'	Disabled (determined by CLI internal config or toggle_mcp_server)

---

Best Practices

1. Write descriptions for the AI: The @tool description determines when the AI selects it. Clearly state “what it does, when to use it, what it should NOT be used for.”
2. Add Annotated to parameters: In simple dict / TypedDict, write Annotated[type, "..."] on fields; the AI uses this information to construct call arguments.
3. Use is_error: True for failures, don’t throw exceptions: Let the AI see the result. For a complete comparison, see Tools Guide - How the SDK handles tool errors.
4. Prefer read-only + readOnlyHint: Be cautious with write operations; pair with can_use_tool or hooks for secondary confirmation.
5. Keep server names short: They appear in tool prefixes; overly long names waste tokens.
6. Place in-process shared state in module scope: Handlers are closures, but each query still reuses the same server instance.
7. Complete OAuth before the first user message: Use mcp_authenticate + mcp_submit_oauth_callback_url, the inbound on_mcp_oauth_required callback, or inject_mcp_token. Completing auth mid-session inevitably breaks the prompt prefix cache.
8. Pull MCP status with get_mcp_status() or on_mcp_status_change: The push channel (status change message) is retained; pick one as needed.
9. Set a reasonable control_request_timeout_ms: Remote server handshakes may take seconds; the default 60s is usually sufficient. Increase it when waiting for user OAuth actions, and set it explicitly in CI environments.
10. Use strict_mcp_config for isolation: Prevent MCP servers declared in the user’s local ~/.qoder/settings.json / .mcp.json from interfering with your application.

---

Complete Example

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())