跳转到主要内容
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 <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", "向某人打招呼。", {"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() 完整签名

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
annotationsMCP 工具注解，详见下表
name（server）server 名（决定工具前缀 mcp__<name>__）
version默认 '1.0.0'
toolsSdkMcpTool 列表
返回值 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 返回值

# 成功返回
{"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。 
@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 协作退出： 
@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"]]    # 可省略，stdio 是默认
    command: str                           # 可执行命令
    args: NotRequired[list[str]]           # 命令参数
    env: NotRequired[dict[str, str]]       # 环境变量
    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' 仍会出现在状态里，但不连接
    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。

​
查询状态

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 回调，每次状态变化都会被调用一次。 
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,   # 默认 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：服务器请求用户输入。

​
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:
    # 在 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」入口时，可以主动调用： 
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

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
Elicitationserver 请求到达时ElicitationHookInput — mcp_server_name / message / mode / elicitation_id? / requested_schema? / url? / title?
ElicitationResultSDK / 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_serversdict[str, McpServerConfig] | str | Path{}server 名 → 配置；或 JSON 配置文件路径
allowed_mcp_server_nameslist[str][]进程类 server 白名单（不影响 in-process）；空列表等于全部放开
strict_mcp_configboolFalse禁止 CLI 从用户配置文件再加载额外 MCP
toolslist[str] | ToolsPreset | NoneNone模型可见工具白名单；不传等于全部内置 + MCP 工具都可见
allowed_toolslist[str][]预授权列表（跳过权限弹窗，不控制可见性）；空列表等于无预授权规则
disallowed_toolslist[str][]明确拒绝的工具，优先于 allow
control_request_timeout_msint60_000control 请求超时（含 mcp 系列），0 禁用
on_mcp_oauth_requiredOnMcpOAuthRequired | NoneNoneCLI 检测到 server 需要 OAuth 时触发
on_mcp_status_changeOnMcpStatusChange | NoneNone每次 server 状态变化时触发；等价于过滤 system/mcp_status_change 流
on_elicitationOnElicitation | NoneNoneserver 通过 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 (
    # 工厂
    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 返回的错误。
  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. 定义业务工具
@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())