☰
MCP 是个啥?用 Python + SQLite 手搓一个最小 MCP Server 配 TaoToken
2026/9/29 6:45:33 网站建设 项目流程

1. 先搞清楚 MCP 到底解决什么问题

如果你最近在折腾 LLM 应用,大概率会反复看到 MCP 这个词。它的全称是 Model Context Protocol,翻译过来叫「模型上下文协议」。你可以把它理解成 LLM 世界里的 USB-C 接口:以前每接一个数据源(数据库、文件系统、内部 API),你都要为这个模型单独写一套适配代码;现在只要数据源这边实现一个标准的 MCP Server,任何支持 MCP 的客户端都能直接调用它。

它到底能做什么?简单说,MCP 让「模型」和「外部工具/数据」之间的连接标准化了。模型不再需要你手动把 SQL 查询结果贴进对话里,而是自己决定「我需要查一下 products 表」,然后通过 MCP 协议发起工具调用,拿到结果后再组织成自然语言回答你。

适合谁?我觉得三类人最该关注:一是刚接触 LLM 工具调用的 Python 开发者,想弄明白 function calling 背后到底怎么串起来的;二是手里有一堆本地 SQLite、CSV、日志文件,想让 AI 直接查的人;三是想给自己 IDE 或 Agent 加自定义工具的工程师。

这篇就聚焦一个最小可跑的场景:用 Python 手搓一个只做 SQLite 查询的 MCP Server,然后通过 TaoToken 的统一 API 通道,在 Cline 里发起一次真实的工具调用,把「模型决定调工具 → 工具执行 → 结果回传 → 模型总结」这条链路完整走一遍。全程本地,代码可复制。

2. 前置准备:TaoToken 统一 Key 与 API 通道

在写 Server 之前,先把「模型侧」的通道准备好。因为后面 Cline 要调用 LLM 来决定是否触发工具,这个 LLM 请求我们统一走 TaoToken 的 API,好处是一个 Key 能覆盖多种模型,配置片段也统一,不用每个客户端改一遍。

你需要先去控制台创建一个 API Key。打开 https://taotoken.net/console ,登录后在 API Keys 页面新建一个,复制出来形如sk-xxxx的字符串,先存到环境变量里,别硬编码进代码。

# Linux / macOS export TAOTOKEN_API_KEY="sk-你的key" # Windows PowerShell $env:TAOTOKEN_API_KEY="sk-你的key"

TaoToken 的 API 基地址是https://taotoken.net/api,兼容 OpenAI 风格的/v1/chat/completions和 Anthropic 风格的/v1/messages。Cline 这类客户端通常让你填 Base URL 和 API Key,Base URL 就填https://taotoken.net/api,Key 填上面那个。

注意:API 地址不要加任何多余路径后缀,客户端一般会自己拼/v1/...。如果你填成https://taotoken.net/api/v1,有些客户端会拼成/api/v1/v1/...导致 404。

想先确认 Key 能用,可以跑一条最小请求:

curl https://taotoken.net/api/v1/chat/completions \ -H "Authorization: Bearer $TAOTOKEN_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "gpt-4o-mini", "messages": [{"role": "user", "content": "ping"}] }'

返回里有choices[0].message.content就说明通道没问题。这一步别跳过,后面 Cline 报错时你能快速判断是 Key 问题还是 MCP 问题。

3. 手搓最小 SQLite MCP Server

3.1 环境与依赖

Python 需要 3.10 以上,因为 MCP 官方 SDK 用到了较新的类型语法。装依赖:

python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate pip install "mcp[cli]" httpx

mcp[cli]会带上官方 SDK 和调试用的 CLI 工具。装完可以python -c "import mcp; print(mcp.__version__)"确认。

3.2 准备一个测试数据库

先造点数据,不然查了个寂寞:

# init_db.py import sqlite3 conn = sqlite3.connect("shop.db") cur = conn.cursor() cur.execute(""" CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, price REAL NOT NULL, stock INTEGER NOT NULL ) """) cur.executemany( "INSERT OR REPLACE INTO products (id, name, price, stock) VALUES (?, ?, ?, ?)", [ (1, "机械键盘", 399.0, 12), (2, "无线鼠标", 129.0, 30), (3, "显示器支架", 259.0, 8), (4, "USB Hub", 89.0, 50), (5, "降噪耳机", 899.0, 5), ], ) conn.commit() conn.close() print("shop.db ready")

跑一下python init_db.py,当前目录会出现shop.db。

3.3 Server 骨架代码

新建sqlite_server.py。核心就三块:声明工具、实现工具、启动 stdio 服务。

# sqlite_server.py import sqlite3 import json from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent DB_PATH = "shop.db" app = Server("sqlite-shop-server") def run_query(sql: str) -> list[dict]: conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row try: cur = conn.execute(sql) rows = [dict(r) for r in cur.fetchall()] return rows finally: conn.close() @app.list_tools() async def list_tools() -> list[Tool]: return [ Tool( name="query_products", description="查询商品表,支持按价格上限和库存下限过滤", inputSchema={ "type": "object", "properties": { "max_price": {"type": "number", "description": "价格上限"}, "min_stock": {"type": "integer", "description": "库存下限"}, }, "required": [], }, ) ] @app.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: if name != "query_products": raise ValueError(f"Unknown tool: {name}") max_price = arguments.get("max_price", 999999) min_stock = arguments.get("min_stock", 0) sql = "SELECT id, name, price, stock FROM products WHERE price <= ? AND stock >= ? ORDER BY price" rows = run_query(sql) if False else None # 占位,见下方参数化版本 conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row cur = conn.execute(sql, (max_price, min_stock)) rows = [dict(r) for r in cur.fetchall()] conn.close() return [TextContent(type="text", text=json.dumps(rows, ensure_ascii=False, indent=2))] async def main(): async with stdio_server() as (read_stream, write_stream): await app.run(read_stream, write_stream, app.create_initialization_options()) if __name__ == "__main__": import asyncio asyncio.run(main())

上面那段run_query(sql) if False else None是我故意留的坑位说明:不要用字符串拼接 SQL,一定要参数化。真实代码里直接用下面那段conn.execute(sql, (max_price, min_stock))就行,把占位那行删掉。

3.4 用 MCP CLI 本地自测

在接客户端之前,先用官方 CLI 验证 Server 能不能正常列出工具:

mcp dev sqlite_server.py

它会启动一个调试界面,能看到query_products工具和它的 inputSchema。如果这里就报错,说明 Server 本身有问题,先别急着接 Cline。

4. 接入 Cline 并发起一次真实工具调用

4.1 配置 Cline 的模型通道

打开 VS Code 里的 Cline 设置,API Provider 选 OpenAI Compatible,Base URL 填https://taotoken.net/api,API Key 填你的 TaoToken Key,模型名填一个支持 function calling 的,比如gpt-4o-mini或claude-3-5-sonnet。保存后 Cline 会做一次连通性检查。

4.2 注册 MCP Server

Cline 的 MCP 配置一般在cline_mcp_settings.json,路径类似:

  • macOS:~/Library/Application Support/Code/User/globalStorage/saoudrizwan.claude-dev/settings/cline_mcp_settings.json
  • Windows:%APPDATA%\Code\User\globalStorage\saoudrizwan.claude-dev\settings\cline_mcp_settings.json

写入:

{ "mcpServers": { "sqlite-shop": { "command": "python", "args": ["/绝对路径/sqlite_server.py"], "env": {} } } }

注意:args里必须是绝对路径,相对路径在客户端启动子进程时工作目录不确定,很容易找不到文件。Windows 下路径用双反斜杠或正斜杠。

保存后重启 Cline,在 MCP 面板里应该能看到sqlite-shop处于 connected 状态,展开能看到query_products。

4.3 发起调用并验证

在 Cline 对话框里输入:

帮我查一下 shop.db 里价格不超过 300 且库存大于等于 10 的商品,按价格从低到高列出来。

正常流程是:Cline 把工具描述和你的问题一起发给 TaoToken 通道上的模型 → 模型返回一个 tool_call,参数是{"max_price": 300, "min_stock": 10}→ Cline 通过 MCP 调用你的 Server → Server 查 SQLite 返回 JSON → Cline 把结果回传模型 → 模型组织成自然语言。

预期返回类似:

[ {"id": 4, "name": "USB Hub", "price": 89.0, "stock": 50}, {"id": 2, "name": "无线鼠标", "price": 129.0, "stock": 30}, {"id": 3, "name": "显示器支架", "price": 259.0, "stock": 8} ]

等等,显示器支架库存是 8,不满足min_stock >= 10,所以正确结果应该只有 USB Hub 和无线鼠标。如果你看到三条,说明参数没传对,回去检查call_tool里min_stock的取值逻辑。这个细节正好用来验证工具调用是真的在执行,而不是模型瞎编。

5. 本篇常见错误排查

报错ModuleNotFoundError: No module named 'mcp':Cline 启动子进程用的 Python 不是你 venv 里的那个。把command改成 venv 里 Python 的绝对路径,比如/Users/you/project/venv/bin/python。

工具列表为空:@app.list_tools()装饰器必须挂在Server实例上,且函数是async。另外确认mcp dev能正常列出,排除 Server 本身问题。

模型不调用工具,直接编答案:换一个 function calling 支持更好的模型,或者在 Cline 里把「Auto-approve」关掉,强制它走工具确认流程。有些小模型对 inputSchema 理解差,会忽略工具。

SQL 报no such table: products:DB_PATH用了相对路径,而子进程工作目录不是项目目录。改成绝对路径,或者用os.path.join(os.path.dirname(__file__), "shop.db")。

TaoToken 返回 401:Key 没读到或复制时带了空格。用echo $TAOTOKEN_API_KEY确认,重新在控制台生成一个也行。

调用超时:SQLite 查询本身很快,超时多半是模型侧网络。检查 Base URL 是否误写成https://taotoken.net/api/v1,正确应为https://taotoken.net/api。

6. 继续往下走

到这里你已经跑通了最小闭环:一个只做 SQLite 查询的 MCP Server,通过 TaoToken 统一通道让 Cline 里的模型自主决定调用它。接下来可以做的扩展方向:把query_products换成更通用的run_sql(记得加只读白名单),或者加一个list_tables工具让模型先探结构再查。

如果你想把这条链路用到长期编码或 Agent 场景,建议直接上 Coding Plan,额度更划算,配置方式跟上面完全一致,只是把模型换成更适合代码的。想先多试几个模型对比工具调用效果,可以去模型对话页面直接测。Key 管理和新建都在 API Keys 页面,接入细节看文档。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询