实验三：高级MCP开发实战 🔧
"真正的专家不是知道所有答案的人，而是知道如何找到答案的人。" - 在MCP开发中，调试和优化技能决定了你能走多远！
🎯 实验概述
欢迎来到MCP开发的"深水区"！在前面的实验中，我们学会了使用现成的MCP服务器。现在，是时候撸起袖子，亲手打造属于自己的MCP服务器了！
就像从使用轮子到制造轮子的转变，这个实验将让你从MCP的"使用者"变成"创造者"。我们将使用最新的工具链，构建一个完整的天气MCP服务器，并掌握专业的调试和优化技巧。
⏱️ 预计用时： 35分钟
🎓 难度等级： ⭐⭐⭐⭐☆
🛠️ 技术栈： Python 3.10+ + MCP SDK 1.9.3 + MCP Inspector + AI Toolkit
📋 学习目标
完成这个实验后，你将成为MCP开发的"工匠大师"：
- [ ] 自定义服务器开发：从零构建生产级MCP服务器
- [ ] 最新SDK掌握：熟练使用MCP Python SDK v1.9.3
- [ ] 专业调试技能：使用MCP Inspector进行高效调试
- [ ] AI Toolkit集成：将自定义服务器集成到Agent Builder
- [ ] 性能优化技巧：编写高效、稳定的MCP代码
- [ ] 错误处理专家：构建健壮的异常处理机制
🏗️ 技术栈深度解析
🐍 MCP Python SDK v1.9.3 - 开发利器
最新版本的MCP Python SDK带来了许多激动人心的新特性：
python
# 新版本SDK的优雅语法
from mcp import Server
from mcp.types import Tool, TextContent, ImageContent
import asyncio
# 创建服务器实例（新语法更简洁）
server = Server("my-awesome-server")
# 装饰器风格的工具注册（更Pythonic）
@server.tool()
async def my_tool(param: str) -> list[TextContent]:
"""工具描述会自动成为schema的一部分"""
return [TextContent(type="text", text=f"处理结果: {param}")]
# 启动服务器（支持多种传输方式）
async def main():
await server.run(transport="stdio") # 或 "sse", "websocket"
if __name__ == "__main__":
asyncio.run(main())🚀 新版本亮点：
| 特性 | v1.8.x | v1.9.3 | 提升 |
|---|---|---|---|
| 装饰器支持 | ❌ | ✅ | 代码更简洁 |
| 类型提示增强 | 基础 | 完整 | IDE支持更好 |
| 错误处理 | 手动 | 自动 | 健壮性提升 |
| 性能优化 | 基础 | 高级 | 响应速度提升30% |
| 调试友好 | 有限 | 丰富 | 开发效率翻倍 |
🔍 MCP Inspector - 调试神器
MCP Inspector是开发者的"显微镜"，让MCP服务器的内部工作一目了然：
🌦️ 实战项目：智能天气助手MCP服务器
🎯 项目需求分析
我们将构建一个功能丰富的天气MCP服务器，具备以下能力：
🔧 核心功能模块
| 功能模块 | 技术实现 | 用户价值 |
|---|---|---|
| 🌤️ 当前天气查询 | OpenWeatherMap API | 实时天气信息 |
| 📅 天气预报 | 7天预报数据 | 出行规划助手 |
| 🌡️ 温度趋势分析 | 数据分析算法 | 科学决策支持 |
| ⚠️ 天气预警 | 极端天气检测 | 安全保障 |
| 📊 数据可视化 | ASCII图表生成 | 直观信息展示 |
| 🌍 多城市支持 | 地理编码服务 | 全球覆盖 |
🎨 用户体验设计
python
# 用户交互示例
"""
用户: 今天北京的天气怎么样？
AI助手: 🌤️ 北京今日天气情况：
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
🌡️ 温度: 25°C (体感 27°C)
☁️ 天气: 多云转晴
💧 湿度: 60%
💨 风速: 3.2 m/s (东南风)
📊 气压: 1013 hPa
🔆 紫外线指数: 6 (中等)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
💡 穿衣建议: 适合穿轻薄外套
🚗 出行建议: 天气良好，适合出行
"""步骤一：环境准备与依赖安装
🔧 创建项目环境
bash
# 创建项目目录
mkdir weather-mcp-server
cd weather-mcp-server
# 创建Python虚拟环境
python -m venv venv
# 激活虚拟环境
# Windows:
venv\Scripts\activate
# macOS/Linux:
source venv/bin/activate
# 升级pip
python -m pip install --upgrade pip📦 安装核心依赖
bash
# 安装MCP SDK和相关依赖
pip install mcp>=1.9.3
pip install httpx>=0.25.0
pip install pydantic>=2.0.0
pip install python-dotenv>=1.0.0
# 安装开发工具
pip install pytest>=7.0.0
pip install black>=23.0.0
pip install mypy>=1.5.0
# 生成requirements.txt
pip freeze > requirements.txt🔍 安装MCP Inspector
bash
# 安装Node.js和MCP Inspector
npm install -g @modelcontextprotocol/inspector@0.14.0
# 验证安装
mcp-inspector --version步骤二：核心服务器架构设计
🏗️ 项目结构规划
weather-mcp-server/
├── src/
│ ├── weather_server.py # 主服务器文件
│ ├── weather_api.py # 天气API封装
│ ├── data_models.py # 数据模型定义
│ ├── utils.py # 工具函数
│ └── config.py # 配置管理
├── tests/
│ ├── test_weather_server.py # 服务器测试
│ └── test_weather_api.py # API测试
├── requirements.txt # 依赖文件
├── .env.example # 环境变量示例
├── README.md # 项目文档
└── pyproject.toml # 项目配置📝 数据模型设计
首先，让我们定义清晰的数据模型：
python
# src/data_models.py
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime
from enum import Enum
class WeatherCondition(str, Enum):
"""天气状况枚举"""
CLEAR = "clear"
CLOUDS = "clouds"
RAIN = "rain"
SNOW = "snow"
THUNDERSTORM = "thunderstorm"
DRIZZLE = "drizzle"
MIST = "mist"
FOG = "fog"
class CurrentWeather(BaseModel):
"""当前天气数据模型"""
city: str = Field(description="城市名称")
country: str = Field(description="国家代码")
temperature: float = Field(description="当前温度(°C)")
feels_like: float = Field(description="体感温度(°C)")
humidity: int = Field(description="相对湿度(%)", ge=0, le=100)
pressure: int = Field(description="大气压强(hPa)")
wind_speed: float = Field(description="风速(m/s)")
wind_direction: int = Field(description="风向(度)", ge=0, le=360)
visibility: int = Field(description="能见度(米)")
uv_index: Optional[float] = Field(description="紫外线指数")
condition: WeatherCondition = Field(description="天气状况")
condition_description: str = Field(description="天气描述")
timestamp: datetime = Field(default_factory=datetime.now)
class WeatherForecast(BaseModel):
"""天气预报数据模型"""
date: str = Field(description="日期")
temperature_max: float = Field(description="最高温度(°C)")
temperature_min: float = Field(description="最低温度(°C)")
condition: WeatherCondition = Field(description="主要天气状况")
condition_description: str = Field(description="天气描述")
humidity: int = Field(description="湿度(%)")
precipitation_probability: int = Field(description="降水概率(%)")
class WeatherAlert(BaseModel):
"""天气预警模型"""
alert_type: str = Field(description="预警类型")
severity: str = Field(description="严重级别")
title: str = Field(description="预警标题")
description: str = Field(description="详细描述")
start_time: datetime = Field(description="开始时间")
end_time: datetime = Field(description="结束时间")步骤三：天气API封装实现
🌐 API客户端设计
python
# src/weather_api.py
import httpx
import asyncio
from typing import Optional, List, Dict, Any
from datetime import datetime, timedelta
import os
from .data_models import CurrentWeather, WeatherForecast, WeatherAlert
from .config import WeatherConfig
class WeatherAPIError(Exception):
"""天气API异常"""
pass
class WeatherAPIClient:
"""天气API客户端"""
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key or os.getenv("OPENWEATHER_API_KEY")
if not self.api_key:
raise WeatherAPIError("缺少OpenWeatherMap API密钥")
self.base_url = "https://api.openweathermap.org/data/2.5"
self.geo_url = "http://api.openweathermap.org/geo/1.0"
# 配置HTTP客户端
self.client = httpx.AsyncClient(
timeout=httpx.Timeout(30.0),
limits=httpx.Limits(max_connections=10)
)
async def get_coordinates(self, city: str, country: str = "CN") -> tuple[float, float]:
"""获取城市坐标"""
try:
response = await self.client.get(
f"{self.geo_url}/direct",
params={
"q": f"{city},{country}",
"limit": 1,
"appid": self.api_key
}
)
response.raise_for_status()
data = response.json()
if not data:
raise WeatherAPIError(f"找不到城市: {city}")
location = data[0]
return location["lat"], location["lon"]
except httpx.HTTPError as e:
raise WeatherAPIError(f"地理编码API请求失败: {str(e)}")
async def get_current_weather(self, city: str, country: str = "CN") -> CurrentWeather:
"""获取当前天气"""
try:
# 1. 获取坐标
lat, lon = await self.get_coordinates(city, country)
# 2. 获取天气数据
response = await self.client.get(
f"{self.base_url}/weather",
params={
"lat": lat,
"lon": lon,
"appid": self.api_key,
"units": "metric",
"lang": "zh_cn"
}
)
response.raise_for_status()
data = response.json()
# 3. 转换为数据模型
return CurrentWeather(
city=data["name"],
country=data["sys"]["country"],
temperature=data["main"]["temp"],
feels_like=data["main"]["feels_like"],
humidity=data["main"]["humidity"],
pressure=data["main"]["pressure"],
wind_speed=data["wind"].get("speed", 0),
wind_direction=data["wind"].get("deg", 0),
visibility=data.get("visibility", 10000),
condition=data["weather"][0]["main"].lower(),
condition_description=data["weather"][0]["description"]
)
except httpx.HTTPError as e:
raise WeatherAPIError(f"天气API请求失败: {str(e)}")
except KeyError as e:
raise WeatherAPIError(f"天气数据解析失败: {str(e)}")
async def get_weather_forecast(
self,
city: str,
days: int = 5,
country: str = "CN"
) -> List[WeatherForecast]:
"""获取天气预报"""
try:
# 获取坐标
lat, lon = await self.get_coordinates(city, country)
# 获取预报数据
response = await self.client.get(
f"{self.base_url}/forecast",
params={
"lat": lat,
"lon": lon,
"appid": self.api_key,
"units": "metric",
"lang": "zh_cn",
"cnt": days * 8 # 每天8个时间点（3小时间隔）
}
)
response.raise_for_status()
data = response.json()
# 按日期分组并提取每日摘要
daily_forecasts = {}
for item in data["list"]:
date = datetime.fromtimestamp(item["dt"]).date().isoformat()
if date not in daily_forecasts:
daily_forecasts[date] = {
"temps": [],
"conditions": [],
"humidity": [],
"precipitation": 0
}
daily_forecasts[date]["temps"].append(item["main"]["temp"])
daily_forecasts[date]["conditions"].append(item["weather"][0])
daily_forecasts[date]["humidity"].append(item["main"]["humidity"])
# 计算降水概率
if "rain" in item or "snow" in item:
daily_forecasts[date]["precipitation"] += 1
# 转换为预报模型
forecasts = []
for date, data in list(daily_forecasts.items())[:days]:
forecast = WeatherForecast(
date=date,
temperature_max=max(data["temps"]),
temperature_min=min(data["temps"]),
condition=data["conditions"][0]["main"].lower(),
condition_description=data["conditions"][0]["description"],
humidity=int(sum(data["humidity"]) / len(data["humidity"])),
precipitation_probability=int((data["precipitation"] / 8) * 100)
)
forecasts.append(forecast)
return forecasts
except Exception as e:
raise WeatherAPIError(f"预报数据获取失败: {str(e)}")
async def close(self):
"""关闭HTTP客户端"""
await self.client.aclose()步骤四：MCP服务器核心实现
🚀 服务器主体代码
python
# src/weather_server.py
import asyncio
import os
from typing import List, Optional
from mcp import Server
from mcp.types import Tool, TextContent, ImageContent
import logging
from .weather_api import WeatherAPIClient, WeatherAPIError
from .data_models import CurrentWeather, WeatherForecast
from .utils import format_weather_display, create_weather_chart
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class WeatherMCPServer:
"""天气MCP服务器"""
def __init__(self, api_key: Optional[str] = None):
# 初始化MCP服务器
self.server = Server("weather-assistant")
self.weather_client = WeatherAPIClient(api_key)
# 注册工具
self._register_tools()
logger.info("🌤️ 天气MCP服务器初始化完成")
def _register_tools(self):
"""注册MCP工具"""
# 当前天气查询工具
current_weather_tool = Tool(
name="get_current_weather",
description="获取指定城市的当前天气信息，支持全球主要城市",
inputSchema={
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，支持中英文，如：北京、Beijing、上海、Shanghai"
},
"country": {
"type": "string",
"description": "国家代码，如：CN（中国）、US（美国）、JP（日本）",
"default": "CN"
},
"include_details": {
"type": "boolean",
"description": "是否包含详细信息（风向、气压、能见度等）",
"default": True
}
},
"required": ["city"]
}
)
# 天气预报工具
forecast_tool = Tool(
name="get_weather_forecast",
description="获取指定城市的天气预报，最多支持7天",
inputSchema={
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称"
},
"days": {
"type": "integer",
"description": "预报天数（1-7天）",
"minimum": 1,
"maximum": 7,
"default": 5
},
"country": {
"type": "string",
"description": "国家代码",
"default": "CN"
},
"format": {
"type": "string",
"description": "输出格式：text（文本）或chart（图表）",
"enum": ["text", "chart"],
"default": "text"
}
},
"required": ["city"]
}
)
# 天气对比工具
compare_weather_tool = Tool(
name="compare_weather",
description="对比多个城市的当前天气状况",
inputSchema={
"type": "object",
"properties": {
"cities": {
"type": "array",
"items": {"type": "string"},
"description": "要对比的城市列表，最多支持5个城市",
"maxItems": 5,
"minItems": 2
},
"country": {
"type": "string",
"description": "国家代码（如果所有城市都在同一国家）",
"default": "CN"
}
},
"required": ["cities"]
}
)
# 注册工具处理函数
self.server.add_tool(current_weather_tool, self.handle_current_weather)
self.server.add_tool(forecast_tool, self.handle_weather_forecast)
self.server.add_tool(compare_weather_tool, self.handle_compare_weather)
async def handle_current_weather(
self,
city: str,
country: str = "CN",
include_details: bool = True
) -> List[TextContent]:
"""处理当前天气查询"""
try:
logger.info(f"🔍 查询天气: {city}, {country}")
# 获取天气数据
weather = await self.weather_client.get_current_weather(city, country)
# 格式化显示内容
display_text = format_weather_display(weather, include_details)
logger.info(f"✅ 成功获取 {city} 的天气信息")
return [TextContent(type="text", text=display_text)]
except WeatherAPIError as e:
error_msg = f"❌ 获取天气信息失败: {str(e)}"
logger.error(error_msg)
return [TextContent(type="text", text=error_msg)]
except Exception as e:
error_msg = f"❌ 系统错误: {str(e)}"
logger.error(error_msg)
return [TextContent(type="text", text=error_msg)]
async def handle_weather_forecast(
self,
city: str,
days: int = 5,
country: str = "CN",
format: str = "text"
) -> List[TextContent]:
"""处理天气预报查询"""
try:
logger.info(f"📅 查询预报: {city}, {days}天")
# 获取预报数据
forecasts = await self.weather_client.get_weather_forecast(
city, days, country
)
if format == "chart":
# 生成图表格式
chart_text = create_weather_chart(forecasts, city)
result_text = f"📊 {city} {days}天天气预报\n\n{chart_text}"
else:
# 生成文本格式
result_text = self._format_forecast_text(forecasts, city, days)
logger.info(f"✅ 成功获取 {city} {days}天预报")
return [TextContent(type="text", text=result_text)]
except WeatherAPIError as e:
error_msg = f"❌ 获取预报失败: {str(e)}"
logger.error(error_msg)
return [TextContent(type="text", text=error_msg)]
except Exception as e:
error_msg = f"❌ 系统错误: {str(e)}"
logger.error(error_msg)
return [TextContent(type="text", text=error_msg)]
async def handle_compare_weather(
self,
cities: List[str],
country: str = "CN"
) -> List[TextContent]:
"""处理天气对比"""
try:
logger.info(f"⚖️ 对比天气: {', '.join(cities)}")
# 并发获取所有城市的天气
weather_data = []
tasks = [
self.weather_client.get_current_weather(city, country)
for city in cities
]
results = await asyncio.gather(*tasks, return_exceptions=True)
# 处理结果
successful_data = []
failed_cities = []
for i, result in enumerate(results):
if isinstance(result, Exception):
failed_cities.append(cities[i])
else:
successful_data.append(result)
if not successful_data:
return [TextContent(
type="text",
text="❌ 无法获取任何城市的天气信息"
)]
# 格式化对比结果
comparison_text = self._format_weather_comparison(
successful_data, failed_cities
)
logger.info(f"✅ 成功对比 {len(successful_data)} 个城市的天气")
return [TextContent(type="text", text=comparison_text)]
except Exception as e:
error_msg = f"❌ 天气对比失败: {str(e)}"
logger.error(error_msg)
return [TextContent(type="text", text=error_msg)]
def _format_forecast_text(
self,
forecasts: List[WeatherForecast],
city: str,
days: int
) -> str:
"""格式化预报文本"""
text = f"📅 {city} {days}天天气预报\n"
text += "=" * 50 + "\n\n"
for i, forecast in enumerate(forecasts, 1):
# 解析日期
from datetime import datetime
date_obj = datetime.fromisoformat(forecast.date)
weekday = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"][date_obj.weekday()]
text += f"📅 第{i}天 ({forecast.date} {weekday})\n"
text += f"🌡️ 温度: {forecast.temperature_min}°C ~ {forecast.temperature_max}°C\n"
text += f"☁️ 天气: {forecast.condition_description}\n"
text += f"💧 湿度: {forecast.humidity}%\n"
text += f"🌧️ 降水概率: {forecast.precipitation_probability}%\n"
# 添加建议
if forecast.precipitation_probability > 70:
text += "☂️ 建议: 记得带伞\n"
elif forecast.temperature_max > 30:
text += "🌞 建议: 注意防晒降温\n"
elif forecast.temperature_min < 5:
text += "🧥 建议: 注意保暖\n"
text += "-" * 30 + "\n"
return text
def _format_weather_comparison(
self,
weather_list: List[CurrentWeather],
failed_cities: List[str]
) -> str:
"""格式化天气对比结果"""
text = "⚖️ 城市天气对比\n"
text += "=" * 60 + "\n\n"
# 表头
text += f"{'城市':<10} {'温度':<10} {'体感':<10} {'天气':<15} {'湿度':<8}\n"
text += "-" * 60 + "\n"
# 数据行
for weather in weather_list:
text += f"{weather.city:<10} "
text += f"{weather.temperature:>6.1f}°C "
text += f"{weather.feels_like:>6.1f}°C "
text += f"{weather.condition_description:<15} "
text += f"{weather.humidity:>5}%\n"
# 分析总结
if len(weather_list) > 1:
temps = [w.temperature for w in weather_list]
hottest = max(weather_list, key=lambda w: w.temperature)
coldest = min(weather_list, key=lambda w: w.temperature)
text += "\n📊 对比分析:\n"
text += f"🔥 最热: {hottest.city} ({hottest.temperature}°C)\n"
text += f"❄️ 最冷: {coldest.city} ({coldest.temperature}°C)\n"
text += f"📈 温差: {max(temps) - min(temps):.1f}°C\n"
# 失败的城市
if failed_cities:
text += f"\n⚠️ 无法获取数据的城市: {', '.join(failed_cities)}\n"
return text
async def start_server(self):
"""启动MCP服务器"""
try:
logger.info("🚀 启动天气MCP服务器...")
await self.server.run()
except KeyboardInterrupt:
logger.info("📴 服务器已停止")
finally:
await self.weather_client.close()
# 启动入口
async def main():
"""主函数"""
# 检查API密钥
api_key = os.getenv("OPENWEATHER_API_KEY")
if not api_key:
print("❌ 请设置 OPENWEATHER_API_KEY 环境变量")
print("💡 获取API密钥: https://openweathermap.org/api")
return
# 创建并启动服务器
server = WeatherMCPServer(api_key)
await server.start_server()
if __name__ == "__main__":
asyncio.run(main())步骤五：工具函数实现
🎨 显示格式化工具
python
# src/utils.py
from typing import List, Dict, Any
from .data_models import CurrentWeather, WeatherForecast
import math
def format_weather_display(weather: CurrentWeather, include_details: bool = True) -> str:
"""格式化天气显示信息"""
# 选择天气图标
weather_icons = {
"clear": "☀️",
"clouds": "☁️",
"rain": "🌧️",
"snow": "❄️",
"thunderstorm": "⛈️",
"drizzle": "🌦️",
"mist": "🌫️",
"fog": "🌫️"
}
icon = weather_icons.get(weather.condition, "🌤️")
# 基础信息
text = f"{icon} {weather.city} 当前天气\n"
text += "━" * 50 + "\n"
text += f"🌡️ 温度: {weather.temperature}°C (体感 {weather.feels_like}°C)\n"
text += f"☁️ 天气: {weather.condition_description}\n"
text += f"💧 湿度: {weather.humidity}%\n"
if include_details:
# 详细信息
text += f"💨 风速: {weather.wind_speed} m/s"
# 风向描述
wind_directions = [
"北", "东北", "东", "东南", "南", "西南", "西", "西北"
]
direction_index = int((weather.wind_direction + 22.5) / 45) % 8
wind_direction_text = wind_directions[direction_index]
text += f" ({wind_direction_text}风)\n"
text += f"📊 气压: {weather.pressure} hPa\n"
text += f"👁️ 能见度: {weather.visibility/1000:.1f} km\n"
if weather.uv_index:
uv_level = get_uv_level(weather.uv_index)
text += f"🔆 紫外线: {weather.uv_index} ({uv_level})\n"
text += "━" * 50 + "\n"
# 添加生活建议
suggestions = generate_weather_suggestions(weather)
if suggestions:
text += "💡 生活建议:\n"
for suggestion in suggestions:
text += f" • {suggestion}\n"
return text
def get_uv_level(uv_index: float) -> str:
"""获取紫外线等级描述"""
if uv_index <= 2:
return "很低"
elif uv_index <= 5:
return "低"
elif uv_index <= 7:
return "中等"
elif uv_index <= 10:
return "高"
else:
return "极高"
def generate_weather_suggestions(weather: CurrentWeather) -> List[str]:
"""生成天气建议"""
suggestions = []
# 温度建议
if weather.temperature < 0:
suggestions.append("气温很低，注意防寒保暖")
elif weather.temperature < 10:
suggestions.append("天气较冷，适合穿厚外套")
elif weather.temperature > 30:
suggestions.append("天气炎热，注意防暑降温")
elif weather.temperature > 25:
suggestions.append("天气温暖，适合穿轻薄衣物")
# 天气条件建议
if weather.condition in ["rain", "drizzle", "thunderstorm"]:
suggestions.append("有降水，记得带伞")
elif weather.condition == "snow":
suggestions.append("有降雪，出行注意安全")
elif weather.condition == "fog":
suggestions.append("有雾，驾驶请减速慢行")
# 湿度建议
if weather.humidity < 30:
suggestions.append("湿度较低，注意补水保湿")
elif weather.humidity > 80:
suggestions.append("湿度较高，注意通风除湿")
# 风速建议
if weather.wind_speed > 10:
suggestions.append("风力较大，外出注意安全")
return suggestions
def create_weather_chart(forecasts: List[WeatherForecast], city: str) -> str:
"""创建ASCII天气图表"""
if not forecasts:
return "无预报数据"
# 提取温度数据
max_temps = [f.temperature_max for f in forecasts]
min_temps = [f.temperature_min for f in forecasts]
# 计算温度范围
all_temps = max_temps + min_temps
temp_min = min(all_temps)
temp_max = max(all_temps)
temp_range = temp_max - temp_min
if temp_range == 0:
temp_range = 1 # 避免除零错误
chart_height = 15
chart_width = len(forecasts) * 8
chart = f"📊 {city} 温度趋势图\n\n"
# 绘制图表
for row in range(chart_height):
line = ""
temp_level = temp_max - (row / (chart_height - 1)) * temp_range
for i, forecast in enumerate(forecasts):
col_start = i * 8
# 检查最高温度点
if abs(forecast.temperature_max - temp_level) <= temp_range / chart_height:
line += "🔴"
# 检查最低温度点
elif abs(forecast.temperature_min - temp_level) <= temp_range / chart_height:
line += "🔵"
else:
line += " "
line += " " # 间距
# 添加温度刻度
line += f" {temp_level:4.1f}°C"
chart += line + "\n"
# 添加日期标签
chart += "\n"
date_line = ""
for forecast in forecasts:
date_str = forecast.date.split("-")[1:3] # 月-日
date_line += f"{'/'.join(date_str):>8}"
chart += date_line + "\n"
# 添加图例
chart += "\n🔴 最高温 🔵 最低温\n"
return chart步骤六：配置管理
python
# src/config.py
import os
from typing import Optional
from pydantic import BaseSettings
class WeatherConfig(BaseSettings):
"""天气服务器配置"""
# API配置
openweather_api_key: Optional[str] = None
api_timeout: int = 30
api_retry_times: int = 3
# 服务器配置
server_name: str = "weather-assistant"
log_level: str = "INFO"
max_concurrent_requests: int = 10
# 缓存配置
enable_cache: bool = True
cache_ttl: int = 300 # 5分钟
class Config:
env_file = ".env"
env_prefix = "WEATHER_"
# 全局配置实例
config = WeatherConfig()步骤七：使用MCP Inspector调试
🔍 启动调试会话
bash
# 1. 启动MCP Inspector
mcp-inspector python src/weather_server.py
# 2. 在浏览器中打开调试界面
# 自动打开: http://localhost:5173🎯 调试界面功能详解
主要调试功能：
实时消息监控
- 查看客户端-服务器通信
- 监控请求和响应格式
- 检测协议错误
工具交互测试
- 直接调用MCP工具
- 测试参数验证
- 查看返回结果
性能分析
- 监控响应时间
- 分析内存使用
- 检测性能瓶颈
错误诊断
- 捕获异常堆栈
- 分析错误原因
- 提供修复建议
🧪 调试测试用例
在Inspector中测试以下场景：
json
// 测试用例1：正常天气查询
{
"tool": "get_current_weather",
"arguments": {
"city": "北京",
"country": "CN",
"include_details": true
}
}
// 测试用例2：错误城市名
{
"tool": "get_current_weather",
"arguments": {
"city": "不存在的城市",
"country": "CN"
}
}
// 测试用例3：天气预报
{
"tool": "get_weather_forecast",
"arguments": {
"city": "上海",
"days": 7,
"format": "chart"
}
}
// 测试用例4：多城市对比
{
"tool": "compare_weather",
"arguments": {
"cities": ["北京", "上海", "广州", "深圳"],
"country": "CN"
}
}步骤八：集成到AI Toolkit
🤖 在Agent Builder中配置
创建新代理
- 打开AI Toolkit Agent Builder
- 创建新的AI代理项目
配置MCP服务器
json{ "mcpServers": { "weather": { "command": "python", "args": ["src/weather_server.py"], "description": "专业天气查询助手", "env": { "OPENWEATHER_API_KEY": "your-api-key-here" } } } }设计代理人格
json{ "systemPrompt": "你是一个专业的天气助手，能够提供准确、详细的天气信息。你的特点是：\n\n🌤️ 专业：基于可靠的气象数据\n📊 详细：提供全面的天气分析\n💡 贴心：给出实用的生活建议\n🎯 准确：确保信息的时效性\n\n在回答时，请用友好、专业的语气，适当使用emoji让信息更生动。", "examples": [ { "user": "明天北京天气怎么样？", "assistant": "我来为你查询明天北京的天气预报！\n\n[调用天气预报工具]\n\n根据最新气象数据，明天北京的天气情况是...\n\n💡 根据天气情况，我建议..." } ] }
🧪 代理功能测试
测试我们的天气代理：
测试对话1：基础查询
用户: 今天杭州天气如何？
天气助手: 🌤️ 我来为你查询杭州今天的天气情况！
[正在获取天气数据...]
☀️ 杭州 当前天气
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
🌡️ 温度: 23°C (体感 25°C)
☁️ 天气: 晴转多云
💧 湿度: 65%
💨 风速: 2.8 m/s (东南风)
📊 气压: 1018 hPa
👁️ 能见度: 10.0 km
🔆 紫外线: 5 (中等)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
💡 生活建议:
• 天气温暖，适合穿轻薄衣物
• 紫外线中等，外出可适当防晒
今天杭州的天气很不错呢！温度适宜，很适合出行。建议穿轻薄的衣服，如果要在户外待很久的话，可以适当做些防晒措施。测试对话2：预报查询
用户: 这周末北京的天气如何？适合出游吗？
天气助手: 🌤️ 我来查询这周末（周六、周日）北京的天气情况，帮你规划出游！
[正在获取天气预报...]
📅 北京 2天天气预报
==================================================
📅 第1天 (2024-07-27 周六)
🌡️ 温度: 22°C ~ 31°C
☁️ 天气: 多云
💧 湿度: 58%
🌧️ 降水概率: 20%
🌞 建议: 注意防晒降温
------------------------------
📅 第2天 (2024-07-28 周日)
🌡️ 温度: 24°C ~ 29°C
☁️ 天气: 晴
💧 湿度: 52%
🌧️ 降水概率: 10%
------------------------------
🎯 出游建议:
✅ 非常适合出游！两天都是好天气
🌞 周六温度较高，注意防晒和补水
☀️ 周日是完美的出游天气
👕 建议穿透气性好的夏装
🧴 记得带防晒霜和遮阳帽
💧 多准备一些水，保持水分
这个周末的天气非常棒，特别适合出游！记得做好防晒准备，祝你玩得开心！🎉步骤九：性能优化与生产部署
⚡ 性能优化技巧
python
# 添加缓存机制
import asyncio
from functools import lru_cache
from datetime import datetime, timedelta
class CachedWeatherClient(WeatherAPIClient):
"""带缓存的天气客户端"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._cache = {}
self._cache_ttl = 300 # 5分钟缓存
async def get_current_weather_cached(self, city: str, country: str = "CN"):
"""带缓存的天气查询"""
cache_key = f"{city}_{country}_current"
now = datetime.now()
# 检查缓存
if cache_key in self._cache:
cache_data, cache_time = self._cache[cache_key]
if (now - cache_time).seconds < self._cache_ttl:
return cache_data
# 获取新数据
weather = await self.get_current_weather(city, country)
# 更新缓存
self._cache[cache_key] = (weather, now)
return weather
# 添加并发限制
import asyncio
from asyncio import Semaphore
class RateLimitedWeatherServer(WeatherMCPServer):
"""限流的天气服务器"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._semaphore = Semaphore(5) # 最多5个并发请求
async def handle_current_weather(self, *args, **kwargs):
"""限流的天气查询处理"""
async with self._semaphore:
return await super().handle_current_weather(*args, **kwargs)🚀 部署配置
bash
# deployment/docker-compose.yml
version: '3.8'
services:
weather-mcp-server:
build: .
environment:
- OPENWEATHER_API_KEY=${OPENWEATHER_API_KEY}
- WEATHER_LOG_LEVEL=INFO
- WEATHER_ENABLE_CACHE=true
volumes:
- ./logs:/app/logs
restart: unless-stopped
healthcheck:
test: ["CMD", "python", "-c", "import asyncio; asyncio.run(health_check())"]
interval: 30s
timeout: 10s
retries: 3🎯 实验成果验证
✅ 功能完整性检查
完成以下检查来验证你的MCP服务器：
基础功能测试
- [ ] 工具注册：所有工具正确注册并可发现
- [ ] 参数验证：输入参数格式和范围验证
- [ ] 数据获取：能够成功调用天气API
- [ ] 结果格式化：返回结果格式正确且用户友好
- [ ] 错误处理：各种异常情况的正确处理
性能指标测试
| 测试项目 | 目标性能 | 实际结果 |
|---|---|---|
| 工具响应时间 | < 3秒 | __ 秒 |
| API调用成功率 | > 95% | __% |
| 并发处理能力 | 5个/秒 | __个/秒 |
| 内存使用 | < 100MB | __MB |
集成测试
- [ ] MCP Inspector：能够在调试器中正常运行
- [ ] AI Toolkit集成：成功集成到Agent Builder
- [ ] 代理对话：AI代理能够正确调用工具
- [ ] 错误恢复：网络异常时的优雅降级
📊 代码质量评估
bash
# 运行代码质量检查
python -m black src/ # 代码格式化
python -m mypy src/ # 类型检查
python -m pytest tests/ -v # 单元测试
python -m pytest tests/ --cov=src --cov-report=html # 覆盖率测试质量目标：
- 代码覆盖率 > 80%
- 类型检查 100% 通过
- 无安全漏洞
- 符合PEP 8规范
🚀 扩展挑战与进阶项目
💡 功能扩展建议
🌟 高级功能模块
历史天气分析
python@server.tool() async def analyze_weather_history( city: str, start_date: str, end_date: str ) -> List[TextContent]: """分析历史天气趋势""" # 实现历史数据分析逻辑 pass天气预警系统
python@server.tool() async def setup_weather_alerts( city: str, alert_types: List[str], notification_method: str ) -> List[TextContent]: """设置天气预警""" # 实现预警系统 pass生活指数查询
python@server.tool() async def get_life_indices( city: str, indices: List[str] # 穿衣、运动、洗车等指数 ) -> List[TextContent]: """获取生活指数建议""" # 实现生活指数计算 pass
🏆 挑战项目
选择一个挑战来进一步提升你的MCP开发技能：
🎯 挑战A：多源天气聚合器（⭐⭐⭐⭐☆）
目标：集成多个天气数据源，提供更准确的预报
技术要点：
- 集成OpenWeatherMap、AccuWeather、Weather.gov
- 实现数据融合算法
- 提供可信度评分
- 支持数据源切换
🎯 挑战B：智能天气助手（⭐⭐⭐⭐⭐）
目标：构建能理解自然语言的智能天气查询系统
技术要点：
- 自然语言处理
- 上下文记忆功能
- 个性化推荐
- 多轮对话支持
🎯 挑战C：企业级天气服务（⭐⭐⭐⭐⭐）
目标：构建企业级的天气微服务
技术要点：
- 分布式架构
- 负载均衡
- 监控告警
- API网关集成
💭 学习总结与反思
🎓 核心技能掌握
通过这个实验，你已经掌握了：
- MCP服务器开发全流程：从设计到部署的完整链路
- 现代Python开发实践：类型提示、异步编程、错误处理
- 专业调试技能：使用MCP Inspector进行高效调试
- AI工具集成：将自定义工具无缝集成到AI代理中
- 生产环境考虑：性能优化、错误处理、监控部署
🔑 关键收获
技术层面：
- 深入理解了MCP协议的工作机制
- 掌握了现代Python异步编程模式
- 学会了专业的API设计和错误处理
- 具备了生产级代码的编写能力
工程层面：
- 学会了从需求分析到代码实现的完整流程
- 掌握了测试驱动开发的实践方法
- 理解了可维护代码的设计原则
- 具备了系统性解决问题的能力
💡 思考与展望
反思问题：
- 在开发过程中遇到了哪些挑战？如何解决的？
- 哪些设计决策可以进一步优化？
- 如何确保服务器在生产环境中的稳定性？
- 还有哪些功能可以增强用户体验？
未来方向：
- 探索更多MCP工具类型（文件操作、数据分析等）
- 学习分布式系统设计模式
- 深入研究AI Agent的架构设计
- 关注MCP生态系统的最新发展
记住：优秀的开发者不是因为写了多少代码，而是因为解决了多少实际问题。你现在具备了构建真正有用的AI工具的能力！
准备好迎接最后的挑战了吗？下一个实验我们将构建一个真正的生产级GitHub克隆MCP服务器！🚀