DEV Community

Miguel Julio
Miguel Julio

Posted on

Building MCP one: A Unified Hub for Model Context Protocol Servers

#ai #mcp #opensource #python

How can we sovlve a integration problem with a centralized approach

The Problem That Started It All

If you've worked with multiple Model Context Protocol (MCP) servers, you know the pain: each server has its own quirks, different endpoint structures, varying response formats, and unique payload requirements.

Imagine you're building an AI application that needs to connect to:

  • A GitHub MCP server (for repository operations)
  • A SQL MCP server (for database queries)
  • A Jupyter MCP server (for code execution)
  • Your custom business logic MCP

There are four different integrations to maintain, monitor, and debug. It may sound simple, and in fact it is, but scale this to more than 10 servers/clients and you'll have a maintenance nightmare.

I know, the math is simple but brutal: M clients × N servers = M×N integration points.

Enter MCP one: From M×N to M+N

Instead of every client integrating with every MCP server directly, what if we had a single hub that normalizes everything?

Before (M×N):
Client A ──┬── MCP Server 1
           ├── MCP Server 2  
           └── MCP Server 3

Client B ──┬── MCP Server 1
           ├── MCP Server 2
           └── MCP Server 3

After (M+N):
Client A ──┐
           ├── MCP Hub ──┬── MCP Server 1
Client B ──┘             ├── MCP Server 2
                         └── MCP Server 3
Enter fullscreen mode Exit fullscreen mode

This is exactly what MCP one does, it acts as a unified gateway that speaks the same language to all your clients while handling the complexity of different MCP implementations behind the scenes.

Key Architecture Decisions

1. Configuration-Driven Approach

Rather than hardcoding server specifics, everything is defined in YAML:

servers:
  - name: github_mcp
    url: http://localhost:3001
    enabled: true
    timeout: 30
    endpoints:
      health: /health
      tools: /api/tools
      call: /api/execute
    response_map:
      tools_key: "available_tools"  # GitHub MCP wraps tools in this key
      tool_name_field: "name"
      tool_desc_field: "description"
    payload_map:
      tool_field: "tool_name"       # GitHub MCP expects this field name
      args_field: "parameters"      # Instead of "arguments"

  - name: sql_mcp  
    url: http://localhost:3002
    enabled: true
    endpoints:
      health: /status
      tools: /tools
      call: /execute
    response_map:
      tools_key: ""                 # SQL MCP returns tools directly
      tool_name_field: "name"
      tool_desc_field: "desc"
    payload_map:
      tool_field: "tool"
      args_field: "args"
Enter fullscreen mode Exit fullscreen mode

This approach means adding a new MCP server is just a few lines of config, no code changes required.

2. Flexible Mapping System

Different MCP servers structure their responses differently. Some return tools directly as an array, others wrap them in an object. Some use description, others use desc.

The mapping system handles this elegantly:

# In registry.py
resp_map = server_info.config.response_map
tools_key = resp_map.get("tools_key", "tools")
name_field = resp_map.get("tool_name_field", "name")
desc_field = resp_map.get("tool_desc_field", "description")

# Handle direct array vs wrapped response
if tools_key:
    raw_tools = raw.get(tools_key, [])
else:
    raw_tools = raw  # Already an array

# Extract fields dynamically
for tool in raw_tools:
    t_name = tool.get(name_field)
    t_desc = tool.get(desc_field, "")
Enter fullscreen mode Exit fullscreen mode

3. Async-First Design

Built with FastAPI and httpx, everything is async from the ground up:

async def execute_tool(self, request: ToolCallRequest) -> ToolCallResponse:
    # Find the tool
    tool = await self.registry.get_tool(request.tool)

    # Get server info
    server_info = await self.registry.get_server_info(tool.server_name)

    # Execute asynchronously
    response = await self._call_mcp_tool(
        server_info.config,
        tool.name,
        request.arguments
    )

    return response
Enter fullscreen mode Exit fullscreen mode

This ensures the hub can handle multiple concurrent requests without blocking.

4. Health Monitoring & Auto-Recovery

The hub continuously monitors all registered servers:

async def _background_refresh_loop(self, interval: int) -> None:
    while not self._shutdown:
        try:
            await self.refresh_all_servers()
            await asyncio.sleep(interval)
        except Exception as e:
            logger.error("background_refresh_error", error=str(e))
            await asyncio.sleep(5)  # Retry in 5 seconds
Enter fullscreen mode Exit fullscreen mode

Servers that go offline are automatically detected and marked as unavailable. When they come back online, they're automatically re-integrated.

The API That Just Works

From the client perspective, MCP one provides a clean, consistent API regardless of the underlying server complexity:

List All Available Tools 

GET /tools
Enter fullscreen mode Exit fullscreen mode

Response:

{
  "tools": [
    {
      "name": "query_database",
      "description": "Execute SQL queries",
      "full_name": "sql_mcp.query_database",
      "server_name": "sql_mcp"
    },
    {
      "name": "create_pr",
      "description": "Create a GitHub pull request", 
      "full_name": "github_mcp.create_pr",
      "server_name": "github_mcp"
    }
  ],
  "total_count": 2
}
Enter fullscreen mode Exit fullscreen mode

Execute Any Tool 

POST /call
{
  "tool": "sql_mcp.query_database",
  "arguments": {
    "query": "SELECT * FROM users LIMIT 10"
  }
}
Enter fullscreen mode Exit fullscreen mode

Response:

{
  "success": true,
  "result": {
    "rows": [...],
    "count": 10
  },
  "server_name": "sql_mcp",
  "execution_time_ms": 45.2
}
Enter fullscreen mode Exit fullscreen mode

Monitor System Health 

GET /status
Enter fullscreen mode Exit fullscreen mode 
{
  "version": "0.1.0",
  "uptime_seconds": 3600,
  "servers_count": 3,
  "servers_online": 2,
  "tools_count": 15,
  "last_refresh": "2024-01-15T10:30:00Z"
}
Enter fullscreen mode Exit fullscreen mode

Real-World Benefits

For Developers

  • Single integration point instead of N different APIs
  • Consistent error handling across all MCP servers
  • Built-in monitoring and health checks
  • Easy configuration, no code changes to add new servers

For Operations

  • Centralized logging with structured logs via structlog
  • Performance metrics with execution time tracking
  • Health monitoring with automatic failure detection
  • Graceful degradation when servers go offline

For Architecture

  • Reduced coupling between clients and MCP servers
  • Easier testing with a single API surface
  • Better scalability with async request handling
  • Configuration as code approach

Implementation Highlights

Clean Separation of Concerns 

# Registry: Manages server lifecycle and tool discovery
class MCPRegistry:
    async def register_server(self, config: MCPServerConfig) -> bool
    async def refresh_all_servers(self) -> None
    async def list_tools(self, server_name: Optional[str] = None) -> List[ToolSchema]

# Router: Handles tool execution
class MCPRouter:
    async def execute_tool(self, request: ToolCallRequest) -> ToolCallResponse

# Main: FastAPI application with dependency injection
app = FastAPI(lifespan=lifespan)
Enter fullscreen mode Exit fullscreen mode

Robust Error Handling 

async def execute_tool(self, request: ToolCallRequest) -> ToolCallResponse:
    start_time = time.time()

    try:
        tool = await self.registry.get_tool(request.tool)
        if not tool:
            return ToolCallResponse(
                success=False,
                error="tool_not_found",  
                execution_time_ms=(time.time() - start_time) * 1000
            )

        # ... execution logic

    except Exception as e:
        logger.error("tool_execution_failed", tool_name=request.tool, error=str(e))
        return ToolCallResponse(
            success=False,
            error="execution_failed",
            execution_time_ms=(time.time() - start_time) * 1000
        )
Enter fullscreen mode Exit fullscreen mode

Type Safety with Pydantic

All data structures are properly typed:

class ToolCallRequest(BaseModel):
    tool: str = Field(..., description="Full tool name (server.tool)")
    arguments: Dict[str, Any] = Field(default_factory=dict)

    @validator('tool')
    def validate_tool_name(cls, v):
        if '.' not in v:
            raise ValueError('Tool name must be in format: server.tool')
        return v
Enter fullscreen mode Exit fullscreen mode

Getting Started

  1. Clone and install: 
git clone https://github.com/your-username/mcp-one.git
cd mcp-one
pip install -r requirements.txt
Enter fullscreen mode Exit fullscreen mode

  1. Configure your servers in src/config.yaml

  2. Run the hub:

cd src
uvicorn app.main:app --reload
Enter fullscreen mode Exit fullscreen mode
  1. Start using the unified API at http://localhost:8000

What's Next?

This is an MVP that solves the core problem, but there's more to build (feel free to contribute 😁) :

  • Authentication & authorization for production deployments
  • Rate limiting to prevent abuse
  • Metrics & monitoring integration (Prometheus, etc.)
  • Load balancing for high-availability MCP servers
  • WebSocket support for real-time tool streaming
  • Plugin system for custom transformations

Why This Matters

As AI applications become more complex, they need to integrate with more external services. The MCP standard is growing, but without proper tooling, we'll face the same integration hell we've seen before.

MCP one represents a different approach: instead of fighting complexity, we embrace it through abstraction. By providing a single, consistent interface to multiple MCP servers, we reduce maintenance burden while enabling more sophisticated AI applications.

The goal isn't to replace MCP servers - it's to make them easier to use together.

Links

Top comments (0)