1. 为什么大模型需要 MCP 这层“外挂”
大模型本身只会“说”,不会“做”。你问它今天数据库里有多少用户,它能给你编一个看起来很像真的数字,但它没法真的去查。MCP(Model Context Protocol)就是解决这个问题的标准协议,它把外部能力(查数据库、调接口、跑脚本)包装成模型能看懂的工具清单,模型自己决定什么时候调用、传什么参数。适合谁?适合正在做本地 Agent、想让 LLM 真正操作数据的开发者,尤其是用 LangChain/LangGraph 这套技术栈的人。
我试过最直观的场景:让模型自己判断“89+2 等于几”这种纯计算题不查库,而“帮我查 users 表里年龄最小的是谁”就自动去执行 SQL。整个过程不需要你写 if-else 路由,模型通过 ReAct 模式自己完成 Reason + Act。这篇就按这个场景,从零把 MCP 服务端、客户端注册、统一 Key 接入、调用验证、报错排查全部跑一遍。
核心链路分三段:第一段写一个 MySQL MCP 服务,用 fastmcp 把 SQL 执行封装成 tool;第二段用 MultiServerMCPClient 把服务注册进客户端,配合 create_react_agent 让模型自主选工具;第三段把大模型请求地址和 Key 统一到 TaoToken,避免每个项目到处散落 sk-xxx。跑通之后你会得到一个闭环:用户输入 → 模型判断 → 调用 MCP → 返回真实结果。
环境准备不复杂,Python 3.10+ 即可,依赖一次装齐:
pip install pymysql fastmcp langchain langchain-openai langgraph langchain-mcp-adapters本地 MySQL 要能连上,建一个 test_db 库和 users 表,随便塞几条数据。这一步别偷懒,后面验证全靠它。MCP 服务端和客户端是两个独立进程,通过 stdio 通信,所以路径、命令、参数三者必须对得上,这是后面 90% 报错的根源。
2. TaoToken 统一 Key 的前置准备与接入文档
在写客户端之前,先把大模型这一侧的接入方式定下来。很多人的痛点是:MCP 工具写好了,但模型请求地址和 Key 散落在各个脚本里,换一个模型就要改一遍 base_url。TaoToken 的作用就是提供一个统一的 Key 和入口,模型对话、编码、Agent 调用都走同一套凭证,配置一次到处复用。
你需要先拿到两样东西:API Key 和 Base URL。Key 在控制台的 API Keys 页面创建,Base URL 统一用https://taotoken.net/api。注意这里不要带任何多余路径,OpenAI 兼容的客户端会自动拼/v1/chat/completions这类后缀。创建 Key 的入口在这里:
控制台 API Keys:https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite
拿到 Key 之后,建议先别急着写 Agent,用最小请求验证一下 Key 和地址是否通。可以直接用 curl 打一发:
curl https://taotoken.net/api/v1/chat/completions \ -H "Authorization: Bearer sk-你的Key" \ -H "Content-Type: application/json" \ -d '{ "model": "gpt-4o", "messages": [{"role": "user", "content": "只回复 ok"}] }'返回里能看到choices[0].message.content就说明 Key 没问题。这一步很关键,因为后面 Agent 报错时你分不清是 MCP 的问题还是 Key 的问题,先把模型侧单独验证掉,排障范围直接砍一半。
模型名称(Model ID)要和你实际要用的保持一致,比如gpt-4o、claude-3-5-sonnet这类。TaoToken 的接入文档里有完整的模型列表和参数说明,配置前扫一眼能省很多试错:
接入文档:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite
如果你后面要做长期编码或 Agent 任务,可以考虑 Coding Plan,额度模型和按量调用不一样,适合高频跑 Agent 的场景:
Coding Plan:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite
前置准备就三件事:Key、Base URL、Model ID。这三个值后面会同时出现在客户端配置里,建议先写进环境变量,别硬编码在脚本里,不然提交代码时容易泄露。
3. 可复制的 MCP 服务端与客户端配置
先写服务端mysql_mcp.py。核心是用 fastmcp 初始化一个服务,把 SQL 执行封装成@app.tool()装饰的函数,模型看到的就是这个函数的描述和参数。数据库连接配置单独抽出来,方便改:
import pymysql from fastmcp import FastMCP import json app = FastMCP("MySQL MCP") DB_CONFIG = { "host": "127.0.0.1", "user": "root", "port": 3308, "password": "123456", "database": "test_db", "charset": "utf8mb4", "cursorclass": pymysql.cursors.DictCursor } def run_query(sql: str): conn = pymysql.connect(**DB_CONFIG) try: with conn.cursor() as cursor: cursor.execute(sql) if sql.strip().lower().startswith("select"): rows = cursor.fetchall() return {"rows": json.loads(json.dumps(rows, default=str))} else: conn.commit() return {"status": "success", "rows_affected": cursor.rowcount} finally: conn.close() @app.tool() def query_mysql(sql: str) -> dict: """ 执行 MySQL 查询语句 参数: sql: 要执行的 SQL 语句 (SELECT / INSERT / UPDATE / DELETE) """ try: return run_query(sql) except Exception as e: return {"error": str(e)} if __name__ == "__main__": app.run(transport="stdio")注意transport="stdio",这是本地进程间通信方式,客户端会用同样的方式启动它。端口 3308 是我本地 MySQL 的映射端口,你按自己的改。test_db和users表要提前建好,否则第一次调用就会返回连接或表不存在的错误。
再写客户端test_mcp.py。这里把 Base URL、Key、Model ID 三件套集中配置,MCP 服务通过MultiServerMCPClient注册,指定启动命令和脚本绝对路径:
import asyncio import os from langchain_mcp_adapters.client import MultiServerMCPClient from langgraph.prebuilt import create_react_agent from langchain_openai import ChatOpenAI BASE_URL = "https://taotoken.net/api" API_KEY = os.getenv("TAOTOKEN_API_KEY", "sk-你的Key") MODEL_NAME = "gpt-4o" async def main(): try: client = MultiServerMCPClient({ "MySQL-MCP": { "command": "python", "args": [os.path.abspath("mysql_mcp.py")], "transport": "stdio" } }) tools = await client.get_tools() if not tools: raise ValueError("未获取到任何工具") llm = ChatOpenAI( base_url=BASE_URL, openai_api_key=API_KEY, model=MODEL_NAME, timeout=60.0, max_retries=2 ) agent = create_react_agent(llm, tools) while True: user_input = input("\n请输入需求(或输入 exit 退出):\n> ") if user_input.strip().lower() == "exit": break async for chunk in agent.astream({"messages": user_input}): print(chunk) except Exception as e: print(f"程序初始化失败: {e}") if __name__ == "__main__": asyncio.run(main())三件套对照表,配置时逐项核对:
| 配置项 | 值 | 说明 |
|---|---|---|
| Base URL | https://taotoken.net/api | 不带 /v1 后缀,客户端自动拼 |
| API Key | 控制台创建 | 建议走环境变量 |
| Model ID | 如gpt-4o | 与文档列表一致 |
MultiServerMCPClient的 key(这里是MySQL-MCP)只是内部标识,可以同时注册多个服务,比如再加一个 Redis 或 Web API,模型会在多个工具里自己挑。create_react_agent负责把工具清单塞进模型的上下文,模型按 Reason + Act 循环决定调用哪个。
4. 验证请求与成功结果
配置写完,先跑服务端确认它能独立启动:
python mysql_mcp.py如果卡住不动是正常的,stdio 模式在等客户端连接。接着另开一个终端跑客户端:
export TAOTOKEN_API_KEY=sk-你的Key python test_mcp.py第一次输入先来个纯计算题,验证模型不会乱调工具:
请输入需求(或输入 exit 退出): > 89+2等于几?预期结果是模型直接回答 91,不触发 MySQL 工具。这一步验证的是模型的判断能力,如果它跑去查库了,说明工具描述写得太宽泛,把query_mysql的 docstring 收窄一点。
再来真实查询:
> 帮我查询users表有哪些数据正常会看到 astream 输出里出现 tool_call,参数是{"sql": "SELECT * FROM users"},然后工具返回 rows,模型把结果整理成自然语言。接着试一个需要推理的:
> 帮我查询users表中年龄最小的是谁?模型会自己生成SELECT * FROM users ORDER BY age ASC LIMIT 1这类 SQL,而不是让你手写。最后试建表写入:
> 帮我创建一个book表并写入一些数据模型会先CREATE TABLE再INSERT,返回rows_affected。到这里闭环就跑通了:用户输入 → 模型判断 → 调用 MCP → 真实执行 → 结果回传。
验证成功的标志有三个:get_tools()返回非空列表、astream 里能看到 tool_call 和 tool_result、数据库里确实多了数据。三个都满足,说明服务端、客户端、Key 三侧全部打通。
5. 本篇常见报错排查清单
排障按“先模型侧、再 MCP 侧、最后路径”的顺序,能最快定位。
401 Unauthorized:Key 错了或没带上。检查Authorization: Bearer sk-xxx是否完整,环境变量是否真的导出成功。用第 2 节的 curl 单独验证,curl 通了说明 Key 没问题,问题在客户端代码。
local proxy failed / connection refused:Base URL 写错,常见的是多写了/v1或少了https。统一用https://taotoken.net/api,让客户端自己拼路径。
reading choices 报错 / 返回体解析失败:通常是模型名不对,或者返回的不是标准 OpenAI 格式。核对 Model ID 是否在文档列表里,别自己编一个。
OAuth / 认证相关报错:如果你用的是 Claude Code 这类工具,认证方式可能不是纯 Key。Claude Code 接入时三件套要写全:Base URL、Key、Model ID,缺一个都会认证失败。参考接入文档里的 ClaudeCodeAnthropic 配置说明。
未获取到任何工具:get_tools()返回空。检查mysql_mcp.py路径是否是绝对路径,command用的python是否在当前环境可用。可以手动python mysql_mcp.py看是否报错。
数据库连接失败:端口、密码、库名对不上。先在本地用mysql -h127.0.0.1 -P3308 -uroot -p手动连一次,确认能连上再跑 MCP。
SQL 执行报错但模型说成功:工具里except把错误吞了返回{"error": ...},模型可能没正确解读。把错误信息打印出来看,别只看模型回复。
排障时记住一个原则:模型侧和 MCP 侧分开验证。curl 验证模型,手动跑服务端验证 MCP,两边都单独通了再合起来,问题一定出在连接处。
6. 从跑通到用起来:统一 Key 与后续扩展
跑通第一个闭环之后,真正省事的地方在于统一 Key。你后面再加 Redis MCP、Web API MCP,模型侧完全不用改,还是同一个 Base URL 和 Key,只在MultiServerMCPClient里多注册一个服务就行。工具越多,模型自主编排的价值越大。
想验证更多模型或对比不同模型在 Agent 场景下的表现,可以直接在模型对话里试:
模型对话:https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite
长期跑编码或 Agent 任务,按量调用成本会上去,Coding Plan 更适合高频场景。Key 和文档入口再放一次,方便你直接跳:
API Keys:https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite 接入文档:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite
最后提醒一句:这个 MySQL MCP 只适合学习,生产环境一定要加 SQL 执行限制,比如只允许 SELECT、禁止 DROP/DELETE,或者加白名单。模型再聪明也可能生成不可逆的语句,权限收窄是底线。把工具描述写清楚、把危险操作挡在服务端,才是能长期用的 Agent 闭环。