大模型LLM通过python调用mcp服务代码示例:TaoToken统一Key接入与config.toml骨架
1. 为什么 Python 调 MCP 总卡在“连不上”这一步如果你正在用 Python 写 LLM 应用并且想让模型自己去调用外部工具那 MCPModel Context Protocol大概率已经出现在你的技术选型里了。它的核心价值很直接把“工具”从你的业务代码里解耦出去变成一个标准化的服务模型通过协议去发现工具、调用工具、拿回结果。网页抓取、文件读取、数据库查询、内部 API 调用都可以包装成 MCP 服务让 LLM 按需触发。但真正动手时很多人会卡在几个很具体的地方。第一是入口不统一LLM 的 API Key、Base URL、模型名散落在各个脚本里换一个模型就要改一遍代码。第二是 MCP 服务的启动方式和 Python 客户端对不上stdio 和 SSE 两种传输模式经常搞混。第三是配置文件没有骨架config.toml和settings.json到底该写哪些字段网上示例五花八门抄过来跑不通。这篇就聚焦一条可落地的路径用 TaoToken 作为统一的 Key 和 API 通道Python 侧通过 MCP 客户端连接一个本地 SSE 服务让 LLM 决策调用工具并整合结果。我会给出config.toml和settings.json的可复制骨架、依赖清单、最小调用脚本以及一次连通性验证动作。适合已经会写 Python、想快速跑通 MCP 调用链路的开发者。整套流程实测下来从零到跑通大概十几分钟。2. TaoToken 前置统一 Key 与 API 通道怎么准备在写代码之前先把“入口”这件事解决掉。MCP 调用链路里LLM 这一侧需要一个兼容 OpenAI 接口的客户端而 TaoToken 提供的就是这个统一通道一个 Key、一个 Base URL后面换模型只改模型名不用动客户端初始化逻辑。你需要先拿到 API Key。进入控制台后创建密钥建议按项目命名方便后面排查是哪个应用在调用。拿到 Key 之后记住两个地址官网是https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentAPI 基础地址是https://taotoken.net/api。注意 API 地址后面不加 UTM 参数直接用于base_url配置。这里有个容易踩的坑很多人把base_url写成官网地址结果请求 404。OpenAI 兼容客户端的base_url要指向 API 根路径也就是https://taotoken.net/apiSDK 会自动拼接/v1/chat/completions这类路径。如果你用的是AsyncOpenAI初始化时把api_key和base_url一起传进去就行。另外MCP 服务本身是独立进程它不依赖 TaoToken 的 Key。Key 只用于 LLM 决策那一层。所以你的架构其实是两段Python 客户端通过 SSE 连本地 MCP 服务同时通过 TaoToken 调 LLM。两段解耦排查问题时可以分别验证。3. 可复制配置config.toml 与 settings.json 骨架先把配置文件搭好后面脚本直接读避免硬编码。我习惯把 MCP 服务定义放在config.toml把运行时参数放在settings.json两者职责分开。config.toml负责描述“有哪些 MCP 服务、怎么启动”。以网页抓取服务为例# config.toml [mcp] # 默认使用的服务名 default_server fetch [mcp.servers.fetch] # 传输方式stdio 或 sse transport sse # SSE 服务地址由 supergateway 暴露 url http://localhost:8000/sse # 启动命令供手动拉起服务时参考 command npx args [-y, supergateway, --stdio, uvx mcp-server-fetch] [mcp.servers.fetch.env] # 如需代理或超时参数在这里补充 MCP_TIMEOUT 30000settings.json负责“LLM 怎么调、用哪个模型”{ llm: { api_key: sk-你的TaoToken密钥, base_url: https://taotoken.net/api, model: deepseek-ai/DeepSeek-V3, temperature: 0.2, max_tokens: 2048 }, mcp: { connect_timeout: 15, tool_call_timeout: 60 }, logging: { level: INFO } }依赖清单用requirements.txt固定住避免版本漂移openai1.30.0 mcp1.0.0 tomli2.0.0Python 3.11 以上自带tomllib如果你用 3.10 或更低装tomli并在代码里做兼容导入。MCP 的 Python SDK 包名就是mcp里面包含ClientSession和sse_client。openai用新版异步客户端旧版openai.AsyncOpenAI的接口差异较大建议直接升到 1.30 以上。4. 最小调用脚本从连接 SSE 到工具结果整合配置就绪后写一个最小可跑的脚本。整体分四步连接 MCP 服务、拉取工具列表、让 LLM 决策、执行工具并整合结果。下面这份代码可以直接复制改一下 Key 和查询内容就能跑。# mcp_agent.py import asyncio import json import os from contextlib import AsyncExitStack from typing import Optional from openai import AsyncOpenAI from mcp import ClientSession from mcp.client.sse import sse_client try: import tomllib except ImportError: import tomli as tomllib class MCPAgent: def __init__(self, config_path: str config.toml, settings_path: str settings.json): self.exit_stack AsyncExitStack() self.session: Optional[ClientSession] None with open(config_path, rb) as f: self.config tomllib.load(f) with open(settings_path, r, encodingutf-8) as f: self.settings json.load(f) llm_cfg self.settings[llm] self.llm AsyncOpenAI( api_keyllm_cfg[api_key], base_urlllm_cfg[base_url], ) self.model llm_cfg[model] async def connect_server(self): server_name self.config[mcp][default_server] server self.config[mcp][servers][server_name] sse_endpoint server[url] transport await self.exit_stack.enter_async_context( sse_client(sse_endpoint) ) self.session await self.exit_stack.enter_async_context( ClientSession(*transport) ) await self.session.initialize() tools (await self.session.list_tools()).tools print(可用工具:, [t.name for t in tools]) return tools async def execute_workflow(self, query: str) - str: tools [ { type: function, function: { name: t.name, description: t.description, parameters: t.inputSchema, }, } for t in (await self.session.list_tools()).tools ] chat_completion await self.llm.chat.completions.create( modelself.model, messages[{role: user, content: query}], toolstools, tool_choiceauto, ) message chat_completion.choices[0].message if not message.tool_calls: return message.content results [] for tool_call in message.tool_calls: args json.loads(tool_call.function.arguments) print(f调用工具: {tool_call.function.name}, 参数: {args}) result await self.session.call_tool( tool_call.function.name, args ) text result.content[0].text print(工具返回:, text[:200], ...) results.append(text) final_response await self.llm.chat.completions.create( modelself.model, messages[ { role: user, content: f原始问题{query}\n工具结果{json.dumps(results, ensure_asciiFalse)}, } ], ) return final_response.choices[0].message.content async def close(self): await self.exit_stack.aclose() async def main(): agent MCPAgent() try: await agent.connect_server() response await agent.execute_workflow( 抓取 https://taotoken.net 页面总结三个关键点 ) print(f\n最终响应{response}) finally: await agent.close() if __name__ __main__: asyncio.run(main())脚本里有两个细节值得说。第一sse_client返回的是一个元组直接解包传给ClientSession不要自己拆。第二tool_call.function.arguments是 JSON 字符串必须json.loads之后再传给call_tool否则 MCP 服务会报参数格式错误。这两处是我踩过的坑网上不少示例省略了。5. 验证请求一次连通性检查与成功结果代码写完后先别急着跑完整流程分两步验证更稳。第一步单独拉起 MCP 服务。在终端执行npx -y supergateway --stdio uvx mcp-server-fetch如果看到类似SSE server listening on http://localhost:8000/sse的输出说明服务已经起来了。这一步依赖 Node 环境和uvx如果uvx找不到先装uv工具链。服务起来后不要关终端另开一个窗口跑 Python。第二步跑脚本。正常输出会分几段先打印可用工具列表比如[fetch]然后打印 LLM 决定调用的工具名和参数接着是工具返回内容的前 200 个字符最后是整合后的最终响应。如果最终响应里包含了对目标页面的总结说明整条链路通了。验证时可以用一个更简单的查询比如“抓取 https://taotoken.net 并返回页面标题”减少 LLM 整合的复杂度先确认工具调用本身没问题。等这一步稳定了再换成需要多步推理的查询。如果你只想先验证 LLM 通道是否正常可以跳过 MCP直接用模型对话页面发一条消息确认 Key 和模型名没问题。这一步能快速排除是 Key 配置问题还是 MCP 连接问题。6. 本篇常见错排查连接、参数与模型名跑不通的时候按下面几个方向排查基本能覆盖大部分情况。连接被拒绝报Connection refused或ConnectError通常是 MCP 服务没起来或者端口不是 8000。检查supergateway的输出确认 SSE 地址和config.toml里的url一致。如果端口被占用换一个端口并在两处同步修改。工具列表为空list_tools返回空数组说明 MCP 服务启动了但没注册工具。检查uvx mcp-server-fetch是否能单独运行有些环境需要先uvx install再执行。参数解析失败报JSONDecodeError或 MCP 返回参数错误多半是arguments没做json.loads或者 LLM 生成的参数结构和inputSchema不匹配。可以在调用前打印args确认。模型名报错如果返回model not found检查settings.json里的model字段。TaoToken 通道下模型名要写完整比如deepseek-ai/DeepSeek-V3不要简写成deepseek。换模型时只改这一个字段客户端不用动。超时抓取大页面时容易超时把settings.json里的tool_call_timeout调大或者在 MCP 服务侧设置更长的超时参数。排查顺序建议从下往上先确认 LLM 通道能通再确认 MCP 服务能连最后看工具调用参数。这样定位最快。7. 下一步把 Key 和接入文档用起来链路跑通之后接下来就是把它接到真实项目里。如果你要长期做编码类或 Agent 类应用建议把 Key 管理、模型切换、调用配额这些事交给统一通道处理避免每个脚本都散落一份配置。API Key 在控制台创建和管理接入文档里有不同语言和框架的示例可以直接对照改。需要长期跑编码任务或 Agent 工作流的话Coding Plan 更适合按项目维度管理调用如果只是临时验证某个模型的效果用模型对话页面发几条消息最快。把这篇的config.toml和settings.json骨架留着后面加新的 MCP 服务只需要在servers下面加一段Python 侧几乎不用改。