资讯详情

从零开始:MCP数据库助手(一)- 基础搭建与TaoToken接入

📅 2026/10/8 22:10:21 | 华诺云谱 👁 阅读
从零开始:MCP数据库助手(一)- 基础搭建与TaoToken接入
1. 为什么要在 Cursor 里给 AI 装一个数据库助手MCP 全称 Model Context Protocol你可以把它理解成一套让 AI 助手调用外部工具的通用插头标准。没有 MCP 的时候AI 只能陪你聊天你问它“我库里 orders 表有几条数据”它只能凭想象编一个数字有了 MCPAI 就能真的去连你的 SQLite 文件把表名、字段、行数一条条读出来再回答你。这个差别用过一次就回不去了。这篇要做的是一个最小可跑的 MCP 数据库助手用 uv 初始化项目用 SQLAlchemy 管连接用 FastMCP 暴露一个list_tables工具最后在 Cursor 里通过 TaoToken 统一 Key/API 通道把它跑通一次真实查询。整套流程面向 Cursor 用户代码全部可复制你跟着敲一遍大概二十分钟能见到结果。适合谁看已经会用 Cursor 写代码、想让 AI 直接读本地 SQLite 的开发者正在评估 MCP 到底能干什么、想先跑个 demo 的人以及被各种“AI 操作数据库”宣传绕晕、想自己动手验证一遍的同学。前置要求只有两个本机装了 Python 3.10 以上以及一个能编辑 JSON 配置的 Cursor。先说清楚边界避免误会。这个助手只暴露你显式写出来的工具函数AI 能做什么完全由你的代码决定它不会自己生成 SQL 去删库。我们这一篇只实现“列出所有表”这一个只读工具先把链路打通下一篇再逐步加查询、加字段描述、加安全校验。链路通了后面加功能就是复制粘贴改几行的事。项目结构我按下面这样组织v1是项目根目录v1/ ├── pyproject.toml # uv 项目配置 ├── README.md # 项目说明 ├── src/ │ └── mcp_datatools/ # 主包 │ ├── __init__.py │ ├── server.py # MCP 服务器 │ └── database.py # 数据库管理 ├── data/ │ ├── init_scripts/ │ │ └── init_sqlite_db.py # 初始化 SQLite 数据库脚本 │ └── test.db # SQLite 测试数据库 └── tests/ # 测试文件目录为什么用src/布局而不是把包直接扔在根目录因为src/能强制你把“安装后的包”和“开发时的源码”分开避免本地跑测试时误导入未安装的代码。这个习惯在 MCP 这种要长期维护的小工具项目里特别值后面加测试、加打包都不会乱。2. 用 uv 初始化项目并接入 TaoToken 统一通道uv 是现在 Python 圈里速度最快的包管理器之一装依赖比 pip 快一个数量级而且自带虚拟环境管理不用你再手动python -m venv。MCP 项目依赖不多但迭代频繁用 uv 能省掉大量等待时间。先装 uvmacOS/Linux 一行命令curl -LsSf https://astral.sh/uv/install.sh | shWindows 用户用 PowerShellpowershell -ExecutionPolicy ByPass -c irm https://astral.sh/uv/install.ps1 | iex装完验证一下uv --version能打印出版本号就说明装好了。接着进项目目录写pyproject.toml[project] name mcp-datatools version 0.1.0 description A MCP server that connects to your database requires-python 3.10 dependencies [ mcp1.0.0, sqlalchemy2.0.0, ] [build-system] requires [hatchling] build-backend hatchling.build然后在v1目录下执行uv syncuv 会自动创建.venv、解析依赖、装好mcp和sqlalchemy。第一次跑会下载包之后基本是秒级。装完你可以uv run python -c import mcp; print(mcp.__version__)确认一下。接下来是 TaoToken 的接入。TaoToken 提供统一的 API 通道把模型调用收敛到一个 Base URL 和一个 Key 上Cursor 里配一次就能用不用每个模型单独填地址。官网入口是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后在控制台生成 Key。具体操作路径打开 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 进入控制台在 API Keys 页面点新建复制生成的 Key 保存好。这个 Key 就是后面 Cursor 配置里要填的凭证。API 的基础地址是 https://taotoken.net/api 注意这个地址不带任何查询参数直接填就行。这里有个容易踩的坑很多人把 Key 直接写进pyproject.toml或者提交到 Git这是大忌。正确做法是放进环境变量或者 Cursor 的 MCP 配置里单独维护。我们后面在 Cursor 配置里用env字段传不落到代码仓库。如果你还想在浏览器里先验证一下 Key 能不能用可以打开模型对话页面 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 发一条消息试试能正常回复说明 Key 和通道都没问题。这一步不是必须的但能帮你把“Key 问题”和“MCP 配置问题”提前分开排障时省很多事。3. 可复制的数据库管理与 MCP server 骨架这一节是全文的核心两个文件database.py管连接server.py暴露工具。先写src/mcp_datatools/database.pydatabase.py - 数据库管理模块 import os from typing import List from contextlib import contextmanager from sqlalchemy import create_engine, inspect, text from sqlalchemy.exc import SQLAlchemyError from mcp.server.fastmcp.utilities.logging import get_logger logger get_logger(__name__) class DatabaseManager: 数据库管理器 def __init__(self): 初始化数据库管理器 self.database_url os.getenv(DB_URL) self.engine None self._connect() def _connect(self) - None: 连接数据库 try: self.engine create_engine(self.database_url, echoFalse) logger.info(f成功连接到数据库: {self.database_url}) except Exception as e: logger.error(f连接数据库时出错: {str(e)}) raise contextmanager def get_connection(self): 获取数据库连接的上下文管理器 if not self.engine: raise RuntimeError(数据库未连接) conn self.engine.connect() try: yield conn finally: conn.close() def get_table_names(self) - List[str]: 获取数据库中的所有表名 try: with self.get_connection() as conn: inspector inspect(conn) return inspector.get_table_names() except SQLAlchemyError as e: logger.error(f获取表名失败: {e}) raise def test_connection(self) - bool: 测试数据库连接 try: with self.get_connection() as conn: result conn.execute(text(SELECT 1)).scalar() return result 1 except Exception as e: logger.error(f测试数据库连接时出错: {str(e)}) return False def close(self) - None: 关闭数据库连接 if self.engine: self.engine.dispose() logger.info(数据库连接已关闭)几个设计点值得说。DB_URL从环境变量读不硬编码这样同一份代码能连测试库也能连正式库。get_connection用contextmanager包了一层保证连接一定被关闭不会因为异常泄漏。test_connection执行SELECT 1这是最轻量的连通性检查比SELECT * FROM some_table安全得多。再写src/mcp_datatools/server.pyserver.py - MCP服务器主文件 import os from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp.utilities.logging import get_logger from .database import DatabaseManager mcp FastMCP(MCP DataTools) logger get_logger(__name__) db_manager None def get_database_manager(): 获取数据库管理器实例 global db_manager if db_manager is None: db_manager DatabaseManager() return db_manager mcp.tool(description查询数据库中的所有表) def list_tables() - str: 获取数据库表列表 try: db_mgr get_database_manager() tables db_mgr.get_table_names() if tables: result f数据库中共有 {len(tables)} 个表:\n\n for i, table in enumerate(tables, 1): result f{i}. {table}\n logger.info(f成功返回 {len(tables)} 个表) return result return 数据库中没有表 except Exception as e: error_msg f获取数据库表列表失败: {str(e)} logger.error(error_msg) return f{error_msg}\n\n请检查数据库连接。 def main(): try: logger.info(启动MCP DataTools服务器) db_mgr get_database_manager() if db_mgr.test_connection(): logger.info(数据库连接测试成功) else: logger.warning(数据库连接测试失败但服务器仍会启动) logger.info(提示运行 python data/init_scripts/init_sqlite_db.py 创建测试数据库) logger.info(当前功能list_tables() - 获取数据库表列表) logger.info(MCP服务器启动成功等待客户端连接...) mcp.run() except Exception as e: logger.error(f服务器启动失败: {str(e)}) raise if __name__ __main__: main()mcp.tool装饰器是 FastMCP 的关键它把普通 Python 函数注册成 AI 可调用的工具description会作为工具说明暴露给模型。list_tables返回的是格式化字符串而不是原始列表因为模型读自然语言比读 JSON 数组更稳不容易在解析上翻车。别忘了src/mcp_datatools/__init__.py空文件即可它的作用是让mcp_datatools成为一个可导入的包。4. 建表脚本与 Cursor 配置验证一次真实查询先准备测试数据。写data/init_scripts/init_sqlite_db.pydata/init_scripts/init_sqlite_db.py - 初始化 SQLite 数据库 import sqlite3 import os from pathlib import Path def create_simple_test_database(): 创建简单测试数据库 data_dir Path(__file__).parent.parent db_path data_dir / test.db print(f数据库路径: {db_path}) if db_path.exists(): os.remove(db_path) print(删除旧数据库文件) conn sqlite3.connect(db_path) cursor conn.cursor() try: cursor.execute( CREATE TABLE users ( id INTEGER PRIMARY KEY, name VARCHAR(50) NOT NULL, email VARCHAR(100) NOT NULL ) ) cursor.execute( CREATE TABLE products ( id INTEGER PRIMARY KEY, name VARCHAR(100) NOT NULL, price DECIMAL(10, 2) NOT NULL ) ) cursor.execute( CREATE TABLE orders ( id INTEGER PRIMARY KEY, user_id INTEGER, total_amount DECIMAL(10, 2) ) ) cursor.execute(INSERT INTO users (name, email) VALUES (Alice, alicetest.com)) cursor.execute(INSERT INTO users (name, email) VALUES (Bob, bobtest.com)) cursor.execute(INSERT INTO products (name, price) VALUES (笔记本电脑, 5999.99)) cursor.execute(INSERT INTO products (name, price) VALUES (鼠标, 129.99)) cursor.execute(INSERT INTO orders (user_id, total_amount) VALUES (1, 6129.98)) cursor.execute(INSERT INTO orders (user_id, total_amount) VALUES (2, 129.99)) conn.commit() print(测试数据库创建成功包含 users / products / orders 三张表) except Exception as e: print(f创建数据库时出错: {e}) conn.rollback() raise finally: conn.close() if __name__ __main__: create_simple_test_database()在v1目录下运行uv run python data/init_scripts/init_sqlite_db.py看到“测试数据库创建成功”就对了data/test.db会出现在目录里。现在配 Cursor。打开 Cursor 设置找到 MCP 配置部分加入下面这段 JSON。注意把/path/to/v1换成你本机v1的绝对路径DB_URL用绝对路径最稳避免 Cursor 工作目录不同导致找不到库{ mcpServers: { mcp-datatools: { command: uv, args: [ run, --project, /path/to/v1, python, -m, mcp_datatools.server ], env: { DB_URL: sqlite:////path/to/v1/data/test.db, PYTHONPATH: /path/to/v1/src } } } }这里三件套要写全Base URL 走 TaoToken 的 https://taotoken.net/api Key 用你在控制台生成的那串Model ID 按你实际要用的模型填。MCP server 本身不直接调模型模型调用是 Cursor 侧的事所以 TaoToken 的配置在 Cursor 的模型设置里而不是这段 MCP JSON 里。两者配合的关系是Cursor 用 TaoToken 通道调模型模型通过 MCP 协议调你的list_tables工具。保存配置后重启 Cursor在对话里问一句“帮我看看数据库里有哪些表”。如果一切正常模型会调用list_tables返回类似数据库中共有 3 个表: 1. orders 2. products 3. users看到这个输出说明从 Cursor 到模型、从模型到 MCP server、从 server 到 SQLite 的整条链路全通了。这一步是整个项目最有成就感的时刻后面加多少工具都是在这个地基上盖楼。5. 常见报错排查401、local proxy failed 与 reading choices链路第一次跑报错几乎必然出现。下面这几个是我实际遇到频率最高的按现象对号入座。401 Unauthorized。这个基本是 Key 的问题。先确认 Cursor 模型设置里填的 Key 是 TaoToken 控制台生成的那串没有多余空格没有把Bearer前缀重复写两遍。如果 Key 刚生成等十几秒再试有时候有短暂生效延迟。还不行就回控制台重新生成一个旧 Key 作废。注意 401 是模型通道的报错跟 MCP server 无关别去翻server.py。local proxy failed / connection refused。这个通常出在 MCP server 启动阶段。先手动在终端跑一遍cd /path/to/v1 DB_URLsqlite:////path/to/v1/data/test.db PYTHONPATH/path/to/v1/src uv run python -m mcp_datatools.server如果终端里能正常打印“MCP服务器启动成功”说明代码没问题是 Cursor 配置里的路径写错了。重点检查--project后面的路径和PYTHONPATH是否指向v1/src以及DB_URL是不是四个斜杠sqlite:////绝对路径是四斜杠sqlite:///相对路径是三斜杠这个特别容易错。reading choices / 解析响应失败。这个报错一般出现在模型返回内容格式异常时。先确认 TaoToken 的 Base URL 填的是 https://taotoken.net/api 结尾没有多余的/v1或斜杠。然后检查 Model ID 是否拼写正确模型名错一个字就会返回非预期结构。如果换了模型才好那就是模型兼容性问题换回稳定模型即可。工具列表里看不到 list_tables。先确认mcp.tool装饰器没写错函数名没被覆盖。然后在终端手动跑 server看启动日志里有没有“当前功能list_tables()”。如果日志有但 Cursor 看不到多半是 Cursor 缓存了旧的 MCP 配置彻底退出 Cursor 再打开或者删掉 MCP 配置重新加一遍。数据库连上了但表是空的。检查DB_URL指向的test.db是不是你刚生成的那个。常见情况是相对路径导致 Cursor 在别的目录找了一个空库。统一用绝对路径四个斜杠问题基本消失。排障的核心思路是分层模型通道的问题看 401 和 reading choicesMCP 启动的问题看 local proxy failed数据的问题看 DB_URL。把这三层分开不要一报错就从头改代码。6. 把这条链路用起来下一步怎么走基础链路跑通后你可以马上做几件事让它更有用。第一照着list_tables的写法加一个describe_table(table_name: str)工具用inspector.get_columns()返回字段名和类型这样 AI 就能回答“users 表有哪些字段”。第二加一个只读的run_select(sql: str)但一定要在函数里校验 SQL 必须以SELECT开头拒绝任何写操作这是安全底线。第三把DB_URL换成环境变量注入测试库和正式库用不同配置避免误操作。如果你打算长期在 Cursor 里做编码和 Agent 类工作可以了解一下 Coding Plan它把模型调用和额度管理打包好适合高频使用场景https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面把 Base URL、Key、Model ID 三件套的填法讲得很清楚配置卡住时对着看一遍比瞎试快。需要新建或管理 Key 就去 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。下一篇我会在这个骨架上加describe_table和安全的run_select并补上单元测试让这个 MCP 数据库助手从“能跑”变成“敢用”。你现在要做的就是把这一篇的代码原样敲一遍亲眼看到那三张表被列出来。链路通了剩下的都是体力活。
📝

华诺云谱内容团队

资深建站顾问 · 行业研究员

10年+企业数字化服务经验,专注智能建站、SEO优化与品牌营销,持续输出建站技巧、行业洞察与营销干货,已帮助5000+企业实现数字化增长。

你可能需要的服务

订阅华诺云谱资讯周报

每周一封,精选建站技巧、SEO与营销干货,直达邮箱。已有 8,000+ 企业主订阅,助你少走弯路。

↑