1. 从 Manus 说起MCP 到底解决了什么问题Manus 这类通用 Agent 产品火起来之后很多人第一反应是它怎么什么都能干——能查资料、能操作浏览器、能读写文件、能调各种 API。拆开看它并不是把几百个工具硬编码进自己的代码里而是通过一套标准协议去发现和调用外部能力。这套协议就是 MCPModel Context Protocol一个让 AI 应用与外部数据源、工具之间用统一方式对话的开放协议。你可以把 MCP 理解成 AI 世界的 USB-C 接口。以前每接一个工具就要为这个工具写一套专属适配代码现在只要工具方按 MCP 规范暴露一个 Server任何支持 MCP 的 Client 都能即插即用。对开发者来说这意味着你写的 Agent 不用再为接高德地图接本地数据库接公司内部 API分别造轮子统一走 MCP 就行。MCP 采用客户端-服务端架构一个 Host宿主程序比如 IDE 插件、桌面助手可以同时连多个 MCP Server。核心角色分四层Host 是用户直接交互的程序Client 由 Host 创建与某个 Server 建立 1:1 连接并负责通信Server 是轻量级程序对外提供标准化的工具或数据访问能力再往下是本地数据源文件、数据库和远程服务各类 API。协议层负责消息封装、请求响应关联和通信模式管理传输层支持两种方式Stdio标准输入输出适合本地进程间通信和 HTTP SSE服务端用 SSE 推、客户端用 HTTP POST 发适合远程通信。所有传输都用 JSON-RPC 2.0 交换消息。消息类型有四种Request期望拿到响应、Result成功响应、Error错误响应带 code 和 message、Notification单向通知无需响应。连接建立类似三次握手Client 发 initialize 请求带协议版本和能力集Server 返回版本与能力信息Client 再发 initialized 通知确认之后进入正常通信阶段。终止则通过主动 close、传输层断开或错误触发。理解了这套原理你就能明白为什么 Manus 能快速扩展能力边界——它本质上是一个 MCP Host把各种 Server 挂上来AI 负责理解用户意图、决定调哪个工具、传什么参数Client 负责把调用真正发出去、把结果拿回来。接下来我们就动手在本地搭一个自己的 MCP Client并用 TaoToken 统一 Key 接入模型跑通一次完整的AI 决定调工具 → Client 执行 → 结果回传 → AI 总结闭环。2. 前置准备TaoToken 统一 Key 与 MCP Client 环境搭建在写代码之前先把模型从哪来这件事解决掉。MCP Client 本身只负责协议通信和工具调度真正做意图理解、决定调不调工具、生成最终回复的是背后的大模型。所以你需要一个能稳定调用模型的入口。TaoToken 提供统一的 API Key兼容 OpenAI 风格的接口Base URL 固定为https://taotoken.net/api这样你的 Client 代码里base_url和api_key两行配置就能搞定不用为不同模型改来改去。先去控制台创建一个 Key。打开 https://taotoken.net/api-keys 登录后新建一个 API Key复制保存好——它只会完整显示一次。这个 Key 就是你后面所有请求的凭证。如果你还没注册从 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 进入即可。环境方面我建议用 Python 3.10因为 MCP 官方 Python SDK 对异步支持比较完整。你需要装这几个包pip install mcp openai python-dotenvmcp是官方协议 SDKopenai用来调模型TaoToken 兼容 OpenAI 接口python-dotenv用来管理环境变量。装完后在项目根目录建一个.env文件把 Key 和 Base URL 写进去OPENAI_API_KEYsk-你的TaoToken密钥 OPENAI_BASE_URLhttps://taotoken.net/api OPENAI_MODELclaude-sonnet-4-20250514这里OPENAI_MODEL填你在 TaoToken 上开通的模型 ID。模型 ID 可以在模型对话页面确认打开 https://taotoken.net/models 能看到可用模型列表和对应的调用名。注意 Base URL 后面不要加/v1SDK 会自己拼路径加了反而容易 404。接下来准备一个 MCP Server 作为被调用的工具。为了让你能立刻跑通我写一个最简单的本地 Server它只暴露一个get_offers工具接收keywords和pageSize两个参数返回模拟的商品数据。真实场景里你可以把它换成查数据库、调内部 API 的 Server协议部分完全一样。# server.py from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent app Server(demo-offers-server) app.list_tools() async def list_tools(): return [ Tool( nameget_offers, descriptionGet product offers from API, inputSchema{ type: object, properties: { keywords: {type: string, description: Keywords to search for products, default: }, pageSize: {type: number, description: Number of items per page, minimum: 1, maximum: 100, default: 10} } } ) ] app.call_tool() async def call_tool(name: str, arguments: dict): if name ! get_offers: raise ValueError(fUnknown tool: {name}) keywords arguments.get(keywords, ) page_size arguments.get(pageSize, 10) offers [ {id: fid-{i}, name: f{keywords} 商品 {i}, price: round(10 i * 3.5, 2), supplier: f供应商 {i}} for i in range(1, min(page_size, 5) 1) ] return [TextContent(typetext, textstr(offers))] async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options()) if __name__ __main__: import asyncio asyncio.run(main())这个 Server 用 Stdio 传输启动后监听标准输入输出等待 Client 发 JSON-RPC 消息。它的list_tools返回工具清单call_tool处理实际调用。真实项目里把offers那段换成你的业务逻辑即可。3. 可复制配置MCP Client 完整代码与关键参数现在写 Client。它的职责是启动时连上 Server、拉取工具列表、把工具转成模型能理解的 function 格式、接收用户输入、让模型决定是否调工具、执行工具调用、把结果回传给模型、拿到最终回复。下面这份代码可以直接复制运行我把它拆成几个关键部分讲。# client.py import asyncio import json import os import sys import traceback from contextlib import AsyncExitStack from typing import Optional from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from openai import OpenAI from dotenv import load_dotenv load_dotenv() class MCPClient: def __init__(self): self.session: Optional[ClientSession] None self.exit_stack AsyncExitStack() self.client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), ) self.model os.getenv(OPENAI_MODEL) self.messages [ { role: system, content: You are a versatile assistant capable of answering questions, completing tasks, and intelligently invoking specialized tools to deliver optimal results. } ] self.available_tools [] 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(Server script must be a .py or .js file) command python if is_python else node server_params StdioServerParameters( commandcommand, args[server_script_path], envNone, ) 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(\nConnected to server with tools:, [tool.name for tool in tools]) async def process_query(self, query: str) - str: self.messages.append({role: user, content: query}) if not self.available_tools: response await self.session.list_tools() self.available_tools [ { type: function, function: { name: tool.name, description: tool.description, parameters: tool.inputSchema, }, } for tool in response.tools ] current_response self.client.chat.completions.create( modelself.model, messagesself.messages, toolsself.available_tools, streamFalse, ) if current_response.choices[0].message.content: print(\nAI:, current_response.choices[0].message.content) while current_response.choices[0].message.tool_calls: for tool_call in current_response.choices[0].message.tool_calls: tool_name tool_call.function.name try: tool_args json.loads(tool_call.function.arguments) except json.JSONDecodeError: tool_args {} print(f\n调用工具 {tool_name}) print(f参数: {tool_args}) result await self.session.call_tool(tool_name, tool_args) print(f\n工具结果: {result}) self.messages.append(current_response.choices[0].message) self.messages.append({ role: tool, tool_call_id: tool_call.id, content: json.dumps(result.content, defaultstr), }) current_response self.client.chat.completions.create( modelself.model, messagesself.messages, toolsself.available_tools, streamFalse, ) self.messages.append(current_response.choices[0].message) return current_response.choices[0].message.content or async def chat_loop(self): print(\nMCP Client Started!) print(Type your queries or quit to exit.) while True: try: query input(\nCommand: ).strip() if query.lower() quit: break response await self.process_query(query) print(\nAI: response) except Exception as e: print(f\nError occurs: {e}) traceback.print_exc() async def cleanup(self): await self.exit_stack.aclose() async def main(): if len(sys.argv) 2: print(Usage: python client.py path_to_server_script) sys.exit(1) client MCPClient() try: await client.connect_to_server(sys.argv[1]) await client.chat_loop() finally: await client.cleanup() if __name__ __main__: asyncio.run(main())几个关键点值得单独说。第一connect_to_server里用AsyncExitStack管理生命周期确保退出时连接被正确关闭避免进程残留。第二available_tools的格式转换是核心——MCP 的inputSchema本身就是 JSON Schema直接塞进 OpenAI 的parameters字段即可不用手动改结构。第三process_query里的while循环处理多轮工具调用模型可能一次返回多个tool_calls也可能调完一个工具后根据结果决定再调下一个循环直到模型不再要求调工具为止。第四工具结果回传时role必须是tool并且带上tool_call_id与请求对应否则模型无法把结果和调用关联起来。如果你更习惯 TypeScript官方也有modelcontextprotocol/sdkClient 侧用StdioClientTransport建立连接client.listTools()拿工具client.callTool()执行调用逻辑和 Python 版一一对应。核心配置同样是三件套Base URL 填https://taotoken.net/api、API Key 填 TaoToken 密钥、Model ID 填你开通的模型名。4. 验证请求跑通一次完整的工具调用闭环代码写好了现在跑起来验证。先启动 Client把 Server 脚本路径作为参数传进去python client.py server.py如果连接成功你会看到Connected to server with tools: [get_offers] MCP Client Started! Type your queries or quit to exit.先输入一句普通对话测试模型通路比如你好。这时模型不会调工具直接返回文本Command: 你好 AI: 你好有什么可以帮你的吗这一步验证的是 TaoToken 的 Key 和 Base URL 配置正确、模型能正常响应。如果这里就报错先别往下走去看第 5 节的排查。接着输入需要调工具的请求帮我找一些手表。这时会发生完整的工具调用链。Client 把用户消息和工具清单一起发给模型模型识别出意图返回一个tool_calls里面指定调用get_offers参数是{keywords: 手表, pageSize: 10}。Client 解析出工具名和参数通过 MCP 协议调用 Server 的call_toolServer 返回商品数据Client 把结果以role: tool的消息追加进对话历史再发给模型。模型拿到数据后生成最终回复Command: 帮我找一些手表 调用工具 get_offers 参数: {keywords: 手表, pageSize: 10} 工具结果: metaNone content[TextContent(typetext, text[{id: id-1, name: 手表 商品 1, price: 13.5, supplier: 供应商 1}, ...])] isErrorFalse AI: 根据您的搜索这里有几款手表供您参考 1. 手表 商品 1价格 13.5供应商 供应商 1 2. 手表 商品 2价格 17.0供应商 供应商 2 ...看到这个输出说明整条链路通了TaoToken 提供模型能力 → 模型理解意图并决定调工具 → MCP Client 执行调用 → MCP Server 返回数据 → 模型总结回复。这就是 Manus 类 Agent 最核心的工作循环。如果你想验证多轮工具调用可以输入一个需要连续调用的请求比如先找手表再根据结果帮我筛选价格低于 20 的。模型可能会先调一次get_offers拿到结果后再决定是否需要二次调用或直接筛选。观察日志里调用工具出现几次就能确认多轮循环是否生效。实测下来最容易出问题的不是协议本身而是模型返回的tool_calls结构不符合预期。有些模型在参数里返回的是字符串化的 JSON需要多一层json.loads有些模型干脆不返回标准tool_calls而是把调用意图写在content里。前者可以在解析时加容错后者只能换模型或加提示词约束。这也是为什么第 2 节强调模型 ID 要选对——工具调用能力强的模型整个闭环才稳。5. 常见报错排查401、local proxy failed 与 choices 解析异常跑不通的时候报错信息往往很直接但原因可能藏在配置里。下面这几个是我和身边开发者最常撞到的按出现频率排。401 Unauthorized。这个几乎都是 Key 的问题。先确认.env里OPENAI_API_KEY填的是 TaoToken 控制台创建的 Key没有多余空格或换行。然后确认OPENAI_BASE_URL是https://taotoken.net/api结尾没有斜杠、没有/v1。如果 Key 是对的还报 401去控制台看这个 Key 是否被禁用、额度是否用完。还有一种情况是环境变量没加载成功——load_dotenv()要在创建OpenAI客户端之前调用顺序反了就读不到。local proxy failed / connection refused。这个报错通常出现在 Client 启动阶段连不上 Server。检查server.py路径是否正确、Python 环境里mcp包是否装好。如果 Server 脚本本身有语法错误进程会直接退出Client 侧看到的就是连接失败。可以单独跑一下python server.py看它能不能正常启动不报错。另外 Windows 上如果python命令指向了 Microsoft Store 的占位程序也会导致启动失败换成python3或完整路径试试。reading choices 相关报错比如NoneType object has no attribute choices或list index out of range。这说明模型返回的响应结构和你代码里取的不一致。常见原因是模型没返回标准 ChatCompletion 格式或者请求本身失败了但没抛异常。在process_query里加一层判断if not current_response or not current_response.choices: raise RuntimeError(f模型返回异常: {current_response})这样能把问题定位到模型调用层而不是在后面解析tool_calls时才崩。如果确认是模型不支持工具调用换一个支持 function calling 的模型 ID 即可。OAuth / 认证跳转类报错。如果你用的是某些需要 OAuth 授权的 MCP Server比如接第三方云服务Server 启动时会要求浏览器授权。本地调试阶段建议先用 Stdio 的简单 Server 跑通协议确认 Client 逻辑没问题再去接需要授权的远程 Server。远程 Server 走 HTTP SSE 传输配置里要填 Server 的 URL 而不是脚本路径别混用。工具调用了但结果没回传。表现是日志里看到调用工具但模型最终回复里没有用到工具数据。检查回传消息的tool_call_id是否和请求里的id完全一致role是否是tool。这两个字段错一个模型就认为工具结果和调用无关直接忽略。另外content字段建议用json.dumps序列化直接塞 Python 对象有些模型解析不了。参数解析失败。模型返回的arguments有时是空字符串、有时是双重转义的 JSON。代码里用try/except json.JSONDecodeError兜底成空字典能避免直接崩溃但工具拿不到参数会返回空结果。更稳的做法是在系统提示词里明确要求调用工具时 arguments 必须是合法 JSON并在解析失败时把原始字符串打出来看看到底长什么样。排查的核心思路是分层定位先确认模型通路普通对话能不能回再确认 Server 通路单独跑 Server 能不能起最后确认协议层工具清单有没有拉到、调用有没有发出、结果有没有回传。每一层都有对应的日志别一上来就怀疑协议本身。6. 从跑通到用好MCP Client 的下一步跑通一次调用只是起点。真正把它用起来有几个方向可以继续。一是把 Server 换成你实际需要的工具——查公司数据库、调内部 API、操作本地文件协议部分不用改只换call_tool里的业务逻辑。二是把 Client 封装成常驻服务配合 Coding Plan 这类长期编码场景让 Agent 在后台持续调用工具链而不是每次手动启动。三是处理多 Server 场景一个 Host 可以连多个 Client每个 Client 对应一个 Server工具清单合并后一起给模型模型会根据描述自动选合适的工具。如果你想把模型调用也统一管理TaoToken 的 Coding Plan 适合长期编码和 Agent 场景模型对话页面适合快速验证某个模型是否支持工具调用接入文档里有各语言的完整示例。配置上始终记住三件套Base URL 用https://taotoken.net/apiKey 用控制台创建的密钥Model ID 用你开通的模型名。这三样对齐了剩下的就是业务逻辑的事。最后留一个实用技巧调试阶段把每次请求的messages完整打印出来尤其是工具调用前后的消息序列。你会清楚看到模型是怎么看到工具清单、怎么决定调用、结果又是怎么喂回去的。看几次之后MCP 就不再是抽象协议而是一条你能完全掌控的数据流。