MCP协议开发实战：从零搭建AI Agent数据库工具链，让Agent真正懂你的数据

qq_39914918

181人浏览 · 2026-07-27 09:39:51

qq_39914918 · 2026-07-27 09:39:51 发布

背景

最近在做一个AI Agent项目，需要让Agent能跟MySQL数据库直接对话。说白了就是——用户用自然语言问一句「上个月销售额最高的三个产品是什么」，Agent自己翻译成SQL，查数据库，再返回结果。

听起来很酷对吧？但坑是真的多。

问题在哪？

一开始我打算让Agent直接调数据库驱动。但仔细一想，这玩意儿风险太大了——Agent自己写SQL，万一来个 DROP TABLE 怎么办？就算不搞破坏，一个 SELECT * FROM orders 查几百万行数据，数据库直接炸了。

后来发现业界有个好东西叫 MCP（Model Context Protocol），专门解决AI模型跟外部工具之间的通信问题。简单说就是给Agent一个「沙箱」，Agent只能通过MCP定义好的接口去操作数据库，不能越界。

实战：搭建MCP数据库工具链

环境准备

我用的是 Python 3.11 + FastMCP 库，MySQL 8.0。

pip install fastmcp pymysql python-dotenv

定义MCP工具

核心思路是：给Agent暴露几个安全的数据库操作接口，全部是只读的：

from fastmcp import FastMCP
import pymysql

mcp = FastMCP("db-agent")

@mcp.tool()
def query_database(sql: str) -> list[dict]:
    """执行SQL查询，注意：仅允许SELECT语句"""
    sql_upper = sql.strip().upper()
    if not sql_upper.startswith("SELECT"):
        return [{"error": "只允许执行SELECT查询"}]
    if len(sql) > 500:
        return [{"error": "查询语句过长，请简化"}]
    conn = pymysql.connect(host="localhost", user="reader", password="***", database="sales_db")
    try:
        with conn.cursor(pymysql.cursors.DictCursor) as cursor:
            cursor.execute(sql)
            rows = cursor.fetchmany(100)
            return rows
    finally:
        conn.close()

关键点：

强制SQL白名单，只允许SELECT
限制查询长度
限制返回行数
用只读账号连接数据库

注册MCP服务

@mcp.tool()
def describe_tables() -> list[dict]:
    """返回数据库所有表的结构信息，帮助Agent理解数据模型"""
    conn = pymysql.connect(host="localhost", user="reader", password="***", database="sales_db")
    try:
        with conn.cursor() as cursor:
            cursor.execute("SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = 'sales_db'")
            return [{"table": row[0], "column": row[1], "type": row[2]} for row in cursor.fetchall()]
    finally:
        conn.close()

@mcp.tool()
def get_table_sample(table_name: str, limit: int = 5) -> list[dict]:
    """获取表的示例数据"""
    safe_table = table_name.replace(';', '').replace('--', '')
    return query_database(f"SELECT * FROM {safe_table} LIMIT {min(limit, 10)}")

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

数据库工具链架构图

效果演示

假设你有一个电商数据库，包含 orders、products、customers 三个表。Agent通过MCP工具获取表结构后，可以这样工作：

用户提问： “今年7月销量TOP5的商品是什么？”

Agent内部流程：

调用 describe_tables() 了解表结构
发现 orders 表有 product_id、quantity、order_date 字段
调用 get_table_sample() 确认数据格式
生成SQL并调用 query_database() 执行

输出结果：

2026年7月销量排名前5的商品为：

无线蓝牙耳机（1,243件）

智能手表Pro（987件）

便携充电宝（765件）

降噪头戴耳机（543件）

桌面支架（412件）

实际查询效果截图

踩坑记录

1. 中文编码问题

Agent生成的SQL里如果有中文，从MySQL查出来是乱码。解决方案：连接时指定charset：

pymysql.connect(..., charset="utf8mb4")

2. Agent生成的SQL太复杂

有时候Agent会生成带子查询、JOIN太多的SQL，性能极差。加个超时限制：

import signal

class TimeoutException(Exception):
    pass

def query_with_timeout(sql, timeout=5):
    signal.signal(signal.SIGALRM, lambda: (_ for _ in ()).throw(TimeoutException("查询超时")))
    signal.alarm(timeout)
    try:
        return query_database(sql)
    finally:
        signal.alarm(0)

3. 上下文窗口问题

表结构太多时，Agent的上下文窗口会不够用。解决方案：只返回最相关的表结构，而不是全部。

@mcp.tool()
def search_tables(keyword: str) -> list[dict]:
    """按关键词搜索相关表和字段"""
    conn = pymysql.connect(...)
    try:
        with conn.cursor() as cursor:
            cursor.execute("""
                SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE
                FROM INFORMATION_SCHEMA.COLUMNS
                WHERE TABLE_SCHEMA = 'sales_db'
                  AND (TABLE_NAME LIKE %s OR COLUMN_NAME LIKE %s)
            """, (f"%{keyword}%", f"%{keyword}%"))
            return [{"table": row[0], "column": row[1], "type": row[2]} for row in cursor.fetchall()]
    finally:
        conn.close()