AI智能体项目实战记录21MCP 使用教程之创建 MCP 客户端1. 引言1.1 MCP 是什么MCPModel Context Protocol是一套让 AI 应用连接外部工具和数据的标准协议。它不负责“决定调用哪个工具”只负责把工具以统一格式暴露出来。AI 应用Host └─ MCP Client ├─ stdio → 本地 MCP Server 子进程 └─ HTTP → 远程 MCP ServerMCP Server 主要提供三类能力类型作用Tools可被模型调用的函数例如读文件、查知识库Resources可读取的数据例如文档、表结构Prompts预置提示词模板本文覆盖两种客户端stdio启动本地 MCP Server 子进程Streamable HTTP连接远程 MCP Server。1.2 本文目标创建两种传输方式的 MCP 客户端通过 FastAPI 接口调试 MCP 工具为下一篇“动态注册到主管智能体”准备可用的客户端。整体链路stdioStreamable HTTP调用方 / 调试接口MCP Client本地 MCP Server 进程远程 MCP ServerMCP 工具如何注册进 LangGraph 主管智能体见第 22 篇。2. 环境与依赖2.1 依赖mcp1.4.1 httpx0.28.1mcp固定 1.4.1 的原因与项目现有pydantic2.7.4、pyjwt2.8.0兼容新版mcp会强制升级pydantic2.11影响现有 LangChain 依赖Streamable HTTP 在客户端内部用httpx实现不依赖新版 SDK。本地 stdio 示例还需要 Node.jsnode--versionnpx--versionWindows 下如果 PowerShell 禁止运行npx.ps1不影响本项目因为客户端直接用npx.cmd启动子进程。2.2 目录结构app/ ├── mcp/ │ ├── __init__.py │ ├── client.py # MCP 客户端stdio / streamable_http │ └── tool.py # MCP 工具包装第 22 篇使用 ├── controller/ │ └── mcp_controller.py # 调试接口 mcp_workspace/ # stdio 文件系统 Server 的沙箱目录2.3 配置项# MCP 开关 MCP_ENABLEDtrue # stdio / streamable_http MCP_TRANSPORTstdio # stdio 配置 MCP_SERVER_COMMANDnpx.cmd MCP_SERVER_ARGS[-y,modelcontextprotocol/server-filesystem,{workspace}] MCP_WORKSPACE_DIRmcp_workspace MCP_SERVER_ENV{} # Streamable HTTP 配置 MCP_HTTP_URLhttps://mcp.deepwiki.com/mcp MCP_HTTP_HEADERS{} MCP_HTTP_PROTOCOL_VERSION2025-03-26 # 超时秒 MCP_TIMEOUT_SECONDS60{workspace}会替换成mcp_workspace的绝对路径。3. 客户端一stdio3.1 工作方式MCPClient → 启动子进程npx.cmd -y modelcontextprotocol/server-filesystem workspace → 通过 stdin/stdout 发送 JSON-RPC → initialize → tools/list → tools/call3.2 核心代码asyncdef_with_session(self,action):paramsStdioServerParameters(commandself.command,argsself.args,envself.env,)asyncwithstdio_client(params)as(read,write):asyncwithClientSession(read,write)assession:awaitsession.initialize()returnawaitasyncio.wait_for(action(session),timeoutself.timeout)asyncdef_list_tools(self)-List[Dict[str,Any]]:asyncdefaction(session):returnawaitsession.list_tools()resultawaitself._with_session(action)return[self._tool_to_dict(tool)fortoolinresult.tools]asyncdef_call_tool(self,name:str,arguments:Dict[str,Any])-Dict[str,Any]:asyncdefaction(session):returnawaitsession.call_tool(name,arguments)resultawaitself._with_session(action)returnself._result_to_dict(result)3.3 Windows 事件循环冲突项目主循环使用的是SelectorEventLooppsycopg3 异步要求但 Windows 下启动子进程只支持ProactorEventLoop。所以 stdio 调用不能直接在主循环里asyncio.run而是放到独立线程的 Proactor 循环def_run_in_proactor_thread(coro):result{}defrunner():loopasyncio.ProactorEventLoop()asyncio.set_event_loop(loop)try:result[value]loop.run_until_complete(coro)exceptBaseExceptionase:result[error]efinally:loop.close()threadthreading.Thread(targetrunner,daemonTrue)thread.start()thread.join()iferrorinresult:raiseunwrap_exception(result[error])returnresult.get(value)def_run_sync(self,coro):ifsys.platformwin32:return_run_in_proactor_thread(coro)returnasyncio.run(coro)3.4 实测transportstdio tools/list 返回 14 个工具 read_file、read_text_file、list_directory、search_files ...4. 客户端二Streamable HTTP4.1 协议流程POST initialize ← 返回 protocolVersion可能带 Mcp-Session-Id POST notifications/initialized POST tools/list POST tools/call响应可能是application/jsontext/event-streamSSE按data:解析4.2 为什么不直接用新版 SDKmcp 1.4.1没有mcp.client.streamable_http升级到新版又会强制升级pydantic影响现有项目。所以在客户端内部用httpx实现核心 JSON-RPC 流程。4.3 核心代码asyncdef_http_initialize(self,client:httpx.AsyncClient)-None:payload{jsonrpc:2.0,id:1,method:initialize,params:{protocolVersion:self.protocol_version,capabilities:{},clientInfo:{name:chat-py-mcp,version:1.0.0},},}responseawaitself._http_post(client,payload,include_protocolFalse)self.protocol_version(response.get(result)or{}).get(protocolVersion)orself.protocol_versionawaitself._http_post(client,{jsonrpc:2.0,method:notifications/initialized},expect_responseFalse,)asyncdef_http_post(self,client,payload,include_protocolTrue,expect_responseTrue):headers{Content-Type:application/json,Accept:application/json, text/event-stream,**self.http_headers,}ifinclude_protocolandself.protocol_version:headers[MCP-Protocol-Version]self.protocol_versionifself._http_session_id:headers[Mcp-Session-Id]self._http_session_id respawaitclient.post(self.http_url,headersheaders,jsonpayload)ifresp.status_code400:raiseRuntimeError(fMCP HTTP{resp.status_code}:{resp.text[:300]})session_idresp.headers.get(mcp-session-id)ifsession_id:self._http_session_idsession_idifnotexpect_responseornotresp.text.strip():return{}iftext/event-streaminresp.headers.get(content-type,):events_parse_sse(resp.text)foreventinevents:ifevent.get(id)payload.get(id):returneventreturnevents[-1]ifeventselse{}returnresp.json()4.4 实测使用公开的 DeepWiki MCP ServerMCP_HTTP_URLhttps://mcp.deepwiki.com/mcptools/list 返回 3 个工具 ask_wiki_question read_wiki_contents read_wiki_structure调用{name:read_wiki_structure,arguments:{repoName:langchain-ai/langgraph}}成功返回 LangGraph 的文档目录。需要鉴权的远程 Server在MCP_HTTP_HEADERS里加MCP_HTTP_HEADERS{Authorization:Bearer token}5. Controller 接口5.1 接口列表方法路径说明GET/chat_py_server/mcp/v1/tools列出 MCP 工具POST/chat_py_server/mcp/v1/call调用 MCP 工具两个接口都要求登录未登录返回RestResp(status-1)。5.2 请求与响应调用工具{name:read_wiki_structure,arguments:{repoName:langchain-ai/langgraph}}响应{status:1,data:{isError:false,content:[{type:text,text:Available pages for ...}],structuredContent:null},message:调用成功}5.3 curl 示例PowerShell 一行curl.exe-X POSThttp://127.0.0.1:5020/chat_py_server/mcp/v1/call-HAuthorization: Bearer token-HContent-Type: application/json-d{name:read_wiki_structure,arguments:{repoName:langchain-ai/langgraph}}CMD 一行curl.exe -X POST http://127.0.0.1:5020/chat_py_server/mcp/v1/call -H Authorization: Bearer token -H Content-Type: application/json -d {\name\:\read_wiki_structure\,\arguments\:{\repoName\:\langchain-ai/langgraph\}}5.4 错误处理anyio/httpx 有时会抛出ExceptionGroup直接打印只能看到unhandled errors in a TaskGroup (1 sub-exception)客户端和控制器都做了异常展开defunwrap_exception(exc:BaseException)-BaseException:currentexcwhilegetattr(current,exceptions,None):currentcurrent.exceptions[0]returncurrent这样能看到真正的ConnectTimeout、ReadTimeout、HTTP 状态码等。6. 配置汇总6.1 stdioMCP_ENABLEDtrue MCP_TRANSPORTstdio MCP_SERVER_COMMANDnpx.cmd MCP_SERVER_ARGS[-y,modelcontextprotocol/server-filesystem,{workspace}] MCP_WORKSPACE_DIRmcp_workspace MCP_SERVER_ENV{} MCP_TIMEOUT_SECONDS60Linux/macOS 下MCP_SERVER_COMMAND可以写npx代码会自动把npx.cmd转成npx。6.2 Streamable HTTPMCP_ENABLEDtrue MCP_TRANSPORTstreamable_http MCP_HTTP_URLhttps://mcp.deepwiki.com/mcp MCP_HTTP_HEADERS{} MCP_HTTP_PROTOCOL_VERSION2025-03-26 MCP_TIMEOUT_SECONDS606.3 其它 stdio Server# memory server MCP_SERVER_COMMANDnpx.cmd MCP_SERVER_ARGS[-y,modelcontextprotocol/server-memory]7. 常见问题Q1报unhandled errors in a TaskGroup (1 sub-exception)怎么办现在客户端会展开异常日志里能看到真正原因。常见是ConnectTimeout/ReadTimeout网络或 Server 太慢调大MCP_TIMEOUT_SECONDSHTTP 401/403远程 Server 需要鉴权配置MCP_HTTP_HEADERSHTTP 404MCP_HTTP_URL写错。Q2PowerShell 提示npx.ps1 cannot be loaded客户端通过subprocess直接执行npx.cmd不经过 PowerShell所以不受执行策略影响。手工测试时也要用npx.cmd。Q3为什么不用最新版mcpSDK新版会强制升级pydantic2.11与项目当前的pydantic2.7.4冲突。Streamable HTTP 已经用httpx实现核心功能不受影响。Q4stdio 第一次调用很慢npx -y首次会下载 npm 包之后走缓存。生产环境建议提前安装包或改为常驻 Server / HTTP 传输。Q5远程 Server 需要登录怎么办MCP_HTTP_HEADERS{Authorization:Bearer token}不要在工具参数里传 token。8. 小结MCP 是工具接入标准不替代 LangGraph 编排stdio 适合本地工具Streamable HTTP 适合远程服务本项目用MCPClient统一封装两种传输外面只看到list_tools / call_toolController 提供调试接口方便先验证工具发现和调用下一篇会把 MCP 工具动态注册到主管智能体并给出读取工作目录文档的完整示例生产环境要限制 MCP Server 白名单、工作目录、超时和鉴权 Header避免把任意本地命令暴露给模型。