企业官网建设流程全解析

1. 为什么我要自己写一个 MCP Server 连数据库

MCP（Model Context Protocol，模型上下文协议）是 2026 年 AI 工具链里绕不开的关键词。从 Cursor 到 Claude Desktop，再到 VS Code 生态的各种插件，主流 AI 开发工具基本都已经支持 MCP 协议。它解决的问题很实际：以前让 AI 查数据库，你得给 GPT 写一套 Function Calling 的 schema，给 Claude 写另一套，换个客户端接口全要重写；MCP 定义了一套通用协议，工具写一次，所有支持 MCP 的客户端都能调用。

打个比方，MCP 对于 AI 工具，相当于 USB 接口对于外设设备。标准统一了，生态才能真正繁荣起来。

但很多开发者还停留在「听过但没实操」的阶段。这篇教程我从零开始，手把手带你用 Python 搭建一个能查询数据库的 MCP Server，并演示如何通过 TaoToken 统一 Key/API 通道接入 AI 工具调用该 Server，全程附完整可运行代码，跟着做就能上手。

MCP 的架构由三个角色组成：MCP Client（发起调用的 AI 应用，比如 Cursor、Claude Desktop 或你自己写的 Agent 程序）、MCP Server（提供具体能力的服务端，比如查数据库、读文件、调第三方 API）、传输层（支持 stdio 本地进程通信和 HTTP+SSE 远程网络服务两种模式）。

MCP Server 可以暴露三类能力：Tools（工具，可执行的函数，如查询数据库、发邮件）、Resources（资源，可读取的数据源，如文件内容、表结构）、Prompts（提示模板，预定义的对话模板）。本教程重点实战 Tools 部分，这也是日常开发中用得最多的。

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

在开始写代码之前，先把 AI 调用通道准备好。开发 MCP Server 的过程中，你需要用不同模型验证 Tool Calling 的兼容性——Claude 准确率高、GPT-4o 响应快、DeepSeek 成本低。如果每个模型都单独装 SDK、单独配 Key，切换成本很高。

我的做法是用 TaoToken 的统一 API 入口，切换模型只需改一个参数。先去官网注册账号：https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=

注册完成后进入控制台创建 API Key：https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=

API Key 管理页面在这里，可以创建、查看、删除 Key：https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=

拿到 Key 之后，API 基础地址是https://taotoken.net/api（注意这个地址不加 UTM 参数）。这个通道兼容 OpenAI SDK 格式，意味着你现有的openaiPython 包可以直接用，只需改base_url和api_key两个字段。

如果你打算长期做编码类 Agent 开发，可以了解下 Coding Plan：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=

想先在网页上验证模型连通性，可以用模型对话页面：https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=

接入文档在这里，遇到参数问题可以查：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=

3. 可复制配置：从零搭建数据库查询 MCP Server

3.1 环境搭建

推荐使用 uv 包管理器，速度快、依赖干净：

# 创建项目 uv init mcp-demo cd mcp-demo # 安装依赖 uv add mcp sqlalchemy

如果还没装 uv，一行搞定：

curl -LsSf https://astral.sh/uv/install.sh | sh

3.2 第一步：Hello World 级别的 MCP Server

先确保链路跑通。创建server.py：

# server.py —— 最小 MCP Server 示例 from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent server = Server("demo-server") @server.list_tools() async def list_tools(): return [ Tool( name="hello", description="打个招呼，验证 MCP 连接是否正常工作", inputSchema={ "type": "object", "properties": { "name": { "type": "string", "description": "你的名字" } }, "required": ["name"] } ) ] @server.call_tool() async def call_tool(name: str, arguments: dict): if name == "hello": return [TextContent( type="text", text=f"你好 {arguments['name']}！MCP Server 连接正常" )] raise ValueError(f"未知工具: {name}") async def main(): async with stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream) if __name__ == "__main__": import asyncio asyncio.run(main())

这就是一个最基本的 MCP Server：注册了一个hello工具，AI 调用时返回一句问候。结构清晰，容易理解。

3.3 第二步：加入数据库查询功能

接下来实现真正有用的功能——让 AI 助手直接查询 SQLite 数据库。创建完整的server.py：

# server.py —— 带数据库查询的 MCP Server import json import sqlite3 from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent server = Server("db-query-server") DB_PATH = "demo.db" def init_db(): """创建示例数据库和测试数据""" conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, price REAL NOT NULL, category TEXT, stock INTEGER DEFAULT 0 ) """) sample_data = [ ("机械键盘", 299.0, "外设", 150), ("4K 显示器", 2499.0, "显示器", 30), ("无线鼠标", 149.0, "外设", 200), ("降噪耳机", 899.0, "音频", 80), ("USB-C 扩展坞", 199.0, "配件", 120), ] cursor.executemany( "INSERT OR IGNORE INTO products (name, price, category, stock) VALUES (?, ?, ?, ?)", sample_data ) conn.commit() conn.close() @server.list_tools() async def list_tools(): return [ Tool( name="query_products", description="按条件查询产品信息。支持分类筛选、价格上限过滤和多种排序方式。", inputSchema={ "type": "object", "properties": { "category": { "type": "string", "description": "产品分类，可选值：外设、显示器、音频、配件" }, "max_price": { "type": "number", "description": "价格上限，只返回不超过该价格的产品" }, "sort_by": { "type": "string", "enum": ["price_asc", "price_desc", "stock_desc"], "description": "排序规则：price_asc 价格升序 / price_desc 价格降序 / stock_desc 库存降序" } } } ), Tool( name="get_stats", description="获取产品库的汇总统计：总数量、平均价格、价格区间、总库存及各分类分布", inputSchema={"type": "object", "properties": {}} ) ] @server.call_tool() async def call_tool(name: str, arguments: dict): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row cursor = conn.cursor() try: if name == "query_products": query = "SELECT * FROM products WHERE 1=1" params = [] if category := arguments.get("category"): query += " AND category = ?" params.append(category) if max_price := arguments.get("max_price"): query += " AND price <= ?" params.append(max_price) sort_map = { "price_asc": "price ASC", "price_desc": "price DESC", "stock_desc": "stock DESC" } if sort_by := arguments.get("sort_by"): query += f" ORDER BY {sort_map.get(sort_by, 'id')}" cursor.execute(query, params) rows = [dict(row) for row in cursor.fetchall()] return [TextContent( type="text", text=json.dumps(rows, ensure_ascii=False, indent=2) )] elif name == "get_stats": cursor.execute(""" SELECT COUNT(*) as total, ROUND(AVG(price), 2) as avg_price, MIN(price) as min_price, MAX(price) as max_price, SUM(stock) as total_stock FROM products """) stats = dict(cursor.fetchone()) cursor.execute(""" SELECT category, COUNT(*) as count FROM products GROUP BY category """) stats["categories"] = { row["category"]: row["count"] for row in cursor.fetchall() } return [TextContent( type="text", text=json.dumps(stats, ensure_ascii=False, indent=2) )] raise ValueError(f"未知工具: {name}") finally: conn.close() async def main(): init_db() async with stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream) if __name__ == "__main__": import asyncio asyncio.run(main())

3.4 第三步：配置 AI 客户端接入

Claude Desktop 配置方法：

编辑配置文件（macOS 路径示例）：

# macOS ~/Library/Application Support/Claude/claude_desktop_config.json # Windows %APPDATA%\Claude\claude_desktop_config.json

写入以下内容：

{ "mcpServers": { "db-query": { "command": "uv", "args": ["run", "server.py"], "cwd": "/your/path/to/mcp-demo" } } }

重启 Claude Desktop，界面上会出现工具图标。试着问它：「帮我查一下 500 块以内的外设有什么」——它会自动调用你的 MCP Server，查询数据库后用自然语言回答。

Cursor 配置：进入 Settings → MCP Servers，添加同样的配置即可。

4. 验证请求：连通性测试与成功结果

4.1 用 TaoToken 通道测试多模型 Tool Calling

开发 MCP Server 时，你需要验证不同模型对 Tool Calling 的兼容性。用 TaoToken 统一入口，切换模型只需改一个参数：

from openai import OpenAI client = OpenAI( api_key="your-taotoken-api-key", base_url="https://taotoken.net/api" ) # 只改 model 参数就能测不同模型 for model in ["claude-sonnet-4-20250514", "gpt-4o", "deepseek-chat"]: response = client.chat.completions.create( model=model, messages=[{"role": "user", "content": "查一下 500 以下的 外设"}], tools=[{ "type": "function", "function": { "name": "query_products", "description": "按条件查询产品信息", "parameters": { "type": "object", "properties": { "category": {"type": "string"}, "max_price": {"type": "number"} } } } }] ) print(f"{model}: {response.choices[0].message.tool_calls}")

一个base_url搞定多模型切换，不用装多个 SDK，也不用处理各家 API 的格式差异。

4.2 验证 MCP Server 是否正常工作

启动 Server 后，在 Claude Desktop 或 Cursor 中提问，观察是否触发工具调用。成功的结果应该是：AI 识别出需要调用query_products工具，传入category="外设"和max_price=500参数，Server 返回 JSON 格式的产品列表，AI 用自然语言总结结果。

如果 AI 没有调用工具，检查以下几点：Server 进程是否正常启动、配置文件路径是否正确、cwd是否指向项目根目录、uv是否在系统 PATH 中。

5. 本篇常见错误排查

5.1 Tool 描述质量直接影响调用准确率

大模型根据description字段判断什么时候调用、传什么参数。我早期把描述写成简单的「查询数据」，结果 AI 频繁误调用。

解决方案：把description当 API 文档写，明确说明适用场景、输入参数含义和返回格式。

5.2 返回数据量太大导致效果变差

MCP Server 返回的内容直接进入大模型上下文窗口。如果一次返回几百条记录，不仅浪费 token，模型的理解和总结能力也会明显下降。

解决方案：在 Server 端实现分页，默认限制 20 条。对于统计类查询，直接返回聚合结果。

5.3 多模型测试的效率问题

开发 MCP Server 的过程中，你可能需要用不同模型验证 Tool Calling 的兼容性。如果每个模型都单独配置环境，效率很低。用统一 API 入口可以快速切换，一个base_url搞定。

5.4 安全防护不能忽视

MCP Server 相当于给 AI 开了一个操作你系统的入口，安全措施必须到位：

SQL 注入防护：始终使用参数化查询（?占位符），绝不拼接 SQL 字符串。

最小权限原则：生产环境使用只读数据库账号。

参数白名单校验：对 AI 传入的参数做严格校验。

调用审计日志：记录每一次 Tool 调用的入参和结果。

import logging logger = logging.getLogger("mcp-audit") @server.call_tool() async def call_tool(name: str, arguments: dict): logger.info(f"Tool called: {name}, args: {arguments}") # ... 执行业务逻辑

5.5 连接超时与进程退出

stdio 模式下，如果 Server 进程异常退出，客户端会显示连接断开。常见原因是代码中有未捕获的异常。建议在main()外层加 try/except，把错误写入日志文件，方便排查。

6. 继续深入：把 MCP Server 接到你的工作流

MCP 协议的核心逻辑可以归纳为三步：注册工具（list_tools）——告诉 AI 你能提供什么能力；执行调用（call_tool）——AI 决定调用时，执行对应逻辑；返回结果——将结果以文本形式传回给 AI。

掌握这个模式后，你可以快速扩展到更多场景：对接公司内部系统（工单、监控、日志查询）、连接个人知识库（Notion、Obsidian）、控制智能家居设备、管理云服务器（查负载、部署、重启）。

如果你在接入过程中遇到 API 报错或配置问题，可以查接入文档：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=

需要创建或管理 API Key，去这里：https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=

想先在网页上验证模型是否正常响应，用模型对话页面：https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=

长期做编码类 Agent 开发的话，Coding Plan 会更划算：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=

Claude Code 相关接入参考：https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=

MCP 生态已经成熟，文档齐全，踩过坑的人也把经验分享出来了。如果你一直在观望，现在正是动手实践的好时机。本文代码基于 mcp Python SDK 1.x 版本测试通过。