syndarix/docs/spikes/SPIKE-001-mcp-integration-pattern.md

SPIKE-001: MCP Integration Pattern

Status: Completed Date: 2025-12-29 Author: Architecture Team Related Issue: #1

---

Objective

Research the optimal pattern for integrating Model Context Protocol (MCP) servers with FastAPI backend, focusing on unified singleton servers with project/agent scoping.

Research Questions

1. What is the recommended MCP SDK for Python/FastAPI?
2. How should we structure unified MCP servers vs per-project servers?
3. What is the best pattern for project/agent scoping in MCP tools?
4. How do we handle authentication between Syndarix and MCP servers?

Findings

1. FastMCP 2.0 - Recommended Framework

FastMCP is a high-level, Pythonic framework for building MCP servers that significantly reduces boilerplate compared to the low-level MCP SDK.

Key Features:

• Decorator-based tool registration (@mcp.tool())
• Built-in context management for resources and prompts
• Support for server-sent events (SSE) and stdio transports
• Type-safe with Pydantic model support
• Async-first design compatible with FastAPI

Installation:

pip install fastmcp

Basic Example:

from fastmcp import FastMCP

mcp = FastMCP("syndarix-knowledge-base")

@mcp.tool()
def search_knowledge(
    project_id: str,
    query: str,
    scope: str = "project"
) -> list[dict]:
    """Search the knowledge base with project scoping."""
    # Implementation here
    return results

@mcp.resource("project://{project_id}/config")
def get_project_config(project_id: str) -> dict:
    """Get project configuration."""
    return config

2. Unified Singleton Pattern (Recommended)

Decision: Use unified singleton MCP servers instead of per-project servers.

Architecture:

┌─────────────────────────────────────────────────────────┐
│                    Syndarix Backend                      │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐      │
│  │   Agent 1   │  │   Agent 2   │  │   Agent 3   │      │
│  │ (project A) │  │ (project A) │  │ (project B) │      │
│  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘      │
│         │                │                │              │
│         └────────────────┼────────────────┘              │
│                          │                               │
│                          ▼                               │
│  ┌─────────────────────────────────────────────────┐    │
│  │              MCP Client (Singleton)              │    │
│  │   Maintains connections to all MCP servers       │    │
│  └─────────────────────────────────────────────────┘    │
└──────────────────────────┬──────────────────────────────┘
                           │
           ┌───────────────┼───────────────┐
           │               │               │
           ▼               ▼               ▼
    ┌────────────┐  ┌────────────┐  ┌────────────┐
    │ Git MCP    │  │   KB MCP   │  │  LLM MCP   │
    │ (Singleton)│  │ (Singleton)│  │ (Singleton)│
    └────────────┘  └────────────┘  └────────────┘

Why Singleton:

• Resource efficiency (one process per MCP type)
• Shared connection pools
• Centralized logging and monitoring
• Simpler deployment (7 services vs N×7)
• Cross-project learning possible (if needed)

Scoping Pattern:

@mcp.tool()
def search_knowledge(
    project_id: str,       # Required - scopes to project
    agent_id: str,         # Required - identifies calling agent
    query: str,
    scope: Literal["project", "global"] = "project"
) -> SearchResults:
    """
    All tools accept project_id and agent_id for:
    - Access control validation
    - Audit logging
    - Context filtering
    """
    # Validate agent has access to project
    validate_access(agent_id, project_id)

    # Log the access
    log_tool_usage(agent_id, project_id, "search_knowledge")

    # Perform scoped search
    if scope == "project":
        return search_project_kb(project_id, query)
    else:
        return search_global_kb(query)

3. MCP Server Registry Architecture

# mcp/registry.py
from dataclasses import dataclass
from typing import Dict

@dataclass
class MCPServerConfig:
    name: str
    port: int
    transport: str  # "sse" or "stdio"
    enabled: bool = True

MCP_SERVERS: Dict[str, MCPServerConfig] = {
    "llm_gateway": MCPServerConfig("llm-gateway", 9001, "sse"),
    "git": MCPServerConfig("git-mcp", 9002, "sse"),
    "knowledge_base": MCPServerConfig("kb-mcp", 9003, "sse"),
    "issues": MCPServerConfig("issues-mcp", 9004, "sse"),
    "file_system": MCPServerConfig("fs-mcp", 9005, "sse"),
    "code_analysis": MCPServerConfig("code-mcp", 9006, "sse"),
    "cicd": MCPServerConfig("cicd-mcp", 9007, "sse"),
}

4. Authentication Pattern

MCP OAuth 2.0 Integration:

from fastmcp import FastMCP
from fastmcp.auth import OAuth2Bearer

mcp = FastMCP(
    "syndarix-mcp",
    auth=OAuth2Bearer(
        token_url="https://syndarix.local/oauth/token",
        scopes=["mcp:read", "mcp:write"]
    )
)

Internal Service Auth (Recommended for v1):

# For internal deployment, use service tokens
@mcp.tool()
def create_issue(
    service_token: str,  # Validated internally
    project_id: str,
    title: str,
    body: str
) -> Issue:
    validate_service_token(service_token)
    # ... implementation

5. FastAPI Integration Pattern

# app/mcp/client.py
from mcp import ClientSession
from mcp.client.sse import sse_client
from contextlib import asynccontextmanager

class MCPClientManager:
    def __init__(self):
        self._sessions: dict[str, ClientSession] = {}

    async def connect_all(self):
        """Connect to all configured MCP servers."""
        for name, config in MCP_SERVERS.items():
            if config.enabled:
                session = await self._connect_server(config)
                self._sessions[name] = session

    async def call_tool(
        self,
        server: str,
        tool_name: str,
        arguments: dict
    ) -> Any:
        """Call a tool on a specific MCP server."""
        session = self._sessions[server]
        result = await session.call_tool(tool_name, arguments)
        return result.content

# Usage in FastAPI
mcp_client = MCPClientManager()

@app.on_event("startup")
async def startup():
    await mcp_client.connect_all()

@app.post("/api/v1/knowledge/search")
async def search_knowledge(request: SearchRequest):
    result = await mcp_client.call_tool(
        "knowledge_base",
        "search_knowledge",
        {
            "project_id": request.project_id,
            "agent_id": request.agent_id,
            "query": request.query
        }
    )
    return result

Recommendations

Immediate Actions

1. Use FastMCP 2.0 for all MCP server implementations
2. Implement unified singleton pattern with explicit scoping
3. Use SSE transport for MCP server connections
4. Service tokens for internal auth (v1), OAuth 2.0 for future

MCP Server Priority

1. LLM Gateway - Critical for agent operation
2. Knowledge Base - Required for RAG functionality
3. Git MCP - Required for code delivery
4. Issues MCP - Required for project management
5. File System - Required for workspace operations
6. Code Analysis - Enhance code quality
7. CI/CD - Automate deployments

Code Organization

syndarix/
├── backend/
│   └── app/
│       └── mcp/
│           ├── __init__.py
│           ├── client.py         # MCP client manager
│           ├── registry.py       # Server configurations
│           └── schemas.py        # Tool argument schemas
└── mcp_servers/
    ├── llm_gateway/
    │   ├── __init__.py
    │   ├── server.py
    │   └── tools.py
    ├── knowledge_base/
    ├── git/
    ├── issues/
    ├── file_system/
    ├── code_analysis/
    └── cicd/

References

• FastMCP Documentation
• MCP Protocol Specification
• Anthropic MCP SDK

Decision

Adopt FastMCP 2.0 with unified singleton servers and explicit project/agent scoping for all MCP integrations.

---

Spike completed. Findings will inform ADR-001: MCP Integration Architecture.