MCP 实战第 2 篇。固定使用官方 Python SDK
mcp==2.3.0,在独立虚拟环境中完成安装、工具发现、实际调用和错误验证。服务只读取虚构缺陷数据,不需要大模型密钥,也不连接真实业务平台。
上一篇已经区分 MCP、API、插件与 Agent,并准备了一个本地缺陷查询函数。接下来要回答更具体的问题:这个函数放进 MCP 服务之后,另一个程序究竟能不能找到它、调用它,并拿到可校验的结果?
这一篇把环境搭建的验收点放在这里。安装命令退出成功,只代表依赖可能已经准备好;只有客户端实际完成连接、发现工具并执行查询,才能确认本篇的链路跑通。
最终我们会得到两个 Python 进程:客户端启动本地服务,通过 stdio 交换 MCP 消息,结束后回收服务进程。整个过程不需要开放 HTTP 端口,也不需要先配置桌面 AI 客户端。
一、先固定版本,再复制代码
核对日期为 2026-10-29。本篇选择官方 Python SDK 2.3.0,使用该版本的MCPServer、Client、StdioServerParameters接口。官方 2.3.0 README
| 项目 | 本篇约定 |
|---|---|
| Python | 本篇脚本要求 3.11+;实际验证为 3.13.3 |
| SDK | mcp==2.3.0 |
| 传输 | 本机 stdio 子进程 |
| 数据 | 内置虚构项目与三条缺陷 |
| 客户端 | 本篇 Python 自检程序 |
| 外部条件 | 安装依赖时需能访问包源;运行样例不访问远程业务服务 |
SDK 包自身声明的 Python 要求是 3.10+,但本篇客户端用到了 Python 3.11 引入的asyncio.timeout,所以整篇采用 3.11+,不把 SDK 的最低要求直接当成示例的最低要求。
不要把其他文章中的 SDK 1.x 代码与本篇随意拼接。例如旧教程中的服务类、客户端会话管理和字段命名,可能与这里不同。本篇使用result.structured_content与result.is_error;配置与代码需要一起遵循固定版本。官方客户端说明
“SDK 2.3.0”与“协议版本”也不是同一个编号。后面会由客户端打印实际协商到的协议版本,不手工将它写成 2.3.0。
二、准备一个独立项目目录
建议使用新目录,不要直接在此前的 Dify 后端虚拟环境里安装。SDK 的间接依赖与其他项目可能有不同要求,独立环境更容易检查问题。
本文目录结构如下,四个文件的完整内容都在下文给出:
mcp-lab/ .venv/ mcp-02-requirements.txt mcp_demo_business.py mcp_demo_server.py mcp_stdio_check.py不要把自己的文件命名为mcp.py、pydantic.py、asyncio.py或json.py,也不要创建同名包目录。Python 搜索导入路径时可能先找到你的文件,出现“装了包却导入失败”的情况。
Windows 用户将文本保存为 UTF-8,文件扩展名应确实为.py,而不是编辑器自动补成.py.txt。macOS 也建议使用纯文本代码编辑器,避免富文本格式混入脚本。
三、Mac 创建与安装环境
先确认本机 Python 版本:
python3--version需要 3.11 或更新版本。若没有,先从 Python 官方下载页安装合适版本,再打开新终端确认解释器。机器上有多个 Python 时,可以把下面命令中的 python3 换成明确的解释器路径。
创建项目与虚拟环境:
mkdirmcp-labcdmcp-lab python3-mvenv .venvvenv为项目创建隔离的 Python 环境,具体机制可查阅 Python venv 官方文档。本文直接调用环境内解释器,后续不依赖终端是否执行过 activate。
将以下内容保存为mcp-02-requirements.txt:
mcp==2.3.0然后安装并确认版本:
.venv/bin/python-mpipinstall-rmcp-02-requirements.txt .venv/bin/python-mpip check .venv/bin/python-c"import sys; from importlib.metadata import version; print(sys.executable); print(sys.version); print('mcp', version('mcp'))"应看到解释器路径位于当前项目的.venv,SDK 版本为 2.3.0。安装时会拉取间接依赖,正常情况下不需要再单独安装 Pydantic。若在受控企业网络中使用内部包源,应按团队已有配置安装,不要遇到证书问题就关闭校验。
四、Windows PowerShell 创建与安装环境
先检查可用解释器:
py-3--version若没有 py 命令,但已有可用的 Python,也可以执行python --version,或者使用解释器完整路径。关键是确认版本满足 3.11+,并用同一个解释器创建环境。
项目初始化:
mkdir mcp-lab cd mcp-lab py-3-m venv.venv把上一节同样的依赖内容保存为mcp-02-requirements.txt,再执行:
.\.venv\Scripts\python.exe-m pip install-r mcp-02-requirements.txt.\.venv\Scripts\python.exe-m pip check.\.venv\Scripts\python.exe-c"import sys; from importlib.metadata import version; print(sys.executable); print(sys.version); print('mcp', version('mcp'))"这里直接调用 python.exe,不需要修改 PowerShell 的脚本执行策略,也不必先运行 Activate.ps1。路径包含空格时,PowerShell 可以使用调用运算符,例如:
&'C:\Your Project\mcp-lab\.venv\Scripts\python.exe'--version该路径只是语法示意,请替换成自己机器上确实存在的位置。Windows 命令依据跨平台运行方式提供,本次没有在 Windows 实机执行;文末会明确区分实际完成的验证。
五、保存业务函数:继续使用虚构数据
保存为mcp_demo_business.py。内容延续第 1 篇的本地练习,但本篇完整给出,不要求读者另行下载旧附件。
"""Local business-function exercise only. Not an MCP server or protocol client."""importargparseimportjson PROJECT="demo-project"ISSUES=({"id":"DEMO-001","title":"演示:提交按钮重复响应","status":"open"},{"id":"DEMO-002","title":"演示:报告标题显示异常","status":"open"},{"id":"DEMO-003","title":"演示:空列表提示缺失","status":"closed"},)deflist_demo_issues(project,status="open",limit=10):# A fixed demo project is a fixture boundary, not a real authorization system.ifproject!=PROJECT:raisePermissionError("demo_project_only")ifnotisinstance(status,str)orstatusnotin{"open","closed"}:raiseValueError("invalid_status")iftype(limit)isnotintornot1<=limit<=20:raiseValueError("invalid_limit")rows=[dict(row)forrowinISSUESifrow["status"]==status][:limit]return{"source":"local-fixture","project":PROJECT,"status":status,"count":len(rows),"issues":rows}defmain():parser=argparse.ArgumentParser(description=__doc__)parser.add_argument("--project",required=True)parser.add_argument("--status",default="open",choices=["open","closed"])parser.add_argument("--limit",default=10,type=int)args=parser.parse_args()try:result=list_demo_issues(args.project,args.status,args.limit)except(ValueError,PermissionError)aserror:print(json.dumps({"error":str(error)}))return2print(json.dumps(result,ensure_ascii=False,indent=2))return0if__name__=="__main__":raiseSystemExit(main())这个模块保持业务职责:检查固定项目、状态、数量范围,返回记录副本。它不负责传输协议,也没有真实用户鉴权。project 的固定值只是演示数据范围,不能当成生产权限机制。
数据里的 DEMO-001~003 都是虚构记录。后续接真实平台时,要替换数据访问实现,并保留独立测试,不要只把注释中的“演示”删掉就认为接入完成。
六、保存真正的 MCP 服务
保存为mcp_demo_server.py:
"""MCP SDK 2.3.0 stdio server; local fictional records only."""fromtypingimportAnnotated,Any,LiteralfrompydanticimportFieldfrommcp.serverimportMCPServerfrommcp.server.mcpserver.exceptionsimportToolErrorfrommcp_demo_businessimportlist_demo_issuesasquery_issues mcp=MCPServer("team-demo",instructions="Only fictional local records are available.")@mcp.tool()defhealth_check()->dict[str,Any]:"""Return this demo service's identity and>return{"service":"team-demo","data_source":"local-fixture","mode":"read-only-demo"}@mcp.tool()deflist_demo_issues(project:Literal["demo-project"],status:Literal["open","closed"]="open",limit:Annotated[int,Field(strict=True,ge=1,le=20)]=10,)->dict[str,Any]:"""List fictional defects in demo-project; never query a real issue tracker."""try:returnquery_issues(project,status,limit)except(PermissionError,ValueError)aserror:raiseToolError(str(error))fromNoneif__name__=="__main__":# stdout belongs to the MCP transport. Do not print startup messages here.mcp.run(transport="stdio")服务暴露两个工具:health_check 返回服务标识和数据来源;list_demo_issues 调用业务函数读取样例。SDK 根据函数参数类型生成输入结构,实际执行时仍会进行校验。
这里有两个容易影响后续接入的细节。
第一,返回值声明为dict[str, Any]。本篇固定版本会据此提供可供程序读取的结构化结果。如果只写一个没有具体键类型的裸dict,不能据此假定structured_content一定存在。需要更严格的输出字段约束时,可进一步定义 Pydantic 模型;本篇先验证链路与结果内容。
第二,mcp.run(transport="stdio")使用标准输入输出传递协议消息。服务侧不要往 stdout 打印“启动成功”等普通文字;诊断日志应使用 stderr 或 SDK 日志机制。否则客户端收到的内容可能不再是有效协议消息。stdio 传输规范
直接在终端启动这个服务,可能只看到它一直等待输入。没有浏览器页面、没有 HTTP 端口,是当前传输方式的正常表现。用于本篇验证时,不必另开终端手工启动它;下一节客户端会创建自己的服务子进程。
七、保存客户端:主动验证,不靠聊天猜测
保存为mcp_stdio_check.py:
"""Python 3.11+ / mcp==2.3.0: actual subprocess + stdio checks, no model."""importargparseimportasynciofromimportlib.metadataimportversionimportjsonfrompathlibimportPathimportsysfrommcpimportClient,StdioServerParametersasyncdefcheck(server_path):checks=[]defrequire(condition,label):ifnotcondition:raiseRuntimeError("check_failed:"+label)checks.append(label)params=StdioServerParameters(command=sys.executable,args=[str(server_path)],cwd=str(server_path.parent),env={"PYTHONUTF8":"1","PYTHONIOENCODING":"utf-8"},)asyncwithClient(params)asclient:listed=awaitclient.list_tools()names=sorted(tool.namefortoolinlisted.tools)require(names==["health_check","list_demo_issues"],"tool_discovery")protocol=client.protocol_version health=awaitclient.call_tool("health_check",{})require(nothealth.is_errorandhealth.structured_content=={"service":"team-demo","data_source":"local-fixture","mode":"read-only-demo"},"health_and_source")opened=awaitclient.call_tool("list_demo_issues",{"project":"demo-project"})data=opened.structured_content require(notopened.is_errorandisinstance(data,dict)anddata.get("source")=="local-fixture"anddata.get("count")==2and[row["id"]forrowindata["issues"]]==["DEMO-001","DEMO-002"],"open_records")closed=awaitclient.call_tool("list_demo_issues",{"project":"demo-project","status":"closed"})require(notclosed.is_errorandclosed.structured_content["issues"][0]["id"]=="DEMO-003","closed_filter")limited=awaitclient.call_tool("list_demo_issues",{"project":"demo-project","limit":1})require(notlimited.is_errorandlimited.structured_content["count"]==1,"return_limit")forlabel,argumentsin[("invalid_project",{"project":"other-project"}),("invalid_status",{"project":"demo-project","status":"all"}),("invalid_limit",{"project":"demo-project","limit":0}),("invalid_limit_type",{"project":"demo-project","limit":True}),]:rejected=awaitclient.call_tool("list_demo_issues",arguments)require(rejected.is_errorisTrue,label)unknown=awaitclient.call_tool("missing_tool",{})require(unknown.is_errorisTrue,"unknown_tool")return{"sdk_version":version("mcp"),"protocol_version":protocol,"transport":"stdio","data_source":"local-fixture","tools":names,"passed":len(checks),"checks":checks}defmain():parser=argparse.ArgumentParser(description=__doc__)parser.add_argument("--server",type=Path,default=Path(__file__).with_name("mcp_demo_server.py"))args=parser.parse_args()server=args.server.resolve()ifnotserver.is_file():parser.error("Server script does not exist")ifversion("mcp")!="2.3.0":parser.error("Install the tutorial's pinned mcp==2.3.0 first")asyncdefrun():# Active connection budget; subprocess cleanup can add time after cancellation.asyncwithasyncio.timeout(30):returnawaitcheck(server)try:result=asyncio.run(run())exceptExceptionaserror:print(json.dumps({"error_type":type(error).__name__,"hint":"Check stderr, interpreter, SDK version and server path."}),file=sys.stderr)return1print(json.dumps(result,ensure_ascii=False,indent=2))return0if__name__=="__main__":raiseSystemExit(main())StdioServerParameters描述需要启动的本地程序。这里的 command 使用sys.executable,意味着客户端启动服务时继续使用当前虚拟环境中的 Python;args 则传入服务脚本的绝对路径。
因此,用户不需要依赖系统 PATH 中恰好排在前面的 python,也不需要把解释器路径写死成作者机器上的位置。cwd 指向服务脚本目录,方便稳定找到同目录业务模块。官方 stdio 客户端说明
本例只显式传入 UTF-8 相关环境变量,没有传递模型 Key 或平台凭据。SDK 对子进程环境有自己的继承策略;将来服务需要凭据时,应按实际 SDK 和部署配置显式提供,不能假定终端里出现过的所有变量都会自动到达服务。
客户端先看is_error,再使用structured_content。工具返回了一个响应,不代表查询业务成功;参数拒绝和未知工具也是需要检查的结果。
客户端设定 30 秒的活动执行预算,避免本篇演示无期限等待。超时后的子进程收尾可能额外花时间,所以不把这个值描述为严格的操作系统总退出时限。它也不是向远程业务系统发送撤销操作的承诺;本例只有只读样例查询。
八、运行并核对结果
Mac:
.venv/bin/python mcp_stdio_check.pyWindows PowerShell:
.\.venv\Scripts\python.exe mcp_stdio_check.py本次 macOS 实际输出如下:
{"sdk_version":"2.3.0","protocol_version":"2026-07-28","transport":"stdio","data_source":"local-fixture","tools":["health_check","list_demo_issues"],"passed":10,"checks":["tool_discovery","health_and_source","open_records","closed_filter","return_limit","invalid_project","invalid_status","invalid_limit","invalid_limit_type","unknown_tool"]}重点看四项证据:SDK 版本为 2.3.0;transport 为 stdio;工具列表包含两个预期名称;passed 为 10。协议版本由这次实际连接产生,本次为 2026-07-28;对不同 SDK 或服务组合,不应只凭显示了同一个日期就推断全部能力兼容。
十项检查分别覆盖:工具发现、服务身份与数据源、open 记录、closed 过滤、数量限制、错误项目、错误状态、越界数量、错误数量类型、未知工具。
程序故意发送几次无效调用,因此 stderr 中出现 rejected arguments 或 Unknown tool 等信息属于这组负向检查的预期行为。判断这份自检是否通过,应结合最终 JSON 和进程退出码;不是见到一行报错文字就认定服务没启动。
Mac 可在执行后检查退出码:
echo$?PowerShell:
$LASTEXITCODE通过时为 0。主程序失败时返回 1,命令参数或固定版本检查失败通常由 argparse 返回 2。若依赖尚未安装,Python 导入阶段就可能退出,应先修复环境,不能只看某个退出码猜根因。
九、从其他目录启动,检查路径是否可靠
自检支持指定服务路径,例如 Mac:
.venv/bin/python mcp_stdio_check.py--server"$PWD/mcp_demo_server.py"Windows PowerShell:
$server=Join-Path(Get-Location)'mcp_demo_server.py'.\.venv\Scripts\python.exe mcp_stdio_check.py--server$server这里仍要保留业务模块与服务文件在同一目录。程序能够定位服务文件,不代表服务自己的其他依赖文件也会被自动复制过去。
本次还把三个脚本复制到包含中文和空格的目录,并从另一个工作目录运行客户端,十项检查再次通过。这验证了当前 macOS 环境中的路径处理,没有据此声称所有 Windows 路径、网络盘和同步目录都已验证。
只运行你自己编写或已经审查的服务文件。stdio 配置会启动本地进程,MCP 并不自动把该进程变成操作系统沙箱。
十、按现象排查,避免重新装一遍碰运气
| 现象 | 优先核对 |
|---|---|
| No module named mcp | 安装与运行是否使用同一个 .venv 解释器 |
| 无法导入 Client 或 MCPServer | 实际安装版本、同名文件遮蔽、是否混用其他教程 |
| No module named mcp_demo_business | 三个脚本是否保存齐全且放在同一目录 |
| 自检提示找不到 server | 文件名、扩展名和 --server 路径 |
| 服务直接启动后等待不退出 | 它在等待 stdio 消息,改运行自检客户端 |
| 协议解析异常 | 服务或导入模块是否向 stdout 打印了普通日志 |
| structured_content 为空 | 先看 is_error,再核对返回类型与 output schema |
| ExceptionGroup | 阅读关联 stderr,逐层区分连接异常与本地检查失败 |
| pip 下载或证书失败 | 网络、可信包源和证书配置;不是工具函数逻辑问题 |
有时最先看到的 ExceptionGroup 只是异步任务组包装后的异常类型,真正的问题在内部异常或前面的服务日志里。自检的固定提示只是起点,没有把所有故障都错误地映射为“网络断开”。
如果需要提交排查信息,优先记录 Python 版本、SDK 版本、解释器位置、文件结构及经过检查的错误片段。不要把整个进程环境、含真实凭据的配置或业务数据直接贴到公开评论区。
十一、保存依赖与验证记录
依赖文件只固定了直接使用的 SDK 版本,间接依赖仍由安装时的解析结果决定。完成验证后,可以保存自己的环境记录:
Mac:
.venv/bin/python-mpip freeze>verified-environment.txtWindows PowerShell:
.\.venv\Scripts\python.exe-m pip freeze|Out-File-Encoding utf8 verified-environment.txt本篇发布包附有作者这次 macOS/Python 3.13.3 的版本记录,供核对使用。它不是已经验证过的 Windows 通用锁文件,也不保证任意 Python 版本都能安装完全相同的间接依赖。
若以后升级 SDK,应在新的环境验证,而不是只修改版本号后直接覆盖正在使用的配置。先运行同一组发现、调用、错误场景,再核对协议、参数及输出差异。
十二、本篇验收与下一步
本篇实际完成:独立环境安装 SDK 2.3.0,pip check 无依赖冲突;本机真实 stdio 子进程通信通过十项检查;中文与空格目录、不同工作目录下的运行验证通过。
这里的“真实”指实际启动了 SDK 服务并传递协议消息,业务数据仍然是虚构样例。没有调用大模型,没有连接缺陷平台,没有使用桌面客户端,也没有在 Windows 实机或远程 HTTP 部署中验证。
下一篇进入工具开发:进一步说明参数约束、结构化结果、可读错误与业务边界。当前两个工具足够证明环境链路成立,后续再围绕团队研发助手逐步增加能力。