MCP Server Architecture

Model Context Protocol (MCP) is the backbone of AI integration in the AltSportsLeagues.ai ecosystem. This guide covers MCP server architecture, implementation patterns, and best practices.

Overview

What is MCP?

Model Context Protocol (MCP) is a standardized protocol for AI model integration that enables:

Tool Discovery: AI models can discover available tools and their capabilities
Resource Access: Structured access to data sources, APIs, and services
Prompt Management: Centralized prompt templates and context injection
Stateful Interactions: Persistent context across multiple AI interactions

MCP in AltSportsLeagues.ai

The project uses MCP for:

Workflow Automation: n8n integration for automated pipelines
Data Access: Direct database operations across Supabase, Neo4j, Firebase
External Integrations: Atlassian (Jira/Confluence), Google Workspace
League Processing: Custom MCP servers for league discovery and onboarding
AI Enhancement: RAG systems, vector search, and intelligent routing

Architecture Layers

Layer 1: MCP Server Infrastructure

┌─────────────────────────────────────────────────────────┐
│                   Claude Code / AI Client                │
└────────────────────┬────────────────────────────────────┘
                     │ MCP Protocol (SSE/stdio)
                     │
┌────────────────────┴────────────────────────────────────┐
│                 MCP Server Layer                         │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐              │
│  │  n8n-mcp │  │ Atlassian│  │ Custom   │              │
│  │          │  │   MCP    │  │  Servers │              │
│  └──────────┘  └──────────┘  └──────────┘              │
└────────────────────┬────────────────────────────────────┘
                     │
┌────────────────────┴────────────────────────────────────┐
│              Service Layer (APIs, Databases)             │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐              │
│  │   n8n    │  │  Jira/   │  │ Supabase │              │
│  │ Workflows│  │Confluence│  │ Neo4j    │              │
│  └──────────┘  └──────────┘  └──────────┘              │
└─────────────────────────────────────────────────────────┘

Layer 2: Transport Mechanisms

Server-Sent Events (SSE):

{
  "command": "npx",
  "args": ["n8n-mcp"],
  "env": {
    "MCP_MODE": "sse",
    "N8N_API_URL": "https://altsportsdata.app.n8n.cloud"
  }
}

stdio (Standard I/O):

{
  "command": "python",
  "args": ["-m", "apps.backend.mcp_servers.servers.league_discovery_cross_comparison_mcp"],
  "env": {}
}

Docker Containers:

{
  "command": "docker",
  "args": [
    "run", "-i", "--rm",
    "ghcr.io/sooperset/mcp-atlassian:latest"
  ]
}

Layer 3: Tool Interfaces

MCP tools provide structured interfaces for AI interactions:

Tool Structure:

{
  name: "mcp__n8n-mcp__search_nodes",
  description: "Search n8n nodes by keyword",
  inputSchema: {
    type: "object",
    properties: {
      query: { type: "string" },
      limit: { type: "number" }
    },
    required: ["query"]
  }
}

MCP Server Implementations

1. n8n-mcp (Workflow Automation)

Architecture:

┌─────────────────────────────────────────────────┐
│              n8n-mcp Server                      │
│                                                  │
│  ┌──────────────────────────────────────────┐  │
│  │        Tool Categories                    │  │
│  │                                           │  │
│  │  • Node Discovery (525+ nodes)           │  │
│  │  • Workflow Management                   │  │
│  │  • Template Library (1000+ templates)    │  │
│  │  • Execution & Monitoring                │  │
│  │  • Validation & Testing                  │  │
│  └──────────────────────────────────────────┘  │
│                                                  │
│  ┌──────────────────────────────────────────┐  │
│  │         n8n API Client                    │  │
│  │  • REST API Communication                │  │
│  │  • WebSocket Connections                 │  │
│  │  • Authentication & Security             │  │
│  └──────────────────────────────────────────┘  │
└─────────────────┬───────────────────────────────┘
                  │
                  │ HTTPS
                  │
      ┌───────────▼──────────────┐
      │  n8n Cloud Instance       │
      │  altsportsdata.app.n8n... │
      └──────────────────────────┘

Key Tools:

// Node Discovery
mcp__n8n-mcp__search_nodes({ query: "slack", limit: 20 })
mcp__n8n-mcp__get_node_documentation({ nodeType: "nodes-base.slack" })
 
// Workflow Management
mcp__n8n-mcp__n8n_create_workflow({ name, nodes, connections })
mcp__n8n-mcp__n8n_list_workflows({ limit: 100 })
mcp__n8n-mcp__n8n_validate_workflow({ id })
 
// Template Access
mcp__n8n-mcp__search_templates({ query: "email automation" })
mcp__n8n-mcp__get_template({ templateId: 123 })
 
// Execution
mcp__n8n-mcp__n8n_trigger_webhook_workflow({ webhookUrl, data })
mcp__n8n-mcp__n8n_get_execution({ id })

Implementation Pattern:

# Python MCP Server (FastMCP)
from fastmcp import FastMCP
 
mcp = FastMCP("league-processor")
 
@mcp.tool()
async def process_league(league_id: str) -> dict:
    """Process league data and trigger n8n workflow."""
    # Validate league data
    league = await validate_league(league_id)
 
    # Trigger n8n workflow
    result = await trigger_workflow(
        webhook_url="https://n8n.../webhook/...",
        data={"league_id": league_id}
    )
 
    return {
        "league": league,
        "workflow_execution": result
    }

2. Atlassian MCP (Jira & Confluence)

Architecture:

┌─────────────────────────────────────────────────┐
│          Atlassian MCP Server                    │
│          (Docker Container)                      │
│                                                  │
│  ┌──────────────────────────────────────────┐  │
│  │           Jira Tools                      │  │
│  │  • Project Management                     │  │
│  │  • Issue CRUD Operations                 │  │
│  │  • Search & Filtering                    │  │
│  │  • Transitions & Workflows               │  │
│  └──────────────────────────────────────────┘  │
│                                                  │
│  ┌──────────────────────────────────────────┐  │
│  │        Confluence Tools                   │  │
│  │  • Page Operations                        │  │
│  │  • Content Search                         │  │
│  │  • Documentation Management              │  │
│  └──────────────────────────────────────────┘  │
└─────────────────┬───────────────────────────────┘
                  │
      ┌───────────▼──────────────┐
      │  Atlassian Cloud          │
      │  altsportsdata.atlassian  │
      └──────────────────────────┘

Key Tools:

// Jira Operations
mcp__atlassian__jira_get_all_projects()
mcp__atlassian__jira_create_issue({
  project: "ASD",
  summary: "New league: PWHL",
  description: "Contract generated",
  issueType: "Task"
})
mcp__atlassian__jira_update_issue({ issueId, fields })
 
// Confluence Operations
mcp__atlassian__confluence_search({ query: "ice hockey" })
mcp__atlassian__confluence_get_page({ pageId })

Use Case: Automated League Onboarding:

// 1. Create Jira ticket
const issue = await mcp__atlassian__jira_create_issue({
  project: "ASD",
  summary: `New League Onboarding: ${leagueName}`,
  description: contractDetails,
  labels: ["league-onboarding", tier]
})
 
// 2. Create Confluence documentation
await mcp__atlassian__confluence_create_page({
  title: `${leagueName} Partnership Details`,
  content: partnershipDoc,
  parentId: "league-partnerships"
})
 
// 3. Trigger n8n workflow for data population
await mcp__n8n-mcp__n8n_trigger_webhook_workflow({
  webhookUrl: "https://n8n.../webhook/league-onboard",
  data: {
    league_name: leagueName,
    jira_ticket: issue.key,
    contract_id: contractId
  }
})

3. Custom MCP Servers

League Discovery Cross-Comparison

Purpose: Discover and compare leagues across multiple data sources

Implementation:

# apps/backend/mcp_servers/servers/league_discovery_cross_comparison_mcp.py
 
from fastmcp import FastMCP
from supabase import create_client
 
mcp = FastMCP("league-discovery")
 
@mcp.tool()
async def discover_leagues(
    sport: str,
    region: str | None = None,
    tier: str | None = None
) -> list[dict]:
    """Discover leagues matching criteria."""
    supabase = create_client(SUPABASE_URL, SUPABASE_KEY)
 
    query = supabase.table("leagues").select("*")
 
    if sport:
        query = query.eq("sport", sport)
    if region:
        query = query.eq("region", region)
    if tier:
        query = query.eq("tier", tier)
 
    result = await query.execute()
    return result.data
 
@mcp.tool()
async def compare_leagues(
    league_ids: list[str]
) -> dict:
    """Cross-compare multiple leagues."""
    leagues = await fetch_leagues(league_ids)
 
    return {
        "leagues": leagues,
        "comparison": {
            "avg_teams": calculate_avg(leagues, "num_teams"),
            "avg_games": calculate_avg(leagues, "games_per_season"),
            "fan_base_total": sum_field(leagues, "fan_base"),
            "market_opportunity": analyze_market(leagues)
        }
    }

Configuration:

{
  "league-discovery-cross-comparison": {
    "command": "python",
    "args": [
      "-m",
      "apps.backend.mcp_servers.servers.league_discovery_cross_comparison_mcp"
    ],
    "env": {
      "SUPABASE_URL": "https://vljfrdsqtmdujhoxwtig.supabase.co",
      "SUPABASE_ANON_KEY": "your-key"
    }
  }
}

MCP Development Patterns

Pattern 1: FastMCP Python Server

Template:

from fastmcp import FastMCP
from pydantic import BaseModel
 
# Initialize MCP server
mcp = FastMCP("my-server")
 
# Define data models
class LeagueData(BaseModel):
    name: str
    sport: str
    tier: int
    teams: int
 
# Define tools
@mcp.tool()
async def process_league(league_id: str) -> LeagueData:
    """Process league data and return structured output."""
    # Implementation
    return LeagueData(...)
 
# Define resources
@mcp.resource("league://{league_id}")
async def get_league_resource(league_id: str) -> str:
    """Get league data as MCP resource."""
    league = await fetch_league(league_id)
    return league.model_dump_json()
 
# Define prompts
@mcp.prompt()
async def analyze_league_prompt(league_name: str) -> str:
    """Generate prompt for league analysis."""
    return f"Analyze the following league: {league_name}"
 
# Run server
if __name__ == "__main__":
    mcp.run()

Pattern 2: TypeScript MCP Server

Template:

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
 
// Initialize server
const server = new Server({
  name: "my-mcp-server",
  version: "1.0.0"
}, {
  capabilities: {
    tools: {},
    resources: {},
    prompts: {}
  }
});
 
// Define tools
server.setRequestHandler("tools/list", async () => ({
  tools: [
    {
      name: "process_league",
      description: "Process league data",
      inputSchema: {
        type: "object",
        properties: {
          league_id: { type: "string" }
        },
        required: ["league_id"]
      }
    }
  ]
}));
 
server.setRequestHandler("tools/call", async (request) => {
  if (request.params.name === "process_league") {
    const result = await processLeague(request.params.arguments.league_id);
    return { content: [{ type: "text", text: JSON.stringify(result) }] };
  }
});
 
// Run server
const transport = new StdioServerTransport();
await server.connect(transport);

Pattern 3: Docker-Based MCP Server

Dockerfile:

FROM python:3.11-slim
 
WORKDIR /app
 
# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
# Copy MCP server
COPY mcp_server.py .
 
# Run with stdio transport
CMD ["python", "mcp_server.py"]

Docker Compose:

version: '3.8'
services:
  mcp-server:
    build: .
    stdin_open: true
    tty: true
    environment:
      - DATABASE_URL=${DATABASE_URL}
      - API_KEY=${API_KEY}
    volumes:
      - ./output:/app/output

Integration Patterns

Pattern 1: n8n Workflow Trigger

Use Case: Trigger n8n workflow from MCP tool

@mcp.tool()
async def onboard_league(league_data: dict) -> dict:
    """Onboard league and trigger n8n workflow."""
 
    # 1. Validate league data
    validated = validate_league_data(league_data)
 
    # 2. Store in Supabase
    league_id = await store_league(validated)
 
    # 3. Trigger n8n workflow
    workflow_result = await http_client.post(
        "https://n8n.../webhook/league-onboard",
        json={
            "league_id": league_id,
            "league_name": validated["name"],
            "tier": validated["tier"]
        }
    )
 
    return {
        "league_id": league_id,
        "workflow_execution_id": workflow_result["execution_id"],
        "status": "onboarding_started"
    }

Pattern 2: Multi-Database Coordination

Use Case: Coordinate data across Supabase, Neo4j, and Firebase

@mcp.tool()
async def sync_league_data(league_id: str) -> dict:
    """Sync league data across all databases."""
 
    # 1. Fetch from Supabase (source of truth)
    league = await supabase.table("leagues").select("*").eq("id", league_id).single()
 
    # 2. Update Neo4j graph
    await neo4j_session.run("""
        MERGE (l:League {id: $id})
        SET l.name = $name, l.sport = $sport
    """, id=league_id, name=league["name"], sport=league["sport"])
 
    # 3. Sync to Firebase for real-time updates
    await firebase_db.reference(f"leagues/{league_id}").set({
        "name": league["name"],
        "updated_at": datetime.now().isoformat()
    })
 
    # 4. Update vector embeddings
    await embed_league_data(league)
 
    return {
        "league_id": league_id,
        "databases_synced": ["supabase", "neo4j", "firebase", "faiss"],
        "status": "success"
    }

Pattern 3: RAG Integration

Use Case: Semantic search with FAISS/ChromaDB

@mcp.tool()
async def semantic_league_search(query: str, limit: int = 10) -> list[dict]:
    """Search leagues using semantic similarity."""
 
    # 1. Generate query embedding
    query_embedding = await generate_embedding(query)
 
    # 2. Search FAISS index
    distances, indices = faiss_index.search(
        query_embedding.reshape(1, -1),
        limit
    )
 
    # 3. Fetch full league data
    league_ids = [league_id_map[idx] for idx in indices[0]]
    leagues = await fetch_leagues(league_ids)
 
    # 4. Enrich with similarity scores
    return [
        {
            **league,
            "similarity_score": float(1 / (1 + distances[0][i]))
        }
        for i, league in enumerate(leagues)
    ]

Best Practices

1. Tool Design

Clear Naming:

# Good
@mcp.tool()
async def search_leagues_by_sport(sport: str) -> list[dict]:
    ...
 
# Bad
@mcp.tool()
async def search(query: str) -> list:
    ...

Structured Output:

from pydantic import BaseModel
 
class LeagueSearchResult(BaseModel):
    leagues: list[dict]
    total_count: int
    query_metadata: dict
 
@mcp.tool()
async def search_leagues(sport: str) -> LeagueSearchResult:
    ...

2. Error Handling

from fastmcp import FastMCP, MCPError
 
@mcp.tool()
async def process_league(league_id: str) -> dict:
    try:
        league = await fetch_league(league_id)
        if not league:
            raise MCPError("LEAGUE_NOT_FOUND", f"League {league_id} not found")
 
        result = await process(league)
        return result
 
    except ValidationError as e:
        raise MCPError("VALIDATION_ERROR", str(e))
    except DatabaseError as e:
        raise MCPError("DATABASE_ERROR", "Failed to access database")

3. Resource Management

@mcp.resource("league://{league_id}/stats")
async def get_league_stats(league_id: str) -> str:
    """Get league statistics as JSON resource."""
    stats = await calculate_stats(league_id)
    return json.dumps(stats, indent=2)

4. Prompt Templates

@mcp.prompt()
async def league_analysis_prompt(
    league_name: str,
    include_financials: bool = False
) -> str:
    """Generate comprehensive league analysis prompt."""
 
    base_prompt = f"Analyze {league_name} focusing on:"
    sections = [
        "- Market size and demographics",
        "- Competitive landscape",
        "- Partnership opportunities"
    ]
 
    if include_financials:
        sections.append("- Revenue potential and pricing")
 
    return base_prompt + "\n" + "\n".join(sections)

Testing MCP Servers

Unit Testing

import pytest
from fastmcp.testing import MCPTestClient
 
@pytest.fixture
async def mcp_client():
    from my_mcp_server import mcp
    async with MCPTestClient(mcp) as client:
        yield client
 
async def test_search_leagues(mcp_client):
    result = await mcp_client.call_tool(
        "search_leagues_by_sport",
        sport="ice hockey"
    )
 
    assert len(result["leagues"]) > 0
    assert result["query_metadata"]["sport"] == "ice hockey"

Integration Testing

async def test_n8n_workflow_integration():
    # 1. Call MCP tool
    result = await mcp_client.call_tool(
        "onboard_league",
        league_data=test_league_data
    )
 
    # 2. Verify workflow triggered
    execution = await n8n_client.get_execution(
        result["workflow_execution_id"]
    )
 
    assert execution["status"] == "success"
 
    # 3. Verify database updated
    league = await supabase.table("leagues").select("*").eq(
        "id", result["league_id"]
    ).single()
 
    assert league["status"] == "onboarded"

Deployment

Docker Deployment

# Build MCP server
docker build -t altsports-mcp-server .
 
# Run locally
docker run -it --rm \
  -e DATABASE_URL=${DATABASE_URL} \
  altsports-mcp-server
 
# Deploy to Cloud Run
gcloud run deploy altsports-mcp-server \
  --image gcr.io/project-id/altsports-mcp-server \
  --platform managed \
  --region us-central1 \
  --allow-unauthenticated

Configuration Management

// .cursor/mcp.json
{
  "mcpServers": {
    "my-mcp-server": {
      "command": "docker",
      "args": [
        "run", "-i", "--rm",
        "-e", "DATABASE_URL",
        "altsports-mcp-server"
      ],
      "env": {
        "DATABASE_URL": "${DATABASE_URL}",
        "API_KEY": "${API_KEY}"
      }
    }
  }
}

Additional Resources

Support

For MCP server development assistance:

Review existing MCP servers in apps/backend/mcp_servers/servers/
Consult MCP development agents: @mcp-engineer, @mcp-developer
Test with n8n integration: /prime-mcp
Deploy with: /deploy:google.cloud-deployment

Development Documentation 📘 Complete API Reference

MCP Server Architecture

Overview

What is MCP?

MCP in AltSportsLeagues.ai

Architecture Layers

Layer 1: MCP Server Infrastructure

Layer 2: Transport Mechanisms

Layer 3: Tool Interfaces

MCP Server Implementations

1. n8n-mcp (Workflow Automation)

2. Atlassian MCP (Jira & Confluence)

3. Custom MCP Servers

League Discovery Cross-Comparison

MCP Development Patterns

Pattern 1: FastMCP Python Server

Pattern 2: TypeScript MCP Server

Pattern 3: Docker-Based MCP Server

Integration Patterns

Pattern 1: n8n Workflow Trigger

Pattern 2: Multi-Database Coordination

Pattern 3: RAG Integration

Best Practices

1. Tool Design

2. Error Handling

3. Resource Management

4. Prompt Templates

Testing MCP Servers

Unit Testing

Integration Testing

Deployment

Docker Deployment

Configuration Management

Additional Resources

Support

Platform

Documentation

Community

Support