Model Context Protocol Server

使用 FastAPI 实现的模型上下文协议服务器： ```python from fastapi import FastAPI, Request, HTTPException from pydantic import BaseModel from typing import Dict, Any, Optional app = FastAPI() # 定义请求和响应模型 class ContextRequest(BaseModel): model_id: str context: Dict[str, Any] request_id: Optional[str] = None # 可选的请求 ID，用于跟踪 class ContextResponse(BaseModel): model_id: str context: Dict[str, Any] request_id: Optional[str] = None # 响应中返回请求 ID # 模拟的模型上下文存储 model_contexts: Dict[str, Dict[str, Any]] = {} @app.post("/context") async def update_context(request: ContextRequest) -> ContextResponse: """ 更新指定模型的上下文。 Args: request: 包含模型 ID 和上下文数据的 ContextRequest 对象。 Returns: 包含更新后的模型 ID 和上下文数据的 ContextResponse 对象。 """ model_id = request.model_id context = request.context request_id = request.request_id # 检查模型 ID 是否存在 if model_id not in model_contexts: model_contexts[model_id] = {} # 更新上下文 model_contexts[model_id].update(context) # 构建响应 response = ContextResponse(model_id=model_id, context=model_contexts[model_id], request_id=request_id) return response @app.get("/context/{model_id}") async def get_context(model_id: str) -> Dict[str, Any]: """ 获取指定模型的上下文。 Args: model_id: 要获取上下文的模型 ID。 Returns: 指定模型的上下文数据。 Raises: HTTPException: 如果模型 ID 不存在。 """ if model_id not in model_contexts: raise HTTPException(status_code=404, detail=f"Model ID '{model_id}' not found") return model_contexts[model_id] @app.delete("/context/{model_id}") async def delete_context(model_id: str) -> Dict[str, str]: """ 删除指定模型的上下文。 Args: model_id: 要删除上下文的模型 ID。 Returns: 一个包含删除结果的消息。 Raises: HTTPException: 如果模型 ID 不存在。 """ if model_id not in model_contexts: raise HTTPException(status_code=404, detail=f"Model ID '{model_id}' not found") del model_contexts[model_id] return {"message": f"Context for model ID '{model_id}' deleted successfully."} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000) ``` **代码解释:** * **`from fastapi import FastAPI, Request, HTTPException`**: 导入 FastAPI 及其相关模块。 * **`from pydantic import BaseModel`**: 导入 Pydantic 的 `BaseModel` 用于定义数据模型。 * **`from typing import Dict, Any, Optional`**: 导入类型提示，提高代码可读性。 * **`app = FastAPI()`**: 创建 FastAPI 应用实例。 * **`ContextRequest` 和 `ContextResponse`**: 使用 Pydantic 定义请求和响应的数据模型。 `ContextRequest` 包含 `model_id` (模型 ID), `context` (上下文数据，一个字典), 和可选的 `request_id`。 `ContextResponse` 包含相同的信息，用于返回更新后的上下文。 * **`model_contexts: Dict[str, Dict[str, Any]] = {}`**: 一个字典，用于存储模型上下文。 键是 `model_id` (字符串)，值是另一个字典，表示该模型的上下文数据。 这只是一个内存中的存储，在实际应用中，你可能需要使用数据库或其他持久化存储。 * **`@app.post("/context")`**: 定义一个 POST 路由，用于更新模型的上下文。 * `async def update_context(request: ContextRequest) -> ContextResponse:`: 定义异步函数 `update_context`，它接收一个 `ContextRequest` 对象作为参数，并返回一个 `ContextResponse` 对象。 * 函数首先从请求中提取 `model_id` 和 `context`。 * 然后，它检查 `model_id` 是否已经存在于 `model_contexts` 中。如果不存在，则创建一个新的条目。 * 接下来，它使用 `update()` 方法将请求中的 `context` 合并到现有的模型上下文中。 * 最后，它创建一个 `ContextResponse` 对象并返回。 * **`@app.get("/context/{model_id}")`**: 定义一个 GET 路由，用于获取指定模型的上下文。 * `async def get_context(model_id: str) -> Dict[str, Any]:`: 定义异步函数 `get_context`，它接收一个 `model_id` 作为参数，并返回一个字典，表示该模型的上下文数据。 * 函数首先检查 `model_id` 是否存在于 `model_contexts` 中。如果不存在，则抛出一个 `HTTPException`，状态码为 404 (Not Found)。 * 如果 `model_id` 存在，则返回 `model_contexts[model_id]`。 * **`@app.delete("/context/{model_id}")`**: 定义一个 DELETE 路由，用于删除指定模型的上下文。 * `async def delete_context(model_id: str) -> Dict[str, str]:`: 定义异步函数 `delete_context`，它接收一个 `model_id` 作为参数，并返回一个字典，包含删除结果的消息。 * 函数首先检查 `model_id` 是否存在于 `model_contexts` 中。如果不存在，则抛出一个 `HTTPException`，状态码为 404 (Not Found)。 * 如果 `model_id` 存在，则使用 `del model_contexts[model_id]` 删除该模型的上下文。 * 最后，返回一个包含成功消息的字典。 * **`if __name__ == "__main__":`**: 确保代码只在直接运行脚本时执行，而不是作为模块导入时执行。 * **`uvicorn.run(app, host="0.0.0.0", port=8000)`**: 使用 Uvicorn 启动 FastAPI 应用。 `host="0.0.0.0"` 表示监听所有网络接口，`port=8000` 表示监听 8000 端口。 **如何运行:** 1. **安装 FastAPI 和 Uvicorn:** ```bash pip install fastapi uvicorn ``` 2. **保存代码:** 将代码保存为 `main.py` (或其他你喜欢的名字)。 3. **运行应用:** ```bash python main.py ``` 或者，如果你安装了 `uvicorn` 作为全局命令： ```bash uvicorn main:app --reload ``` `--reload` 标志会在代码更改时自动重新加载服务器，方便开发。 **如何使用:** 你可以使用 `curl` 或任何 HTTP 客户端来与服务器交互。 * **更新上下文 (POST):** ```bash curl -X POST -H "Content-Type: application/json" -d '{"model_id": "my_model", "context": {"key1": "value1", "key2": 123}, "request_id": "req123"}' http://localhost:8000/context ``` * **获取上下文 (GET):** ```bash curl http://localhost:8000/context/my_model ``` * **删除上下文 (DELETE):** ```bash curl -X DELETE http://localhost:8000/context/my_model ``` **重要注意事项:** * **错误处理:** 代码包含基本的错误处理 (例如，检查模型 ID 是否存在)。 在实际应用中，你需要更完善的错误处理机制，例如记录错误日志、返回更详细的错误信息等。 * **安全性:** 此代码没有包含任何安全措施。 在生产环境中，你需要添加身份验证、授权、输入验证等安全措施。 * **持久化存储:** 此代码使用内存中的字典来存储模型上下文。 这意味着，当服务器重启时，所有上下文数据都会丢失。 在实际应用中，你需要使用数据库或其他持久化存储来保存上下文数据。 常见的选择包括 Redis、PostgreSQL、MongoDB 等。 * **并发:** FastAPI 是一个异步框架，可以处理并发请求。 但是，如果你的模型上下文存储是线程不安全的，你需要采取措施来保护它，例如使用锁。 * **模型集成:** 此代码只是一个模型上下文协议服务器的框架。 你需要将其与你的实际模型集成。 这可能涉及到加载模型、调用模型进行推理、以及将模型的结果存储到上下文中。 * **数据验证:** 你可以使用 Pydantic 的验证功能来确保上下文数据符合预期的格式和类型。 This provides a basic implementation of a Model Context Protocol server using FastAPI. Remember to adapt it to your specific needs and consider the important notes above for production deployments.