当前位置：首页 > news >正文

将天气查询API封装为MCP服务

news 来源：原创 2025/4/24 8:44:43

下面我将展示如何将一个天气查询API封装为符合MCP协议的服务。我们将使用Python实现，包括服务端和客户端。

## 1. 服务端实现

```python
# weather_mcp_server.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Dict, List, Optional, Any
import httpx
import json
from datetime import datetime
import uuid

app = FastAPI()

# 模拟天气数据服务
class WeatherService:
def __init__(self):
self.api_key = "your_weather_api_key"
self.base_url = "https://api.weatherapi.com/v1"

async def get_weather(self, location: str, date: Optional[str] = None) -> Dict:
"""获取天气数据"""
async with httpx.AsyncClient() as client:
try:
response = await client.get(
f"{self.base_url}/forecast.json",
params={
"key": self.api_key,
"q": location,
"days": 1
}
)
response.raise_for_status()
return response.json()
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))

# MCP协议处理器
class MCPHandler:
def __init__(self):
self.weather_service = WeatherService()

def validate_mcp_request(self, request: Dict) -> bool:
"""验证MCP请求格式"""
required_fields = ["protocol_version", "instruction", "content"]
return all(field in request for field in required_fields)

def extract_parameters(self, request: Dict) -> Dict:
"""从MCP请求中提取参数"""
content = request.get("content", {})
instruction = request.get("instruction", {})

return {
"location": content.get("location"),
"date": content.get("date"),
"format": instruction.get("output_format", "json")
}

def create_mcp_response(self,
data: Any,
request: Dict,
status: str = "success") -> Dict:
"""创建MCP响应"""
return {
"protocol_version": request.get("protocol_version", "1.0"),
"session_id": request.get("session_id", str(uuid.uuid4())),
"timestamp": datetime.now().isoformat(),
"status": status,
"content": {
"output": data
},
"metadata": {
"processing_time": "0.1s",
"source": "weather_api"
}
}

mcp_handler = MCPHandler()

@app.post("/weather/mcp")
async def weather_mcp_endpoint(request: Dict):
"""MCP天气查询端点"""
try:
# 验证请求
if not mcp_handler.validate_mcp_request(request):
raise HTTPException(status_code=400, detail="Invalid MCP request format")

# 提取参数
params = mcp_handler.extract_parameters(request)

# 获取天气数据
weather_data = await mcp_handler.weather_service.get_weather(
params["location"],
params["date"]
)

# 格式化响应
formatted_data = format_weather_data(weather_data, params["format"])

# 创建MCP响应
response = mcp_handler.create_mcp_response(formatted_data, request)

return response

except Exception as e:
return mcp_handler.create_mcp_response(
{"error": str(e)},
request,
status="error"
)

def format_weather_data(data: Dict, format: str) -> Any:
"""根据指定格式格式化天气数据"""
if format == "json":
return data
elif format == "text":
current = data.get("current", {})
location = data.get("location", {})
return f"""
天气报告 - {location.get('name')}, {location.get('country')}
时间: {current.get('last_updated')}
温度: {current.get('temp_c')}°C
体感温度: {current.get('feelslike_c')}°C
天气状况: {current.get('condition', {}).get('text')}
湿度: {current.get('humidity')}%
风速: {current.get('wind_kph')} km/h
"""
else:
return data
```

## 2. 客户端实现

```python
# weather_mcp_client.py
from mcp_client import MCPClient, MCPTaskType, MCPOutputFormat
from typing import Optional, Dict, Any

class WeatherMCPClient:
def __init__(self, api_endpoint: str, api_key: str):
self.mcp_client = MCPClient(api_endpoint, api_key)

def get_weather(self,
location: str,
date: Optional[str] = None,
output_format: MCPOutputFormat = MCPOutputFormat.JSON) -> Dict:
"""
获取天气信息

Args:
location: 位置（城市名或坐标）
date: 可选，查询日期
output_format: 输出格式

Returns:
天气信息
"""
# 构建MCP请求选项
options = self.mcp_client.create_mcp_request(
prompt="", # 这里不需要prompt
task_type=MCPTaskType.CUSTOM,
output_format=output_format,
custom_parameters={
"service": "weather",
"operation": "query"
}
)

# 设置内容
options["content"] = {
"location": location,
"date": date
}

# 发送请求
response = self.mcp_client.send_request(options)
return response

# 使用示例
def weather_example():
# 初始化客户端
client = WeatherMCPClient(
api_endpoint="http://localhost:8000/weather/mcp",
api_key="your_api_key"
)

# 获取JSON格式的天气信息
weather_json = client.get_weather(
location="Beijing",
output_format=MCPOutputFormat.JSON
)
print("JSON格式天气信息:", weather_json)

# 获取文本格式的天气信息
weather_text = client.get_weather(
location="Shanghai",
output_format=MCPOutputFormat.TEXT
)
print("文本格式天气信息:", weather_text)

# 获取特定日期的天气信息
weather_date = client.get_weather(
location="Guangzhou",
date="2024-03-20",
output_format=MCPOutputFormat.STRUCTURED
)
print("特定日期天气信息:", weather_date)

if __name__ == "__main__":
weather_example()
```

## 3. 配置文件

```python
# config.py
class Config:
# 服务配置
SERVICE_HOST = "0.0.0.0"
SERVICE_PORT = 8000

# 天气API配置
WEATHER_API_KEY = "your_weather_api_key"
WEATHER_API_BASE_URL = "https://api.weatherapi.com/v1"

# MCP配置
MCP_VERSION = "1.0"
MCP_ALLOWED_FORMATS = ["json", "text", "structured"]

# 安全配置
API_KEY_HEADER = "X-API-Key"
RATE_LIMIT = 100 # 每分钟请求限制
```

## 4. 启动脚本

```python
# run_server.py
import uvicorn
from weather_mcp_server import app
from config import Config

if __name__ == "__main__":
uvicorn.run(
app,
host=Config.SERVICE_HOST,
port=Config.SERVICE_PORT
)
```

## 5. 使用说明

1. **安装依赖**：
```bash
pip install fastapi uvicorn httpx pydantic
```

2. **配置服务**：
- 在`config.py`中设置天气API密钥
- 配置服务端口和其他参数

3. **启动服务**：
```bash
python run_server.py
```

4. **使用客户端**：
```python
from weather_mcp_client import WeatherMCPClient

# 创建客户端实例
client = WeatherMCPClient(
api_endpoint="http://localhost:8000/weather/mcp",
api_key="your_api_key"
)

# 查询天气
weather = client.get_weather("Beijing")
print(weather)
```

## 6. MCP协议适配要点

1. **请求格式标准化**：
```json
{
"protocol_version": "1.0",
"session_id": "uuid",
"instruction": {
"task_type": "custom",
"output_format": "json",
"execution_parameters": {
"service": "weather",
"operation": "query"
}
},
"content": {
"location": "Beijing",
"date": "2024-03-20"
}
}
```

2. **响应格式标准化**：
```json
{
"protocol_version": "1.0",
"session_id": "uuid",
"timestamp": "2024-03-20T10:00:00Z",
"status": "success",
"content": {
"output": {
"location": {...},
"current": {...},
"forecast": {...}
}
},
"metadata": {
"processing_time": "0.1s",
"source": "weather_api"
}
}
```

3. **错误处理**：
```json
{
"protocol_version": "1.0",
"session_id": "uuid",
"timestamp": "2024-03-20T10:00:00Z",
"status": "error",
"content": {
"output": {
"error": "Invalid location",
"error_code": "LOCATION_INVALID"
}
}
}
```

## 7. 扩展功能

1. **缓存支持**：
```python
from functools import lru_cache
import time

class WeatherService:
@lru_cache(maxsize=100)
def get_weather_cached(self, location: str, timestamp: int) -> Dict:
"""带缓存的天气查询"""
return self.get_weather(location)
```

2. **批量查询**：
```python
async def batch_get_weather(self, locations: List[str]) -> List[Dict]:
"""批量获取天气信息"""
tasks = [self.get_weather(loc) for loc in locations]
return await asyncio.gather(*tasks)
```

3. **数据转换器**：
```python
class WeatherDataConverter:
@staticmethod
def to_celsius(kelvin: float) -> float:
return kelvin - 273.15

@staticmethod
def to_fahrenheit(celsius: float) -> float:
return celsius * 9/5 + 32
```

这个示例展示了如何将一个天气查询API封装为MCP服务，包括：
- 服务端实现
- 客户端实现
- 配置管理
- 错误处理
- 数据格式化
- 扩展功能

通过这种方式，我们可以将任何现有的API服务转换为符合MCP协议的服务，实现标准化的接口和更好的可控性。