企业官网建设流程全解析

AI智能体项目实战记录（21）：MCP 使用教程之创建 MCP 客户端

1. 引言

1.1 MCP 是什么

MCP（Model Context Protocol）是一套让 AI 应用连接外部工具和数据的标准协议。它不负责“决定调用哪个工具”，只负责把工具以统一格式暴露出来。

AI 应用（Host） └─ MCP Client ├─ stdio → 本地 MCP Server 子进程 └─ HTTP → 远程 MCP Server

MCP Server 主要提供三类能力：

本文覆盖两种客户端：

1. stdio：启动本地 MCP Server 子进程；
2. Streamable HTTP：连接远程 MCP Server。

1.2 本文目标

1. 创建两种传输方式的 MCP 客户端；
2. 通过 FastAPI 接口调试 MCP 工具；
3. 为下一篇“动态注册到主管智能体”准备可用的客户端。

整体链路：

MCP 工具如何注册进 LangGraph 主管智能体，见第 22 篇。

2. 环境与依赖

2.1 依赖

mcp==1.4.1 httpx==0.28.1

mcp固定 1.4.1 的原因：

• 与项目现有pydantic==2.7.4、pyjwt==2.8.0兼容；
• 新版mcp会强制升级pydantic>=2.11，影响现有 LangChain 依赖；
• Streamable HTTP 在客户端内部用httpx实现，不依赖新版 SDK。

本地 stdio 示例还需要 Node.js：

node--versionnpx--version

Windows 下如果 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_ENABLED=true # stdio / streamable_http MCP_TRANSPORT="stdio" # stdio 配置 MCP_SERVER_COMMAND="npx.cmd" MCP_SERVER_ARGS='["-y","@modelcontextprotocol/server-filesystem","{workspace}"]' MCP_WORKSPACE_DIR="mcp_workspace" MCP_SERVER_ENV='{}' # Streamable HTTP 配置 MCP_HTTP_URL="https://mcp.deepwiki.com/mcp" MCP_HTTP_HEADERS='{}' MCP_HTTP_PROTOCOL_VERSION="2025-03-26" # 超时（秒） MCP_TIMEOUT_SECONDS=60

{workspace}会替换成mcp_workspace的绝对路径。

3. 客户端一：stdio

3.1 工作方式

MCPClient → 启动子进程：npx.cmd -y @modelcontextprotocol/server-filesystem <workspace> → 通过 stdin/stdout 发送 JSON-RPC → initialize → tools/list → tools/call

3.2 核心代码

asyncdef_with_session(self,action):params=StdioServerParameters(command=self.command,args=self.args,env=self.env,)asyncwithstdio_client(params)as(read,write):asyncwithClientSession(read,write)assession:awaitsession.initialize()returnawaitasyncio.wait_for(action(session),timeout=self.timeout)asyncdef_list_tools(self)->List[Dict[str,Any]]:asyncdefaction(session):returnawaitsession.list_tools()result=awaitself._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)result=awaitself._with_session(action)returnself._result_to_dict(result)

3.3 Windows 事件循环冲突

项目主循环使用的是SelectorEventLoop（psycopg3 异步要求），但 Windows 下启动子进程只支持ProactorEventLoop。

所以 stdio 调用不能直接在主循环里asyncio.run，而是放到独立线程的 Proactor 循环：

def_run_in_proactor_thread(coro):result={}defrunner():loop=asyncio.ProactorEventLoop()asyncio.set_event_loop(loop)try:result["value"]=loop.run_until_complete(coro)exceptBaseExceptionase:result["error"]=efinally:loop.close()thread=threading.Thread(target=runner,daemon=True)thread.start()thread.join()if"error"inresult:raiseunwrap_exception(result["error"])returnresult.get("value")def_run_sync(self,coro):ifsys.platform=="win32":return_run_in_proactor_thread(coro)returnasyncio.run(coro)

3.4 实测

transport=stdio tools/list 返回 14 个工具： read_file、read_text_file、list_directory、search_files ...

4. 客户端二：Streamable HTTP

4.1 协议流程

POST initialize ← 返回 protocolVersion，可能带 Mcp-Session-Id POST notifications/initialized POST tools/list POST tools/call

响应可能是：

• application/json
• text/event-stream（SSE，按data:解析）

4.2 为什么不直接用新版 SDK

mcp 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"},},}response=awaitself._http_post(client,payload,include_protocol=False)self.protocol_version=(response.get("result")or{}).get("protocolVersion")orself.protocol_versionawaitself._http_post(client,{"jsonrpc":"2.0","method":"notifications/initialized"},expect_response=False,)asyncdef_http_post(self,client,payload,include_protocol=True,expect_response=True):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 resp=awaitclient.post(self.http_url,headers=headers,json=payload)ifresp.status_code>=400:raiseRuntimeError(f"MCP HTTP{resp.status_code}:{resp.text[:300]}")session_id=resp.headers.get("mcp-session-id")ifsession_id:self._http_session_id=session_idifnotexpect_responseornotresp.text.strip():return{}if"text/event-stream"inresp.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 Server：

MCP_HTTP_URL="https://mcp.deepwiki.com/mcp"

tools/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 接口列表

两个接口都要求登录，未登录返回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 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"}}'

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:current=excwhilegetattr(current,"exceptions",None):current=current.exceptions[0]returncurrent

这样能看到真正的ConnectTimeout、ReadTimeout、HTTP 状态码等。

6. 配置汇总

6.1 stdio

MCP_ENABLED=true MCP_TRANSPORT="stdio" MCP_SERVER_COMMAND="npx.cmd" MCP_SERVER_ARGS='["-y","@modelcontextprotocol/server-filesystem","{workspace}"]' MCP_WORKSPACE_DIR="mcp_workspace" MCP_SERVER_ENV='{}' MCP_TIMEOUT_SECONDS=60

Linux/macOS 下MCP_SERVER_COMMAND可以写npx；代码会自动把npx.cmd转成npx。

6.2 Streamable HTTP

MCP_ENABLED=true MCP_TRANSPORT="streamable_http" MCP_HTTP_URL="https://mcp.deepwiki.com/mcp" MCP_HTTP_HEADERS='{}' MCP_HTTP_PROTOCOL_VERSION="2025-03-26" MCP_TIMEOUT_SECONDS=60

6.3 其它 stdio Server

# memory server MCP_SERVER_COMMAND="npx.cmd" MCP_SERVER_ARGS='["-y","@modelcontextprotocol/server-memory"]'

7. 常见问题

Q1：报unhandled errors in a TaskGroup (1 sub-exception)怎么办？

现在客户端会展开异常，日志里能看到真正原因。常见是：

• ConnectTimeout/ReadTimeout：网络或 Server 太慢，调大MCP_TIMEOUT_SECONDS；
• HTTP 401/403：远程 Server 需要鉴权，配置MCP_HTTP_HEADERS；
• HTTP 404：MCP_HTTP_URL写错。

Q2：PowerShell 提示npx.ps1 cannot be loaded？

客户端通过subprocess直接执行npx.cmd，不经过 PowerShell，所以不受执行策略影响。手工测试时也要用npx.cmd。

Q3：为什么不用最新版mcpSDK？

新版会强制升级pydantic>=2.11，与项目当前的pydantic==2.7.4冲突。Streamable HTTP 已经用httpx实现，核心功能不受影响。

Q4：stdio 第一次调用很慢？

npx -y首次会下载 npm 包，之后走缓存。生产环境建议提前安装包，或改为常驻 Server / HTTP 传输。

Q5：远程 Server 需要登录怎么办？

MCP_HTTP_HEADERS='{"Authorization":"Bearer <token>"}'

不要在工具参数里传 token。

8. 小结

• MCP 是工具接入标准，不替代 LangGraph 编排；
• stdio 适合本地工具，Streamable HTTP 适合远程服务；
• 本项目用MCPClient统一封装两种传输，外面只看到list_tools / call_tool；
• Controller 提供调试接口，方便先验证工具发现和调用；
• 下一篇会把 MCP 工具动态注册到主管智能体，并给出读取工作目录文档的完整示例；
• 生产环境要限制 MCP Server 白名单、工作目录、超时和鉴权 Header，避免把任意本地命令暴露给模型。