前言

说白了，AI Agent 现在最大的问题不是模型不够聪明，而是它够不着你的数据。

你让它查个数据库、读个文件、调个 API，它只能干瞪眼。除非你手动写一堆胶水代码把这些能力接进去——但每换一个模型、每加一个工具，就得重写一遍。

MCP（Model Context Protocol）就是来解决这个问题的。它是 Anthropic 在 2024 年底开源的一套协议，目标很明确：让 AI Agent 能用统一的方式连接外部工具和数据源。

今天这篇文章，手把手带你用 Python 从零写一个 MCP Server，实现一个能让 AI Agent 直接查询数据库的工具服务器。

一、MCP 到底是什么？

你可以把 MCP 想象成 AI 界的 USB-C 接口。

以前每个外设都有自己的接口（USB-A、Micro-USB、Lightning），乱七八糟。USB-C 统一了之后，一根线搞定所有设备。

MCP 也是这个思路：

• 以前：每个 AI 工具都得写专门的对接代码
• 现在：所有工具都按 MCP 协议暴露能力，任何支持 MCP 的 AI Agent 都能直接调用

它的核心概念就三个：

概念	说明	类比
Tool	Agent 可以调用的函数	REST API 的 endpoint
Resource	Agent 可以读取的数据	数据库里的表
Prompt	预设的提示词模板	API 的文档说明

二、环境准备

先装好需要的依赖：

# 创建虚拟环境
python -m venv mcp-env
source mcp-env/bin/activate  # Windows: mcp-env\Scripts\activate

# 安装 MCP SDK
pip install mcp

# 如果你要做数据库相关的工具，还需要
pip install aiosqlite

注意：MCP SDK 要求 Python 3.10+，低于这个版本会报错。用 python --version 先确认一下。

三、从零搭建一个 MCP Server

3.1 最小可运行示例

先跑通一个最简单的 MCP Server，确认环境没问题：

# server.py
from mcp.server.fastmcp import FastMCP

# 创建 MCP Server 实例
mcp = FastMCP("我的第一个MCP服务器")

# 定义一个 Tool
@mcp.tool()
def add(a: int, b: int) -> int:
    """两数相加"""
    return a + b

# 定义一个 Resource
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """返回个性化问候语"""
    return f"你好，{name}！欢迎使用 MCP Server。"

if __name__ == "__main__":
    mcp.run()

运行：

python server.py

就这么简单。@mcp.tool() 装饰器把一个普通 Python 函数变成了 AI Agent 可以调用的工具。

3.2 实战：搭建数据库查询工具

光说不练假把式。来搞个有实际价值的——让 AI Agent 能直接查 SQLite 数据库。

# db_server.py
import aiosqlite
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("数据库查询助手")

DB_PATH = "example.db"

async def init_db():
    """初始化示例数据库"""
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS employees (
                id INTEGER PRIMARY KEY,
                name TEXT NOT NULL,
                department TEXT NOT NULL,
                salary REAL NOT NULL,
                hire_date TEXT NOT NULL
            )
        """)
        # 插入测试数据
        data = [
            (1, '张三', '技术部', 15000, '2023-01-15'),
            (2, '李四', '市场部', 12000, '2023-03-20'),
            (3, '王五', '技术部', 18000, '2022-11-01'),
            (4, '赵六', '人事部', 10000, '2024-02-28'),
            (5, '钱七', '技术部', 22000, '2021-06-10'),
        ]
        await db.executemany(
            "INSERT OR IGNORE INTO employees VALUES (?,?,?,?,?)", data
        )
        await db.commit()

@mcp.tool()
async def query_employees(sql: str) -> str:
    """
    执行 SQL 查询员工表。
    表结构: employees(id, name, department, salary, hire_date)
    只支持 SELECT 查询，不允许修改数据。
    """
    # 安全检查：只允许 SELECT
    if not sql.strip().upper().startswith("SELECT"):
        return "错误：只允许 SELECT 查询，不支持修改操作。"

    try:
        async with aiosqlite.connect(DB_PATH) as db:
            db.row_factory = aiosqlite.Row
            async with db.execute(sql) as cursor:
                rows = await cursor.fetchall()
                if not rows:
                    return "查询结果为空。"
                # 格式化输出
                result = []
                for row in rows:
                    result.append(dict(row))
                return str(result)
    except Exception as e:
        return f"查询出错: {str(e)}"

@mcp.tool()
async def get_table_info() -> str:
    """获取数据库中所有表的结构信息"""
    async with aiosqlite.connect(DB_PATH) as db:
        async with db.execute(
            "SELECT name FROM sqlite_master WHERE type='table'"
        ) as cursor:
            tables = await cursor.fetchall()
            info = []
            for (table_name,) in tables:
                async with db.execute(f"PRAGMA table_info({table_name})") as col_cursor:
                    columns = await col_cursor.fetchall()
                    col_info = [
                        f"  {col[1]} ({col[2]})" for col in columns
                    ]
                    info.append(f"表 {table_name}:\n" + "\n".join(col_info))
            return "\n\n".join(info)

if __name__ == "__main__":
    import asyncio
    asyncio.run(init_db())
    mcp.run()

3.3 代码解读

几个关键点：

1. @mcp.tool() 装饰器：把函数变成 MCP Tool。函数的 docstring 会自动变成工具的描述，AI Agent 靠这个决定什么时候调用你。
2. 类型注解：sql: str 和 -> str 不是摆设，MCP SDK 会用这些信息生成 JSON Schema，告诉 Agent 参数格式。
3. 安全检查：query_employees 里限制了只允许 SELECT，防止 Agent 误操作删库跑路。
4. 异步支持：MCP SDK 原生支持 async/await，数据库操作用 aiosqlite 异步执行。

四、如何测试你的 MCP Server

4.1 用 Claude Desktop 测试

如果你用的是 Claude Desktop，编辑配置文件：

// ~/Library/Application Support/Claude/claude_desktop_config.json (macOS)
// %APPDATA%\Claude\claude_desktop_config.json (Windows)
{
  "mcpServers": {
    "数据库助手": {
      "command": "python",
      "args": ["/path/to/your/db_server.py"]
    }
  }
}

重启 Claude Desktop，你就能在对话中直接让它查数据库了：

“帮我查一下技术部薪资最高的人是谁？”

Claude 会自动调用 query_employees 工具，执行 SQL，返回结果。

4.2 用 MCP Inspector 调试

MCP 官方提供了一个调试工具：

npx @modelcontextprotocol/inspector python db_server.py

它会启动一个 Web 界面，你可以手动调用 Tool、查看 Resource，非常方便调试。

五、进阶：添加 Resource 和 Prompt

Tool 是让 Agent 做事，Resource 是让 Agent 读数据，Prompt 是给 Agent 模板。

# 在 db_server.py 基础上添加

@mcp.resource("schema://employees")
def get_schema() -> str:
    """返回员工表的完整结构定义"""
    return """
    表名: employees
    字段:
    - id: INTEGER (主键)
    - name: TEXT (员工姓名)
    - department: TEXT (所属部门)
    - salary: REAL (月薪)
    - hire_date: TEXT (入职日期，格式 YYYY-MM-DD)
    """

@mcp.prompt()
def analyze_salary() -> str:
    """薪资分析提示词模板"""
    return """
    你是一个数据分析助手。请根据员工数据：
    1. 计算各部门平均薪资
    2. 找出薪资异常值
    3. 给出薪资优化建议
    使用 query_employees 工具获取数据。
    """

六、常见问题 Q&A

Q1: MCP 和 Function Calling 有什么区别？

Function Calling 是模型厂商各自实现的，OpenAI 一套、Google 一套、Anthropic 一套。MCP 是统一协议，写一次，到处用。

Q2: MCP Server 支持哪些传输方式？

目前支持两种：

• stdio：通过标准输入输出通信，适合本地工具
• SSE（Server-Sent Events）：通过 HTTP 通信，适合远程服务

Q3: 生产环境能用吗？

说实话，MCP 还比较新（2024年底才开源），SDK 还在快速迭代。个人项目、内部工具完全没问题，核心业务建议先观望。

Q4: 除了 Python 还有其他语言的 SDK 吗？

有。TypeScript SDK 同样官方维护，Go、Rust 社区也有第三方实现。

七、总结

MCP 解决的是 AI Agent 生态的连接问题。以前每个工具都要写专门的对接代码，现在统一用 MCP 协议暴露能力，Agent 就能即插即用。

今天带你从零实现了：

• 一个最小 MCP Server
• 一个数据库查询工具
• Resource 和 Prompt 的用法
• Claude Desktop 集成和调试方法

下一步你可以试试：

• 接入真实的数据库（MySQL、PostgreSQL）
• 把内部 API 封装成 MCP Tool
• 用 SSE 传输方式做远程 MCP Server

代码已上传 GitHub，欢迎 Star。

---

本文为实战教程，所有代码均已在本地验证通过。如有问题欢迎评论区交流。