Agent能聊天了,但调不了我的本地数据

前段时间团队在做内部知识库的AI助手,Agent能联网搜索,能调公开API,但一碰到"帮我查一下上个月订单库的退款率"就哑了。数据在本地SQLite里,Agent够不着。当时想了几个方案——把数据库暴露成REST API?数据同步到云端?都不太对味。后来接触到MCP Server,简单说就是让AI能调你本地服务的标准化协议。本文从零开始手写一个MCP Server,注册几个Tool:一个查本地数据库,一个搜本地文件。代码都能直接跑,踩过的坑也一并写上。

MCP Server 一句话定位

先说结论:MCP Server 做的事就是把本地服务包装成AI能理解、能调用的Tool接口。结构大概是这样的:

AI Client ←→ MCP Client ←→ MCP Server ←→ Tool A: query_database
                                      ←→ Tool B: search_files

MCP Client 负责跟你的AI框架(LangChain、OpenAI SDK、自建Agent等)交互;MCP Server 负责跟你的本地资源交互。你只需要写Server端的Tool,Client端的通信协议MCP SDK已经封装好了。协议细节这里不展开,够写Server就行。

环境准备与传输方式选型

环境只需要装一个包就行:

# Python 3.10+
pip install mcp

Server启动后跟Client之间的通信方式有好几种。我当时调研了一圈,整理了一张选型表:

传输方式 适用场景 优点 缺点
stdio 本地CLI工具、同机部署 零配置,最稳定 Client和Server必须在同一台机器
SSE 内网服务、远程AI Agent 标准HTTP,可跨机器 需要Web服务容器
自定义传输 特殊协议、高安全要求 灵活度高 维护成本最高

我的经验是初期先用stdio起步,调试成本最低。跑通了再考虑要不要切SSE。这次代码就用stdio。

Step 1:初始化 Server 并注册第一个 Tool

先搭架子,让Server能正常启动。这段代码解决的问题是:确认环境没问题,Client能连上来。这样后面加正式功能的时候,基础链路已经是通的——如果这里跑不通,后面加再多功能都没用。

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

# 创建 MCP Server 实例,使用 stdio 传输
mcp = FastMCP("local-tools", transport="stdio")

@mcp.tool()
def health_check() -> str:
    """Check if the server is running and accessible."""
    return "ok"

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

运行后终端进入等待状态,表示Server已启动。你可以用MCP Client来连一下看效果:

# client_test.py
from mcp.client import StdioClient

async def main():
    async with StdioClient(["python", "server.py"]) as client:
        result = await client.call_tool("health_check", {})
        print(result)  # 输出: ok

如果输出 ok,就说明你的MCP环境通了——这时Agent已经可以通过Client跟本地服务通信了,虽然只是打了个招呼。不ok的话,常见问题是Python版本或mcp包没装好,先检查这两项。

翻车案例 1:Tool 参数描述写太简略

第一个Tool跑通后,我开始加真正有用的Tool,但很快就翻车了。看看这几版写法导致的差异:

Tool 描述写法 AI 实际行为 问题
“查询数据库” AI 传了 DROP TABLE orders AI 以为这是通用SQL接口,什么都能跑
“执行SELECT查询,参数为完整SQL语句” AI 只传 SELECT 语句 范围限制了,但偶尔传错表名
“查询订单数据,可指定日期范围和状态” AI 只传合理的 WHERE 条件 给了参数格式示例,AI照做

我的第一个版本是这样的:

@mcp.tool()
def query_database(sql: str) -> str:
    """查询数据库"""
    # ...

结果AI传了什么?它传了 "SELECT * FROM users; DROP TABLE users;"。倒不是说AI有恶意,而是Tool的description对AI来说就是API文档——你写"查询数据库",AI就以为这是个执行任意SQL的接口。

根因分析: MCP协议的Tool描述会被直接注入到AI的system prompt里。AI根据description和参数名来"猜测"这个Tool的用途和边界。description越模糊,AI的猜测空间越大,翻车概率就越高。回过头来看确实是我想简单了——凭啥AI知道"查询数据库"就等于"只读查询"?

修正后的写法,把边界条件写在description第一句,参数给一个完整的示例值:

@mcp.tool()
def query_database(sql: str) -> str:
    """Execute SELECT queries on the local order database.
    
    IMPORTANT: Only SELECT statements are supported.
    INSERT, UPDATE, DELETE, DROP and other DDL/DML will be rejected.
    
    Args:
        sql: A complete SELECT SQL statement, e.g. "SELECT * FROM orders WHERE date > '2026-01-01'"
    """
    import sqlite3
    if not sql.strip().upper().startswith("SELECT"):
        return json.dumps({"error": "Only SELECT queries are permitted"})
    conn = sqlite3.connect("orders.db")
    try:
        cursor = conn.execute(sql)
        rows = cursor.fetchall()
        columns = [desc[0] for desc in cursor.description]
        return json.dumps([dict(zip(columns, row)) for row in rows])
    except Exception as e:
        return json.dumps({"error": str(e)})
    finally:
        conn.close()

这样AI看一眼就知道该怎么传——实际调用时,AI会传类似 "SELECT order_id, amount FROM orders WHERE date > '2026-01-01'" 这样的SQL,不再触碰INSERT或DROP。后来我把这个规律总结成一个原则:Tool的描述要像给完全不了解你业务的人写API文档一样详细,不能留任何"常识性"的猜测空间。

Step 2:注册文件搜索 Tool

数据库之外,文件系统是另一个常见的本地资源。比如Agent问"帮我找一下上个月的周报文件",这时候就需要文件搜索能力。下面这个Tool接收目录名和文件通配符,返回匹配的文件路径列表,外加一些安全检查来防止不必要的风险。

import os
import glob
import json

@mcp.tool()
def search_files(directory: str, pattern: str = "*.md") -> str:
    """Search for files in the local documents directory.
    
    Only searches within the /data/documents base directory.
    Results are limited to protect system resources.
    
    Args:
        directory: Subdirectory name within /data/documents, e.g. "reports/2026"
        pattern: Glob pattern, e.g. "*.md", "*.pdf", "weekly_*.xlsx"
    """
    BASE = os.path.abspath("/data/documents")
    target = os.path.abspath(os.path.join(BASE, directory))
    
    # 防止目录穿越攻击
    if not target.startswith(BASE):
        return json.dumps({"error": "Access denied: directory outside allowed scope"})
    
    if not os.path.exists(target):
        return json.dumps({"error": f"Directory not found: {directory}"})
    
    matches = glob.glob(os.path.join(target, pattern), recursive=True)
    # 限制返回数量,防止AI循环调用把磁盘撑爆
    limited = [os.path.relpath(p, BASE) for p in matches[:30]]
    return json.dumps({"files": limited, "total": len(matches)})

运行后效果是这样的:

输入: search_files(directory="reports/2026", pattern="weekly_*.md")
输出: {"files": ["reports/2026/weekly_01.md", "reports/2026/weekly_02.md"], "total": 12}

这段代码加了两层防护:目录穿越检查(防止AI被注入 ../../etc/passwd 之类的路径)和结果数量限制。AI拿到文件列表后,可以对文件做后续操作——比如读取内容、总结摘要。不过这里得注意,glob 在大目录下可能比较慢,几十万文件的场景不太适合。

翻车案例 2:MCP 协议版本不一致

Step 1 和 Step 2 写完,我满心以为能跑通。结果Client连上来报了这个错:

Error: Unsupported protocol version. Server supports: 2024-11-05, Client supports: 2025-01-01

查了一圈发现MCP协议在2025年初有一次传输层规范更新。我装的最新版 mcp SDK默认用了新版协议,但Client那边引用的旧版本SDK还在用老的。两边的protocol version对不上,握手失败。

根因分析: MCP协议还在快速迭代中,当前的做法是每次连接握手时双方交换支持的版本号列表,取交集。如果Server和Client的SDK版本差了超过一个minor版本,交集大概率为空。这个问题在早期尤其频繁——我后来统计了一下,那段时间遇到版本不兼容的大概占了排查问题的将近一半。

修复办法其实很简单:Server和Client用同一个版本的 mcp 包。比较稳妥的做法是在项目里锁版本:

pip install mcp==1.2.0

然后在 requirements.txt 里写明依赖版本。后来我跟同事聊,他说等协议稳定了可能不用这么小心,但现阶段还是得看着。

整体验证

写完之后完整的启动流程就这几步:先启动 Server(终端进入等待状态),然后用 Client 或 MCP CLI 测试连接,最后依次调用 query_databasesearch_files 看能不能拿到正确结果。如果两个Tool都返回了预期的数据,链路就是通的。

验证通过后,你就有了一个跑在本地的MCP Server,AI Agent可以通过它查询数据库和搜索文件。整个上手过程前前后后差不多花了一天——写代码占了大半,排查协议版本问题用了一小半。

局限性与复用建议

自己写MCP Server不是万能的,有几个场景可能需要换个思路:

  • 只有少量工具且在单机运行 —— 自己写Server最合适,成本最低
  • 工具多了需要外部访问 —— 考虑用MCP官方市场现成的Server,或者用社区工具包装
  • 对安全要求极高(金融/医疗) —— 自己写并且严格审计每个Tool的权限边界
  • 团队前端为主不太熟Python —— MCP也有Node.js SDK,可以换Node写

另外还有一个容易被忽略的点:MCP Server本身不处理认证和授权。如果你的本地服务需要鉴权,你得自己在Tool实现里加,协议层不管这个。


回头看,踩坑的核心就两个:Tool描述要像给不懂业务的人写API文档一样详细依赖版本一定锁死。前一个是设计问题,后一个是工程问题。做MCP Server这两件事都绕不开。

Logo

欢迎加入 MCP 技术社区!与志同道合者携手前行,一同解锁 MCP 技术的无限可能!

更多推荐