资讯动态

MCP Client/Server 实战:Agent 工具调用从空壳到天气联调

发布时间:2026/10/30 5:08:50 来源:尧图企业网站定制
简介《从零到一MCP快速入门实战》面向希望掌握模型上下文协议MCP的开发者与AI应用实践者尤其适合具备Python基础、准备把大模型接入外部工具却受困于Function calling门槛的初中级读者。内容从技术体系切入回顾Function calling与Agent开发脉络再以uv工具搭建虚拟环境逐步完成MCP客户端项目创建、基础代码结构编写与运行并演示接入OpenAI、DeepSeek在线模型及本地ollama、vLLM模型的配置方式随后以天气查询服务器为例讲清Server通讯机制、依赖安装、代码编写与Inspector调试最后延伸到客户端与服务端的进阶功能。整份资料为单个PDF文件约47.66MB页面信息密度较高适合按章节顺序阅读并同步动手复现。已有1127人学习可作为建立MCP开发整体认知、少走环境配置弯路的入门参考。1. MCP 不是新框架它把 Agent 调外部工具的重复劳动砍掉了很多人第一次写 Agent卡住的地方不是模型而是「查个天气」这种小功能一个外部函数写上上百行再手写一份 JSON Schema 描述给模型看还要调提示词模板提高命中率。MCPModel Context Protocol模型上下文协议要解决的就是这段重复劳动。它把大模型运行环境叫 MCP Client把外部工具运行环境叫 MCP Server两端按同一套规范通信——统一名称、统一调用格式、统一提示词模板。好处很直接天气查询、数据库检索、网页抓取这类通用能力一个人写好 Server其他人拷进项目就能用不用每人重写一遍 function calling。这篇按「先跑通空壳客户端 → 接在线/本地模型 → 写天气 Server 联调 → 排错 → 进阶」的顺序拆适合已经会写 Python、想快速把 MCP 用到自己项目里的人。2. uv 与最小客户端先把空壳跑通再谈接模型我见过太多人一上来就把模型、Server、工具全堆进一个文件跑不通就无从下手。MCP 这条链路上有两个独立的进程客户端和服务器通过标准输入输出通信任何一个环节出问题现象都长得差不多。所以第一步只做一件事让一个不连服务器、不接模型的客户端跑起来确认 uv 环境、包导入、异步入口这三件事没问题。2.1 为什么这类项目我一般先用 uv 而不是 pip 或 condaMCP 的 Python 依赖不多但迭代频繁你在调客户端的时候会反复加依赖、删依赖、换模型 SDK。pip 的问题在于每次装包都要重新解析整棵依赖树装一个 openai 可能要等十几秒conda 的问题在于它管的是全局环境为每个 MCP 小项目建一个 conda 环境磁盘和激活时间都划不来。uv 是 Rust 写的 Python 依赖管理工具定位上同时替代 pip、venv 和 pip-tools 三件套。落到 MCP 场景我看重它三点一是装包速度快反复增删依赖时体感差别很明显二是它会自动识别当前项目主目录并在其中创建虚拟环境不用你再记python -m venv的路径参数三是它原生支持pyproject.toml依赖清单和项目元数据放一个文件里把项目拷给别人时不用附带一堆说明。安装按你的系统选一条# 方式一本机已有 pip直接装 pip install uv # 方式二没有 pip 的环境用官方安装脚本 curl -LsSf https://astral.sh/uv/install.sh | sh # 装完验证 uv --version提示安装脚本会装到用户级目录装完如果提示uv: command not found重新开一个终端或者手动把安装输出里提示的路径加进PATH。常用命令和 pip 的对应关系我整理成一张表改脚本的时候直接查目的pip 写法uv 写法安装依赖pip install mcpuv add mcp按文件安装pip install -r requirements.txtuv pip install -r requirements.txt创建虚拟环境python -m venv .venvuv venv在项目里执行脚本先激活再python x.pyuv run python x.pyuv add和uv pip install的区别值得说清楚前者会写入pyproject.toml并更新锁文件是「这个项目需要它」后者只往当前环境里塞包不记录。自己写 MCP 项目一律用uv add否则过两天你换台机器就说不清当时装了什么。2.2 从 uv init 到能跑的 client.py项目初始化四步一条命令一步# 1. 创建项目骨架会生成 pyproject.toml 和 main.py uv init mcp-client cd mcp-client # 2. 在当前目录创建虚拟环境 .venv uv venv # 3. 激活环境Windows 用 .venv\Scripts\activate source .venv/bin/activate # 4. 安装 MCP 官方 SDK uv add mcp激活这步别省。后面用StdioServerParameters启动子进程时如果command写成裸的python子进程继承的是系统解释器而不是.venv里的那个mcp包就找不到——这是新手最容易翻的车之一第 5 章会展开。接着在项目根目录建client.py写一个只打印模拟回复的版本import asyncio from contextlib import AsyncExitStack from mcp import ClientSession class MCPClient: def __init__(self): # 暂不连接真实服务器先把会话对象占位 self.session None # 统一管理后续所有异步上下文资源 self.exit_stack AsyncExitStack() async def connect_to_mock_server(self): print(MCP 客户端已初始化尚未连接服务器) async def chat_loop(self): print(MCP 客户端已启动输入 quit 退出) while True: try: query input(\nQuery: ).strip() if query.lower() quit: break # 此时没有接模型只做回显 print(f[Mock Response] 你说的是: {query}) except Exception as e: print(f发生错误: {e}) async def cleanup(self): # 关闭 AsyncExitStack 里注册的所有资源 await self.exit_stack.aclose() async def main(): client MCPClient() try: await client.connect_to_mock_server() await client.chat_loop() finally: # 无论正常退出还是异常都走一遍资源清理 await client.cleanup() if __name__ __main__: asyncio.run(main())运行uv run python client.py能输入、能回显、输入quit能干净退出这一步就算过了。这段代码有三个地方值得单独讲因为它们决定了后面接模型和接服务器时改哪里第一是AsyncExitStack。MCP 客户端连服务器时会打开子进程、标准输入输出管道、会话对象三层资源正常退出要按逆序关掉异常退出也得关。手写try/finally一层层嵌套很快就乱了。AsyncExitStack的用法是「用enter_async_context注册最后aclose()一次性逆序释放」所以你现在看到的self.exit_stack虽然什么都没装但它是后面所有连接的容器别删。第二是input()在异步函数里的阻塞问题。input()是同步阻塞调用会把整个事件循环卡住期间任何await都不会推进。这个空壳版本只有输入没有任何并发所以看不出问题一旦你接上 stdio 传输服务器那边可能在推送消息而主线程正卡在input()上就会出现「工具调用没响应」的玄学现象。常见做法是把输入单独丢进线程池或者干脆换成语义更清晰的命令行参数模式。这一点在第 5 章的排查清单里还会再出现。第三是入口写法。asyncio.run(main())只能在一个没有运行中事件循环的线程里调用命令行跑没问题但在 Jupyter 里直接运行会报asyncio.run() cannot be called from a running event loop。在 Notebook 里调试时改成直接await main()更省事。2.3 代码结构对齐为什么类里先留两个空方法很多人写这类入门代码时会顺手把类拍扁成函数跑是能跑但接模型的时候要重构一遍。MCPClient这个类的两个方法分别对应两种不同的职责connect_to_mock_server管「连接层」chat_loop管「交互层」。后面接真实服务器时你只需要把前者换成stdio_clientClientSession的初始化接模型时只动后者两者互不干扰。这种切分在调试时价值很大工具列不出来是连接层的问题工具列出来了但模型不调用是交互层的问题边界一清二楚。self.session现在还是None接上真实服务器后它会变成ClientSession实例这个对象提供了几个核心方法initialize()完成握手并协商协议版本list_tools()拉取服务器声明的所有工具名称、描述、参数 Schemacall_tool(name, arguments)真正执行工具并拿回结果。记住这三个方法名第 4 章的联调就是围绕它们展开的。3. 接模型在线 API 与本地推理的两条接入路径空壳跑通之后下一个问题是「谁来理解用户的话并决定调哪个工具」。答案是大模型但模型跑在哪差别很大在线 API 省事本地推理省钱且数据不出机器。这一段把两条路都走一遍因为它们的客户端代码骨架几乎一样差别集中在base_url、api_key和模型对工具调用tool calling的支持程度上。3.1 在线模型接入.env、依赖与 client.py 的最小改动先把密钥从代码里挪出去。在项目根目录建.env# .env API_KEYsk-xxxxxxxxxxxxxxxx BASE_URLhttps://api.deepseek.com MODELdeepseek-chat.env必须加进.gitignore这条别嫌啰嗦。密钥进版本库是安全事故不是「下次注意」的级别。安装依赖uv add openai python-dotenv这里用的是 OpenAI 兼容的 SDK几乎所有提供 chat/completions 接口的服务都能指向自己的base_url使用换供应商只需要改.env三行不用动代码。然后改造client.py把connect_to_mock_server换成真实连接把chat_loop换成带工具调用的对话循环。import os import json import asyncio from contextlib import AsyncExitStack from dotenv import load_dotenv from openai import OpenAI from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client load_dotenv() def mcp_tools_to_openai(mcp_tools): 把 MCP 的工具描述转成 OpenAI 兼容的 tools 结构 return [ { type: function, function: { name: t.name, description: t.description or , parameters: t.inputSchema, }, } for t in mcp_tools ] class MCPClient: def __init__(self): self.session None self.exit_stack AsyncExitStack() self.llm OpenAI( api_keyos.getenv(API_KEY), base_urlos.getenv(BASE_URL), ) self.model os.getenv(MODEL, deepseek-chat) async def connect_to_server(self, server_script: str): 启动 MCP Server 子进程并建立会话 params StdioServerParameters( commandpython, args[server_script], envNone, ) read, write await self.exit_stack.enter_async_context(stdio_client(params)) self.session await self.exit_stack.enter_async_context( ClientSession(read, write) ) await self.session.initialize() tools await self.session.list_tools() print(已连接服务器可用工具:, [t.name for t in tools.tools]) async def process_query(self, query: str) - str: 一轮对话模型判断是否调工具调完把结果回灌给模型 messages [{role: user, content: query}] tools mcp_tools_to_openai((await self.session.list_tools()).tools) while True: resp self.llm.chat.completions.create( modelself.model, messagesmessages, toolstools, tool_choiceauto, ) msg resp.choices[0].message messages.append(msg.model_dump(exclude_noneTrue)) # 模型不调工具直接给最终回答 if not msg.tool_calls: return msg.content or # 逐个执行工具并把结果以 tool 角色回灌 for call in msg.tool_calls: args json.loads(call.function.arguments or {}) result await self.session.call_tool( call.function.name, argumentsargs ) text \n.join( c.text for c in result.content if hasattr(c, text) ) messages.append( { role: tool, tool_call_id: call.id, content: text, } ) async def chat_loop(self): print(输入 quit 退出) while True: query input(\nQuery: ).strip() if query.lower() quit: break print(await self.process_query(query)) async def cleanup(self): await self.exit_stack.aclose()几个关键点tool_choiceauto表示让模型自己判断调不调工具。如果你想强制它必须先查天气再回答可以设成{type: function, function: {name: get_weather}}但强制指定后模型不会跳过遇到寒暄类问题也会硬调一次实际项目里一般还是用auto。msg.model_dump(exclude_noneTrue)是把模型返回的对象转成字典再塞回消息列表。这一步必须做而且要去掉值为None的字段——部分服务端对tool_calls: null这类显式空值会报 400。循环用while True是必要的不是保守写法。模型可能在拿到工具结果后再决定调第二个工具比如先查城市编码再查天气。每轮都要把工具结果以roletool回灌并且必须带上tool_call_id否则服务端无法把结果和请求对上。inputSchema是 MCP 工具自带的 JSON Schema格式和 OpenAI 的parameters完全一致所以转换函数几乎是直接搬运。这也是 MCP 省事的地方——工具的参数定义只写一份。3.2 本地 ollama 与 vLLM接口看起来一样参数不一样本地推理有两条常见路线。ollama 适合快速验证一条命令拉模型就能用vLLM 适合要吞吐、要并发、要显存利用率的场景。ollama 提供 OpenAI 兼容端点改.env就行# 指向本机 ollama API_KEYollama BASE_URLhttp://localhost:11434/v1 MODELqwen2.5:7bapi_key随便填ollama 不校验但 SDK 要求这个字段非空。模型名用ollama list里看到的完整名字。vLLM 把模型服务化之后同样是 OpenAI 兼容# 启动推理服务端口默认 8000 vllm serve /path/to/your/model --port 8000 --max-model-len 8192# .env API_KEYvllm BASE_URLhttp://localhost:8000/v1 MODEL/path/to/your/model注意 vLLM 的MODEL字段通常要填启动时用的那个模型路径或注册名填错会返回 404 而不是模型不存在之类的明确报错。两条本地路线的真实差别不在接入代码在模型能力上。工具调用依赖模型本身的 function calling 训练七 B 以下的小模型经常会「该调的时候不调」「参数乱填」「一次调三个工具」。同样的代码在线大模型一通就过换成小模型就要在提示词上补很多约束。这不是客户端的问题别在代码里找原因。稳妥的做法是先用在线模型把链路验完确认 Server 没问题再换本地模型做压测。另一个参数是max_tokens。本地小模型的上下文窗口普遍吃紧工具返回的长文本比如网页抓取结果塞进去两三轮就爆。建议在chat.completions.create里显式设一个上限并在工具返回端做截断别指望模型自己忽略。3.3 工具多了怎么办先筛后塞当服务器工具数量上去以后一次性把所有工具的 Schema 塞进tools参数会让每次请求的 token 数暴涨。我一般做两级处理按会话主题先分组比如查数据库的一组、写文件的一组当前对话只挂相关组再给工具描述加「什么时候该用」的句子让模型的选择更准。MCP 本身没有规定工具必须全量暴露客户端调list_tools拿回来之后是可以自己过滤的这个自由度别浪费。4. 天气查询 Server从 stdio 通信到工具联调客户端能连、模型能调剩下就是把工具那一端写出来。这一章用一个天气查询 Server 走完整条链路重点不在业务逻辑而在「工具是怎么被声明的」「客户端怎么发现它」「怎么在不写客户端的情况下验证它」。4.1 服务器概念与通讯机制MCP Server 的本质是一个进程对外声明若干工具tools、资源resources和提示模板prompts并响应客户端的调用请求。传输方式上本地开发用得最多的是 stdio客户端启动服务器进程通过子进程的标准输入输出传 JSON-RPC 消息。这个机制带来一条硬约束值得单独拎出来标准输出是协议通道不是日志通道。你在服务器里随手print(debug)客户端收到的就是一坨不符合 JSON-RPC 格式的文本轻则报解析错误重则直接卡死在初始化阶段。调试信息一律走stderr或者写文件这条踩过的人都懂。新建一个文件weather_server.pyfrom mcp.server.fastmcp import FastMCP # 参数是服务器名称客户端和 Inspector 里会显示 mcp FastMCP(weather) # 示例数据生产上换成你自己的气象数据源 WEATHER_DB { 北京: {cond: 晴, temp: 26}, 上海: {cond: 多云, temp: 24}, 广州: {cond: 雷阵雨, temp: 30}, } mcp.tool() def get_weather(city: str) - str: 查询指定城市的实时天气。 Args: city: 城市中文名称例如 北京 上海不要传拼音或英文 item WEATHER_DB.get(city) if item is None: # 返回可读的错误说明而不是抛异常 return f暂不支持该城市: {city}目前支持: {list(WEATHER_DB)} return f{city} 当前{item[cond]}气温 {item[temp]} 摄氏度 if __name__ __main__: mcp.run()代码很短但每一处都有讲究。mcp.tool()装饰器负责把一个普通 Python 函数注册成 MCP 工具。FastMCP 会从函数签名自动生成参数 Schema从 docstring 自动生成工具描述。这意味着 docstring 不是写给人看的注释它是模型看到的说明书写得好不好直接决定调用准不准。city: str这个类型注解必须有没有它生成的 Schema 里参数类型是空的模型不知道该传字符串还是对象很容易传成{city: {name: 北京}}这种嵌套结构然后你的函数拿到一个 dict 直接报错。返回字符串而不是抛异常是有意的设计。工具内部抛异常时异常会以错误结果的形式回传给模型模型看到一堆 traceback 通常不知道怎么处理可能反复重试同一个调用。返回一句可读的失败说明模型往往能自己纠正比如换个城市再查。mcp.run()默认走 stdio不需要额外配置。这就是为什么客户端能直接用一个python weather_server.py命令把它拉起来。4.2 客户端联调initialize、list_tools、call_tool 三步把第 3 章的client.py里的connect_to_server指向weather_server.pyasync def main(): client MCPClient() try: await client.connect_to_server(weather_server.py) await client.chat_loop() finally: await client.cleanup() if __name__ __main__: asyncio.run(main())运行之后按照三步验证链路是否真的通了第一步看握手。initialize()之后没有报错说明子进程起来了、协议版本协商通过。这一步失败九成是解释器路径或依赖问题。第二步看工具列表。代码里会打印已连接服务器可用工具: [get_weather]。列表是空的说明mcp.tool()没生效或者服务器在run()之前就崩了。第三步看调用。输入「北京今天天气怎么样」观察三件事模型有没有主动发起tool_calls、call_tool有没有真的执行、回灌之后模型有没有把结果组织成自然语言。这三件任何一件没发生问题定位方向完全不同——第一件是工具描述问题第二件是通信问题第三件是消息拼装问题。想把过程看得更清楚可以在process_query里临时加两行日志到 stderrimport sys print(f[debug] tool_calls{msg.tool_calls}, filesys.stderr)注意是filesys.stderr。这一行如果写成普通print你的客户端会立刻挂掉。4.3 Inspector不写客户端也能把服务器验完每次改服务器都重新跑一遍客户端、再接一次模型费时还费额度。更省事的做法是先用 Inspector 直接把服务器点亮。MCP Python SDK 自带一个开发命令uv run mcp dev weather_server.py它会起一个本地 Web 界面并把你的服务器接进去界面上能看到服务器声明的所有工具、每个工具的参数 Schema还能手动填参数点执行、看原始返回。服务器这边改完代码重启一下界面刷新就能验新一轮。这个工具的价值在于把「服务器对不对」和「客户端/模型对不对」切开。工具在 Inspector 里调不通就别去客户端里找原因了问题一定在服务器侧Inspector 里调得通而客户端不行问题就在消息拼装或模型能力上。二分排查比盯着一个完整流程猜要快得多。5. 避坑与排查五条踩过的记录链路能跑通之后真正耗时间的都在细节上。以下五条是我在实际项目里反复遇到的按「现象 → 原因 → 解决」写症状对得上就直接照做。5.1 连接与进程类记录一客户端的工具列表永远是空的现象list_tools()返回空列表没有报错Inspector 里却能看到工具。原因StdioServerParameters的command写的是裸的python子进程用的是系统解释器里面没装mcp包。服务器进程启动即崩但错误输出在 stderr客户端没捕获表现为「连上了但什么都没有」。解决command显式指向虚拟环境内的解释器Linux/macOS 用.venv/bin/pythonWindows 用.venv\\Scripts\\python.exe。或者统一在uv run下启动客户端让子进程继承同一个环境。记录二服务器加了一行 print 之后初始化卡住不动现象initialize()一直挂起或者抛出 JSON 解析错误。原因stdio 传输把标准输出当成协议通道。你的print输出混在 JSON-RPC 消息里客户端解析失败。这个坑最阴的地方在于加 print 之前一切正常加完之后报错信息指向协议层完全看不出跟那行 print 有关。解决所有调试输出改走stderr或者用logging配一个FileHandler写到文件里。养成习惯MCP Server 代码里不出现裸的print。记录三子进程拿不到环境变量现象服务器代码里读os.getenv(SOME_TOKEN)永远是None但同样的变量在客户端进程里能读到。原因StdioServerParameters(envNone)表示不额外注入环境变量子进程只继承部分基础变量你在 shell 里export的那些不一定传得过去。解决需要透传时显式构造env{**os.environ, SOME_TOKEN: os.getenv(SOME_TOKEN, )}。别只传自己那一个变量整个os.environ展开一层更稳否则PATH丢了子进程里的其他命令可能找不到。5.2 模型与工具调用类记录四工具描述写得太含糊模型死活不调用现象问「北京天气」模型直接编一个答案压根不发起tool_calls。原因inputSchema是对的但description只写了「查询天气」四个字。模型判断要不要调工具主要看描述和用户意图的匹配度描述太抽象时它宁可自己编。参数描述缺失还会导致传参格式错。解决docstring 里写清楚三件事——这个工具干什么、参数格式和单位、什么场景下该用。像city: 城市中文名称不要传拼音或英文这种约束句写与不写调用准确率差别很明显。记录五多轮对话后上下文爆掉请求直接失败现象前两轮正常第三轮开始返回超长错误或者 token 超限。原因工具返回的长文本网页正文、查询结果集被完整塞进messages每一轮都累加几轮就撑满窗口。模型不会自己帮你截断。解决在把工具结果写进messages之前做截断比如保留前 2000 字符并追加一行「结果已截断」。如果业务确实需要长内容改成让模型基于摘要继续或者把大结果存成文件、只在消息里给路径。6. 进阶把 Server 做稳的三个具体动作工具数量的膨胀速度往往超出预期一个 Server 从 1 个工具长到 10 个工具是几周的事。我后来固定做三件事。第一控制工具粒度。粗粒度和细粒度各有代价粒度示例优点代价粗query_db(sql)工具少模型易选参数自由度高易生成不合法 SQL细get_user_by_id(uid)参数受约束成功率高工具多选择成本上升我的取舍是读操作用细粒度把常见查询封装成固定参数的工具分析型、探索型的需求留一个粗粒度入口但在这个入口的工具描述里明确写出库表结构和使用限制。第二给工具加超时和降级。MCP 的工具调用是异步的一个卡住的外部请求会把整轮对话拖死。在 Server 侧用asyncio.wait_for包一层超时返回「服务暂时不可用请稍后再试」而不是让客户端无限等待。import asyncio mcp.tool() async def get_weather(city: str) - str: 查询指定城市的实时天气city 传中文城市名。 try: return await asyncio.wait_for(_fetch_weather(city), timeout5) except asyncio.TimeoutError: return 天气服务响应超时请稍后重试timeout5这个值按你的下游接口定原则是比下游自身的超时略短一点让失败尽早暴露成一句可读的话而不是让用户看着转圈。第三把错误信息写成模型能读懂的形式。前面提过不要抛异常更进一步错误信息里带上「下一步能做什么」。比如查不到城市时返回支持的城市列表模型下一轮往往就能自己修正只返回KeyError: xxx模型会无所适从。验证方法也很固定每次改完 Server先uv run mcp dev weather_server.py在 Inspector 里把所有工具的正例、反例、边界值各点一遍再看客户端里的表现。这套流程我后来做成了肌肉记忆——从那以后我每新起一个 MCP 项目都强制自己先单独把 Server 在 Inspector 里跑通再写客户端代码省下来的排查时间比写代码的时间还多。希望帮到你。本文还有配套的精品资源点击获取

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑