MCP 协议实战：让你的 AI Agent 像调用函数一样操作本地文件、数据库和 API

读完这篇，你就能搭一个能读写文件、查数据库、调外部 API 的 AI Agent——不用写一堆胶水代码，靠 MCP 协议一个标准搞定。

---

一、为什么你需要关注 MCP？

2024 年底，Anthropic 开源了 MCP（Model Context Protocol）。简单说，它是 AI 模型与外部工具之间的"USB-C 接口"——定义了统一的客户端-服务端通信协议，让任何 AI 应用都能用同一种方式调用任何工具。

以前想让 GPT 读你的本地文件？ 你得自己写一套文件操作函数 → 注册成 function calling tool → 处理权限 → 格式化返回。每个工具都这样来一遍，累死。

MCP 之后？ 社区已经提供了几十个现成的 MCP Server——文件系统、数据库、GitHub、浏览器、Slack……你只需一行配置，Agent 就能直接调用。自己写一个 MCP Server 也不到 100 行代码。

本文用 Python 带你从零搭一个能用的 MCP Server，然后连上 AI Agent，让它操作你的本地文件。

---

二、MCP 协议架构：三板斧讲明白

AI 应用 (Host)
  |
  +-- MCP Client (文件工具)
  |     |
  |     +-- JSON-RPC over stdio --> 文件系统 Server
  |
  +-- MCP Client (数据库工具)
  |     |
  |     +-- JSON-RPC over stdio --> 数据库 Server
  |
  +-- MCP Client (API工具)
        |
        +-- JSON-RPC over HTTP/SSE --> 第三方 API Server

角色	做什么	类比
Host	AI 应用本体（Claude Desktop / 你的 Python 脚本）	手机
MCP Client	与 Server 通信的协议客户端，每个 Client 连一个 Server	USB-C 数据线
MCP Server	提供具体能力的服务端（文件 / 数据库 / API）	外设（U盘 / 键盘 / 显示器）

三个核心概念：

概念	说明	你的理解方式
Resources	Server 暴露的"只读数据"	相当于 GET 请求，Agent 读取文件内容、查表结构
Tools	Server 暴露的"可执行操作"	相当于 POST 请求，Agent 创建文件、执行 SQL、调 API
Prompts	Server 预置的提示词模板	相当于快捷指令，Agent 可以直接用它来引导对话

通信协议是 JSON-RPC 2.0，传输层支持两种：

• stdio：本机进程通信，适合本地工具
• HTTP + SSE：远程服务，适合部署到服务器

---

三、从零搭建 MCP Server：让 Agent 操作你的文件系统

3.1 安装 MCP SDK

pip install mcp

3.2 先写一个最小 Server：读取文件内容

server.py：

import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

# 创建 MCP Server 实例
server = Server("my-filesystem-tools")

# ===== 注册一个 Tool：读取文件 =====
@server.list_tools()
async def handle_list_tools() -> list[Tool]:
    return [
        Tool(
            name="read_file",
            description="读取指定路径的文件内容。返回文件中的全部文本。",
            inputSchema={
                "type": "object",
                "properties": {
                    "filepath": {
                        "type": "string",
                        "description": "要读取的文件绝对路径"
                    }
                },
                "required": ["filepath"]
            }
        )
    ]

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "read_file":
        filepath = arguments["filepath"]
        try:
            with open(filepath, "r", encoding="utf-8") as f:
                content = f.read()
            return [TextContent(type="text", text=content)]
        except FileNotFoundError:
            return [TextContent(type="text", text=f"错误：文件不存在 - {filepath}")]
        except PermissionError:
            return [TextContent(type="text", text=f"错误：没有读取权限 - {filepath}")]
    
    raise ValueError(f"未知工具: {name}")

# ===== 启动 Server =====
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, server.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

运行测试——用 MCP 自带的 Inspector 工具调试：

npx @modelcontextprotocol/inspector python server.py

浏览器打开会看到一个调试界面，你可以直接调用 read_file 工具，输入路径，看返回值。

3.3 扩展：加上写文件和列出目录

# 在 handle_list_tools 中添加两个新 Tool
Tool(
    name="write_file",
    description="将内容写入指定路径的文件。会覆盖已有文件。",
    inputSchema={
        "type": "object",
        "properties": {
            "filepath": {"type": "string", "description": "要写入的文件绝对路径"},
            "content": {"type": "string", "description": "要写入的文本内容"}
        },
        "required": ["filepath", "content"]
    }
),
Tool(
    name="list_directory",
    description="列出指定目录下的所有文件和子目录",
    inputSchema={
        "type": "object",
        "properties": {
            "dirpath": {"type": "string", "description": "目录绝对路径，留空则列出当前工作目录"}
        }
    }
)

# 在 handle_call_tool 中添加对应处理
elif name == "write_file":
    filepath = arguments["filepath"]
    content = arguments["content"]
    os.makedirs(os.path.dirname(filepath) or ".", exist_ok=True)
    with open(filepath, "w", encoding="utf-8") as f:
        f.write(content)
    return [TextContent(type="text", text=f"✅ 已写入 {len(content)} 个字符到 {filepath}")]

elif name == "list_directory":
    dirpath = arguments.get("dirpath") or os.getcwd()
    entries = os.listdir(dirpath)
    result = "\n".join(f"  {'📁' if os.path.isdir(os.path.join(dirpath, e)) else '📄'} {e}" for e in entries)
    return [TextContent(type="text", text=f"目录 {dirpath} 的内容：\n{result}")]

现在你的 MCP Server 已经能读、写、列目录了——总共不到 80 行。

---

四、接入 AI Agent：让 Claude 操作你的文件

MCP Server 写好了，接下来让 AI 用起来。这里演示两个主流方案。

4.1 方案一：Claude Desktop 直接接入（最省事）

Claude Desktop 原生支持 MCP。编辑配置文件：

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

{
  "mcpServers": {
    "my-filesystem": {
      "command": "python",
      "args": ["/path/to/server.py"]
    }
  }
}

重启 Claude Desktop，你会看到工具图标亮起 🔨。然后直接对 Claude 说：

“帮我读取 ~/projects/README.md 的内容，总结出项目结构，然后新建一个 NOTES.md 把总结写进去。”

Claude 会自动：

1. 调用你的 read_file 读 README
2. 分析内容生成总结
3. 调用你的 write_file 写入 NOTES.md

全程不需要你写任何胶水代码。

4.2 方案二：用 Python 代码自主控制

如果你想在自己的应用中集成 MCP，不用 Claude Desktop：

# client.py - 在你的 Python 应用中调用 MCP Server
import asyncio
from mcp.client import Client
from mcp.client.stdio import stdio_client, StdioServerParameters

async def main():
    # 连接到本地 MCP Server
    server_params = StdioServerParameters(
        command="python",
        args=["server.py"]
    )
    
    async with stdio_client(server_params) as (read, write):
        client = Client(read, write)
        await client.initialize()
        
        # 列出可用工具
        tools = await client.list_tools()
        print(f"可用工具: {[t.name for t in tools]}")
        
        # 调用 read_file 工具
        result = await client.call_tool("read_file", {
            "filepath": "/Users/me/test.txt"
        })
        print(f"文件内容: {result.content[0].text}")

asyncio.run(main())

---

五、进阶：接入数据库，Agent 能查 MySQL 了

真正有用的 Agent 必须能跟数据打交道。写一个 SQLite MCP Server：

import sqlite3
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

server = Server("sqlite-tools")
DB_PATH = "./app.db"

@server.list_tools()
async def handle_list_tools() -> list[Tool]:
    return [
        Tool(
            name="execute_sql",
            description="执行一条 SQL 查询语句（仅支持 SELECT）。返回查询结果的 JSON 格式。",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "要执行的 SELECT SQL 语句"
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="list_tables",
            description="列出数据库中所有表及其字段信息",
            inputSchema={"type": "object", "properties": {}}
        )
    ]

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[TextContent]:
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    
    if name == "list_tables":
        cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
        tables = cursor.fetchall()
        result = []
        for (table_name,) in tables:
            cursor.execute(f"PRAGMA table_info({table_name})")
            columns = cursor.fetchall()
            cols_str = ", ".join(f"{c[1]} ({c[2]})" for c in columns)
            result.append(f"📊 {table_name}: {cols_str}")
        conn.close()
        return [TextContent(type="text", text="\n".join(result) or "数据库为空")]
    
    elif name == "execute_sql":
        query = arguments["query"].strip()
        if not query.upper().startswith("SELECT"):
            conn.close()
            return [TextContent(type="text", text="⚠️ 安全限制：仅允许 SELECT 查询")]
        try:
            cursor.execute(query)
            rows = cursor.fetchall()
            columns = [desc[0] for desc in cursor.description]
            result = [dict(zip(columns, row)) for row in rows]
            conn.close()
            return [TextContent(type="text", text=str(result))]
        except Exception as e:
            conn.close()
            return [TextContent(type="text", text=f"SQL 错误: {str(e)}")

配上 Claude Desktop 配置：

{
  "mcpServers": {
    "sqlite-db": {
      "command": "python",
      "args": ["/path/to/sqlite_server.py"]
    }
  }
}

然后你就能对 Claude 说：

“数据库里的 users 表，帮我查一下注册超过 30 天但还没激活的用户数量。”

Agent 会先调 list_tables 看表结构，再调 execute_sql 写查询，全程自动。

---

六、MCP 现状：能打了吗？

维度	状态	备注
协议成熟度	✅ 稳定	JSON-RPC 2.0，无 breaking changes
Python SDK	✅ 可用	pip install mcp 即可
TypeScript SDK	✅ 可用	@modelcontextprotocol/sdk
社区 Server 数量	🟡 30+	文件系统、GitHub、Postgres、Slack、Brave Search 等
Claude Desktop 集成	✅ 原生	配置即用
GPT / OpenAI 原生支持	❌ 暂不支持	需通过 LangChain / 自建 Client 桥接
远程部署 (HTTP)	🟡 实验性	2026 年中刚出，生产慎用

结论：本地 Agent 场景已经完全可以投产。 远程部署还需观望，但本地文件/数据库/API 的工具链已经足够你搭出很能打的 Agent。

---

七、三个生产环境注意点

1. 权限隔离

上面那个文件系统 Server 可以读写任意路径——Agent 如果被注入恶意 prompt，可能读你的 ~/.ssh/id_rsa。上线前必须加沙箱：

ALLOWED_DIRS = ["/home/user/projects", "/home/user/downloads"]

def is_safe_path(filepath: str) -> bool:
    abs_path = os.path.abspath(filepath)
    return any(abs_path.startswith(d) for d in ALLOWED_DIRS)

2. 并发处理

MCP Server 默认单线程。如果 Agent 同时调两个工具，会串行执行。高并发场景用 asyncio 或上 HTTP 传输：

# 用 HTTP + SSE 替代 stdio
from mcp.server.sse import SseServerTransport

3. 错误粒度

Agent 的决策质量高度依赖工具返回的错误信息。别只返回 "操作失败"，要给 Agent 足够的信息去调整策略：

# ❌ 差
return [TextContent(type="text", text="操作失败")]

# ✅ 好
return [TextContent(type="text", text=f"文件 {filepath} 不存在。同目录下有以下文件：{siblings}。要读取其中某一个吗？")]

---

八、总结

MCP 的价值一句话：你不用再为每个工具写胶水代码了。 定义好 Tool 的 schema → Agent 自己决定什么时候调用、传什么参数、怎么利用返回值。

下一步值得做的是：

• 给你的工具加 Resources（比如把数据库表结构暴露为 resource，Agent 启动时就能感知上下文）
• 尝试 Multi-Server 架构——一个 Agent 同时连文件系统 Server + 数据库 Server + API Server，协作完成复杂任务
• 关注 MCP 的 Remote Server 进展——一旦稳定，你就能把 MCP Server 部署到云上，多个 Agent 共享同一套工具

---

下一篇预告：用 LangGraph 把 MCP Server 串成 Multi-Agent 流水线——爬虫 Agent 采集数据 → 分析 Agent 处理 → 报告 Agent 输出，全程自动。关注不迷路。