Retrieval Gateway MCP

A Model Context Protocol (MCP) server that provides ordered, Kafka-based resource retrieval with automatic buffering for out-of-order sequences.

Table of Contents

• Overview
• Architecture
• Requirements
• Setup
• Running the Services
• Running Tests
• MCP Server Usage
• API Reference
• Code Quality

Overview

This service implements a retrieval gateway that:

• Ensures ordered processing of requests per ordering key (e.g., user ID)
• Buffers out-of-order requests automatically
• Provides idempotency using request IDs
• Operates via Kafka
• Exposes functionality through an MCP server for AI assistant integration

Architecture

┌─────────────────┐
│   MCP Client    │  (Claude Desktop, Inspector, etc.)
│  (STDIO/SSE)    │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│   MCP Server    │  FastMCP-based tool server
│   (retrieve)    │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  Kafka Topics   │  Request/Response flow
│  mcp.retrieval  │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ Consumer Worker │  RequestProcessor
│   + Database    │  PostgreSQL state tracking
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ Mock Upstream   │  Simulated resource service
│   (FastAPI)     │
└─────────────────┘

Key Components

1. RequestProcessor (src/retrieval_gateway_mcp/req_processor.py)

   • Core business logic for ordered processing
   • Handles buffering, idempotency, and state management
   • Database-backed with PostgreSQL

2. MCP Server (src/retrieval_gateway_mcp/mcp_server.py)

   • Exposes retrieve tool via MCP protocol
   • Produces requests to Kafka
   • Polls for responses (primary + buffered)

3. Kafka Consumer (src/retrieval_gateway_mcp/kafka/consumer.py)

   • Consumes requests from Kafka topics
   • Calls upstream service
   • Produces responses back to Kafka

4. Mock Upstream (src/retrieval_gateway_mcp/mock_upstream.py)

   • FastAPI server simulating a resource service
   • Supports contacts, orders, users
   • Includes configurable delays

Error Handling

The system handles failures with a DLQ (Dead Letter Queue) using Kafka topics:

• Malformed messages → mcp.retrieval.dlq topic
  • JSON decode errors
  • Schema validation failures
  • Processing exceptions

Design choice: DLQ uses Kafka topic only (not database). Rationale: malformed messages often lack valid request_id or schema, which would require synthetic IDs and/or placeholder values to force into the database. Kafka topic provides durable storage for forensic analysis without compromising data integrity.

Requirements

• Python: >=3.12, <4.0
• Poetry: Dependency management
• Docker: For PostgreSQL and Kafka
• Docker Compose: Multi-container orchestration

Setup

1. Clone and Install Dependencies

poetry install

2. Configure Environment Variables

Copy the example environment file and update with your values:

cp .env.example .env

Edit .env to set your credentials (the defaults work for local development):

# Database credentials
POSTGRES_USER=mcp_user
POSTGRES_PASSWORD=mcp_password
POSTGRES_DB=mcp_gateway
POSTGRES_DB_TEST=mcp_gateway_test

# Database ports
POSTGRES_PORT=5433
POSTGRES_TEST_PORT=5434

# Kafka configuration
KAFKA_CLUSTER_ID=MkU3OEVBNTcwNTJENDM2Qk

3. Start Infrastructure

Start PostgreSQL and Kafka:

docker-compose up -d

4. Initialize Database

poetry run python -c "from src.retrieval_gateway_mcp.models import *; from src.retrieval_gateway_mcp.config import Config; init_database(create_database_engine(Config.DATABASE_URL))"

5. Configure Pre-commit (Optional)

poetry run pre-commit install

Running the Services

You need three separate terminals for the full system:

Terminal 1: Mock Upstream Server

poetry run python -m retrieval_gateway_mcp.mock_upstream

Runs on http://localhost:8001

Terminal 2: Kafka Consumer Worker

poetry run python -m retrieval_gateway_mcp.kafka.consumer

Processes requests from Kafka and calls upstream.

Terminal 3: MCP Server

poetry run python -m retrieval_gateway_mcp.mcp_server

Exposes the retrieve tool via STDIO transport.

Running Tests

All Tests

poetry run pytest

Unit Tests Only

poetry run pytest tests/unit/

Integration Tests

# Requires infrastructure running (docker-compose up -d)
poetry run pytest tests/integration/

Test Categories

1. Unit Tests (tests/unit/test_req_processor.py) - 14 tests

   • Core business logic
   • Buffering scenarios
   • Error handling
   • Idempotency

2. RequestProcessor Integration (tests/integration/test_req_processor.py) - 2 tests

   • Database interactions
   • End-to-end request flow

3. Kafka Integration (tests/integration/test_kafka.py) - 2 tests

   • Idempotency with duplicate delivery
   • Offset replay behavior

4. MCP Integration (tests/integration/test_mcp.py) - 2 tests

   • Simple retrieve tool usage
   • Out-of-order buffering with MCP client

MCP Server Usage

With MCP Inspector

npx @modelcontextprotocol/inspector poetry run python -m retrieval_gateway_mcp.mcp_server

Opens web UI at http://localhost:5173 for interactive testing.

Programmatic Usage

See test_mcp_client.py for example Python client code.

API Reference

MCP Tool: retrieve

Retrieve a resource with ordered processing.

Parameters:

• ordering_key (string, required): Key for ordering (e.g., user ID)
• sequence (int, required): Sequence number for this ordering key
• resource_type (string, required): Resource type (e.g., "contact")
• resource_key (string, required): Resource identifier
• tenant_id (string, optional): Tenant ID (default: "default")
• client_id (string, optional): Client ID (default: "mcp")

Returns:

{
  "res": {
    "request_id": "01KD1...",
    "status": "OK|ACCEPTED",
    "payload": { ... },
    "served_at": 1234567890
  },
  "processed_from_buffer": [
    {
      "request_id": "01KD2...",
      "status": "OK",
      "payload": { ... }
    }
  ]
}

Status Codes:

• OK: Request processed successfully
• ACCEPTED: Request buffered (out of order)

Database Schema

requests table

• request_id (PK): Unique ULID
• ordering_key: Grouping key
• sequence: Order within group
• status: PENDING | COMPLETED | FAILED
• response_payload: JSON response
• created_at, updated_at: Timestamps

ordering_state table

• ordering_key (PK): Grouping key
• last_completed_sequence: Last processed sequence
• updated_at: Timestamp

buffered_requests table

• id (PK): Auto-increment
• ordering_key, sequence: Buffered request
• request_id: Reference to requests
• raw_request: Original JSON payload

Code Quality

This project uses ruff for linting and formatting via pre-commit.

Manual Checks

# Run on all files
poetry run pre-commit run --all-files

# Run ruff only
poetry run ruff check .
poetry run ruff format .

Auto-formatting

Pre-commit hooks run automatically on git commit if installed.

Configuration

Environment variables (.env or shell):

# Database
DATABASE_URL=postgresql://mcp_user:mcp_password@localhost:5432/mcp_gateway

# Kafka
KAFKA_BOOTSTRAP_SERVERS=localhost:29092
KAFKA_REQUEST_TOPIC=mcp.retrieval.request.v1
KAFKA_RESPONSE_TOPIC=mcp.retrieval.response.v1

# Mock Upstream
UPSTREAM_BASE_URL=http://localhost:8001

# Logging
LOG_LEVEL=INFO

Project Structure

retrieval-gateway-mcp/
├── src/retrieval_gateway_mcp/
│   ├── req_processor.py      # Core processing logic
│   ├── mcp_server.py          # MCP tool server
│   ├── models.py              # SQLAlchemy models
│   ├── config.py              # Configuration
│   ├── mock_upstream.py       # Test upstream server
│   └── kafka/
│       ├── consumer.py        # Kafka consumer worker
│       └── producer.py        # Kafka producer
├── tests/
│   ├── unit/                  # Unit tests
│   └── integration/           # Integration tests
├── docker-compose.yml         # Infrastructure
├── pyproject.toml             # Dependencies
└── README.md                  # This file

Known Issues

1. MCP Buffering Test: test_retrieve_tool_buffering has ~95% reliability due to Kafka async timing

License

MIT