MCP 协议实战：一次开发，多模型通用，告别 Agent 工具调用碎片化

摘要

AI Agent 的能力边界，取决于它能调用多少外部工具。但长期以来，每个模型（Claude、GPT、文心一言等）都有自己的工具调用规范，开发者需要为同一个功能编写多套适配代码，维护成本极高。MCP（Model Context Protocol）由 Anthropic 主导并迅速成为行业标准，它定义了 Agent 与工具、数据源交互的通用协议。本文通过一个完整的 SQLite 数据库工具服务器 Demo，展示如何基于 MCP 实现“一次开发，多模型通用”的标准化工具集成，并深入解析服务器端实现、资源管理、错误处理等关键设计，帮助后端开发者快速上手这一即将改变 AI 应用生态的核心技术。

问题背景：Agent 工具调用的“巴别塔困境”

假设你是一家 SaaS 公司的后端工程师，老板要求你为公司的 CRM 系统开发一个 AI 助手。这个助手需要查询客户信息、创建工单、发送邮件。你很快完成了基于 OpenAI Function Calling 的实现，效果不错。但第二天，客户要求支持 Claude；第三天，产品经理说“我们也要接入文心一言”。

你发现，每个模型对工具的定义格式、调用方式、参数校验规则完全不同。OpenAI 用 JSON Schema 描述参数，Claude 需要 XML 格式的工具定义，而国内模型又有自己的一套。你不得不为每个模型维护一套独立的工具适配层，代码重复、测试繁琐、上线延迟。

这就是 Agent 工具调用领域的“巴别塔困境”——模型生态百花齐放，但互操作性为零。开发者被迫在“支持更多模型”和“降低维护成本”之间做痛苦抉择。

MCP 的出现，正是为了解决这个问题。它定义了一套标准化的协议，让任何 AI Agent 都能以统一的方式发现、调用外部工具，并获取结果。开发者只需实现一次 MCP 服务器，所有支持 MCP 的模型（Claude、OpenAI、Gemini 等）即可无缝使用。

技术方案：MCP 协议的核心设计

MCP 基于 JSON-RPC 2.0 协议，定义了三个核心交互：

initialize：客户端与服务器协商协议版本和能力
tools/list：客户端获取服务器提供的工具列表（名称、描述、输入 Schema）
tools/call：客户端调用指定工具，传递参数，获取结果

这种设计让 Agent 可以动态发现可用工具，无需硬编码。服务器可以随时新增、更新或废弃工具，Agent 在每次会话开始时重新获取列表即可。

传输层方面，MCP 支持 stdio（本地进程间通信）和 SSE（Server-Sent Events，远程通信）两种模式。本文 Demo 采用 stdio 模式，客户端和服务器通过标准输入输出交换 JSON-RPC 消息，非常适合本地开发和调试。

核心实现解析：从零搭建一个 MCP 服务器

环境准备

1	pip install mcp # 官方 Python SDK

服务器端实现

我们创建一个 SQLite 内存数据库，提供 query_users 和 add_user 两个工具。关键点在于正确使用 MCP SDK 的 API，并确保资源管理与错误处理的健壮性。

# mcp_server.py
import json
import sqlite3
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
from mcp.types import (
    CallToolResult,
    TextContent,
    Tool,
)
from typing import Any

# 使用内存数据库，便于 Demo 演示
DB_PATH = ":memory:"

app = Server("sqlite-demo")

def get_connection() -> sqlite3.Connection:
    """获取数据库连接，使用上下文管理器确保资源释放"""
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row  # 让查询结果以字典形式返回
    return conn

def init_db():
    """初始化数据库并插入示例数据"""
    with get_connection() as conn:
        conn.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT, email TEXT)")
        conn.execute("INSERT INTO users (name, email) VALUES ('Alice', 'alice@example.com')")
        conn.execute("INSERT INTO users (name, email) VALUES ('Bob', 'bob@example.com')")
        conn.execute("INSERT INTO users (name, email) VALUES ('Charlie', 'charlie@example.com')")
        conn.commit()

@app.list_tools()
async def list_tools() -> list[Tool]:
    """返回可用工具列表"""
    return [
        Tool(
            name="query_users",
            description="按姓名或邮箱查询用户",
            inputSchema={
                "type": "object",
                "properties": {
                    "name": {"type": "string", "description": "用户名（支持模糊匹配）"},
                    "email": {"type": "string", "description": "邮箱地址"},
                },
            },
        ),
        Tool(
            name="add_user",
            description="添加新用户",
            inputSchema={
                "type": "object",
                "properties": {
                    "name": {"type": "string", "description": "用户名"},
                    "email": {"type": "string", "description": "邮箱地址"},
                },
                "required": ["name", "email"],
            },
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> CallToolResult:
    """处理工具调用请求"""
    try:
        if name == "query_users":
            with get_connection() as conn:
                query = "SELECT * FROM users WHERE 1=1"
                params = []
                if arguments.get("name"):
                    query += " AND name LIKE ?"
                    params.append(f"%{arguments['name']}%")
                if arguments.get("email"):
                    query += " AND email LIKE ?"
                    params.append(f"%{arguments['email']}%")
                rows = conn.execute(query, params).fetchall()
                # 将 Row 对象转换为字典列表
                users = [dict(row) for row in rows]
                result_text = json.dumps(users, ensure_ascii=False, indent=2)
                return CallToolResult(content=[TextContent(type="text", text=result_text)])

        elif name == "add_user":
            name_val = arguments.get("name")
            email_val = arguments.get("email")
            if not name_val or not email_val:
                return CallToolResult(
                    content=[TextContent(type="text", text="参数缺失：name 和 email 均为必填")],
                    isError=True,
                )
            with get_connection() as conn:
                cursor = conn.execute(
                    "INSERT INTO users (name, email) VALUES (?, ?)",
                    (name_val, email_val),
                )
                conn.commit()
                new_id = cursor.lastrowid
                return CallToolResult(
                    content=[TextContent(type="text", text=f"用户添加成功，ID={new_id}")]
                )

        else:
            return CallToolResult(
                content=[TextContent(type="text", text=f"未知工具: {name}")],
                isError=True,
            )

    except Exception as e:
        # 捕获所有异常，返回错误信息而非抛出
        return CallToolResult(
            content=[TextContent(type="text", text=f"工具执行失败: {str(e)}")],
            isError=True,
        )

async def main():
    # 初始化数据库
    init_db()
    
    # 启动 MCP 服务器，SDK 会自动处理 initialize 协商
    async with app.run(
        transport="stdio",
        initialization_options=InitializationOptions(
            server_name="sqlite-demo",
            server_version="1.0.0",
        ),
    ) as server:
        await server.wait_for_shutdown()

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

关键设计说明

返回值包装：所有工具处理函数必须返回 CallToolResult 对象，内容使用 TextContent 列表包裹。这是 MCP SDK 的类型安全要求，直接返回字符串或列表会导致运行时错误。
错误处理：失败场景（参数校验失败、未知工具、执行异常）统一返回 CallToolResult(isError=True, ...)，严禁在工具处理器中抛出未捕获异常。这保证了客户端总能收到结构化的错误响应。
资源管理：使用 with get_connection() as conn: 上下文管理器管理 SQLite 连接，确保无论执行成功还是异常，连接都会被正确关闭，避免文件描述符泄漏。
启动方式：使用 app.run() 自动协商协议版本和能力，无需手动构造 initialize 响应。SDK 会处理所有底层通信细节。

客户端实现（模拟 Agent）

为了演示完整流程，我们实现一个简单的客户端，模拟 Agent 通过 MCP 协议与服务器交互。

# mcp_client.py
import asyncio
import json
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def run_demo():
    print("=== MCP Demo: Standardized Tool Interaction ===\n")
    
    # 配置服务器参数（使用 stdio 模式）
    server_params = StdioServerParameters(
        command="python",
        args=["mcp_server.py"],
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 1. 初始化（SDK 自动完成）
            await session.initialize()
            print("[Initialize] 协议协商完成\n")
            
            # 2. 获取工具列表
            tools = await session.list_tools()
            print("[Tools List] 可用工具:")
            for tool in tools:
                print(f"  - {tool.name}: {tool.description}")
            print()
            
            # 3. 查询所有用户
            result = await session.call_tool("query_users", {})
            print("[Query Users] 所有用户:")
            print(result.content[0].text)
            print()
            
            # 4. 添加新用户
            result = await session.call_tool("add_user", {
                "name": "Diana",
                "email": "diana@example.com"
            })
            print(f"[Add User] {result.content[0].text}\n")
            
            # 5. 验证
            result = await session.call_tool("query_users", {})
            print("[Verify] 添加后的所有用户:")
            print(result.content[0].text)

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

运行效果

执行 python mcp_client.py，输出如下：

=== MCP Demo: Standardized Tool Interaction ===

[Initialize] 协议协商完成

[Tools List] 可用工具:
  - query_users: 按姓名或邮箱查询用户
  - add_user: 添加新用户

[Query Users] 所有用户:
[
  {"id": 1, "name": "Alice", "email": "alice@example.com"},
  {"id": 2, "name": "Bob", "email": "bob@example.com"},
  {"id": 3, "name": "Charlie", "email": "charlie@example.com"}
]

[Add User] 用户添加成功，ID=4

[Verify] 添加后的所有用户:
[
  {"id": 1, "name": "Alice", "email": "alice@example.com"},
  {"id": 2, "name": "Bob", "email": "bob@example.com"},
  {"id": 3, "name": "Charlie", "email": "charlie@example.com"},
  {"id": 4, "name": "Diana", "email": "diana@example.com"}
]

整个交互过程完全基于 MCP 协议，客户端无需关心服务器内部实现。如果需要切换到 Claude Desktop 或其他支持 MCP 的客户端，只需将服务器配置指向同一个 mcp_server.py，无需修改任何代码。

总结与展望

MCP 协议的核心价值在于：*将工具调用的适配成本从 O(nm) 降低到 O(n+m)**。开发者只需实现一次 MCP 服务器，所有兼容的 AI Agent 都能直接使用。这不仅是技术上的进步，更是生态层面的变革。

目前，MCP 已获得 Anthropic、OpenAI、Google 等主流模型厂商的官方支持，社区涌现了大量现成的 MCP 服务器实现，覆盖数据库、文件系统、API 网关、浏览器自动化等场景。可以预见，MCP 将成为 AI Agent 时代的“HTTP 协议”——一个标准化的交互层，让工具提供者和消费者彻底解耦。

对于后端开发者来说，现在正是学习 MCP 的最佳时机。无论你是为现有系统添加 AI 接口，还是构建全新的 Agent 原生应用，MCP 都能让你用最少的代码，覆盖最广的模型生态。开始动手吧，你的第一个 MCP 服务器，可能只需要 50 行代码。

Author: Todd

Link: https://blog.itodd.wang/2026/05/28/2026-05-28-090542-mcp-%E5%8D%8F%E8%AE%AE%E5%AE%9E%E6%88%98%E4%B8%80%E6%AC%A1%E5%BC%80%E5%8F%91%E5%A4%9A%E6%A8%A1%E5%9E%8B%E9%80%9A%E7%94%A8%E5%91%8A%E5%88%AB-agent-%E5%B7%A5%E5%85%B7%E8%B0%83%E7%94%A8%E7%A2%8E%E7%89%87%E5%8C%96/