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

下面我将展示如何将一个天气查询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协议的服务，实现标准化的接口和更好的可控性。