news 2026/9/29 11:50:01

MCP协议实战:用Python+SQLite给AI Agent装上“手脚“,TaoToken统一Key接入

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
MCP协议实战:用Python+SQLite给AI Agent装上“手脚“,TaoToken统一Key接入

1. 为什么你的 AI Agent 只会聊天,不会干活

很多人第一次用 Claude Desktop 或 Cursor 里的 Agent 功能,都会有一种落差感:明明模型很聪明,能写代码、能分析逻辑,但一到"帮我查一下数据库里上周的销售数据"这种任务,它就开始编——要么给你一段看起来像 SQL 但根本跑不通的伪代码,要么直接说"我无法访问你的数据库"。

这不是模型不行,是它缺"手脚"。大模型本身只能处理文本,它没有眼睛看你的文件系统,没有手去连你的数据库,也没有嘴去调你的内部 API。你让它查数据,它只能靠猜。而 MCP(Model Context Protocol)解决的正是这个问题:它给 AI 装上一套标准化的"手脚接口",让模型能真正调用外部工具、拿到真实结果、再基于结果继续推理。

MCP 是 Anthropic 开源的一套协议,核心思路很朴素——把"AI 能调用的工具"抽象成统一的 Server,任何兼容 MCP 的客户端(Claude Desktop、Cursor、各类 Agent 框架)都能通过同一套协议发现工具、调用工具、拿到结构化返回。你写一次 Server,Claude 能用,Cursor 能用,以后换别的 Agent 也能用。这就是它比"每个工具单独写适配"强的地方。

这篇文章我带你从零跑通一个完整闭环:用 Python 写一个 MCP Server,后端接本地 SQLite,暴露"查询销售数据"和"生成汇总报告"两个工具,然后通过 TaoToken 统一 Key 接入模型侧,让 Agent 真的能查能写。全程可复制,代码直接跑。

适合谁看:写过一点 Python、想让 AI Agent 真正操作本地数据、又不想折腾一堆 SDK 适配的人。你不需要懂 MCP 协议细节,跟着配就行。

先说清楚这个闭环长什么样:你的 Agent 客户端(比如 Claude Desktop)通过 MCP 协议连上你写的server.py,server.py里用 sqlite3 读写test.db。当你在客户端里说"帮我查所有销售数据",客户端把这句话发给模型,模型看到 MCP Server 暴露的工具列表,决定调用query_sales,客户端把调用请求转发给你的 Server,Server 执行 SQL 返回 JSON,模型拿到 JSON 后组织成人话回给你。整条链路里,模型只负责"决定调哪个工具、传什么参数",真正的数据操作在你本地完成,Key 和数据都不出你的机器。

2. TaoToken 统一 Key 接入前的环境准备与 MCP SDK 安装

在写 Server 之前,先把模型侧的接入方式定下来。因为 MCP Server 本身只是"工具端",它需要一个能理解 MCP 协议的模型客户端来驱动。这里我用 TaoToken 作为统一接入层——它的好处是一个 Key 能覆盖 Claude、GPT 等多个模型,省得你为每个模型单独配一套凭证。

TaoToken 的定位是模型 API 的统一入口,官网在 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 端点是 https://taotoken.net/api 。你注册后在控制台生成一个 API Key,后面配置客户端时填进去就行。注意 API 地址不要加 UTM 参数,直接用https://taotoken.net/api。

环境要求不复杂:

Python 3.10 以上,推荐 3.12,因为 MCP SDK 用到了较新的类型注解语法。pip 包管理器随 Python 自带。一个支持 MCP 的客户端,Claude Desktop 最省事,Cursor 也行。SQLite 在 Mac 和 Linux 上系统自带,Windows 需要单独装一下,或者直接用 Python 内置的 sqlite3 模块——对,Python 标准库就带 sqlite3,你甚至不用额外装数据库。

安装 MCP SDK 只要一条命令:

pip install mcp

这个包是官方 SDK,体积很小,装完就能用。如果你用虚拟环境(推荐),先python -m venv venv再激活,然后装。我试过在干净的 3.12 环境里装,几秒钟就完事,没有乱七八糟的依赖。

建项目目录:

mkdir mcp-sqlite-demo cd mcp-sqlite-demo touch server.py init_db.py

MCP 项目结构可以极简,一个server.py就够。不需要框架、不需要配置文件、不需要路由。这是它比传统 Web 服务轻的地方——MCP Server 通过标准输入输出(stdio)和客户端通信,不占端口、不起 HTTP 服务。

在动手写代码前,先把 TaoToken 的 Key 拿到手。进控制台创建 API Key,复制保存。这个 Key 后面在客户端配置里会用到。如果你用的是 Claude Desktop,它的模型接入和 MCP Server 配置是两回事:模型接入走 TaoToken 的 API 端点,MCP Server 走本地进程。两者在同一个配置文件里配,但字段不同,别搞混。

顺便说下模型选择。MCP 工具调用对模型的"函数调用"能力有要求,Claude 系列在这块比较稳,GPT 系列也行。你在 TaoToken 控制台能看到可用模型列表,选一个支持 tool use 的就行。Model ID 要填对,比如claude-sonnet-4-20250514这种格式,具体以控制台显示为准。

3. 可复制的 MCP Server 配置:Python + SQLite 完整代码

这一节是全文核心,代码直接复制就能跑。先建数据库和测试数据。

init_db.py:

import sqlite3 conn = sqlite3.connect("test.db") cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS sales ( id INTEGER PRIMARY KEY AUTOINCREMENT, product TEXT NOT NULL, amount REAL NOT NULL, sale_date TEXT NOT NULL ) """) test_data = [ ("iPhone 15", 5999.00, "2026-08-01"), ("MacBook Pro", 14999.00, "2026-08-02"), ("AirPods Pro", 1899.00, "2026-08-02"), ("iPad Air", 4399.00, "2026-08-03"), ("iPhone 15", 5999.00, "2026-08-03"), ] cursor.executemany( "INSERT INTO sales (product, amount, sale_date) VALUES (?, ?, ?)", test_data ) conn.commit() conn.close() print("数据库初始化完成,插入了 5 条测试数据。")

跑一下python init_db.py,看到输出就说明test.db建好了。

然后是server.py,这是 MCP Server 本体:

from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent import sqlite3 import json app = Server("sqlite-assistant") DB_PATH = "test.db" @app.list_tools() async def list_tools() -> list[Tool]: return [ Tool( name="query_sales", description="查询销售数据。可以按产品名筛选,也可以查全部。返回按日期倒序排列的记录。", inputSchema={ "type": "object", "properties": { "product": { "type": "string", "description": "要查询的产品名称,如 'iPhone 15'。留空则查全部。" } }, "required": [] } ), Tool( name="get_sales_summary", description="获取销售汇总结计:总销售额、各产品销量与金额、最近销售日期。", inputSchema={ "type": "object", "properties": {}, "required": [] } ) ] @app.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: conn = sqlite3.connect(DB_PATH, timeout=10) cursor = conn.cursor() if name == "query_sales": product = arguments.get("product", "") if product: cursor.execute( "SELECT * FROM sales WHERE product LIKE ? ORDER BY sale_date DESC", (f"%{product}%",) ) else: cursor.execute("SELECT * FROM sales ORDER BY sale_date DESC") rows = cursor.fetchall() columns = [desc[0] for desc in cursor.description] results = [dict(zip(columns, row)) for row in rows] conn.close() return [TextContent( type="text", text=json.dumps(results, ensure_ascii=False, indent=2) )] elif name == "get_sales_summary": cursor.execute("SELECT SUM(amount) FROM sales") total = cursor.fetchone()[0] or 0 cursor.execute(""" SELECT product, COUNT(*) as count, SUM(amount) as total FROM sales GROUP BY product ORDER BY total DESC """) by_product = cursor.fetchall() cursor.execute("SELECT MAX(sale_date) FROM sales") latest = cursor.fetchone()[0] conn.close() summary = { "total_amount": total, "by_product": [ {"product": r[0], "count": r[1], "total": r[2]} for r in by_product ], "latest_date": latest } return [TextContent( type="text", text=json.dumps(summary, ensure_ascii=False, indent=2) )] else: conn.close() return [TextContent(type="text", text=f"未知工具:{name}")] async def main(): async with stdio_server() as (read_stream, write_stream): await app.run(read_stream, write_stream) if __name__ == "__main__": import asyncio asyncio.run(main())

代码里几个关键点。@app.list_tools()注册工具清单,客户端启动时会调这个,拿到你能提供哪些工具、每个工具要什么参数。@app.call_tool()处理实际调用,根据工具名分发到不同逻辑。stdio_server()是通信层,通过标准输入输出和客户端对话,所以你的 Server 不需要监听端口。

description字段特别重要,模型就是靠它判断"什么时候该调这个工具"。我写的是"查询销售数据。可以按产品名筛选,也可以查全部",比干巴巴的"查询数据库"强很多。后面排障那节会细说。

现在配置客户端。以 Claude Desktop 为例,配置文件路径:

Mac:~/Library/Application Support/Claude/claude_desktop_config.jsonWindows:%APPDATA%\Claude\claude_desktop_config.json

内容如下,注意把路径换成你自己的绝对路径:

{ "mcpServers": { "sqlite-assistant": { "command": "python3", "args": ["/绝对路径/mcp-sqlite-demo/server.py"], "cwd": "/绝对路径/mcp-sqlite-demo/", "env": { "TAOTOKEN_API_KEY": "你的_TaoToken_Key", "TAOTOKEN_BASE_URL": "https://taotoken.net/api" } } } }

这里三件套要写全:Base URL 是https://taotoken.net/api,Key 是你控制台生成的,Model ID 在客户端模型选择里填。如果你用的是 Cursor,在设置 → MCP 里点 "Add MCP Server",Type 选 Command,Command 填python3 /绝对路径/mcp-sqlite-demo/server.py,环境变量同样加上。

配置里cwd必须设对,否则 Server 找不到test.db。这是新手最容易踩的坑,下一节细讲。

4. 验证请求:让 Agent 真的查一次数据库

配置保存后重启 Claude Desktop。左下角如果出现一个工具图标,说明 MCP Server 连上了。如果没出现,先别急着怀疑代码,去看客户端日志——Claude Desktop 的日志在~/Library/Logs/Claude/mcp.log(Mac),里面会写 Server 启动失败的原因。

连上之后,直接在对话框里输入:

帮我查一下所有销售数据,按日期倒序排列。

正常的话,你会看到模型先"思考"一下,然后触发工具调用,界面上会显示它调用了query_sales,参数是空(查全部)。返回结果类似:

[ {"id": 5, "product": "iPhone 15", "amount": 5999.00, "sale_date": "2026-08-03"}, {"id": 4, "product": "iPad Air", "amount": 4399.00, "sale_date": "2026-08-03"}, {"id": 3, "product": "AirPods Pro", "amount": 1899.00, "sale_date": "2026-08-02"}, {"id": 2, "product": "MacBook Pro", "amount": 14999.00, "sale_date": "2026-08-02"}, {"id": 1, "product": "iPhone 15", "amount": 5999.00, "sale_date": "2026-08-01"} ]

模型拿到这个 JSON 后,会用人话总结给你:"共 5 条销售记录,最近一笔是 8 月 3 日的 iPhone 15,金额 5999 元……"

再试一个带参数的:

帮我查一下 iPhone 15 的销售记录。

这次模型会调用query_sales并传{"product": "iPhone 15"},Server 里走LIKE %iPhone 15%分支,返回两条记录。

再试汇总:

给我一份销售汇总报告,包含总销售额和各产品排名。

模型调用get_sales_summary,返回:

{ "total_amount": 33295.0, "by_product": [ {"product": "MacBook Pro", "count": 1, "total": 14999.0}, {"product": "iPhone 15", "count": 2, "total": 11998.0}, {"product": "iPad Air", "count": 1, "total": 4399.0}, {"product": "AirPods Pro", "count": 1, "total": 1899.0} ], "latest_date": "2026-08-03" }

到这里,一个"能查能写"的闭环就跑通了。注意整个过程中你只说了人话,没告诉模型用哪个工具、写什么 SQL。MCP 协议负责让模型"看见"可用工具,模型自己决定调哪个、传什么参数。这就是它和"复制提示词粘贴到对话框"的本质区别——AI 从"说"变成了"做"。

如果你想验证写入能力,可以再加一个insert_sale工具,逻辑和query_sales类似,只是执行INSERT。不过生产环境里写操作要谨慎,建议加权限校验或只读副本,别让 Agent 直接写主库。

5. 本篇常见报错排查:401、路径、工具不触发

跑通之后,我把踩过的坑整理一下,你遇到报错可以对照。

报错一:401 Unauthorized 或 invalid api key

这个通常出在模型侧接入,不是 MCP Server 本身。检查 TaoToken 的 Key 有没有填对、有没有多余空格、Base URL 是不是https://taotoken.net/api(注意不要带 UTM 参数)。如果客户端里模型调用报 401,但 MCP 工具列表能正常显示,说明 Server 没问题,是模型凭证的事。去控制台重新生成一个 Key 试试。

报错二:local proxy failed 或 connection refused

这种多半是客户端配置里 Base URL 写错,或者网络环境导致连不上 API 端点。确认TAOTOKEN_BASE_URL是https://taotoken.net/api,不要写成官网地址。如果客户端有代理设置,检查是否冲突。

报错三:FileNotFoundError: test.db not found

这是 MCP Server 工作目录问题。MCP Server 启动时的工作目录默认不是你的项目目录,而是系统根目录或客户端指定目录。解决方式是在客户端配置里加"cwd": "/绝对路径/mcp-sqlite-demo/",或者在server.py里把DB_PATH改成绝对路径。我建议两个都做,双保险。

报错四:工具被"看见"但没被调用,模型直接编答案

现象是你问"查销售数据",模型不调工具,直接给你一段编造的 JSON。原因是工具的description写得太模糊,模型不知道什么时候该用它。解决方式是把 description 写具体,比如"查询销售数据。可以按产品名筛选,也可以查全部"就比"查询数据库"好得多。模型是根据 description 决定是否调用的,写得越清楚,触发越准。

报错五:database is locked

MCP Server 进程和你手动开的 sqlite3 命令行同时操作一个文件时会锁。解决方式是在sqlite3.connect()里加timeout=10,或者启用 WAL 模式:cursor.execute("PRAGMA journal_mode=WAL")。代码里我已经加了 timeout,WAL 你可以按需加。

报错六:reading 'choices' 相关错误

这个一般出现在模型返回格式解析阶段,可能是 Model ID 填错,或者选的模型不支持 tool use。去 TaoToken 控制台确认模型列表,换一个支持函数调用的模型。如果用的是 Claude Code 类客户端,检查auth.json或 settings 里的模型配置是否和 Base URL、Key 匹配。

报错七:OAuth 相关报错

部分客户端在接入时会走 OAuth 流程,如果配置里混用了 API Key 和 OAuth,会冲突。确认你用的是 API Key 模式,不要同时开 OAuth。如果客户端强制 OAuth,去它的设置里关掉,改用 Key 认证。

排查顺序建议:先看客户端日志确认 Server 有没有启动,再看工具列表有没有加载,最后看模型调用有没有发出。三步定位,比瞎猜快。

6. 从 SQLite 到生产:MCP 接入的下一步

跑通 SQLite 版本后,你会发现换后端几乎不用改 MCP 层代码。比如换成 MySQL,只需要把sqlite3.connect换成mysql.connector.connect,SQL 语法微调,工具定义和调用逻辑原封不动。这就是 MCP 的价值——协议统一,后端随便换。

接入企业内部 API 同理,把 SQL 查询替换成requests.get("https://your-api/endpoint"),返回 JSON 直接透传给模型。模型不关心你后端是数据库还是 HTTP 服务,它只关心工具名、参数、返回结构。

如果你想让 Agent 长期跑编码任务或复杂 Agent 流程,可以考虑 TaoToken 的 Coding Plan,一个 Key 覆盖多模型,省得来回切换。模型对话调试可以去 https://taotoken.net/api-keys 管理 Key,接入文档在 https://taotoken.net/doc 有详细说明。需要快速验证模型响应,用模型对话页面最直接。

最后说个实用技巧:MCP Server 的description值得反复打磨。你可以把它当成"给模型看的 API 文档",写得越像人话、越具体,模型调用越准。我一般会先写一版,跑几个测试问题,看模型有没有调错工具,再回来改 description。这个迭代过程比调代码参数见效快得多。

代码都在上面了,复制、改路径、重启客户端,你就能拥有一个真正能查数据库的 AI Agent。

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/9/29 11:49:41

服装智能制造大会上的AI质检案例分享

1. AI服装制造场景 在服装智能制造大会上,AI质检成为最受关注的议题之一。传统人工质检依赖老师傅的经验与肉眼判断,效率低、漏检率高、招工难,已成为制约服装工厂产能与品质的瓶颈。随着计算机视觉与深度学习技术的成熟,AI质检正…

作者头像 李华
网站建设 2026/9/29 11:47:34

STM32CubeMX安装与配置全攻略:从下载到代码生成

1. 为什么STM32CubeMX值得你花时间折腾如果你刚开始接触STM32,或者从标准库时代一路走过来,第一次听说STM32CubeMX这个名字的时候大概率会有点懵——这玩意儿到底是干嘛的?简单说,它是ST官方推出的一款图形化配置工具,…

作者头像 李华
网站建设 2026/9/29 11:47:26

模型优化实战:从性能剖析到剪枝量化蒸馏的组合拳

接到一个模型优化任务,很多人第一反应是掏出剪枝、量化、蒸馏的论文挨个试一遍,结果折腾两周,精度掉了三个点,推理速度没快多少,最后还得灰溜溜回滚。这种事我见过太多次了。今天这篇就围绕“Model-Optimizer”这个主题…

作者头像 李华
网站建设 2026/9/29 11:42:48

RISC18架构国产单片机实战指南:VS Code开发与产线落地

1. 为什么“8位RISC18项目”要优先看英锐恩?——不是营销话术,是产线实测出来的选型逻辑做嵌入式开发的同行应该都经历过这种场景:一个温控小家电项目,功能不复杂——按键LED指示ADC采温PWM调风扇,主控资源需求极低&am…

作者头像 李华
网站建设 2026/9/29 11:41:17

ABAP 里的绝世好剑,是一套能承受业务重压的开发体系

海外子公司的采购员打开一张待审批的采购单,页面显示可用库存充足。等审批完成、订单落库,仓库却发现库存早已被另一笔业务占用。总部系统运行正常,海外页面偶尔超时,开发团队的第一反应往往是给查询加缓存,或者把审批程序改成异步执行。 这种时候,系统真正需要的,未必…

作者头像 李华