MCP协议开发实战：从零搭建AI Agent工具链

tuddy789464

94人浏览 · 2026-07-21 02:00:58

tuddy789464 · 2026-07-21 02:00:58 发布

一、引言：为什么需要MCP协议？

1.1 AI Agent工具链的挑战

异构工具集成困难
上下文管理与资源消耗
安全与权限控制

1.2 MCP协议的核心价值

标准化工具定义与调用
解耦Agent核心与工具实现
提升开发效率与可维护性

二、 MCP协议核心概念解析

2.1 协议架构概览

Server（工具提供方）
Client（Agent/调用方）
Transport Layer（通信层）

2.2 核心资源与工具

Resource（数据资源）
Tool（可执行操作）
Prompt（提示模板）

2.3 通信模型与生命周期

初始化与发现
请求/响应模式
错误处理与状态管理

三、实战准备：环境与工具栈

3.1 开发环境搭建

Node.js/Python 环境配置
MCP SDK 安装与初始化

3.2 核心依赖介绍

@modelcontextprotocol/sdk (JavaScript/TypeScript)
mcp (Python)
调试与测试工具

四、从零构建你的第一个MCP Server

4.1 项目初始化与配置

创建项目结构
定义server manifest

4.2 实现核心工具（Tool）

定义工具schema（输入/输出）
编写工具执行逻辑
示例：天气查询工具

下面是一个完整的 Python 代码示例，演示如何实现一个天气查询工具，包含工具 schema 定义、执行逻辑和模拟 API 调用：

import asyncio
from typing import Any, Dict, Optional
from mcp import Server, Tool, ToolResult
from mcp.types import Tool as ToolSchema
class WeatherQueryTool:
"""天气查询工具实现类"""
def __init__(self, api_key: Optional[str] = None):
    """
    初始化天气查询工具
Args:
    api_key: 模拟的 API 密钥，实际项目中替换为真实天气服务 API 密钥
"""
self.api_key = api_key or "demo_key"
self.cache = {}  # 简单的缓存，避免重复查询相同城市
def get_tool_schema(self) -&gt; ToolSchema:
"""
定义工具 schema，描述工具的输入参数和输出格式
Returns:
    ToolSchema: 工具 schema 对象
"""
return ToolSchema(
    name="get_weather",
    description="查询指定城市的当前天气信息",
    inputSchema={
        "type": "object",
        "properties": {
            "city": {
                "type": "string",
                "description": "城市名称，例如：北京、上海、New York"
            },
            "country_code": {
                "type": "string",
                "description": "国家代码（可选），例如：CN、US",
                "default": "CN"
            }
        },
        "required": ["city"]
    }
)
async def execute(self, arguments: Dict[str, Any]) -&gt; ToolResult:
"""
执行天气查询工具
Args:
    arguments: 工具调用参数，包含 city 和可选的 country_code
    
Returns:
    ToolResult: 工具执行结果
    
Raises:
    ValueError: 当城市名称为空或无效时
"""
# 1. 参数验证与提取
city = arguments.get("city", "").strip()
country_code = arguments.get("country_code", "CN").upper()

if not city:
    raise ValueError("城市名称不能为空")

# 2. 检查缓存（模拟）
cache_key = f"{city}_{country_code}"
if cache_key in self.cache:
    return ToolResult(
        content=[{
            "type": "text",
            "text": f"【缓存结果】{city}({country_code})的天气：{self.cache[cache_key]}"
        }],
        isError=False
    )

# 3. 模拟 API 调用（实际项目中替换为真实天气 API）
try:
    weather_data = await self._simulate_weather_api_call(city, country_code)
    
    # 4. 处理 API 响应
    result_text = self._format_weather_result(weather_data)
    
    # 5. 更新缓存（模拟）
    self.cache[cache_key] = result_text
    
    return ToolResult(
        content=[{
            "type": "text",
            "text": result_text
        }],
        isError=False
    )
    
except Exception as e:
    # 6. 错误处理
    return ToolResult(
        content=[{
            "type": "text",
            "text": f"查询天气失败：{str(e)}"
        }],
        isError=True
    )
async def _simulate_weather_api_call(self, city: str, country_code: str) -&gt; Dict[str, Any]:
"""
模拟天气 API 调用
Args:
    city: 城市名称
    country_code: 国家代码
    
Returns:
    Dict: 模拟的天气数据
    
Note:
    实际项目中应替换为真实的天气服务 API 调用，如 OpenWeatherMap、和风天气等
"""
# 模拟网络延迟
await asyncio.sleep(0.5)

# 模拟不同城市的天气数据
weather_templates = {
    "北京": {"temp": 25, "humidity": 60, "condition": "晴", "wind_speed": 3.5},
    "上海": {"temp": 28, "humidity": 75, "condition": "多云", "wind_speed": 4.2},
    "广州": {"temp": 32, "humidity": 80, "condition": "阵雨", "wind_speed": 5.0},
    "深圳": {"temp": 30, "humidity": 78, "condition": "阴", "wind_speed": 3.8},
    "杭州": {"temp": 27, "humidity": 70, "condition": "晴转多云", "wind_speed": 2.5}
}

# 返回模拟数据或默认数据
if city in weather_templates:
    return weather_templates[city]
else:
    # 默认模拟数据
    return {
        "temp": 26 + hash(city) % 10,  # 模拟温度变化
        "humidity": 65 + hash(city) % 20,
        "condition": ["晴", "多云", "阴", "小雨"][hash(city) % 4],
        "wind_speed": 3.0 + (hash(city) % 30) / 10
    }
def _format_weather_result(self, weather_data: Dict[str, Any]) -&gt; str:
"""
格式化天气查询结果
Args:
    weather_data: 原始天气数据
    
Returns:
    str: 格式化的天气信息字符串
"""
return f"""🌤️ 天气查询结果：
• 温度：{weather_data['temp']}°C
• 湿度：{weather_data['humidity']}%
• 天气状况：{weather_data['condition']}
• 风速：{weather_data['wind_speed']} m/s
（数据来源：模拟天气服务，实际使用时请接入真实 API）"""
示例：在 MCP Server 中注册工具
async def create_weather_server():
"""创建包含天气查询工具的 MCP Server"""
server = Server("weather-server")
weather_tool = WeatherQueryTool(api_key="your_api_key_here")
注册工具
await server.register_tool(
name=weather_tool.get_tool_schema().name,
description=weather_tool.get_tool_schema().description,
input_schema=weather_tool.get_tool_schema().inputSchema,
handler=weather_tool.execute
)
return server
if name == "main":
启动服务器示例
server = asyncio.run(create_weather_server())
print("天气查询 MCP Server 已启动")

代码说明：

工具 Schema 定义：通过 get_tool_schema() 方法定义工具的输入参数（城市名称、国家代码）和输出格式。
执行逻辑：execute() 方法包含完整的工具执行流程：参数验证 → 缓存检查 → API 调用 → 结果处理 → 错误处理。
模拟 API 调用：_simulate_weather_api_call() 模拟了真实的网络请求和延迟，实际项目中可替换为 OpenWeatherMap、和风天气等真实 API。
MCP 集成：create_weather_server() 展示了如何在 MCP Server 中注册这个工具，使其能够被 Client 发现和调用。
错误处理：包含参数验证、API 调用异常处理，确保工具的健壮性。