ARTICLE DETAIL

资讯详情

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

MCP 实战入门:用 TaoToken 统一 Key 跑通 agent 调用工具 demo

MCP 实战入门:用 TaoToken 统一 Key 跑通 agent 调用工具 demo 1. 从一次本地工具调用说起MCP 到底解决了什么问题如果你最近在折腾 agent大概率会遇到一个很具体的场景agent 需要查监控、读数据库、拉工单、检索知识库每接一个系统就得写一套适配代码。MCPModel Context Protocol想做的事情就是把这堆适配收敛成一套统一插座——工具方按 MCP 的方式暴露能力agent 侧按 MCP 的方式调用双方不用再为每个系统单独约定协议。我这次要跑通的 demo 很小一个 MCP Server 暴露get_monitor_metrics工具返回一份模拟的监控指标一个 agent 通过 MCP client 连上它拿到指标后交给大模型分析最后输出一段结论。整条链路会涉及两个关键点一是 agent 怎么调用工具二是 stdio 与 SSE/HTTP 两种传输方式在实际配置上差在哪。适合谁看已经会用 Python 写点脚本、想搞明白 MCP 调用链路、准备把本地工具接进 agent 的开发者。读完你能拿到可复制的 MCP server 配置片段、agent 侧调用代码以及一次端到端验证动作。整个过程不需要公网服务本地就能跑。先说结论stdio 适合本地单 agent 调用SSE/HTTP 适合远程多客户端复用。选哪种取决于你的工具跑在哪、有几个 agent 要用、要不要鉴权审计。下面按可跟做的顺序拆开。2. TaoToken 前置准备统一 Key 与 API 通道怎么配在写 MCP 代码之前先把模型调用这条链路准备好。agent 拿到监控指标后要交给大模型分析这一步需要一个稳定的 API 通道。我用 TaoToken 来统一管理 Key好处是后面不管换哪个模型agent 侧代码里的 Base URL 和 Key 都不用动。先拿到 Key。打开 https://taotoken.net/api-keys 登录后创建一个 API Key复制保存。注意这个 Key 只在创建时完整显示一次丢了就重新建一个。接着确认你要用的模型 ID。打开 https://taotoken.net/models 可以看到当前可用的模型列表把模型 ID 记下来比如常见的对话模型 ID。这个 ID 后面会写进 agent 代码。如果你打算长期跑编码类 agent可以看下 Coding Plan 页面 https://taotoken.net/coding-plan 它更适合高频调用场景。只是跑这个 demo 的话按量用 API 就够了。配置上agent 侧需要三个东西Base URL、API Key、Model ID。Base URL 用https://taotoken.net/api注意这里不加任何查询参数。Key 用刚才创建的那串Model ID 用你选定的那个。如果你用的是 Claude Code 这类工具配置会落在 settings 文件里。一个可复制的片段长这样{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_API_KEY: sk-你的Key, ANTHROPIC_MODEL: 你的模型ID } }如果你用的是 Codex 系工具配置落在auth.json里三件套同样是 Base URL、Key、Model ID{ base_url: https://taotoken.net/api, api_key: sk-你的Key, model: 你的模型ID }如果你用 Cline 或带 MCP 的编辑器插件MCP server 的配置通常写在mcp_settings.json或插件的 MCP 配置区格式类似{ mcpServers: { monitor: { command: python3, args: [mcp_server.py] } } }这里先记住一个原则Base URL、Key、Model ID 这三件套在哪个工具里都是核心缺一个就连不上。MCP server 的配置则是另一层负责工具怎么被拉起。两层不要混。3. 可复制配置stdio 与 SSE/HTTP 两种 MCP server 写法这一节直接给可复制的代码。先写 stdio 版的 MCP server它通过标准输入输出通信agent 启动它作为子进程。# mcp_server.py import json import sys from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent app Server(monitor-server) app.list_tools() async def list_tools(): return [ Tool( nameget_monitor_metrics, description获取当前系统监控指标, inputSchema{type: object, properties: {}} ) ] app.call_tool() async def call_tool(name: str, arguments: dict): if name get_monitor_metrics: metrics { timestamp: 1770000000, cpu_usage: 86.42, memory_usage: 91.33, pod_restart_count: 8, api_error_rate: 12.5, qps: 3200, latency_ms: 1800 } return [TextContent(typetext, textjson.dumps(metrics, ensure_asciiFalse))] raise ValueError(f未知工具: {name}) 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())这段代码的关键在最后stdio_server()把 server 挂到标准输入输出上agent 侧只要用StdioServerParameters指定commandpython3和args[mcp_server.py]就能拉起这个子进程。再看 SSE/HTTP 版。它把 server 跑成一个独立 Web 服务agent 通过 URL 访问# mcp_server_http.py import json from mcp.server import Server from mcp.server.sse import SseServerTransport from mcp.types import Tool, TextContent from starlette.applications import Starlette from starlette.routing import Route, Mount app Server(monitor-server-http) sse SseServerTransport(/messages/) app.list_tools() async def list_tools(): return [ Tool( nameget_monitor_metrics, description获取当前系统监控指标, inputSchema{type: object, properties: {}} ) ] app.call_tool() async def call_tool(name: str, arguments: dict): if name get_monitor_metrics: metrics {cpu_usage: 86.42, memory_usage: 91.33, pod_restart_count: 8} return [TextContent(typetext, textjson.dumps(metrics, ensure_asciiFalse))] raise ValueError(f未知工具: {name}) async def handle_sse(request): async with sse.connect_sse(request.scope, request.receive, request._send) as streams: await app.run(streams[0], streams[1], app.create_initialization_options()) starlette_app Starlette( routes[ Route(/sse, endpointhandle_sse), Mount(/messages/, appsse.handle_post_message), ] ) if __name__ __main__: import uvicorn uvicorn.run(starlette_app, host0.0.0.0, port8000)两种写法的差异在传输层stdio 版不需要端口agent 拉起子进程即可SSE/HTTP 版跑在 8000 端口agent 通过http://localhost:8000/sse连接。配置片段上stdio 用commandargsSSE/HTTP 用url{ mcpServers: { monitor-http: { url: http://localhost:8000/sse } } }对照表如下传输模式配置字段适合场景主要缺点stdiocommand args本地工具、单 agent不适合远程复用SSE/HTTPurl远程服务、团队共享运维与鉴权复杂度高WebSocketurl长连接实时双向交互连接管理复杂WebSocket 在 MCP 实践里不如前两者常见它更适合远程 IDE、实时状态订阅这类双向低延迟场景。这个 demo 不展开先把 stdio 和 SSE/HTTP 跑通。4. 验证请求agent 侧调用代码与端到端结果现在写 agent 侧代码通过 stdio 连接 MCP server调用工具再把结果交给大模型分析。# agent.py import asyncio import json import os from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from openai import OpenAI client OpenAI( base_urlhttps://taotoken.net/api, api_keyos.environ[TAOTOKEN_API_KEY] ) async def main(): server_params StdioServerParameters( commandpython3, args[mcp_server.py] ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() result await session.call_tool(get_monitor_metrics) metrics_text result.content[0].text print( * 50) print(Step1: 获取监控数据) print( * 50) print(metrics_text) prompt f以下是系统监控指标请分析异常并给出排查建议\n{metrics_text} resp client.chat.completions.create( model你的模型ID, messages[{role: user, content: prompt}] ) print( * 50) print(Step2: Agent 分析) print( * 50) print(resp.choices[0].message.content) if __name__ __main__: asyncio.run(main())运行前设置环境变量export TAOTOKEN_API_KEYsk-你的Key python3 agent.py实测下来输出会分两段。第一段是工具返回的原始指标 Step1: 获取监控数据 {timestamp: 1770000000, cpu_usage: 86.42, memory_usage: 91.33, pod_restart_count: 8, api_error_rate: 12.5, qps: 3200, latency_ms: 1800}第二段是大模型的分析结论 Step2: Agent 分析 当前系统存在异常风险。主要问题包括 1. CPU 使用率较高 2. 内存使用率较高 3. Pod 重启次数偏多 4. 接口错误率和延迟偏高 建议下一步优先查看 Pod 重启原因、应用错误日志、接口依赖服务状态以及最近发布记录。到这里stdio 链路就跑通了。如果你想验证 SSE/HTTP 版把 agent 里的stdio_client换成 SSE 连接方式指向http://localhost:8000/sse先启动python3 mcp_server_http.py再跑 agent。两条链路调用的工具和返回结构一致差别只在连接方式。5. 本篇常见错排查401、local proxy failed 与 reading choices跑这个 demo 时报错基本集中在两类模型调用失败和 MCP 连接失败。下面按真实报错对照排查。401 Unauthorized。这个最常见说明 Key 不对或没传。检查TAOTOKEN_API_KEY环境变量是否设置Key 是否复制完整。如果你把 Key 写死在代码里确认没有多余空格。Base URL 必须是https://taotoken.net/api多一个斜杠或少一个路径都可能出问题。local proxy failed。这个报错通常出现在网络层说明请求没到达目标地址。先确认 Base URL 拼写正确再确认本机网络能正常访问。如果你在容器里跑检查容器网络是否放通。这个报错和 Key 无关别急着换 Key。reading choices 相关报错。比如KeyError: choices或reading choices说明返回体结构和你预期的不一样。常见原因是模型 ID 写错或者请求根本没成功、返回的是错误对象。打印完整resp看结构确认model字段用的是有效模型 ID。MCP 连接失败。stdio 模式下如果报找不到mcp_server.py检查args里的路径是不是相对当前工作目录。SSE/HTTP 模式下如果连不上先确认 server 已经启动、端口没被占用再确认 agent 里的 URL 和 server 监听地址一致。工具调用返回空。检查call_tool里的工具名是否和list_tools里声明的一致大小写敏感。返回内容要包成TextContent直接返回 dict 会解析失败。一个实用技巧先在终端单独跑python3 mcp_server.py确认 server 能正常启动不报错再跑 agent。这样能把 server 侧和 agent 侧的问题分开定位。6. 继续往下走把工具接进你的真实链路demo 跑通后下一步通常是把真实工具接进来。本地文件检索、Git 操作、数据库查询这类继续用 stdio 就行配置简单、暴露面小。如果工具在远端、多个 agent 要复用就换成 SSE/HTTP把 server 部署成独立服务再通过网关加鉴权和审计。模型调用这条链路统一用 TaoToken 的 Key 和 Base URL换模型时只改 Model IDagent 代码不用动。需要看当前可用模型去 https://taotoken.net/models 要管理 Key去 https://taotoken.net/api-keys 接入细节看文档 https://taotoken.net/doc 。长期跑编码类 agent 的话Coding Plan 页面 https://taotoken.net/coding-plan 有更合适的方案。最后留一个我踩过的坑stdio 模式下 agent 退出子进程通常也会结束别指望它常驻。如果你需要 server 一直活着给多个 agent 用一开始就选 SSE/HTTP省得后面重构。
返回列表