MCP Federated Multi-Tool Aggregator

A production-grade MCP server that aggregates tools from 55+ MCP servers, resolves naming conflicts via automatic namespacing, executes multi-step dependency chains, and exposes a unified interface to LLMs through both a FastAPI REST API and a native MCP stdio server.

Workflows run with or without an Anthropic API key — a built-in deterministic fallback engine keeps everything operational offline.

Quick Start (4 commands)

git clone <your-repo-url> && cd mcp-aggregator
python -m venv .venv && source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install -r requirements.txt
python main.py

Open http://localhost:8000 — the browser UI loads immediately.

Anthropic API key is optional. To enable Claude-driven orchestration:

echo "ANTHROPIC_API_KEY=sk-ant-..." >> .env
python main.py

Without a key, the workflow engine runs in deterministic fallback mode — all tool aggregation, conflict resolution, dependency chains, and workflow execution continue to work normally.

Run automated demos (in a second terminal):

python demo/run_demos.py

Run tests:

pytest tests/ -v

Architecture

┌─────────────────────────────────────────────────────────────────┐
│          LLM Layer  (Anthropic — Optional)                       │
│   When configured: Claude plans tool usage                       │
│   When absent:     Deterministic fallback planner takes over     │
└──────────────────────────┬──────────────────────────────────────┘
                           │
┌──────────────────────────▼──────────────────────────────────────┐
│           LangGraph Orchestration Engine                         │
│                                                                  │
│   ┌─────────────────┐       ┌──────────────────────────────┐    │
│   │ Anthropic Planner│       │ Deterministic Fallback Planner│   │
│   │ (when API key   │       │ (always available — offline  │    │
│   │  is present)    │       │  dependency chains)          │    │
│   └────────┬────────┘       └──────────────┬───────────────┘    │
│            └──────────────┬────────────────┘                     │
│                           ▼                                      │
│      StateGraph: llm_call → router → tool_call → llm_call       │
│      Staged dependency chains: output of Tool A → input of B    │
└──────────────────────────┬──────────────────────────────────────┘
                           │
┌──────────────────────────▼──────────────────────────────────────┐
│                  AggregatorCore                                  │
│   • Conflict resolver (namespacing: server__tool)                │
│   • Tool router (qualified_name → upstream client, O(1))         │
│   • Execution history + health monitor                           │
└──────────────────────────┬──────────────────────────────────────┘
                           │
┌──────────────────────────▼──────────────────────────────────────┐
│               Async MCP Client Pool                              │
│   One MCPClientWrapper per server — stdio transport              │
│   All servers connect in parallel; failures are isolated         │
└──┬──────┬──────┬──────┬──────┬──────┬──────┬──────┬────────────┘
   │      │      │      │      │      │      │      │
echo   math  util  files   git  mock  mock  mock  … (55 total)
(custom)(custom)(custom)(npx) (npx) -01   -02   -50

Screenshots

API Documentation (FastAPI + OpenAPI)

Interactive OpenAPI documentation exposing tool execution, registry inspection, workflow orchestration, and execution history APIs.

Registry Statistics

Runtime statistics demonstrating successful aggregation across 55 MCP servers and 234 tools.

Tool Discovery and Execution

Automated demo validating server discovery, tool registration, direct execution, and unified routing.

Workflow Orchestration

Successful execution of a dependency-aware workflow using the deterministic fallback orchestration engine.

Key Design Decisions

| Decision | Rationale | |---|---| | MCP Python SDK (official) | Real stdio transport — not a simulation | | Qualified names (server__tool) | Zero ambiguity when routing; LLM sees full context | | Dual-mode orchestration | System stays operational without Anthropic API dependency | | SQLite registry | Survives restarts; inspectable with any DB viewer | | LangGraph StateGraph | Industry-standard DAG executor; loop + branch topology | | AsyncIO connection pool | All servers connect in parallel; O(1) routing | | FastAPI | Async-native; auto-generates OpenAPI docs at /docs | | Mock server scalability layer | Validates system behaviour at 50+ server scale |

Project Structure

mcp-aggregator/
├── main.py                   # Entry point (HTTP or MCP stdio mode)
├── requirements.txt
├── .env.example
│
├── aggregator/
│   ├── core.py               # AggregatorCore: startup, routing, health
│   ├── registry.py           # SQLite-backed tool/server/conflict store
│   ├── conflict.py           # ConflictResolver: namespacing + dedup
│   ├── mcp_client.py         # Async MCP client wrapper (per server)
│   └── mcp_server.py         # Exposes aggregator AS an MCP server
│
├── orchestration/
│   ├── state.py              # WorkflowState TypedDict
│   ├── nodes.py              # llm_call_node (Anthropic + fallback),
│   │                         # tool_call_node, router_node
│   ├── graph.py              # LangGraph StateGraph builder
│   └── chains.py             # Pre-built demo workflow chains
│
├── api/
│   ├── app.py                # FastAPI factory + lifespan
│   ├── models.py             # Pydantic schemas
│   └── routes/
│       ├── tools.py          # GET /tools, POST /tools/{name}/call
│       ├── workflows.py      # POST /workflows/run, GET /workflows
│       └── registry.py       # GET /registry/stats|conflicts|servers
│
├── servers/
│   ├── config.py             # All server configs (add more here)
│   └── custom/
│       ├── echo_server.py    # echo, reverse_string, word_count
│       ├── math_server.py    # add, multiply, compute_expression, search*
│       ├── utility_server.py # get_time, format_json, hash_text, search*
│       └── mock_server.py    # Scalability validation server (×50)
│                             # (* search exists on 2 servers → conflict demo)
├── db/
│   └── schema.sql
│
├── demo/
│   ├── run_demos.py          # Automated demo runner
│   └── index.html            # Browser UI
│
└── tests/
    ├── test_registry.py
    └── test_conflict.py

API Reference

| Method | Endpoint | Description | |--------|----------|-------------| | GET | /tools | List all aggregated tools (filter: ?server=name) | | GET | /tools/{name} | Get tool metadata | | POST | /tools/{name}/call | Call a tool {"arguments": {...}} | | GET | /workflows | List available workflow chains | | POST | /workflows/run | Run a workflow {"workflow": "...", "input": {...}} | | POST | /workflows/run/stream | SSE streaming workflow execution | | GET | /registry/stats | Aggregation statistics | | GET | /registry/conflicts | All conflict records + resolutions | | GET | /registry/servers | All configured servers + status | | GET | /registry/history | Recent tool execution log | | GET | /health | Health check + stats | | GET | /docs | Interactive OpenAPI docs (Swagger UI) |

Conflict Resolution

When two or more servers expose a tool with the same name, the aggregator:

1. Assigns every tool its qualified name: server_name__tool_name
2. Records the conflict in SQLite with the list of offending servers
3. Marks the highest-priority server's tool as the winner
4. All tools remain callable via their qualified names — nothing is dropped

Example — both math-server and utility-server have a search tool:

math-server__search    → routes to math-server
utility-server__search → routes to utility-server
conflict record: tool=search, servers=[math-server, utility-server], winner=math-server__search

Inspect all conflicts:

curl http://localhost:8000/registry/conflicts

Workflow Execution

The aggregator supports two execution modes that are selected automatically:

Mode 1 — Anthropic-Assisted (when ANTHROPIC_API_KEY is set)

Claude reads the full tool registry, plans the sequence of tool calls, interprets intermediate results, and drives the LangGraph loop until the task is complete.

User prompt → Claude → tool_call_node → Claude (reads result) → … → final summary

Mode 2 — Deterministic Fallback (no API key required)

A built-in planner executes pre-defined workflow chains without any LLM involvement. Critically, this is not a simulation — it runs through the same tool_call_node execution path and builds real dependency chains from actual tool outputs:

Fallback planner → tool_call_node (Tool A) → read real result → tool_call_node (Tool B with A's output) → …

Example: time_and_hash in fallback mode

Step 1: call utility-server__get_time  → "2024-01-15T10:30:00+00:00"   (real tool call)
Step 2: call utility-server__hash_text with the actual timestamp above  (real dependency)
        → sha256("2024-01-15T10:30:00+00:00") = "a3f9..."

Both modes produce identical API response shapes — callers cannot tell the difference.

Dependency Chains (Tool A → Tool B)

The LangGraph orchestration layer enables multi-step workflows where the output of one tool becomes the input of the next. Dependency chains work in both Anthropic and fallback mode.

Run a chain:

curl -X POST http://localhost:8000/workflows/run \
  -H "Content-Type: application/json" \
  -d '{"workflow": "time_and_hash", "input": {}}'

Available chains:

| Chain | Steps | Notes | |-------|-------|-------| | echo_reverse | echo → reverse_string | Basic 2-step chain | | math_pipeline | add → multiply → compute_expression | 3-step chained math (each step uses prior result) | | file_analysis | read_file → word_count → LLM summary | Requires filesystem server + API key | | time_and_hash | get_time → hash_text | Live timestamp feeds hash input | | conflict_demo | math-server__search + utility-server__search | Demonstrates namespaced routing |

Scaling to 50+ Servers

Adding more servers requires one block in servers/config.py:

ServerConfig(
    name="github",
    command="npx",
    args=["-y", "@modelcontextprotocol/server-github"],
    env={"GITHUB_PERSONAL_ACCESS_TOKEN": os.getenv("GITHUB_TOKEN", "")},
    priority=8,
),

The aggregator handles the rest automatically:

• Connects in parallel (asyncio)
• Discovers tools automatically
• Resolves any new conflicts
• Updates the SQLite registry
• Routes calls with O(1) dict lookup

No code changes needed in any other module.

Ready-to-add servers (just uncomment in servers/config.py):

• @modelcontextprotocol/server-brave-search (needs BRAVE_API_KEY)
• @modelcontextprotocol/server-github (needs GITHUB_TOKEN)
• @modelcontextprotocol/server-slack (needs SLACK_BOT_TOKEN)
• mcp-server-aws (needs AWS credentials)
• Any custom Python MCP server following the pattern in servers/custom/

System Scale Demonstration

| Metric | Value | |--------|-------| | Configured MCP Servers | 58 | | Successfully Connected | 55 | | Aggregated Tools | 234 | | Tool Name Conflicts Resolved | 21 | | Workflow Executions Tested | 22 |

Server composition:

• 3 custom MCP servers (echo, math, utility — written from scratch)
• 5 external MCP server integrations (filesystem, git, fetch, sqlite, memory via npx)
• 50 mock MCP servers for scalability validation (async pool + conflict resolution at scale)

The mock servers validate that the aggregator's async connection pool, conflict resolver, and routing layer operate correctly under load — they are not counted as independent production integrations.

Claude Desktop Integration

To use the aggregator as a single MCP server in Claude Desktop:

// ~/.config/claude/claude_desktop_config.json
{
  "mcpServers": {
    "aggregator": {
      "command": "python",
      "args": ["/path/to/mcp-aggregator/main.py", "--mcp-stdio"]
    }
  }
}

Claude Desktop will see all aggregated tools under their qualified names.

Implemented Features

• [x] Real MCP SDK connections (stdio transport, proper lifecycle)
• [x] 3 custom MCP servers written from scratch (echo, math, utility)
• [x] 5 external MCP servers supported via npx (filesystem, git, fetch, sqlite, memory)
• [x] 50 mock MCP servers for scalability validation
• [x] SQLite-backed tool registry (servers, tools, conflicts, executions, workflows)
• [x] Conflict detection + namespaced resolution (server__tool)
• [x] Unified tool routing (O(1) dict lookup)
• [x] LangGraph StateGraph with loop + branch topology
• [x] Tool dependency chains (A→B output chaining)
• [x] Offline deterministic workflow execution (no API key required)
• [x] Anthropic-free workflow fallback mode
• [x] Dynamic staged dependency chaining in fallback mode
• [x] FastAPI REST API with 10+ endpoints
• [x] SSE streaming workflow execution
• [x] Execution history persistence
• [x] Background health monitor (30s ping)
• [x] Browser demo UI (single HTML file, no framework)
• [x] Automated demo script
• [x] --mcp-stdio mode (acts as MCP server for Claude Desktop)
• [x] 11 unit tests (registry + conflict resolver)

Tools & Frameworks

| Tool | Version | Why | |------|---------|-----| | mcp (official SDK) | ≥1.0 | Real MCP protocol — stdio transport, proper ClientSession lifecycle | | FastAPI | ≥0.111 | Async-native, auto OpenAPI docs, streaming responses (SSE) | | LangGraph | ≥0.1 | DAG-based agentic orchestration; built-in loop/branch topology | | langchain-anthropic | ≥0.1 | Claude API integration (claude-opus-4-5) | | SQLAlchemy / aiosqlite | ≥2.0 | Async SQLite for registry persistence | | Pydantic v2 | ≥2.0 | Request/response validation | | httpx | ≥0.27 | Async HTTP for demo scripts | | uvicorn | ≥0.29 | ASGI server with hot-reload support | | pytest | ≥8.0 | Unit tests |

Design Challenges & Solutions

| Challenge | Solution | |-----------|----------| | Tool name collisions across servers | server__tool namespacing — every tool gets a globally unique qualified name; conflicts are recorded in SQLite with winner designation | | Workflow execution without LLM access | Deterministic fallback planner in nodes.py — executes the same tool_call_node path, reads real tool outputs, and stages subsequent calls as genuine dependency chains | | Scaling to 50+ servers without blocking | asyncio connection pool — all servers connect in parallel; each server runs in an isolated MCPClientWrapper; failures are non-fatal | | Dependency chains that chain real outputs | Staged scheduling in fallback: initial pass enqueues only the first tool; after tool_call_node writes the result, the next pass reads it and builds the downstream call | | Routing across hundreds of tools efficiently | O(1) dict lookup on qualified name — no iteration, no string matching at call time |

Known Limitations

1. No auth/TLS — endpoints are open. Add OAuth2 or API keys before production use.
2. Shared asyncio loop — all upstream MCP servers run in the same process. A frozen subprocess blocks the loop until the CALL_TIMEOUT (30s) expires.
3. stdio transport only — SSE-transport MCP servers (remote) require adding an SSE client branch to mcp_client.py.
4. npx servers require Node.js — if Node is not installed, those 5 servers silently fail and only the 3 custom Python servers connect.
5. LangGraph circuit-breaker at 10 steps — deep chains are capped. Raise step_count >= 10 in nodes.py for more complex workflows.
6. No persistent message history across HTTP calls — each /workflows/run call starts a fresh LangGraph state. Add Redis or SQLite message store for multi-turn sessions.
7. Tool schema passthrough — input schemas are stored and forwarded as-is. Validation against schema before dispatch is not yet enforced.
8. Scalability demonstration uses mock servers — the 50-server scale test validates async pooling and conflict resolution behaviour; it does not represent 50 independent production integrations.

Future Enhancements

The current implementation demonstrates MCP federation, conflict resolution, workflow orchestration, and large-scale tool aggregation. Future improvements could further extend the platform toward production readiness.

Planned Enhancements

• Support additional MCP transport protocols such as SSE and WebSockets.
• Enable dynamic server registration and hot reloading without restarting the aggregator.
• Add authentication and authorization mechanisms (OAuth2, API Keys, RBAC).
• Introduce distributed execution across multiple aggregator instances.
• Expand workflow capabilities with a visual workflow builder and custom workflow definitions.
• Add observability features including metrics dashboards, tracing, and health monitoring.
• Integrate additional real-world MCP servers such as GitHub, Slack, cloud providers, and search services.
• Implement advanced workflow scheduling and event-driven execution.
• Improve caching and persistence strategies for high-volume deployments.
• Add comprehensive integration and performance testing for larger server deployments.

Long-Term Vision

The long-term goal is to evolve the MCP Federated Aggregator into a scalable tool orchestration platform capable of unifying hundreds of MCP servers, providing intelligent routing, workflow automation, and a single access layer for LLM-powered applications.