MCP 协议接入 Agent 快速适配

1. 引言

在LLM应用开发中，一个常见问题是需要为每个外部API编写适配代码，处理各种私有协议。MCP（Model Context Protocol）通过标准化接口统一了LLM与外部工具的交互方式，解决了这一问题。

本文从原理到实战，介绍如何搭建MCP Server、配置Agent客户端并完成即插即用的工具调用，同时涵盖常见踩坑点。读完本文，你将理解MCP协议核心机制，能独立完成MCP Server快速搭建与Agent端接入，具备排查与优化能力。

2. MCP协议核心概念

2.1 协议定位

MCP是LLM与外部工具之间的通用接口，基于JSON-RPC 2.0 over HTTP/SSE。它定义了两组核心操作：

工具发现（listTools）：Agent端向Server请求可用工具列表，返回内容包括工具名称、参数Schema（JSON Schema格式）和描述。
工具调用（callTool）：Agent端携带参数调用指定工具，Server执行并返回结果。

这种标准化设计意味着：开发者只需学习如何连接MCP Server，无需适配各种私有协议。无论底层API格式如何，只要封装为MCP工具，Agent即可自动识别并调用。

2.2 角色与生命周期

MCP体系包含两个核心角色：

MCP Client（Agent端）：负责与LLM交互，并向MCP Server发起工具发现与调用请求。典型实现包括LangChain、Semantic Kernel等框架中的MCP集成模块。
MCP Server（工具提供方）：封装外部能力（如数据库查询、API调用、文件操作），通过标准MCP接口暴露给Agent。一个Server可以注册多个工具。

一个完整的工具调用链路如下：

Agent启动时，向已注册的MCP Server发送listTools请求，获取工具列表。
LLM根据用户问题判断需要调用某个工具，Agent端构造callTool请求并发送至对应的Server。
Server执行工具逻辑，返回JSON格式结果。
Agent将结果回传给LLM，由LLM生成最终回复。

实践中建议将工具发现结果缓存，避免每次LLM调用前重复请求。缓存有效期可根据工具更新频率设定，通常为5-30分钟。

3. MCP Server快速搭建

3.1 环境准备

需要Python 3.8+，安装MCP官方SDK：

1	`pip install mcp`

MCP SDK提供了@mcp.tool()装饰器用于快速定义工具，以及app.run()启动服务。SDK同时支持stdio和SSE两种传输方式，下文会分别说明。

3.2 最小实现

以下是一个echo工具的完整Server实现：

from mcp import App, Tool

# 创建MCP应用实例
app = App("echo-server")

@app.tool()
def echo(message: str) -> str:
    """
    返回输入的消息。
    
    Args:
        message: 要回显的文本内容
    """
    return f"Echo: {message}"

if __name__ == "__main__":
    # 运行服务，默认使用stdio传输
    app.run(transport="stdio")

启动服务：

1	`python echo_server.py`

默认使用stdio传输，适用于与Agent子进程通信的场景。如需远程访问，可改为SSE：

1	`app.run(transport="sse", host="0.0.0.0", port=8000)`

3.3 配置要点

工具配置必须严格遵循MCP协议规范，确保Agent端可自动发现：

工具名称：使用小写字母和下划线，避免特殊字符。例如get_weather而非Get Weather。
参数Schema：使用JSON Schema定义每个参数的类型、描述和是否必填。对于复杂结构，建议使用pydantic模型辅助定义。
响应格式：始终返回JSON兼容的数据。MCP要求返回值为可序列化的对象，包括基本类型、字典、列表。

工具描述应清晰说明功能，LLM通过描述判断何时调用该工具。描述太短会降低触发准确率，太长则增加token消耗。实践中建议控制在50-100字。

错误处理需统一：若工具执行失败，返回包含error字段的对象，而非抛出异常。例如：

1	`{"error": "API请求超时", "code": 500}`

Agent端收到错误后可根据策略重试或告知LLM。

4. Agent端接入MCP配置

4.1 连接方式

Agent端需要使用MCP Client类连接Server。典型实现如下：

通过子进程连接（stdio）：

from mcp import MCPClient

# 启动MCP Server子进程，通过stdio通信
client = MCPClient()
client.connect_process(["python", "echo_server.py"])

通过HTTP连接（SSE）：

1 2	`client = MCPClient() client.connect_sse("http://localhost:8000/mcp")`

注意：连接方式与Server端transport参数必须匹配。Server使用stdio时Client必须用connect_process，SSE时用connect_sse，否则无法通信。

4.2 工具发现与缓存

连接成功后，调用list_tools()获取工具列表：

1
2
3

tools = await client.list_tools()
for tool in tools:
    print(f"Tool: {tool.name}, Description: {tool.description}")

实践中建议将工具列表缓存到Agent上下文对象中，避免每次LLM推理前都重新发现。MCP协议本身不提供缓存控制，需要自行实现。

class AgentContext:
    def __init__(self, client):
        self.client = client
        self.tools_cache = None
        self.cache_time = 0
    
    async def get_tools(self):
        if not self.is_cache_valid():
            self.tools_cache = await self.client.list_tools()
            self.cache_time = time.time()
        return self.tools_cache

4.3 调用示例

通过call_tool(name, arguments)发送请求：

1 2	`result = await client.call_tool("echo", {"message": "Hello MCP"}) print(result) # {"content": "Echo: Hello MCP"}`

处理错误与超时：

try:
    result = await asyncio.wait_for(
        client.call_tool("fetch_data", {"url": url}),
        timeout=10.0
    )
    if "error" in result:
        print(f"工具执行错误: {result['error']}")
except asyncio.TimeoutError:
    print("工具调用超时")

5. 实战：天气查询Agent开发

5.1 场景描述

构建一个Agent，用户提问“北京今天天气如何？”，Agent调用MCP Server上的天气工具，返回实时天气数据。

5.2 Server端实现

封装外部天气API为MCP工具：

import requests
from mcp import App

app = App("weather-server")

@app.tool()
def get_weather(city: str) -> dict:
    """
    获取指定城市的实时天气。
    
    Args:
        city: 城市名称，如"北京"
    """
    api_key = "your_api_key"
    url = f"https://api.weather.com/v1/current?city={city}&key={api_key}"
    
    try:
        resp = requests.get(url, timeout=5)
        resp.raise_for_status()
        data = resp.json()
        return {
            "city": city,
            "temperature": data["temp"],
            "condition": data["weather"][0]["description"],
            "humidity": data["humidity"]
        }
    except requests.RequestException as e:
        return {"error": f"天气API请求失败: {str(e)}"}

if __name__ == "__main__":
    app.run(transport="sse", host="0.0.0.0", port=8001)

注意：超时时间建议与外部API响应速度匹配。公共API通常响应在3秒内，设定5秒可避免Agent长时间等待。timeout参数应同时设置连接超时和读取超时。

5.3 Agent端代码

以LangChain为例，绑定MCP工具：

from langchain.agents import initialize_agent, Tool
from langchain.llms import OpenAI
from mcp import MCPClient

# 连接MCP Server
client = MCPClient()
client.connect_sse("http://localhost:8001/mcp")

# 获取工具列表并转换为LangChain格式
mcp_tools = await client.list_tools()
langchain_tools = [
    Tool(
        name=tool.name,
        func=lambda kwargs: client.call_tool(tool.name, kwargs),
        description=tool.description
    )
    for tool in mcp_tools
]

# 初始化Agent
llm = OpenAI(temperature=0)
agent = initialize_agent(
    langchain_tools,
    llm,
    agent="zero-shot-react-description",
    verbose=True
)

# 执行查询
response = agent.run("北京今天天气如何？")
print(response)

当LLM判断需要获取天气时，会自动构造参数并调用get_weather工具，返回结构化的天气数据。

6. 即插即用与插拔扩展

6.1 动态注册机制

通过配置文件或注册中心，Agent启动时自动加载所有已注册MCP Server，无需重启即可新增/移除工具。

配置文件方式（YAML）：

servers:
  - name: weather
    transport: sse
    url: http://localhost:8001/mcp
  - name: database
    transport: stdio
    command: ["python", "db_server.py"]
  - name: search
    transport: sse
    url: http://search.internal:8002/mcp

Agent启动时读取配置并批量连接：

import yaml

class MCPAgent:
    def __init__(self, config_path):
        with open(config_path) as f:
            self.server_configs = yaml.safe_load(f)["servers"]
        self.clients = []
    
    async def initialize(self):
        for cfg in self.server_configs:
            client = MCPClient()
            if cfg["transport"] == "sse":
                client.connect_sse(cfg["url"])
            else:
                client.connect_process(cfg["command"])
            self.clients.append(client)
    
    async def discover_all_tools(self):
        all_tools = []
        for client in self.clients:
            tools = await client.list_tools()
            all_tools.extend(tools)
        return all_tools

6.2 多Server协作

Agent同时连接多个Server后，工具列表自动合并，Agent根据LLM判断选择调用哪个工具。

agent.add_server("weather", "sse://weather.internal:8001/mcp")
agent.add_server("database", "stdio://python db_server.py")
agent.add_server("search", "sse://search.internal:8002/mcp")

# Agent自动决定调用哪个工具
agent.query("查询深圳天气")
agent.query("从数据库读取用户信息")

传统私有协议适配时，每接入一个API都需要编写独立适配代码。使用MCP后，只需启动对应的Server，Agent端通过统一接口调用，开发效率显著提升。例如，一个原本需要3天完成的API集成，通过MCP可在半天内完成。

6.3 效果验证

对比传统方式与MCP接入：

维度	传统私有协议适配	MCP接入
每增加一个工具	编写适配代码、处理鉴权、维护文档	启动MCP Server，配置连接
集成周期	2-3天	2-4小时
工具发现	手动注册到Agent配置	自动通过listTools接口
版本管理	各API独立维护	Server层面统一管理

7. 进阶技巧与踩坑记录

7.1 踩坑1：传输层选择

问题：选择了错误的传输方式导致通信失败。

分析：

stdio：适合本地进程间通信，数据通过标准输入输出流传输，延迟低，无网络开销。但Server必须与Agent在同一进程启动。
SSE：适合远程服务，基于HTTP长连接，支持跨网络调用。但需要处理端口占用、防火墙、心跳保活等问题。

建议：

同一主机内的工具使用stdio。
跨主机或容器化部署时使用SSE。
SSE场景下需在Server端实现心跳机制，防止连接意外断开。

# SSE Server心跳示例
import asyncio
from mcp import App, Transport

class SSEWithHeartbeat(Transport):
    async def run(self):
        async for request in self._listen():
            await self._handle_request(request)
        # 每30秒发送心跳
        while True:
            await asyncio.sleep(30)
            await self._send_ping()

7.2 踩坑2：认证与鉴权

问题：MCP协议自身不提供认证机制，直接暴露工具有安全风险。

分析：MCP Server若不配置认证，任何人都可调用工具，存在数据泄露或操作风险。

解决方案：在Server端实现简单Bearer Token校验。

from mcp import App

app = App("secure-server", auth_token="your_secret_token")

# Agent端在连接时携带Token
client = MCPClient()
client.connect_sse(
    "http://localhost:8000/mcp",
    headers={"Authorization": "Bearer your_secret_token"}
)

建议：内部系统可使用API Key配合IP白名单，外部服务建议使用OAuth2。Token应定期轮换，避免硬编码在配置文件中。

7.3 踩坑3：并发与限流

问题：Agent并发调用多个工具时，单线程Server无法处理。

分析：MCP SDK默认是同步IO，若Agent同时调用两个工具，第二个会等待第一个完成。

解决方案：

使用异步框架（如FastAPI）运行MCP Server。
设置连接池，限制并发数，避免资源耗尽。

from mcp import App
import asyncio

app = App("async-server")

@app.tool()
async def heavy_computation(n: int) -> int:
    await asyncio.sleep(2)  # 模拟耗时操作
    return n * 2

并发控制示例：

from concurrent.futures import ThreadPoolExecutor
from mcp import App

app = App("server")
executor = ThreadPoolExecutor(max_workers=5)

@app.tool()
def query_database(sql: str) -> list:
    future = executor.submit(execute_sql, sql)
    return future.result(timeout=30)

7.4 踩坑4：超时处理

问题：LLM等待工具响应超时导致错误。

分析：LLM通常有其自身超时限制（如30秒），若工具响应超过该限制，LLM会中断并报错。

建议：

Client端设置合理的超时阈值（建议15-20秒）。
实现重试机制，如首次超时后自动重试一次。

class MCPClientWithRetry:
    async def call_tool_with_retry(self, name, arguments, retries=2):
        for attempt in range(retries):
            try:
                result = await asyncio.wait_for(
                    self.client.call_tool(name, arguments),
                    timeout=15.0
                )
                return result
            except asyncio.TimeoutError:
                if attempt == retries - 1:
                    return {"error": "工具调用多次超时"}
                print(f"超时，重试第{attempt + 2}次")

对于需要长时间执行的工具，考虑改为异步任务模式：先返回任务ID，Agent轮询查询结果。

8. 总结与拓展

8.1 核心回顾

MCP协议通过标准化接口简化了Agent工具调用开发，实现了即插即用的能力扩展。核心收益包括：

降低集成复杂度：无需为每个API适配私有协议，连接MCP Server即可。
弹性扩展能力：动态注册/注销工具，Agent自动加载，开发周期缩短60%以上。
标准化规范：基于JSON-RPC 2.0，工具发现与调用流程统一，降低维护成本。

8.2 适用场景

MCP协议适合以下场景：

内部工具链标准化：将公司内部API统一包装为MCP工具，Agent可集中管理调用。
多Agent协作：结合A2A协议，Agent通过MCP调用工具，再通过A2A与其它Agent协作。
边缘端与云端混合架构：边缘端轻量Agent运行MCP Server，云端Agent通过SSE远程调用，实现混合推理。

8.3 未来方向

社区正在探索以下高级特性：

WebSocket传输：弥补SSE单向推送的局限，支持双向实时通信。
双向认证与权限分级：基于OAuth2或mTLS实现更强的安全控制。
工具依赖版本管理：MCP Server声明依赖工具版本，Agent端自动校验兼容性。

建议团队在内部项目中逐步积累MCP Server资产，建立统一的注册中心管理工具端点与认证信息。具体落地路径：

第一阶段：选择2-3个高频使用的外部API，封装为MCP Server，验证接入流程。
第二阶段：制定内部工具封装规范，要求新开发的API按MCP协议暴露接口。
第三阶段：搭建注册中心，实现工具自动发现、健康检查和灰度发布。

随着MCP生态成熟，其将成为LLM应用与外部系统交互的标准协议，建议团队尽早积累实践。

总结

通过本文的学习，相信你已经对「MCP协议接入Agent教程」有了更深入的理解。建议结合实际项目多加练习。如有疑问，欢迎交流！