mcp-weather-server

好的，这是提供天气数据给 LLM 的一个示例模型上下文协议服务器： ```python import asyncio import json import os from typing import Any, Dict, List, Optional from fastapi import FastAPI, HTTPException from pydantic import BaseModel # 模拟天气数据 WEATHER_DATA = { "San Francisco": {"temperature": 15, "condition": "Cloudy"}, "New York": {"temperature": 22, "condition": "Sunny"}, "London": {"temperature": 18, "condition": "Rainy"}, "Tokyo": {"temperature": 25, "condition": "Clear"}, } class ContextRequest(BaseModel): """ LLM 请求上下文信息的请求体。 """ query: str location: Optional[str] = None # 可选的位置信息 class ContextResponse(BaseModel): """ 服务器返回给 LLM 的上下文信息。 """ context: Dict[str, Any] app = FastAPI() @app.post("/context") async def get_context(request: ContextRequest) -> ContextResponse: """ 根据 LLM 的查询请求，提供上下文信息。 """ print(f"Received query: {request.query}") print(f"Received location: {request.location}") location = request.location if not location: # 如果没有提供位置，则尝试从查询中提取 # 这是一个非常简单的示例，实际应用中需要更复杂的 NLP 处理 if "San Francisco" in request.query: location = "San Francisco" elif "New York" in request.query: location = "New York" elif "London" in request.query: location = "London" elif "Tokyo" in request.query: location = "Tokyo" else: raise HTTPException(status_code=400, detail="Location not specified and could not be inferred from query.") if location not in WEATHER_DATA: raise HTTPException(status_code=404, detail=f"Weather data not found for location: {location}") weather = WEATHER_DATA[location] context = { "location": location, "temperature": weather["temperature"], "condition": weather["condition"], } print(f"Returning context: {context}") return ContextResponse(context=context) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000) ``` **代码解释:** 1. **导入必要的库:** - `asyncio`: 用于异步操作。 - `json`: 用于处理 JSON 数据。 - `os`: 用于操作系统相关的功能。 - `typing`: 用于类型提示。 - `fastapi`: 用于创建 API。 - `pydantic`: 用于数据验证和序列化。 2. **模拟天气数据:** - `WEATHER_DATA`: 一个字典，存储了不同城市的天气数据。 这只是一个模拟数据，实际应用中需要从外部 API 或数据库获取。 3. **定义数据模型:** - `ContextRequest`: 定义了 LLM 请求上下文信息的请求体，包含 `query` (LLM 的查询) 和可选的 `location` (位置信息)。 - `ContextResponse`: 定义了服务器返回给 LLM 的上下文信息，包含一个 `context` 字典。 4. **创建 FastAPI 应用:** - `app = FastAPI()`: 创建一个 FastAPI 应用实例。 5. **定义 `/context` 接口:** - `@app.post("/context")`: 定义一个 POST 请求的接口，路径为 `/context`。 - `async def get_context(request: ContextRequest) -> ContextResponse:`: 定义处理请求的异步函数。 - `request: ContextRequest`: 接收请求体，并将其解析为 `ContextRequest` 对象。 - `-> ContextResponse`: 指定函数返回 `ContextResponse` 对象。 6. **处理请求逻辑:** - **打印接收到的查询和位置信息:** 用于调试和日志记录。 - **获取位置信息:** - 首先尝试从 `request.location` 中获取位置信息。 - 如果 `request.location` 为空，则尝试从 `request.query` 中提取位置信息。 这是一个非常简单的示例，实际应用中需要使用更复杂的 NLP 技术来提取位置信息。 - 如果无法获取位置信息，则返回一个 HTTP 400 错误。 - **获取天气数据:** - 检查 `location` 是否在 `WEATHER_DATA` 中。 - 如果 `location` 不在 `WEATHER_DATA` 中，则返回一个 HTTP 404 错误。 - 从 `WEATHER_DATA` 中获取天气数据。 - **构建上下文信息:** - 创建一个 `context` 字典，包含 `location`、`temperature` 和 `condition`。 - **返回上下文信息:** - 创建一个 `ContextResponse` 对象，并将 `context` 字典赋值给它。 - 返回 `ContextResponse` 对象。 7. **运行 FastAPI 应用:** - `if __name__ == "__main__":`: 确保代码只在直接运行脚本时执行，而不是在被导入为模块时执行。 - `uvicorn.run(app, host="0.0.0.0", port=8000)`: 使用 Uvicorn 运行 FastAPI 应用。 - `host="0.0.0.0"`: 允许从任何 IP 地址访问应用。 - `port=8000`: 指定应用监听的端口为 8000。 **如何运行:** 1. **安装依赖:** ```bash pip install fastapi uvicorn pydantic ``` 2. **运行脚本:** ```bash python your_script_name.py ``` 3. **测试接口:** 可以使用 `curl` 或其他 HTTP 客户端来测试接口。 例如： ```bash curl -X POST -H "Content-Type: application/json" -d '{"query": "What is the weather in San Francisco?", "location": "San Francisco"}' http://localhost:8000/context ``` 或者，如果省略 `location`，服务器会尝试从 `query` 中推断： ```bash curl -X POST -H "Content-Type: application/json" -d '{"query": "What is the weather in San Francisco?"}' http://localhost:8000/context ``` **重要说明:** * **真实数据源:** 这个示例使用模拟的天气数据。 在实际应用中，你需要使用真实的天气 API (例如 OpenWeatherMap, AccuWeather) 或数据库来获取数据。 * **NLP 处理:** 从查询中提取位置信息的部分非常简单。 在实际应用中，你需要使用更复杂的 NLP 技术 (例如命名实体识别) 来准确地提取位置信息。 * **错误处理:** 这个示例只包含基本的错误处理。 在实际应用中，你需要添加更完善的错误处理机制，例如日志记录和重试机制。 * **安全性:** 在生产环境中，你需要考虑安全性问题，例如身份验证和授权。 * **可扩展性:** 如果需要处理大量的请求，你需要考虑使用负载均衡和缓存等技术来提高可扩展性。 * **模型上下文协议:** 这个示例符合模型上下文协议的基本要求，即接收 LLM 的查询请求，并返回相关的上下文信息。 你需要根据 LLM 的具体要求来调整请求和响应的格式。 这个示例提供了一个基本的框架，你可以根据自己的需求进行修改和扩展。 希望这个示例对你有所帮助!