MCP 集成

MCP（Model Context Protocol）是 AI Agent 调用外部工具的开放协议。Python SDK 内置了 MCP 客户端能力，宿主应用只需要描述「有哪些 MCP 服务器」，SDK 会自动完成连接、工具发现、消息路由、OAuth、状态同步等工作。

​

一图看懂

┌────────────────────────────────────────────────────────────┐
│  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：工具就是一个 Python async 函数，运行在你自己的进程里。create_sdk_mcp_server 产物通过 SDK 的 control channel 与 CLI 通信，不会再起一个子进程。
• External：你在配置里声明子进程或远端 URL，CLI 负责连接、发现、调用。

---

​

三种接入方式

方式	配置项 type	进程边界	适用场景
In-Process	'sdk'（由 create_sdk_mcp_server 创建）	同进程	自定义业务工具，需要直接访问 host 状态
Stdio	'stdio'（可省略）	子进程	已有 MCP 工具包（@modelcontextprotocol/server-*）
SSE / HTTP	'sse' / 'http'	远程	远端服务、SaaS 工具、需要 OAuth 的服务

三种方式可以混用——在同一个 query() / QoderSDKClient 里同时注册多个不同类型的服务器。

💡 mcp_servers 也可以传 str / pathlib.Path：指向一个 JSON 配置文件路径，SDK 会以 --mcp-config <path> 透传给 CLI。

---

​

In-Process Server（推荐）

In-process 工具是最直接的扩展方式：定义一个普通的 async 函数，用装饰器声明 schema，就能被 Agent 调用。详细的 @tool() / schema / handler 行为见 tools.md，本节只覆盖与 MCP server 装配相关的部分。

​

30 秒上手

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() 完整签名

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

参数	说明
name（tool）	工具名，全限定名将是 mcp__<server>__<name>
description	给模型看的说明，决定 AI 何时调用——写清楚 What/When
input_schema	简单 dict / TypedDict / 完整 JSON Schema dict 三种写法，详见 Tools Reference - input_schema
annotations	MCP 工具注解，详见下表
name（server）	server 名（决定工具前缀 mcp__<name>__）
version	默认 '1.0.0'
tools	SdkMcpTool 列表

返回值 McpSdkServerConfig 形如 {"type": "sdk", "name": ..., "instance": ...}，直接塞进 options.mcp_servers 即可。

​

annotations 实际支持

下列三个字段会被 SDK 真正消费,并通过 get_mcp_status().mcpServers[i].tools[i].annotations 回传到宿主侧:

字段	作用	宿主侧读法
readOnlyHint	声明工具是只读的。只读工具可以并发执行(同批次内不互相阻塞);TUI 工具详情会渲染 [read-only] 徽章	annotations.readOnly
destructiveHint	声明工具会执行破坏性操作。TUI 工具详情会渲染 [destructive] 徽章	annotations.destructive
openWorldHint	声明工具会触达外部世界(联网搜索、调用第三方 API)。TUI 工具详情会渲染 [open-world] 徽章	annotations.openWorld

注意宿主侧字段名是去掉 Hint 后缀的:readOnlyHint → annotations.readOnly,依此类推。annotations 对象只包含被显式设置的字段。

⚠️ 这三个字段不会影响 auto 模式的权限决策。CLI 把 server 自声明的 annotation 视为不可验证的提示信息(server 可以随意 under-/over-declare),不会把它们带进权限管线,以免变相替 server 的自我标榜背书。要硬性拒绝某些工具,请用 allowed_tools 白名单或 hooks 拦截——annotation 仅用于宿主侧识别(get_mcp_status)和 TUI 展示。

idempotentHint 和 title 目前不被 SDK 消费——传了不会报错,但既不影响 CLI 行为、也不会出现在 get_mcp_status() 的回传里。如果你的应用需要这些信息,请在宿主侧自行维护映射。

💡 关于 maxResultSizeChars：Python SDK 通过 ToolAnnotations(maxResultSizeChars=...) 把 anthropic/maxResultSizeChars 写到工具的 _meta，CLI 据此放宽默认 50K 的返回长度限制。该字段是 Python 端的增量能力（TS 通过同名 annotation 暴露，wire 一致）。

​

handler 返回值

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

业务失败请用 is_error: True 而不是抛异常。完整的 content 类型说明、Python 端与 TS 端的几处行为差异（resource_link 降级为文本、顶层 _meta 不透传、binary embedded resource 被 skip）见 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 取消信号

handler 可以选择接收第二个参数 ToolInvocationContext，在 CLI 取消当前调用时通过 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"}]}

⚠️ 不要复用同一个 server 配置跨多次 query()：每次 query 会绑定独立的 transport。重复使用没有副作用，但你也不会得到「跨 query 共享状态」的能力——共享状态请放在 handler 闭包外的模块作用域里。

---

​

Stdio Server

通过子进程的 stdin/stdout 与 MCP 服务器通信。NPM 上 @modelcontextprotocol/server-* 系列都是 stdio 实现。

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

command 不可达或启动失败时不会拖垮整个 query —— 对应 server 的 status 会保持非 'connected'，其它 server 不受影响。

---

​

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

远端 URL 不可达时同样不会让 query 挂掉，server 状态非 'connected'，其它 server 不受影响。需要 OAuth 的远端服务请看 OAuth 认证。

---

​

工具命名与白名单

CLI 在向模型暴露 MCP 工具时统一加前缀：

mcp__<server_name>__<tool_name>

例如 server 名 my_tools、工具名 greet，模型看到的工具名是 mcp__my_tools__greet。Server 名允许含连字符等特殊字符（my-tools → mcp__my-tools__<tool>）。

​

tools：限制模型可见的工具集合

想让模型只看到部分工具，用 tools。CLI 会把所有未列出的内置工具加进 disallow 列表，等于”白名单”语义：

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

⚠️ 不传 tools 等于全部放开：所有内置工具 + 所有已连接 MCP server 的工具都会暴露给模型。生产环境建议显式列出，按需收口。

​

allowed_tools：预授权（不是可见性白名单）

allowed_tools 把列出的工具加入”自动放行”规则——调用时跳过权限弹窗，但不会把没列出来的工具藏起来。常用于让低风险的 MCP 工具免审批：

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

不传 allowed_tools 仅意味着没有预授权规则——模型仍能看到/调用所有工具，只是写操作会按 permission_mode 走审批流程。完整语义详见 Permissions 文档。

​

allowed_mcp_server_names：进程类 server 白名单

只过滤进程类（stdio/sse/http）服务器，不影响 in-process server。配合 strict_mcp_config=True 可以拒绝 CLI 加载本地额外配置：

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
)

⚠️ 不传 allowed_mcp_server_names 等于全部放开：所有声明的进程类 server 都会连接；想收口必须显式列出。in-process server 始终不受此字段影响。

---

​

运行时管理（QoderSDKClient）

query() 是一次性的迭代器，无法在中途变更 server 或鉴权。运行时管理 MCP 必须使用 QoderSDKClient，它把状态查询、OAuth、增删 server、reconnect / toggle 等都暴露为公开方法。

⚠️ 缓存原则：MCP server 配置 / 鉴权状态变更会重建 tools 列表，会话中途变更会破坏 prompt prefix 缓存。SDK 提供「查询状态 + 首条消息前完成鉴权」的方法；server 集合本身请通过 options.mcp_servers 在启动时一次性配置，必要时新建一个 QoderSDKClient。

​

查询状态

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

💡 MCP 握手发生在 CLI 完成 initialize 之后、第一次用户消息之前。QoderSDKClient.connect() 已经等到 initialize 返回；握手 IO 可能要几百毫秒，必要时自己轮询 get_mcp_status() 直到 connected。

​

订阅状态变化

两种方式任选其一：

方式 1（推荐）：在 options 上挂 on_mcp_status_change 回调，每次状态变化都会被调用一次。

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

方式 2：消费消息流，过滤 system/mcp_status_change。回调和消息流是同一份 payload，回调只是免去过滤的便利。

​

运行时增删 server / 重连 / 启停

方法	用途
client.set_mcp_servers(servers)	用全量 desired 映射替换当前 MCP 配置；返回 {added, removed, errors}
client.reconnect_mcp_server(name)	重连指定 server，typically 用于从 'failed' 状态恢复
client.toggle_mcp_server(name, enabled)	启用 / 禁用某个 server；禁用会断开连接并下线其工具

⚠️ 这三个方法都会触发 tools 列表重建，因此都会破坏 prompt prefix 缓存。生产环境优先在启动时配齐 mcp_servers，把这些 API 留给调试和本地开发场景。

​

控制请求超时

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

超时后 SDK 会自动写一条 control_cancel_request，并 reject 当前 Future。

---

​

OAuth 认证

远端 MCP 服务器（HTTP/SSE）经常需要 OAuth。CLI 内置完整的 OAuth 2.0 + PKCE + Dynamic Client Registration（RFC 7591）实现。Python SDK 暴露 inbound（CLI 主动让宿主完成 OAuth） 与 outbound（宿主主动触发 OAuth） 两条路径，可按宿主形态选择。

⚠️ 缓存原则：OAuth 完成后 CLI 会重连 server、重新发现 tools，会话中途完成鉴权必然破坏 prompt prefix 缓存。建议在首次用户消息发出之前完成鉴权，tools 列表稳定下来后再开聊。

💡 本节只覆盖 CLI 主导的 OAuth：CLI 自己做 metadata discovery、PKCE、token 交换、token 持久化。还有另一条服务器主导的鉴权链路——server 用 MCP elicitation/create 让 client 跳转去某个 URL 完成授权。两条链路独立，不会同时触发。详见 Elicitation：服务器请求用户输入。

​

Inbound：on_mcp_oauth_required 回调

CLI 在握手中检测到 server 需要 OAuth 时，会通过 control_request 把 McpOAuthRequest 推给 SDK，SDK 调用宿主的 on_mcp_oauth_required 回调。宿主返回以下任一种 resolution：

返回类型	含义
OAuthToken 或 {"token": OAuthToken}	宿主自己跑完整个 OAuth 流程，直接给 CLI 注入 token
{"callbackUrl": "..."}	宿主只拿到完整的回调 URL（含 code / state），CLI 解析并换 token
{"code": "...", "state": "..."}	宿主自己解出了 code，直接交还 CLI
None	拒绝，CLI 把该 server 标为 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：宿主主动驱动鉴权

宿主自己 UI 有「Sign in」入口时，可以主动调用：

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

方法	用途	调用时机
client.mcp_authenticate(name, redirect_uri=None)	拉起 OAuth；返回 {authUrl?, requiresUserAction}；静默续期时 requiresUserAction=False 无需 UI	首条 user message 前
client.mcp_submit_oauth_callback_url(name, callback_url)	提交完整回调 URL（含 code/state）	首条 user message 前
client.inject_mcp_token(name, token)	宿主自己跑完整个 OAuth，把 OAuthToken 直接塞回 CLI	首条 user message 前
client.mcp_clear_auth(name)	删除 CLI 存的 OAuth 凭据，相当于「sign out」	任意时刻；下次工具调用会触发重新鉴权

redirect_uri 可选，覆盖默认 OAuth 回调目标（Electron 自定义协议、企业内网回调地址等）。

CLI 默认把 token 存到系统 Keychain（macOS / Linux Secret Service），回退到 ~/.qoder/mcp-oauth-tokens.json（权限 0o600 + 跨进程锁）。

---

​

Elicitation：服务器请求用户输入

MCP elicitation/create 是 server → client 方向的请求，用来让 client 在用户面前展示一段交互（form 模式收结构化输入；url 模式让用户去某个 URL 完成操作）。

✅ Python SDK 现已对齐 TS SDK：QoderAgentOptions.on_elicitation 接收一个返回 ElicitationResult 的异步回调，签名与 TS 版本一致。未设置回调时，SDK 仍按默认契约自动答 {"action": "cancel"}。Elicitation / ElicitationResult 两类 hook 事件仍会并行触发，作为只读观察通道。

⚠️ 当前 CLI 不 advertise elicitation.url capability。server 端 elicit({mode: 'url'}) 会被 CLI 直接拒绝（MCP error -32602: Client does not support URL-mode elicitation requests），因此 URL 模式 elicit 不会到达 SDK，system/elicitation_complete 通知在当前 CLI 上也不会触发。等 CLI 开启 URL capability 后该路径会自动接通。

​

用 on_elicitation 应答 elicit

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

• 字段名遵循 TS SDK 的 camelCase（serverName / elicitationId / requestedSchema / displayName），CLI 的 snake_case payload 由 SDK 自动转换。
• 返回 None 等价于 {"action": "cancel"}，方便宿主在 fallback 路径里直接放弃。
• 也可以返回 mcp.types.ElicitResult Pydantic 模型（SDK 会 model_dump）。

​

观察 elicitation（hook 通道）

on_elicitation 落地后，Elicitation / ElicitationResult hook 仍然会并行触发——它们是只读观察通道，不承担决策。

Hook 事件	时机	payload TypedDict
Elicitation	server 请求到达时	ElicitationHookInput — mcp_server_name / message / mode / elicitation_id? / requested_schema? / url? / title?
ElicitationResult	SDK / host 响应完成后	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])],
    },
)

​

与 OAuth 链路的边界

• CLI 主导 OAuth（mcp_authenticate / inject_mcp_token / on_mcp_oauth_required）：token 落 qodercli Keychain；get_mcp_status() 在 needs-auth 时驱动；不触发 Elicitation hook。
• 服务器主导 elicit：token 在 server 内部；get_mcp_status() 不会标 needs-auth；由 on_elicitation 回调决策（未注册时 SDK 自动 cancel）。

两条链路不互斥但也不重叠：同一个 server 通常只走其中一条。

---

​

Options 速查

字段	类型	默认	说明
mcp_servers	dict[str, McpServerConfig] | str | Path	{}	server 名 → 配置；或 JSON 配置文件路径
allowed_mcp_server_names	list[str]	[]	进程类 server 白名单（不影响 in-process）；空列表等于全部放开
strict_mcp_config	bool	False	禁止 CLI 从用户配置文件再加载额外 MCP
tools	list[str] | ToolsPreset | None	None	模型可见工具白名单；不传等于全部内置 + MCP 工具都可见
allowed_tools	list[str]	[]	预授权列表（跳过权限弹窗，不控制可见性）；空列表等于无预授权规则
disallowed_tools	list[str]	[]	明确拒绝的工具，优先于 allow
control_request_timeout_ms	int	60_000	control 请求超时（含 mcp 系列），0 禁用
on_mcp_oauth_required	OnMcpOAuthRequired | None	None	CLI 检测到 server 需要 OAuth 时触发
on_mcp_status_change	OnMcpStatusChange | None	None	每次 server 状态变化时触发；等价于过滤 system/mcp_status_change 流
on_elicitation	OnElicitation | None	None	server 通过 MCP elicitation/create 请求用户输入时由宿主决策；未设置时 SDK 自动 cancel
hooks['Elicitation']	list[HookMatcher]	–	server 请求用户输入时的只读观察 hook（决策走 on_elicitation）

​

QoderSDKClient 上的方法

方法	说明	调用时机
get_mcp_status()	拿当前所有 MCP server 状态	任意时刻
set_mcp_servers(servers)	全量替换 MCP server 配置；返回 {added, removed, errors}	任意时刻（会破坏前缀缓存）
reconnect_mcp_server(name)	重连指定 server	任意时刻
toggle_mcp_server(name, enabled)	启用 / 禁用 server	任意时刻
mcp_authenticate(name, redirect_uri=None)	主动启动 OAuth	首条 user message 前
mcp_submit_oauth_callback_url(name, callback_url)	提交 OAuth 回调 URL	首条 user message 前
inject_mcp_token(name, token)	宿主自己跑完整 OAuth 后注入 token	首条 user message 前
mcp_clear_auth(name)	删除已存的 OAuth 凭据	任意时刻

---

​

类型参考

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 枚举（McpServerConnectionStatus）：

值	含义
'pending'	已注册，未开始连接
'connecting'	正在握手
'connected'	已连接，工具可调用
'failed'	连接失败（看 error 字段）
'needs-auth'	需要 OAuth，请走认证流程
'disabled'	被禁用（CLI 内部配置或 toggle_mcp_server 决定）

---

​

最佳实践

1. 描述写给 AI 看：@tool 的 description 决定 AI 何时选用它。说清楚「做什么、什么时候用、不该用于什么」。
2. 参数加 Annotated：在简单 dict / TypedDict 中给字段写 Annotated[type, "..."]，AI 用这些信息构造调用参数。
3. 失败用 is_error: True，不抛异常：让 AI 看见结果。完整对比见 工具使用指南 - SDK 如何处理 tool 返回的错误。
4. 优先只读 + readOnlyHint：写操作要谨慎，搭配 can_use_tool 或 hooks 二次确认。
5. server 名简短：会出现在工具前缀里，太长的名字浪费 token。
6. In-process 共享状态放模块作用域：handler 是闭包，但每次 query 仍会 reuse 同一个 server 实例。
7. OAuth 在首条 user message 前完成：用 mcp_authenticate + mcp_submit_oauth_callback_url，或 on_mcp_oauth_required inbound 回调，或 inject_mcp_token。会话中途完成鉴权必然破坏 prompt prefix 缓存。
8. MCP 状态走 get_mcp_status() 或 on_mcp_status_change：push 通道（status change message）保留，按需选一种即可。
9. 设置合理的 control_request_timeout_ms：远端 server 握手可能上秒，默认 60s 通常够；OAuth 等待用户操作时要调大；CI 环境记得显式给。
10. strict_mcp_config 用于隔离：避免用户本地的 ~/.qoder/settings.json / .mcp.json 里声明的 MCP server 干扰你的应用。

---

​

完整示例

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