Sameet_Assignment by sameetjathan - MCP Server

MCP Server - Model Context Protocol Server

A complete, production-ready MCP (Model Context Protocol) server built with FastAPI, featuring data fetching, real-time updates via WebSockets, and historical data queries. The server follows clean architecture principles with comprehensive test coverage.

Features

Data Fetching Endpoints: Retrieve latest data, items by ID, and filtered data
Real-time Updates: WebSocket support for live data updates
Historical Queries: Query past records by date range or item ID
Caching Layer: In-memory LRU cache for improved performance
Error Handling: Structured JSON error responses
Comprehensive Logging: Detailed logging for all operations
Full Test Coverage: 70-90% test coverage with pytest

Architecture

The project follows a clean, modular architecture:

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

Key Components

Routers: Handle HTTP requests and WebSocket connections
Services: Contain business logic, separate from route handlers
Models: Internal data structures
Schemas: Pydantic models for request/response validation
Cache: LRU cache implementation for performance optimization
Errors: Custom exception classes with structured error responses

Setup Instructions

Prerequisites

Python 3.8 or higher
pip (Python package manager)

Installation

Clone or navigate to the project directory:
```
cd sameet
```

Create a virtual environment (recommended):

python -m venv venv

# On Windows:
venv\Scripts\activate

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

Install dependencies:
```
pip install -r requirements.txt
```

Running the Server

Development Mode

Run the server with auto-reload enabled:

python main.py

Or using uvicorn directly:

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

The server will start on http://localhost:8000

Production Mode

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

API Documentation

Once the server is running, access the interactive API documentation:

Swagger UI: http://localhost:8000/docs
ReDoc: http://localhost:8000/redoc

Running Tests

Run All Tests

pytest

Run Tests with Coverage Report

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

This will:

Run all tests
Generate a coverage report in the terminal
Create an HTML coverage report in htmlcov/index.html
Ensure coverage is at least 70%

Run Specific Test Files

# 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

Run Tests Verbosely

pytest -v

API Usage Examples

Data Fetching Endpoints

Get Latest Data Items

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

Response:

{
  "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"
}

Get Item by ID

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

Filter Data Items

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
  }'

Create New Data Item

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"}
  }'

Historical Queries

Query Historical Data by Date Range

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
  }'

Get History for Specific Item

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

Real-time Updates

Connect via WebSocket

Using 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())

Broadcast Update

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"}
  }'

All connected WebSocket clients will receive the update automatically.

Assumptions and Design Decisions

In-Memory Storage: The server uses in-memory storage for simplicity. In production, this should be replaced with a persistent database (PostgreSQL, MongoDB, etc.).
LRU Cache: The cache uses a Least Recently Used (LRU) eviction policy with a default capacity of 1000 items. This can be configured.
WebSocket Real-time Updates: Real-time updates are implemented using WebSockets. Server-Sent Events (SSE) could be an alternative for one-way communication.

Error Handling: All errors return structured JSON responses with consistent format:

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

Logging: Logs are written to both console and file (logs/mcp_server.log). Log level can be configured.
Type Hints: All code uses Python type hints for better code quality and IDE support.
PEP8 Compliance: Code follows PEP8 style guidelines.
Test Coverage: Tests aim for 70-90% coverage, including:
- Unit tests for services and utilities
- Integration tests for API endpoints
- WebSocket connection tests
- Error handling tests
- Cache behavior tests

Configuration

Environment Variables

You can configure the server using environment variables:

LOG_LEVEL: Logging level (DEBUG, INFO, WARNING, ERROR) - default: INFO
CACHE_CAPACITY: LRU cache capacity - default: 1000
PORT: Server port - default: 8000
HOST: Server host - default: 0.0.0.0

Logging Configuration

Logging is configured in utils/logging_config.py. Logs are written to:

Console (stdout)
File: logs/mcp_server.log (if configured)

Development

Code Structure

Business Logic: All business logic is in the services/ directory, keeping routes thin
Validation: Request/response validation using Pydantic schemas
Error Handling: Centralized error handling with custom exceptions
Caching: Transparent caching layer that can be easily replaced

Adding New Endpoints

Create schema in schemas/ if needed
Add business logic to services/data_service.py or create new service
Add route handler in appropriate router file
Write tests in tests/
Update API documentation (auto-generated from docstrings)

Troubleshooting

Port Already in Use

If port 8000 is already in use:

uvicorn main:app --port 8001

Import Errors

Make sure you're in the project root directory and the virtual environment is activated.

Test Failures

Ensure all dependencies are installed:

pip install -r requirements.txt

License

This project is provided as-is for educational and development purposes.

Contributing

When contributing:

Follow PEP8 style guidelines
Add type hints to all functions
Write tests for new features
Update documentation
Ensure test coverage remains above 70%