Sameet Assignment

MCP服务器-模型上下文协议服务器

一个完整的、生产就绪的MCP（模型上下文协议）服务器，使用FastAPI构建，具有数据获取、通过WebSockets实时更新和历史数据查询的功能。服务器遵循干净的架构原则，具有全面的测试覆盖率。

特性

• 数据获取端点：按ID检索最新数据、项目和筛选数据
• 实时更新：WebSocket支持实时数据更新
• 历史查询：按日期范围或项目ID查询过去的记录
• 缓存层：内存中的LRU缓存可提高性能
• 错误处理：结构化JSON错误响应
• 综合录井：所有操作的详细日志记录
• 全面测试覆盖：pytest的测试覆盖率为70-90%

建筑

该项目遵循干净的模块化架构：

mcp-server/
├── cache/              # Caching utilities (LRU cache)
├── errors/             # Custom exception classes
├── models/             # Internal data models
├── routers/            # API route handlers
│   ├── data.py        # Data fetching endpoints
│   ├── realtime.py    # WebSocket real-time updates
│   └── historical.py  # Historical query endpoints
├── schemas/            # Pydantic schemas for validation
├── services/           # Business logic layer
├── tests/              # Comprehensive test suite
├── utils/              # Utility functions and helpers
└── main.py             # FastAPI application entry point

关键组件

1. 路由器：处理HTTP请求和WebSocket连接
2. 服务：包含业务逻辑，与路由处理程序分开
3. 模型：内部数据结构
4. 模式：用于请求/响应验证的Pydantic模型
5. 缓存：实现LRU缓存以优化性能
6. 错误：具有结构化错误响应的自定义异常类

安装说明

先决条件

• Python 3.8或更高版本
• pip（Python包管理器）

安装

1. 克隆或导航到项目目录：

   cd sameet

1. 创建虚拟环境（推荐）：

   python -m venv venv

   # On Windows:
   venv\Scripts\activate

   # On Linux/Mac:
   source venv/bin/activate

1. 安装依赖项：

   pip install -r requirements.txt

运行服务器

发展模式

在启用自动重新加载的情况下运行服务器：

python main.py

或者直接使用uvicorn：

uvicorn main:app --reload --host 0.0.0.0 --port 8000

服务器将于启动 http://localhost:8000

生产模式

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

API 文档

服务器运行后，访问交互式API文档：

• Swagger 用户界面: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc

运行测试

运行所有测试

pytest

使用覆盖率报告运行测试

pytest --cov=. --cov-report=term-missing --cov-report=html

这将：

• 运行所有测试
• 在终端中生成覆盖报告
• 在中创建HTML覆盖率报告 htmlcov/index.html
• 确保覆盖率至少为70%

运行特定测试文件

# Test data endpoints
pytest tests/test_data_endpoints.py

# Test cache
pytest tests/test_cache.py

# Test services
pytest tests/test_data_service.py

# Test real-time updates
pytest tests/test_realtime_endpoints.py

# Test historical queries
pytest tests/test_historical_endpoints.py

口头运行测试

pytest -v

API使用示例

数据获取端点

获取最新数据项

curl http://localhost:8000/api/v1/data/latest?limit=10

答复：

{
  "success": true,
  "data": [
    {
      "id": "item_abc123",
      "name": "Temperature Sensor",
      "value": 23.5,
      "category": "sensor",
      "timestamp": "2024-01-15T10:30:00Z",
      "metadata": {"unit": "celsius"}
    }
  ],
  "count": 1,
  "message": "Retrieved 1 latest items"
}

按ID获取项目

curl http://localhost:8000/api/v1/data/item/item_abc123

筛选数据项

curl -X POST http://localhost:8000/api/v1/data/filter \
  -H "Content-Type: application/json" \
  -d '{
    "category": "sensor",
    "min_value": 0.0,
    "max_value": 100.0,
    "limit": 50
  }'

创建新数据项

curl -X POST http://localhost:8000/api/v1/data/item \
  -H "Content-Type: application/json" \
  -d '{
    "name": "New Sensor",
    "value": 42.5,
    "category": "sensor",
    "metadata": {"location": "room_1"}
  }'

历史查询

按日期范围查询历史数据

curl -X POST http://localhost:8000/api/v1/historical/query \
  -H "Content-Type: application/json" \
  -d '{
    "start_date": "2024-01-01T00:00:00Z",
    "end_date": "2024-01-31T23:59:59Z",
    "category": "sensor",
    "limit": 100
  }'

获取特定项目的历史记录

curl http://localhost:8000/api/v1/historical/item/item_abc123?limit=1000

实时更新

通过WebSocket连接

使用Python：

import asyncio
import websockets
import json

async def connect_websocket():
    uri = "ws://localhost:8000/api/v1/realtime/ws"
    async with websockets.connect(uri) as websocket:
        # Receive welcome message
        message = await websocket.recv()
        print(f"Received: {message}")
        
        # Send ping
        await websocket.send(json.dumps({"type": "ping"}))
        response = await websocket.recv()
        print(f"Pong: {response}")

asyncio.run(connect_websocket())

广播更新

curl -X POST http://localhost:8000/api/v1/realtime/broadcast \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Live Update",
    "value": 99.9,
    "category": "broadcast",
    "metadata": {"source": "api"}
  }'

所有连接的WebSocket客户端都将自动接收更新。

假设和设计决策

1. 内存存储：为了简单起见，服务器使用内存存储。在生产环境中，这应该被持久数据库（PostgreSQL、MongoDB等）所取代。

1. LRU缓存：缓存使用最近最少使用（LRU）驱逐策略，默认容量为1000个项目。这是可以配置的。

1. WebSocket实时更新：使用WebSockets实现实时更新。服务器发送事件（SSE）可以作为单向通信的替代方案。

1. 错误处理：所有错误都返回格式一致的结构化JSON响应：

   {
     "success": false,
     "error": "ERROR_CODE",
     "message": "Human-readable message",
     "details": {}
   }

1. 日志记录：日志同时写入控制台和文件(logs/mcp_server.log).可以配置日志级别。

1. 类型提示：所有代码都使用Python类型提示，以获得更好的代码质量和IDE支持。

1. PEP8合规性：代码遵循PEP8风格指南。

1. 测试覆盖率：测试的目标是70-90%的覆盖率，包括：

- 服务和公用设施的单元测试 - API端点的集成测试 - WebSocket连接测试 - 错误处理测试 - 缓存行为测试

配置

环境变量

您可以使用环境变量配置服务器：

• LOG_LEVEL：日志记录级别（调试、信息、警告、错误）-默认值：信息
• CACHE_CAPACITY：LRU缓存容量-默认值：1000
• PORT：服务器端口-默认值：8000
• HOST：服务器主机-默认值：0.0.0.0

日志记录配置

日志记录已在中配置 utils/logging_config.py日志被写入：

• 控制台（stdout）
• 文件： logs/mcp_server.log （如果已配置）

发展

编码结构

• 业务逻辑：所有业务逻辑都在 services/ 目录，保持路由精简
• 验证：使用Pydantic模式进行请求/响应验证
• 错误处理：具有自定义异常的集中错误处理
• 缓存：可轻松替换的透明缓存层

添加新端点

1. 在中创建架构 schemas/ 如有需要
2. 添加业务逻辑 services/data_service.py 或创建新服务
3. 在相应的路由器文件中添加路由处理程序
4. 在中编写测试 tests/
5. 更新API文档（从文档字符串自动生成）

故障排除

端口已在使用中

如果端口8000已在使用中：

uvicorn main:app --port 8001

导入错误

确保您位于项目根目录中，并且虚拟环境已激活。

测试失败

确保安装了所有依赖项：

pip install -r requirements.txt

许可证

本项目按原样提供，用于教育和发展目的。

贡献

贡献时：

1. 遵循PEP8风格指南
2. 为所有函数添加类型提示
3. 为新功能编写测试
4. 更新文档
5. 确保测试覆盖率保持在70%以上