MCP 协议开发实战:从零搭建 AI Agent 工具链
1. 引言
随着大语言模型能力的持续增强,AI Agent 正在从「能聊天」走向「能干活」。然而,要让 Agent 真正调用外部工具、访问业务数据,开发者面临一个核心难题:如何让模型与工具之间高效、标准化地通信?MCP(Model Context Protocol,模型上下文协议)正是为解决这一问题而生的开放协议。
本文将带你从零开始,理解 MCP 的核心概念,动手实现一个完整的 MCP Server,并将其接入 AI Agent,最终搭建出一条可复用的工具链。无论你是后端工程师、AI 应用开发者,还是对 Agent 架构感兴趣的爱好者,本文都能为你提供一条清晰、可落地的实战路径。
2. MCP 协议核心概念
在动手写代码之前,先建立对 MCP 的整体认知。本节介绍 MCP 是什么、解决什么问题,以及它的核心架构与通信模型。
2.1 什么是 MCP
MCP 是由 Anthropic 于 2024 年底提出的开放协议,旨在为 AI 应用(Host)与外部工具/数据源(Server)之间建立标准化的连接方式。你可以把它理解为「AI 世界的 USB-C 接口」:只要设备支持这个标准,就能即插即用。
2.2 MCP 解决的核心问题
在没有 MCP 之前,每个 Agent 接入一个工具,都需要定制一套 API 封装、鉴权逻辑和调用协议,重复造轮子且难以复用。MCP 通过统一协议,将「工具定义、调用、结果返回」标准化,让工具开发者只需实现一次 Server,即可被任意支持 MCP 的 Host 复用。
2.3 核心架构与角色
MCP 架构中包含三个关键角色:
- Host:AI 应用本身,如 Claude Desktop、自研 Agent,负责与用户交互并调度模型。
- Client:Host 内部的连接组件,负责与 Server 建立会话、收发消息。
- Server:工具/数据源的提供方,暴露标准化的工具、资源和提示词。
三者关系如下图所示:
2.4 通信模型与消息格式
MCP 基于 JSON-RPC 2.0 进行通信,支持两种传输方式:
- stdio:Server 作为子进程启动,通过标准输入输出通信,适合本地开发。
- HTTP + SSE:Server 作为独立服务部署,通过 HTTP 长连接通信,适合生产环境。
每条消息包含 jsonrpc、method、params、id 等字段,协议定义了初始化、工具列表、工具调用等标准方法。
3. 环境准备与项目初始化
本节搭建开发环境,创建项目骨架,为后续编码做好准备。
3.1 开发环境要求
- Python 3.10+ 或 Node.js 18+
- 一个支持 MCP 的 Host(如 Claude Desktop,或使用官方 MCP Inspector 调试)
- 包管理工具(pip / npm)
3.2 初始化项目
以 Python 为例,创建项目目录并安装官方 SDK:
mkdir mcp-agent-toolkit
cd mcp-agent-toolkit
python -m venv .venv
source .venv/bin/activate
pip install mcp
3.3 验证 SDK 安装
python -c "import mcp; print(mcp.__version__)"
4. 实现第一个 MCP Server
本节从零实现一个功能完整的 MCP Server,包含工具定义、注册与运行。
4.1 定义工具
我们实现一个「获取天气」的工具,演示工具定义、参数校验与结果返回:
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
app = Server("weather-server")
@app.list_tools()
async def list_tools():
return [
Tool(
name="get_weather",
description="查询指定城市的实时天气",
inputSchema={
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"}
},
"required": ["city"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "get_weather":
city = arguments["city"]
# 这里替换为真实天气 API 调用
result = f"{city} 今日晴,25°C,微风"
return [TextContent(type="text", text=result)]
raise ValueError(f"未知工具: {name}")
4.2 启动 Server
async def main():
async with stdio_server() as (read_stream, write_stream):
await app.run(read_stream, write_stream)
if __name__ == "__main__":
import asyncio
asyncio.run(main())
4.3 用 MCP Inspector 调试
MCP 官方提供了可视化调试工具 Inspector,可以快速验证 Server 的工具列表与调用结果:
mcp dev server.py
5. 将 Server 接入 AI Agent
Server 就绪后,本节演示如何将其接入一个真实的 AI Agent,让模型能够自主调用工具。
5.1 配置 Host 连接
以 Claude Desktop 为例,在配置文件中声明 MCP Server:
{
"mcpServers": {
"weather": {
"command": "python",
"args": ["server.py"],
"cwd": "/path/to/mcp-agent-toolkit"
}
}
}
5.2 在 Agent 中调用工具
在自研 Agent 中,通过 SDK 连接 Server 并获取工具列表:
from mcp.client.stdio import stdio_client
from mcp.client.session import ClientSession
async def query_tool():
async with stdio_client(["python", "server.py"]) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
tools = await session.list_tools()
result = await session.call_tool("get_weather", {"city": "北京"})
print(result)
5.3 完整工具链流程
6. 进阶:构建多工具组合与状态管理
单个工具只是起点,真实场景往往需要多个工具协作。本节介绍如何扩展 Server、管理工具间状态,并处理错误。
6.1 注册多个工具
在 list_tools 中返回多个 Tool 对象,并在 call_tool 中按 name 分发即可。建议将每个工具封装为独立函数,便于维护与测试。
6.2 工具间共享状态
当工具需要共享数据(如用户会话、缓存)时,可在 Server 内部维护一个状态对象:
class ToolContext:
def __init__(self):
self.cache = {}
context = ToolContext()
@app.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "save_note":
context.cache[arguments["key"]] = arguments["value"]
return [TextContent(type="text", text="已保存")]
if name == "get_note":
value = context.cache.get(arguments["key"], "未找到")
return [TextContent(type="text", text=value)]
6.3 错误处理与超时
工具调用可能失败或超时,Server 应返回结构化错误信息,Host 侧也应设置合理的超时与重试策略:
@app.call_tool()
async def call_tool(name: str, arguments: dict):
try:
# 业务逻辑
pass
except Exception as e:
return [TextContent(type="text", text=f"工具执行失败: {str(e)}")]
7. 生产环境部署与安全
从本地原型走向生产,需要解决部署形态、鉴权、审计等工程问题。
7.1 部署形态选择
- 本地 stdio:适合个人工具、开发调试。
- 远程 HTTP + SSE:适合团队共享、服务化部署,需配合网关做鉴权与限流。
7.2 安全最佳实践
- 所有工具调用必须经过鉴权,禁止 Server 暴露敏感操作。
- 对工具入参做严格校验,防止注入攻击。
- 记录完整调用日志,便于审计与排障。
- 对第三方 API 调用设置超时与熔断。
7.3 可观测性
为 Server 接入日志、指标与链路追踪,便于定位问题:
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp-server")
@app.call_tool()
async def call_tool(name: str, arguments: dict):
logger.info("调用工具 %s, 参数 %s", name, arguments)
# ...
8. 总结与展望
本文从 MCP 的核心概念出发,完整走通了「定义工具 → 实现 Server → 接入 Agent → 多工具协作 → 生产部署」的实战链路。MCP 的价值在于标准化:一次实现,处处复用,让 AI Agent 的工具生态得以快速生长。
未来,MCP 生态会持续演进,工具市场、跨组织共享、更丰富的资源类型都将成为现实。建议你从一个小工具开始,亲手跑通这条链路,再逐步扩展自己的 Agent 工具链。
更多推荐


所有评论(0)