> ## Documentation Index > Fetch the complete documentation index at: https://docs.qoder.com/llms.txt > Use this file to discover all available pages before exploring further. # MCP 集成 MCP（Model Context Protocol）是 AI Agent 调用外部工具的开放协议。Python SDK 内置了 MCP 客户端能力，宿主应用只需要描述「有哪些 MCP 服务器」，SDK 会自动完成连接、工具发现、消息路由、OAuth、状态同步等工作。
## 一图看懂 ``` ┌────────────────────────────────────────────────────────────┐ │ 你的 Python 应用 (SDK Host) │ │ │ │ ┌──────────────────────────┐ │ │ │ create_sdk_mcp_server(...) │ ← In-Process 工具 │ │ │ + @tool(...) │ 直接定义、零进程 │ │ └──────────────────────────┘ │ │ │ │ │ ▼ │ │ ┌──────────────────────────┐ │ │ │ query({mcp_servers}) │── stdio ─▶ qodercli 子进程 │ │ │ / 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 ` 透传给 CLI。
## In-Process Server（推荐） In-process 工具是最直接的扩展方式：定义一个普通的 `async` 函数，用装饰器声明 schema，就能被 Agent 调用。详细的 `@tool()` / schema / handler 行为见 [tools.md](/zh/cli/sdk/python/tools)，本节只覆盖与 MCP server 装配相关的部分。
### 30 秒上手 ```python theme={null} import asyncio from typing import Annotated from qoder_agent_sdk import ( QoderAgentOptions, create_sdk_mcp_server, query, tool, ) @tool("greet", "向某人打招呼。", {"name": Annotated[str, "对方的姓名"]}) async def greet(args): return {"content": [{"type": "text", "text": f"你好，{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="用 greet 工具向 Alice 打招呼", options=options): print(msg) asyncio.run(main()) ```
### `@tool()` / `create_sdk_mcp_server()` 完整签名 ```python theme={null} 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____` | | `description` | 给模型看的说明，决定 AI 何时调用——**写清楚 What/When** | | `input_schema` | 简单 dict / `TypedDict` / 完整 JSON Schema dict 三种写法，详见 [Tools Reference - `input_schema`](/zh/cli/sdk/python/references#input_schema) | | `annotations` | MCP 工具注解，详见下表 | | `name`（server） | server 名（决定工具前缀 `mcp____`） | | `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 返回值 ```python theme={null} # 成功返回 {"content": [{"type": "text", "text": "结果"}]} # 业务失败：使用 is_error 而不是抛异常 {"content": [{"type": "text", "text": "错误描述"}], "is_error": True} ``` **业务失败请用 `is_error: True`** 而不是抛异常。完整的 content 类型说明、Python 端与 TS 端的几处行为差异（`resource_link` 降级为文本、顶层 `_meta` 不透传、binary embedded resource 被 skip）见 [Tools Reference - `CallToolResult`](/zh/cli/sdk/python/references#calltoolresult)。 ```python theme={null} @tool( "query_db", "只读 SQL 查询。", {"sql": Annotated[str, "SQL 查询语句"]}, 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": "只允许 SELECT 语句"}], } rows = await db.query(sql) return {"content": [{"type": "text", "text": json.dumps(rows)}]} ```
### Handler 取消信号 handler 可以选择接收第二个参数 `ToolInvocationContext`，在 CLI 取消当前调用时通过 `extra.signal` 协作退出： ```python theme={null} @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 实现。 ```python theme={null} class McpStdioServerConfig(TypedDict): type: NotRequired[Literal["stdio"]] # 可省略，stdio 是默认 command: str # 可执行命令 args: NotRequired[list[str]] # 命令参数 env: NotRequired[dict[str, str]] # 环境变量 tools: NotRequired[list[McpServerToolPolicy]] ``` ```python theme={null} 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 ```python theme={null} 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]] ``` ```python theme={null} 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 认证](#oauth-认证)。
## 工具命名与白名单 CLI 在向模型暴露 MCP 工具时统一加前缀： ``` mcp____ ``` 例如 server 名 `my_tools`、工具名 `greet`，模型看到的工具名是 `mcp__my_tools__greet`。Server 名允许含连字符等特殊字符（`my-tools` → `mcp__my-tools__`）。
### `tools`：限制模型可见的工具集合 想让模型**只看到部分工具**，用 `tools`。CLI 会把所有未列出的内置工具加进 disallow 列表，等于"白名单"语义： ```python theme={null} 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 工具免审批： ```python theme={null} 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 文档](/zh/cli/sdk/python/permissions#控制工具范围toolsallowed_toolsdisallowed_tools)。
### `allowed_mcp_server_names`：进程类 server 白名单 只过滤**进程类**（stdio/sse/http）服务器，**不影响 in-process server**。配合 `strict_mcp_config=True` 可以拒绝 CLI 加载本地额外配置： ```python theme={null} options = QoderAgentOptions( mcp_servers={ "keep": {"command": "..."}, "drop": {"command": "..."}, }, allowed_mcp_server_names=["keep"], # 'drop' 仍会出现在状态里，但不连接 strict_mcp_config=True, # 不从 settings.json / .mcp.json 加载 MCP 服务器 ) ``` > ⚠️ **不传 `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`。
### 查询状态 ```python theme={null} async with QoderSDKClient(options) as client: status = await client.get_mcp_status() # 返回 McpStatusResponse：{"mcpServers": [McpServerStatus, ...]} # 每个 McpServerStatus 包含： # 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` 回调，每次状态变化都会被调用一次。 ```python theme={null} 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 留给调试和本地开发场景。
### 控制请求超时 ```python theme={null} options = QoderAgentOptions( control_request_timeout_ms=20_000, # 默认 60_000，传 0 禁用 ) ``` 超时后 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：服务器请求用户输入](#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 | ```python theme={null} async def handle_oauth(request: McpOAuthRequest) -> McpOAuthResolution | None: # 在 Electron BrowserWindow / 系统浏览器里开 request['auth_url'] 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, # 用户授权可能要点时间 ) ```
### Outbound：宿主主动驱动鉴权 宿主自己 UI 有「Sign in」入口时，可以主动调用： ```python theme={null} 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) # 静默路径（cached client + 有效 refresh token）：requiresUserAction=False # 无需弹任何 UI，server 直接进入 connected # tools 列表已稳定，可以放心发用户消息 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 ```python theme={null} from qoder_agent_sdk import ElicitationRequest, ElicitationResult, QoderAgentOptions async def on_elicitation(req: ElicitationRequest) -> ElicitationResult: # form 模式：req["requestedSchema"] 是 JSON Schema；返回 content 必须匹配。 if req.get("mode") == "form": return {"action": "accept", "content": {"token": "xxx"}} # 任何不便处理的情况返回 decline / cancel，CLI 会把结果回送给 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?` | ```python theme={null} 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 凭据 | 任意时刻 |
## 类型参考 ```python theme={null} from qoder_agent_sdk import ( # 工厂 create_sdk_mcp_server, tool, SdkMcpTool, # 配置 McpServerConfig, McpStdioServerConfig, McpSSEServerConfig, McpHttpServerConfig, McpSdkServerConfig, McpServerToolPolicy, # 状态 McpServerStatus, McpServerStatusConfig, McpServerConnectionStatus, McpServerInfo, McpToolInfo, McpToolAnnotations, McpStatusResponse, McpStatusChangeMessage, # 运行时变更 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 返回的错误](/zh/cli/sdk/python/tools#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 干扰你的应用。
## 完整示例 ```python theme={null} 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. 定义业务工具 @tool( "get_user_orders", "查询用户的订单，可按状态过滤。", { "user_id": Annotated[str, "用户 UUID"], "status": Annotated[str, "按订单状态过滤：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"查询失败: {e}"}], } # 2. 组装 server crm = create_sdk_mcp_server(name="crm", tools=[get_user_orders]) # 3. 准备 inbound OAuth 回调（远端 server 需要鉴权时由 CLI 调） 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. 启动 client，先完成鉴权再发首条消息 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 兜底：CLI 没主动 inbound 时，也能拉一次状态自己驱动 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. 消费消息（此时 tools 列表已稳定，前缀缓存可正确建立） await client.query("查 user-123 最近的已支付订单") 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("完成，cost=", msg.total_cost_usd) else: print("失败:", msg.subtype) asyncio.run(main()) ```