AliQambari/mcp-compose

🚀 Modular MCP Composition Architecture

🎯 Features

⚡ Lightning Fast

• Redis Caching: 100x faster responses after first query
• Connection Pooling: Efficient database connection management
• Async/Await: Non-blocking operations for maximum throughput

🏗️ Production Ready

• Modular Architecture: Clean separation of concerns
• Type Safety: Comprehensive type hints prevent runtime errors
• Error Handling: Structured exception hierarchy with user-friendly messages
• Observability: Detailed logging with cache hit/miss tracking

🔌 Modular Composition Architecture

• Server Composition: Combine multiple MCP servers into one endpoint
• Pluggable Services: Database, Math, and custom service modules
• Dynamic Loading: Enable/disable servers via configuration
• Tag-based Filtering: Fine-grained control over tool availability
• Extensible Framework: Add new MCP servers without touching core code

🚀 Quick Start

1️⃣ Installation

# Clone the repository
git clone <your-repo-url>
cd mcp_composition_server

# Install dependencies
pip install -e .

# Install Redis (if not already installed)
# Windows: Download from https://redis.io/download
# Linux: sudo apt install redis-server
# macOS: brew install redis

2️⃣ Configuration

Create your .env file:

# PostgreSQL Configuration
POSTGRES_HOST=localhost
POSTGRES_PORT=5432
POSTGRES_USER=postgres
POSTGRES_PASSWORD=your_password

# MySQL Configuration  
MYSQL_HOST=localhost
MYSQL_PORT=3306
MYSQL_USER=root
MYSQL_PASSWORD=your_password

# Redis Cache Configuration
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0

# Application Settings
LOG_LEVEL=INFO
SAVE_FILES=false

3️⃣ Start the Server

# Start Redis server
redis-server

# Start the composed MCP server
python main.py

4️⃣ Test with Client

from fastmcp import Client

async def test_database():
    client = Client("http://localhost:8000/mcp")
    async with client:
        # List tables in PostgreSQL database
        result = await client.call_tool("list_tables", {
            "db_type": "postgres",
            "db_name": "your_database"
        })
        print(result.content[0].text)

import asyncio
asyncio.run(test_database())

📊 Performance Comparison

🏗️ Architecture Overview

graph TB
    Client[MCP Client] --> Composer[MCP Composition Server]
    Composer --> DB[Database Server]
    Composer --> Math[Math Server]
    Composer --> Custom[Custom Server N...]
    
    DB --> Cache{Redis Cache}
    Math --> Cache
    
    Cache -->|Hit| Return[Cached Response]
    Cache -->|Miss| Process[Process Request]
    
    Process --> MySQL[(MySQL)]
    Process --> Postgres[(PostgreSQL)]
    Process --> Store[Store in Cache]
    Store --> Return

🎨 Design Patterns Used

• 🔗 Composition Pattern: Multiple MCP servers composed into unified interface
• 🏭 Factory Pattern: Database client creation without exposing instantiation logic
• 📚 Repository Pattern: Abstract database operations through consistent interfaces
• 💉 Dependency Injection: Services receive dependencies through constructors
• 🎯 Single Responsibility: Each server/service has one clear purpose
• 🔄 Strategy Pattern: Different implementations with same interface
• 🏗️ Builder Pattern: Dynamic server composition based on configuration

🔧 Configuration

📁 Project Structure

mcp-composition-server/
├── 📁 database/              # Database layer
│   ├── __init__.py          # Package exports
│   ├── base.py              # Abstract interfaces
│   ├── mysql_client.py      # MySQL implementation
│   ├── postgres_client.py   # PostgreSQL implementation
│   └── factory.py           # Database factory
├── 📁 services/             # Business logic layer
│   ├── __init__.py          # Package exports
│   ├── table_service.py     # Table operations
│   └── math_service.py      # Math operations
├── 📁 servers/              # Server modules
│   ├── __init__.py          # Package exports
│   ├── database_server.py   # Database MCP server
│   └── math_server.py       # Math MCP server
├── 📁 utils/                # Utility modules
│   ├── __init__.py          # Package exports
│   ├── logger.py           # Logging utilities
│   ├── file_utils.py       # File operations
│   └── cache_decorator.py   # Redis caching with aiocache
├── 📁 outputs/              # Output files (optional)
├── config.py               # Configuration management
├── main.py                 # Composed MCP server
├── test_client.py          # Test client example
├── pyproject.toml          # Project configuration & dependencies
├── .env                    # Environment variables
└── README.md              # This documentation

⚙️ Environment Variables

🛠️ Available Tools

📋 Database Tools

list_tables

List all tables in a database with caching support.

Parameters:

• db_type (string): Database type - "mysql" or "postgres"
• db_name (string): Name of the database

Example:

result = await client.call_tool("list_tables", {
    "db_type": "postgres",
    "db_name": "my_database"
})

🧮 Math Tools

add

Add two numbers with caching support.

Parameters:

• a (float): First number
• b (float): Second number

Example:

result = await client.call_tool("add", {
    "a": 5.5,
    "b": 3.2
})

📈 Monitoring & Logging

🔍 Cache Monitoring

The server provides detailed logging for cache operations:

# Cache Miss (First Call)
💾 CACHE MISS: list_tables:postgres:mydb - fetching from database
🔍 Fetching tables from postgres database: mydb
✅ Database query completed: 25 tables from mydb
✅ CACHED: list_tables:postgres:mydb (TTL: 300s)

# Cache Hit (Subsequent Calls)  
🎯 CACHE HIT: list_tables:postgres:mydb

📊 Performance Metrics

Monitor your server performance:

• Cache Hit Rate: Track how often cache is used vs database queries
• Response Times: Monitor query execution times
• Connection Pool: Database connection utilization
• Error Rates: Track and alert on failures

🚀 Extending the Architecture

🔌 Adding New MCP Servers

1. Create your server in servers/my_server.py:

from fastmcp import FastMCP

def create_my_server() -> FastMCP:
    server = FastMCP("my-service")
    
    @server.tool()
    async def my_tool(param: str) -> str:
        return f"Processed: {param}"
    
    return server

2. Add to servers/__init__.py:

from .my_server import create_my_server
__all__ = [..., "create_my_server"]

3. Update main.py:

if "my_service" in enabled_servers:
    await main_server.import_server(create_my_server())

🛠️ Configuration-Driven Composition

# Enable only specific servers
ENABLED_SERVERS=database,math,my_service

# Filter by tags
SERVER_INCLUDE_TAGS=production
SERVER_EXCLUDE_TAGS=admin,debug

check EXTENDING.md for detailed examples

🎯 Adding Caching to Any Method

from utils.cache_decorator import cached

class MyService:
    @cached(ttl=600)  # 10 minutes cache
    async def expensive_operation(self, param: str) -> Result:
        # Automatically cached!
        pass

🐳 Docker Support

📦 Docker Compose

version: '3.8'
services:
  mcp-composition-server:
    build: .
    ports:
      - "8000:8000"
    environment:
      - POSTGRES_HOST=postgres
      - REDIS_HOST=redis
      - ENABLED_SERVERS=database,math
    depends_on:
      - postgres
      - redis

  postgres:
    image: postgres:15
    environment:
      POSTGRES_DB: mydb
      POSTGRES_USER: user
      POSTGRES_PASSWORD: password

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  - postgres
  - redis

postgres: image: postgres:15 environment: POSTGRES_DB: mydb POSTGRES_USER: user POSTGRES_PASSWORD: password

redis: image: redis:7-alpine ports: - "6379:6379"

### 🚀 **Quick Docker Start**

```bash
# Build and start all services
docker-compose up -d

# View logs
docker-compose logs -f mcp-composition-server

# Stop services
docker-compose down

🧪 Testing

Use the included test client to verify functionality:

# Test the composed server with all tools
python test_client.py

# Test specific server composition
ENABLED_SERVERS=math python test_client.py

🤝 Contributing

We welcome contributions! Here's how to get started:

📝 Contribution Guidelines

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -m 'Add amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Open a Pull Request

🙏 Acknowledgments

• FastMCP - For the excellent MCP framework
• aiomysql & asyncpg - For async database drivers
• aiocache - For Redis caching made simple
• Redis - For blazing-fast caching