接 MCP：微软 AI Agent 入门课第 11 课的 client 与 server 最小实现

2026-08-18

很多人第一次接 MCP 的卡点不是概念，是不知道两端各自最少要写哪几行。协议文档讲的是消息格式，读完仍然不知道「我这个跑十分钟的任务怎么变成一个能被发现的 tool」。ai-agents-for-beginners 第 11 课的 11-agentic-protocols/code_samples/mcp-agents/ 里有一份两端俱全的可读示例，server/server.py 与 client/client.py 各是一端的入口文件，这篇就沿着这两个文件走一遍。

需要先说清楚：这个目录里的两个 agent 是模拟的，travel_agent 不会真订机票，research_agent 也不查资料，它们只是把「长任务 + 中途要人确认 + 中途要模型帮忙」这三种形态用 anyio.sleep 撑起来，好让你看清协议这一侧的代码长什么样。

前置条件

这一段别跳过，示例的目录结构对启动方式有硬要求。

依赖方面，仓库根的 requirements.txt 里列了 mcp[cli]，第 11 课这两个文件用到的 mcp.ClientSession、mcp.server.Server、mcp.server.streamable_http 都来自它。server/server.py 另外 import 了 uvicorn、starlette、anyio、pydantic，client/client.py 用 rich 做终端输出。00-course-setup/README.md 的 Requirements 一节写明 Python 3.12+，并建议先建虚拟环境：

python -m venv venv

激活方式仓库里分了两种写法，Windows 侧是 Command Prompt 的：

venv\Scripts\activate

zsh/bash 侧是：

source venv/bin/activate

然后在仓库根执行 pip install -r requirements.txt。

还有一个目录位置的要求，是从代码里读出来的：server/server.py 里写的是 from .event_store import SimpleEventStore，client/client.py 里写的是 from .utils import TokenManager, cast_input_value，两处都是包内相对 import。所以启动时的工作目录必须是 mcp-agents/，让 server 和 client 分别作为顶层包被识别，这也正是 mcp-agents/README.md 里给的启动命令的形态：

# Start the server with event store for resumption
python -m server.server --port 8006

# In another terminal, run the interactive client
python -m client.client --url http://127.0.0.1:8006/mcp

Windows 下同样需要两个窗口，两个窗口都要各自激活虚拟环境；换成 PowerShell 时激活脚本与上面的 Command Prompt 写法不同，仓库里只给了 Command Prompt 那一种，其余请以你本机 Python 的安装说明为准。

server 端：工具是怎么被暴露出去的

server/server.py 定义了 class ResumableServer(Server)，构造函数的默认 name 是 "resumable_mcp_server"（这是仓库当前代码里的默认值，随版本可能变动）。注册工作全部发生在 __init__ 内部，用的是两个装饰器：

@self.list_tools()
async def handle_list_tools() -> list[Tool]:

@self.call_tool()
async def handle_call_tool(name: str, args: dict) -> list[TextContent]:

这就是 MCP server 侧最小的两件事：一个函数负责回答「我有什么」，另一个函数负责回答「你调的这个怎么执行」。

handle_list_tools 返回的是 Tool 对象列表，每个 Tool 三个字段：name、description、inputSchema。inputSchema 是原样的 JSON Schema，例如 travel_agent 的写法是 {"type": "object", "properties": {"destination": {"type": "string", "description": "Travel destination", "default": "Paris"}}}——"Paris" 是仓库示例里的取值，不是什么约定。long_running_agent 的 inputSchema 则是 {"type": "object", "properties": {}}，即不收参数。

这里有个值得留意的细节：三个 Tool 的 inputSchema 都没有写 required 键。这会直接影响 client 端的参数收集逻辑，下一节会接上。

handle_call_tool 的骨架是按 name 分支的 if/elif，末尾兜底：

else:
    raise ValueError(f"Unknown tool: {name}")

分支内部拿到的 args 是普通 dict，取值方式是 args.get("destination", "Paris")。也就是说同一个默认值在 schema 和函数体里各写了一遍，改的时候两处都得动，这是读代码时容易踩空的一处。

传输层在 create_server_app() 里组装：StreamableHTTPSessionManager(app=server, event_store=event_store, json_response=False, security_settings=security_settings)，再由 Starlette 用 Mount("/mcp", app=session_manager.handle_request) 挂到 /mcp 路径上，lifespan 传的是 lambda app: session_manager.run()。json_response=False 这一行后面跟的注释是 # Use SSE streams，也就是走流式而不是一次性 JSON 响应。

client 端：发现与一次往返

client/client.py 的连接入口是 streamablehttp_client(server_url, headers=..., terminate_on_close=False)，它解包出三样东西：

async with streamablehttp_client(
    server_url,
    headers=headers if headers else None,
    terminate_on_close=False  # Enable resumption
) as (read_stream, write_stream, get_session_id):

前两个流交给 ClientSession，第三个 get_session_id 是个可调用对象，后面拿 session id 用。ClientSession 的构造在这份示例里挂了三个回调：message_handler、sampling_callback、elicitation_callback。这三个回调是服务端反向找客户端要东西时的接线口——message_handler 里按 types.LoggingMessageNotification、types.ProgressNotification、types.ResourceUpdatedNotification 分类型处理；elicitation_callback 返回 types.ElicitResult(action="accept" | "decline", content={...})；sampling_callback 返回 types.CreateMessageResult。

握手与发现是两句：

result = await session.initialize()
tools_result = await session.list_tools()
tools = tools_result.tools
tools_dict = {tool.name: tool for tool in tools}

initialize() 的返回值里，示例用到了 result.serverInfo.name 和 result.protocolVersion。list_tools() 拿到的每个 tool 就是服务端那边 Tool(...) 的对侧，tool.inputSchema 原封不动传了过来——这就是「工具发现」的全部，client 并不需要事先知道服务端有什么。

交互循环里，用户敲了一个 tool 名之后，参数是照着 schema 现问的：

if tool.inputSchema and tool.inputSchema.get("properties"):
    for prop_name, prop_info in tool.inputSchema["properties"].items():
        required = prop_name in tool.inputSchema.get("required", [])
        default = prop_info.get("default", "")
        desc = prop_info.get("description", "")

输入拿到后过一道 cast_input_value(value, prop_info)，这个函数在 client/utils.py 里，按 prop_info["type"] 分支：integer 转 int、number 转 float、boolean 判断字符串是否在一组真值里，其余原样返回；ValueError 时回退成原字符串。终端拿到的永远是 str，schema 里的类型信息就是靠这个函数落地的。

把上一节的坑接上：因为服务端三个 Tool 都没写 required，required 这个变量在示例里恒为 False，那句「缺参数就 break」的分支实际不会触发。这段用的还是 Python 的 for ... else 结构——只有循环没被 break 时才执行 else 里的调用。

真正发请求的地方在 execute_tool_with_resumption()，它没有用 session.call_tool()，而是走了更底层的 send_request：

result = await session.send_request(
    types.ClientRequest(
        types.CallToolRequest(
            method="tools/call",
            params=types.CallToolRequestParams(
                name=command,
                arguments=args
            ),
        )
    ),
    types.CallToolResult,
    metadata=metadata,
)

绕这一圈是为了带上 metadata，即 ClientMessageMetadata，它承载 resumption_token 或 on_resumption_token_update 回调。返回的 result 再交给 extract_text_content()，那个函数遍历 result.content，取第一个带 text 属性的元素，都没有就返回 "No text content available"。

断线之后的那一次往返走的是另一条路径，值得单独看。client 一启动就先 token_manager.load_tokens()，如果读到了东西，会在建连之前把两个头塞进 headers：

headers[MCP_SESSION_ID_HEADER] = existing_tokens["session_id"]
if "protocol_version" in existing_tokens:
    headers[MCP_PROTOCOL_VERSION_HEADER] = existing_tokens["protocol_version"]

这两个常量是从 mcp.server.streamable_http 里导入的——客户端文件引用服务端模块里的头名，读的时候别以为找错了地方。接着在 ClientSession 内部，示例判断 resumption_token、last_tool、last_args 三样是否齐全，齐了就跳过 session.initialize()，直接拿缓存下来的工具名和参数重发一次 tools/call，控制台对应那句是 ✅ Skipping initialization, directly resuming...。恢复失败时才回落到正常的 initialize() 流程。

token 是怎么存下来的也在这条链上：新会话时 metadata 挂的是 on_resumption_token_update 回调，服务端一给出 token 就顺手把 session id、protocol version、工具名与参数一起写盘。client/utils.py 的 save_tokens() 在写之前有两道拦截——session_id 或 resumption_token 为空直接返回 False，token 以 pending_ 或 temp_ 开头也拒绝，注释自述是不想留下占位性质的 token 文件。任务正常结束后示例会调 token_manager.delete_tokens() 清掉。

回程方向上，服务端主动推的消息用的是 ctx.session.send_progress_notification(progress_token=..., progress=..., total=100, message=..., related_request_id=...) 与 ctx.session.send_log_message(level="info", data=..., logger="long_running_agent", related_request_id=...)，落到客户端就是 message_handler 里的那两个 isinstance 分支。要人确认走 ctx.session.elicit(...)，其中 requestedSchema 传的是 PriceConfirmationSchema.model_json_schema()——一个带 confirm: bool 和 notes: str 两个字段的 pydantic BaseModel，schema 是现生成的。要模型帮忙走 ctx.session.create_message(messages=[SamplingMessage(...)], max_tokens=..., related_request_id=...)。这几处调用里的具体取值都是仓库示例里写死的，只说明参数位置，不是什么推荐配置。

边界：这份示例明确不做的事

client 的 sampling 是假的。sampling_callback 里返回的是一段硬编码字符串，mcp-agents/README.md 对应处写的是 demo 用途的 mock response。想接真模型得自己在这个回调里发请求，示例里没有这部分。
只允许本机连。TransportSecuritySettings(allowed_hosts=["127.0.0.1:*", "localhost:*"], allowed_origins=["http://127.0.0.1:*", "http://localhost:*"]) 写死在 create_server_app() 里，换成局域网地址访问需要改这处配置。
默认的事件存储不落盘。server/event_store.py 里 SimpleEventStore 的 docstring 自述是 “Simple in-memory event store for testing resumption functionality”；同文件另有一个基于 sqlite3 的 PersistentEventStore，但 run_server() 里实例化的是前者。
代码与注释对不上一处。handle_call_tool 的循环里写的是 await anyio.sleep(2)，紧跟的注释却是 # Fixed 0.5 second delay between steps，两处不一致，以代码为准。
server/__init__.py 的自述与文件名对不上。它的 docstring 提到 resumable_server.py，而目录里实际只有 server.py 和 event_store.py。
异常里有一条乐观分支。工具调用抛错时，示例会判断错误信息里是否同时含有 "ValidationError" 和 "JSON"，命中就打印「连接有问题但很可能已经成功」并照样清掉 token。这一段是示例代码里的处理方式，不是判断任务成败的可靠依据，自己写 client 时别照抄。
关于 Windows 上 PowerShell 的激活脚本、关于把这套东西部署到公网、关于鉴权，仓库这一课里没有找到相关说明。

怎么确认两端配对成功

按顺序看四处输出就够了：

server 侧 run_server() 会打 f"Starting Resumable HTTP MCP Server on http://127.0.0.1:{port}/mcp"，端口对不对在这里看。
紧接着有一个二选一的日志：带事件存储时是 "Event store enabled - resumption supported"，加了 --no-event-store 则是 "Event store disabled - no resumption support"。这一行决定后面断线能不能续。
client 侧握手成功会打 ✅ Connected to: 加上 result.serverInfo.name，这个名字对应的就是 ResumableServer.__init__ 里那个默认 name（仓库当前代码里是 resumable_mcp_server，随版本可能变动）；下一行是 📋 Protocol: 加 result.protocolVersion。
然后 display_tools(tools) 会把服务端返回的工具名逐个列出来。此时在提示符下敲 list 可以随时重列，敲 help 会额外把名字里含 agent 的工具单独列一遍。

如果第 3 步的 name 出不来而是走到 ❌ Connection error:，问题在传输层而不在工具定义。如果 name 出来了但工具列表是空的，那就该回去看 handle_list_tools 的返回值。

客户端还带一个独立开关：python -m client.client --clean-tokens 只做一件事——删掉 client/utils.py 里 DEFAULT_TOKEN_FILE = "resumption_tokens.json" 指向的那个文件然后退出，不会连服务器。上一次的会话中断过、这次一启动就提示在尝试恢复的时候，用它清干净重来。

以上代码片段均摘自仓库对应文件；组合展示的部分为按仓库代码中的接口语义组合的示例，未经实测，以仓库最新代码为准。这一课的示例与 MCP Python SDK 的接口都在演进，tools/call 之外的原语（resources、prompts）在这份示例里没有出现，需要时请以仓库最新内容为准。

本文依据 github.com/microsoft/ai-agents-for-beginners 仓库于 2026-08-18 的公开内容整理，事实来自仓库内的课程正文与代码示例。我们没有跑过文中涉及的代码，因此不涉及运行结果、耗时与 Agent 实际表现的任何描述。该课程持续更新，文中涉及的文件路径、依赖与接口写法随版本变动，请以仓库最新内容为准。文中涉及的云端服务调用会产生费用并可能上传数据，请自行评估密钥与数据边界。

安全与合规相关做法请结合自身环境评估，本文不构成安全方案建议。