1. 从零理解 MCP:Agent 工具调用为什么需要 Model Context Protocol
刚接触 Agent 开发的人,大概率都写过这样的代码:把天气查询、数据库读写、文件操作这些函数直接塞进项目里,然后在 prompt 里告诉模型「你有这些工具可以用」。项目小的时候没问题,一旦工具超过十个,代码就开始失控——工具定义散落在各个文件,换个 Agent 项目就得把同一套工具重写一遍,模型还经常搞混参数格式。
我试过在一个项目里集成十几个工具函数,结果模型调用时把city参数传成了location,排查了半天才发现是工具描述写得不一致。这类问题的根源不是模型笨,而是工具的管理方式太原始。
MCP(Model Context Protocol,模型上下文协议)就是来解决这个问题的。它是 Anthropic 开源的一套标准化协议,核心思路很简单:把工具、提示词模板、文档资源从 Agent 项目里抽出来,放到一个独立的「MCP 服务器」里统一管理。Agent 作为「MCP 客户端」,通过标准接口去访问这些工具,自己不存储任何工具实现。
打个比方,以前的 Agent 像是一个什么都自己扛的个体户,工具就是他的私人工具箱;MCP 模式下,Agent 变成了一个会打电话的调度员,工具箱放在专门的仓库(MCP 服务器)里,需要什么就按标准流程去取。仓库可以服务多个调度员,工具改一次所有 Agent 都生效。
这套协议适合谁?如果你正在做 Agent 应用,工具数量超过五个,或者希望工具能在多个项目间复用,MCP 就是值得投入的方向。它不绑定具体模型,OpenAI 兼容接口、Claude、本地部署的模型都能对接。下面我用 FastMCP 搭一个最小可用的天气查询工具服务,从服务端注册到客户端调用完整跑一遍。
2. FastMCP 环境准备与 TaoToken 接入配置
FastMCP 是 MCP 官方 Python SDK 里封装度最高的开发框架,用装饰器就能把普通函数变成 MCP 工具,省去了手写 JSON Schema 的麻烦。安装只需要一行:
pip install fastmcp openai python-dotenv这里fastmcp负责服务端,openai用于客户端调用大模型,python-dotenv管理配置。如果你用的是本地 Ollama,openai包同样能用,只要把 base_url 指向本地端口即可。
接下来是模型接入部分。客户端需要调用一个大模型来做 Function Calling,你可以用本地模型,也可以用云端 API。如果走云端,TaoToken 提供了 OpenAI 兼容的接口,配置方式和标准 OpenAI SDK 一致。先到控制台创建一个 API Key:
访问 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite 生成密钥,注意保存后页面不再显示完整 Key。
拿到 Key 之后,在项目根目录建一个.env文件:
OPENAI_API_KEY=sk-你的TaoToken密钥 OPENAI_BASE_URL=https://taotoken.net/api OPENAI_MODEL=claude-sonnet-4-20250514Base URL 填https://taotoken.net/api,不要加多余的路径后缀。模型 ID 根据你实际要用的填,比如claude-sonnet-4-20250514或gpt-4o。如果你不确定有哪些模型可用,可以到模型对话页面先试一下:
https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite
在对话页面选好模型发一条消息,确认能正常返回,再回到代码里配置。这一步能帮你排除掉 Key 无效或模型 ID 写错的问题。
环境准备好之后,目录结构建议这样组织:
mcp-demo/ ├── .env ├── weather_server.py # MCP 服务端 └── mcp_client.py # MCP 客户端服务端和客户端分开文件,方便后续把服务端独立部署。FastMCP 支持 stdio 和 SSE 两种传输方式,本地开发用 stdio 最简单,客户端直接以子进程方式启动服务端脚本。
3. 可复制的 FastMCP 服务端配置与工具注册
服务端的核心就是三件事:初始化 FastMCP 实例、用@mcp.tool()装饰器注册工具、启动服务。先看完整代码:
# weather_server.py from mcp.server.fastmcp import FastMCP mcp = FastMCP("WeatherServer") @mcp.tool() async def query_weather(city: str) -> str: """ 输入指定城市的英文名称,返回今日天气查询结果。 :param city: 城市名称(需使用英文) :return: 格式化后的天气信息 """ print(f"[Server] 收到查询请求: {city}") # 实际项目中这里替换为真实天气 API 调用 weather_data = { "beijing": "晴,气温 18-26 摄氏度,微风", "shanghai": "多云,气温 20-28 摄氏度,东南风 3 级", "tokyo": "小雨,气温 15-22 摄氏度", } result = weather_data.get(city.lower(), f"暂未收录 {city} 的天气数据") return f"{city} 今日天气:{result}" if __name__ == "__main__": print("启动 MCP 服务...") mcp.run(transport="stdio")这段代码里几个关键点值得展开。FastMCP("WeatherServer")里的名字是服务标识,客户端连接后能看到。@mcp.tool()装饰器做的事情比看起来多:它把函数名query_weather作为工具名,把 docstring 作为工具描述,把参数类型注解city: str自动转成 JSON Schema。模型看到的工具定义大概长这样:
{ "name": "query_weather", "description": "输入指定城市的英文名称,返回今日天气查询结果。", "input_schema": { "type": "object", "properties": { "city": {"type": "string", "description": "城市名称(需使用英文)"} }, "required": ["city"] } }docstring 写得越清楚,模型调用时参数填得越准。我踩过的坑是描述里没写「需使用英文」,结果模型传了中文城市名进来,查询直接落空。所以参数说明一定要写全。
如果你不想用 docstring,也可以直接在装饰器里指定描述:
@mcp.tool(description="根据城市英文名称获取实时天气,返回字符串格式的天气信息") async def query_weather(city: str) -> str: ...两种方式选一种即可,同时写的话装饰器参数优先级更高。启动服务:
python weather_server.pystdio 模式下服务端启动后不会输出太多信息,它等待客户端通过标准输入输出建立连接。你可以先单独跑一下确认没有语法错误,看到「启动 MCP 服务...」就说明服务端本身没问题。
4. 客户端调用验证:从工具注册到 Agent 成功执行
客户端要做的事情是:启动服务端子进程、建立 MCP 会话、拉取工具列表、把工具转成 OpenAI Function Calling 格式、让模型决定调用哪个工具、执行工具、把结果回传给模型生成最终回答。完整代码如下:
# mcp_client.py import asyncio import os import json import sys from typing import Optional from contextlib import AsyncExitStack from openai import OpenAI from dotenv import load_dotenv from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client load_dotenv() class MCPClient: def __init__(self): self.exit_stack = AsyncExitStack() self.api_key = os.getenv("OPENAI_API_KEY") self.base_url = os.getenv("OPENAI_BASE_URL", "https://taotoken.net/api") self.model = os.getenv("OPENAI_MODEL", "claude-sonnet-4-20250514") if not self.api_key: raise ValueError("未找到 OPENAI_API_KEY,请检查 .env 文件") self.client = OpenAI(api_key=self.api_key, base_url=self.base_url) self.session: Optional[ClientSession] = None async def connect_to_server(self, server_script_path: str): is_python = server_script_path.endswith(".py") is_js = server_script_path.endswith(".js") if not (is_python or is_js): raise ValueError("服务端脚本必须是 .py 或 .js 文件") command = sys.executable if is_python else "node" server_params = StdioServerParameters( command=command, args=[server_script_path], env=None, ) stdio_transport = await self.exit_stack.enter_async_context( stdio_client(server_params) ) self.stdio, self.write = stdio_transport self.session = await self.exit_stack.enter_async_context( ClientSession(self.stdio, self.write) ) await self.session.initialize() response = await self.session.list_tools() tools = response.tools print("\n已连接到 MCP 服务器,可用工具:", [t.name for t in tools]) async def process_query(self, query: str) -> str: messages = [{"role": "user", "content": query}] response = await self.session.list_tools() available_tools = [ { "type": "function", "function": { "name": tool.name, "description": tool.description, "parameters": tool.inputSchema, }, } for tool in response.tools ] response = self.client.chat.completions.create( model=self.model, messages=messages, tools=available_tools, ) choice = response.choices[0] if choice.finish_reason == "tool_calls": tool_call = choice.message.tool_calls[0] tool_name = tool_call.function.name tool_args = json.loads(tool_call.function.arguments) print(f"\n[调用工具] {tool_name} 参数: {tool_args}") result = await self.session.call_tool(tool_name, tool_args) messages.append(choice.message.model_dump()) messages.append({ "role": "tool", "content": result.content[0].text, "tool_call_id": tool_call.id, }) final = self.client.chat.completions.create( model=self.model, messages=messages, ) return final.choices[0].message.content return choice.message.content async def chat_loop(self): print("\nMCP 客户端已启动,输入 quit 退出") while True: try: query = input("\n你: ").strip() if query.lower() == "quit": break response = await self.process_query(query) print(f"\nAI: {response}") except Exception as e: print(f"\n发生错误: {str(e)}") async def cleanup(self): await self.exit_stack.aclose() async def main(): current_dir = os.path.dirname(os.path.abspath(__file__)) server_path = os.path.join(current_dir, "weather_server.py") if not os.path.exists(server_path): raise FileNotFoundError(f"找不到服务端脚本: {server_path}") client = MCPClient() try: await client.connect_to_server(server_path) await client.chat_loop() finally: await client.cleanup() if __name__ == "__main__": asyncio.run(main())运行客户端:
python mcp_client.py启动后你会看到「已连接到 MCP 服务器,可用工具: ['query_weather']」,说明工具注册成功。然后输入「北京今天天气怎么样」,客户端会打印出工具调用日志,最终返回类似「北京今日天气:晴,气温 18-26 摄氏度,微风」的结果。
这里有个细节要注意:tool.inputSchema直接作为parameters传给 OpenAI 接口,FastMCP 生成的 Schema 格式和 OpenAI 要求的格式是兼容的,不需要额外转换。如果你用的是其他模型接口,Schema 字段名可能不同,比如有些要求input_schema而不是parameters,按对应文档调整即可。
5. 常见报错排查:401、local proxy failed 与 choices 读取异常
跑通之后,实际部署时还会遇到几类典型报错,这里集中说一下排查思路。
401 认证失败。最常见的原因是.env文件没被正确加载,或者 Key 里带了多余空格。检查方式是在客户端初始化后打印一下self.api_key[:8],确认前缀正确。另外注意 Base URL 不要写成https://taotoken.net/api/v1,多出来的/v1会导致路径拼接错误。如果确认 Key 没问题还是 401,到控制台重新生成一个 Key 试试,旧 Key 可能已失效。
local proxy failed 或连接超时。这类报错通常出现在客户端启动服务端子进程的阶段。先确认weather_server.py能单独运行不报错,再检查sys.executable是否指向了正确的 Python 解释器。如果你在虚拟环境里跑客户端,但服务端脚本依赖装在了系统 Python 里,就会因为找不到mcp包而启动失败。解决办法是统一用同一个解释器,或者在StdioServerParameters里显式指定command为虚拟环境的 python 路径。
读取 choices 时报 IndexError 或 NoneType。这个错误一般发生在response.choices[0]这一行。原因可能是模型返回了空响应,或者接口返回格式和预期不符。排查时先把response完整打印出来看结构。如果用的是非 OpenAI 官方接口,有些实现会在choices为空时把内容放在其他字段里。另外finish_reason的判断也要注意,有些模型返回的是tool_calls,有些返回function_call,需要按实际返回调整判断逻辑。
工具调用后模型不生成最终回答。检查messages.append的顺序,tool角色的消息必须紧跟在带tool_calls的 assistant 消息之后,且tool_call_id要对应上。如果顺序错了,模型会认为工具结果和调用无关,直接忽略。
OAuth 或权限相关报错。如果你接入的是需要 OAuth 的服务,注意 token 过期时间。TaoToken 的 API Key 是长期有效的,不涉及 OAuth 刷新流程,配置好之后不用管过期问题。如果遇到权限不足的提示,到控制台确认 Key 的权限范围是否覆盖了你要调用的模型。
6. 从最小示例到生产可用:MCP 工具服务的扩展方向
上面这个天气查询示例只有单个工具,实际项目里你可能会注册十几个甚至几十个工具。FastMCP 支持在同一个服务里注册多个工具,客户端list_tools会一次性拉取全部。工具多了之后,建议按功能拆分多个 MCP 服务器,比如「数据库服务」「文件服务」「外部 API 服务」各一个,客户端按需连接。
传输方式上,stdio 适合本地开发,生产环境建议用 SSE 或 streamable HTTP,这样服务端可以独立部署,多个客户端共享。FastMCP 启动时把transport改成"sse"即可,客户端连接方式也要相应调整。
如果你打算长期做 Agent 开发,把工具层用 MCP 标准化之后,换模型、换框架都不用重写工具代码。模型接入方面,TaoToken 的 Coding Plan 适合需要频繁调用模型的编码场景:
https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite
接入文档里有各语言 SDK 的完整示例,遇到配置问题可以先翻文档:
https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite
最后留一个实用建议:工具描述里把参数格式、取值范围、示例都写清楚,模型调用准确率会明显提升。我现在的习惯是每个工具 docstring 至少写三行——功能说明、参数说明、返回值说明,看起来啰嗦,但省去了大量调试时间。