ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

AI王炸:MCP服务端客户端的完整实现与TaoToken统一接入

AI王炸:MCP服务端客户端的完整实现与TaoToken统一接入 1. 从零理解 MCP服务端与客户端到底在解决什么问题MCP 全称 Model Context Protocol你可以把它理解成 AI 世界里的 USB-C 接口。以前每个 AI 工具想调用外部能力都得自己写一套对接逻辑查天气写一套、查数据库写一套、发邮件再写一套模型厂商和工具开发者各写各的重复劳动特别多。MCP 做的事情就是把这层对接标准化——服务端按协议暴露能力客户端按协议发现并调用能力双方不用互相认识只要都遵守同一份协议就能握手。它适合谁如果你正在给 AI 工具搭建标准化的工具调用通道比如让本地大模型能查内部数据库、让 IDE 里的编码助手能读 GitLab MR、让聊天客户端能调你自己的业务 API那 MCP 就是当前最省心的路径。核心概念只有三个Resources 是结构化数据比如文件内容、API 响应Tools 是可执行函数比如查询数据库、发送邮件Prompts 是预设的交互模板。你不需要一次全用上大多数落地场景先把 Tools 跑通就够了。很多人会问 MCP 和 Function Calling 有什么区别。简单说Function Calling 是模型层面的能力你告诉模型有哪些函数模型决定调哪个MCP 是工程层面的协议它管的是这些函数怎么被注册、被发现、被跨进程调用。两者不冲突实际项目里经常是 MCP 服务端暴露 Tools客户端把 Tools 转成 Function Calling 的格式喂给模型。理解这一点后面的代码你就能看懂每一层在干什么。这一篇的目标很明确写一个能跑的 MCP 服务端写一个能连上它的客户端然后通过 TaoToken 统一 Key 和 API 通道完成鉴权与调用端到端一次跑通。下面每一步都有可复制的命令和配置你跟着做就行。2. TaoToken 前置准备统一 Key 与 API 通道怎么配在写代码之前先把鉴权通道准备好。MCP 客户端最终要调用大模型来决定调哪个工具这一步需要一个稳定的 API 入口。TaoToken 提供统一的 Key 和 API 通道你只需要在一个地方管理凭证不用为每个模型厂商单独配一套。第一步打开官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册并登录。登录后进入控制台地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 。在控制台里你能看到账户概览和用量情况。第二步创建 API Key。进入 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 点新建复制生成的 Key。这个 Key 就是后面客户端里api_key字段要填的值。注意 Key 只在创建时完整显示一次先存到安全的地方。第三步确认 Base URL。TaoToken 的 API 入口是 https://taotoken.net/api 注意这个地址不带任何查询参数。客户端里base_url填这个值后面拼上/v1就是标准的 OpenAI 兼容路径。如果你不确定该用哪个模型可以先到模型对话页面 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 试一下确认通道通了再写代码。对于长期做编码和 Agent 的场景可以了解 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 它更适合高频调用。这里有个关键点MCP 客户端调用大模型时base_url指向 TaoToken 的 API 入口api_key用你刚创建的 Keymodel填你要用的模型 ID。这三件套配齐鉴权就完成了。下面进入代码环节。3. 可复制配置MCP 服务端注册与客户端连接先写服务端。用 Python 为例安装依赖pip install mcp pip install mcp[cli] pip install httpx0.27如果你用 uv 管理环境也可以直接uv add mcp httpx0.27。注意 httpx 锁 0.27 是为了避开部分版本和 mcp 的兼容问题这个坑后面排障会讲。服务端代码server.py核心思路是把普通 Python 函数加上mcp.tool()注解就变成了 MCP 可发现的工具from mcp.server.fastmcp import FastMCP mcp FastMCP(demo-tools) mcp.tool() def filter_by_rate(rate: str) - list: 根据评级条件筛选标的 print(评级过滤条件, rate) return [标的A, 标的B, 标的C] mcp.tool() def filter_by_type(t: str) - list: 根据类型条件筛选标的 print(类型过滤条件, t) return [标的A, 标的C, 标的D] mcp.tool() def filter_by_range(low: float, high: float) - list: 根据区间条件筛选标的 print(区间过滤条件, low, high) return [标的A2, 标的C2, 标的D2] if __name__ __main__: mcp.run(transportsse)启动服务端mcp dev server.py或者直接python server.pySSE 模式默认监听本地端口客户端连http://localhost:8000/sse。客户端这边关键是把 MCP 的 Tools 转成 OpenAI 兼容的 Function Calling 格式再通过 TaoToken 通道发给模型。配置文件settings.json里把三件套写清楚{ mcpServers: { demo-tools: { url: http://localhost:8000/sse, transport: sse } }, llm: { base_url: https://taotoken.net/api/v1, api_key: 你的TaoToken Key, model: 你的模型ID } }客户端主逻辑client.pyimport asyncio import json from contextlib import AsyncExitStack from mcp.client.sse import sse_client from mcp import ClientSession from openai import AsyncOpenAI class MCPClient: def __init__(self): self.session None self.exit_stack AsyncExitStack() self.client AsyncOpenAI( api_key你的TaoToken Key, base_urlhttps://taotoken.net/api/v1 ) async def connect_to_sse_server(self, server_url: str): streams await self.exit_stack.enter_async_context( sse_client(urlserver_url) ) self.session await self.exit_stack.enter_async_context( ClientSession(*streams) ) await self.session.initialize() response await self.session.list_tools() print(已连接可用工具, [t.name for t in response.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] resp await self.client.chat.completions.create( model你的模型ID, messagesmessages, toolsavailable_tools ) message resp.choices[0].message final_text [message.content or ] if message.tool_calls: for tool_call in message.tool_calls: tool_name tool_call.function.name tool_args json.loads(tool_call.function.arguments) result await self.session.call_tool(tool_name, tool_args) final_text.append(f[调用 {tool_name} 参数 {tool_args}]) messages.append({ role: assistant, tool_calls: [{ id: tool_call.id, type: function, function: { name: tool_name, arguments: json.dumps(tool_args) } }] }) messages.append({ role: tool, tool_call_id: tool_call.id, content: str(result.content) }) resp await self.client.chat.completions.create( model你的模型ID, messagesmessages, toolsavailable_tools ) if resp.choices[0].message.content: final_text.append(resp.choices[0].message.content) return \n.join(final_text) async def chat_loop(self): print(MCP 客户端已启动输入 quit 退出) while True: query input(\nQuery: ).strip() if query.lower() quit: break print(\n await self.process_query(query)) async def cleanup(self): await self.exit_stack.aclose() async def main(): client MCPClient() try: await client.connect_to_sse_server(http://localhost:8000/sse) await client.chat_loop() finally: await client.cleanup() if __name__ __main__: asyncio.run(main())如果你用的是 Claude Code 这类工具配置方式类似把 Base URL、Key、Model ID 三件套填进对应的 settings 文件即可。Claude Code 的接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有完整的字段说明。4. 验证请求端到端跑通与成功结果确认配置写完先确认服务端在跑。开一个终端执行python server.py看到监听日志后另开终端跑客户端python client.py。如果连接成功你会看到类似输出已连接可用工具 [filter_by_rate, filter_by_type, filter_by_range] MCP 客户端已启动输入 quit 退出然后输入一个会触发工具调用的查询比如「帮我按评级 A 筛选一下」。正常流程是客户端把工具列表发给模型模型返回 tool_calls客户端执行session.call_tool把结果回填后再问一次模型最后打印自然语言回答。你会看到类似[调用 filter_by_rate 参数 {rate: A}] 根据评级 A 筛选返回标的标的A、标的B、标的C同时服务端终端会打印评级过滤条件 A说明调用真的打到了服务端函数里。这一步能跑通端到端链路就成立了。再验证一下 TaoToken 通道本身。你可以单独发一个最小请求确认 Key 和 Base URL 没问题curl https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer 你的TaoToken Key \ -H Content-Type: application/json \ -d {model:你的模型ID,messages:[{role:user,content:ping}]}返回里有choices字段就说明通道正常。如果这一步就报错先别怀疑 MCP 代码问题在鉴权配置上。实测下来最容易出问题的不是协议本身而是环境细节。比如服务端用 SSE 模式时端口被占用、客户端连的 URL 少了/sse后缀、模型 ID 填错导致返回空 tool_calls。建议每改一处配置就单独验证一次别一次性全改完再排查。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth第一个高频错误是 401。报错长这样openai.AuthenticationError: Error code: 401 - {error: {message: Invalid API key}}原因通常是 Key 复制时带了空格或者把官网地址误填进了base_url。检查两点api_key是不是完整的 TaoToken Keybase_url是不是https://taotoken.net/api/v1。注意 API 入口不带查询参数别把带 UTM 的官网地址填进去。第二个是local proxy failed或连接被拒httpx.ConnectError: [Errno 111] Connection refused这基本是服务端没起来或者客户端连的端口不对。先确认python server.py还在前台跑着再确认客户端 URL 是http://localhost:8000/sse。如果你改了服务端端口客户端要同步改。第三个是reading choices相关报错典型信息是KeyError: choices或response.choices为空。这通常意味着返回体不是标准 OpenAI 格式可能是模型 ID 写错、通道返回了错误页或者请求体里tools字段格式不对。先用第 4 节的 curl 验证通道再检查available_tools的构造是否和tool.inputSchema一致。第四个是 OAuth 相关报错出现在某些客户端工具首次连接远程 MCP 服务时OAuth flow required / invalid_client如果你连的是本地 SSE 服务一般不会触发 OAuth如果连的是需要授权的远程服务按客户端提示走授权流程即可。本地开发阶段建议先用无鉴权的 SSE 服务把链路跑通再叠加授权层。还有一个隐蔽的坑httpx版本过高导致 mcp 客户端握手异常。如果你遇到RuntimeError: Attempted to use a closed client之类的问题把 httpx 降到 0.27 再试。这个在依赖安装那一步已经锁好了但如果你环境里已有其他版本记得确认一下。排障时记住一个原则先分层再定位。通道层用 curl 验服务端层用mcp dev验客户端层单独打印list_tools结果。三层各自通了端到端自然通。6. 把 MCP 通道接进你的日常工具链链路跑通之后真正省事的地方在于复用。服务端你只需要维护一份工具注册代码客户端可以换成任何支持 MCP 的工具。比如你在 Cherry Studio 里加一个 MCP Server填上 SSE 地址问答时选中它就能用在编码工具里配置 MCP把 Base URL、Key、Model ID 三件套填进 settings就能让助手直接调你的内部工具。如果你要长期跑编码和 Agent 任务建议把 Key 管理集中到 TaoToken 控制台用量和额度在一个地方看不用在多个厂商后台之间切换。模型对话页面适合快速验证某个模型能不能正确触发工具调用接入文档里有各客户端的字段对照表配置时对着填不容易错。最后留一个实用习惯每次新增一个 MCP 工具先在服务端单独print一下入参确认参数解析没问题再接到客户端让模型去调。模型返回的tool_calls参数是 JSON 字符串字段名和类型必须和inputSchema严格对应差一个下划线都会导致调用失败。这个细节踩过一次后面就会顺手很多。
返回列表