Building a Distributed Enterprise MCP Gateway in Python: Remote SSE, Bearer Authentication, Dynamic Tool Discovery, and LangGraph Swarms

"""
Enterprise Remote FastMCP Server Implementation
TechJunkGigs Production Blueprint - Python 3.11+
"""

import os
import sys
import logging
from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field
from mcp.server.fastmcp import FastMCP

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("mcp_gateway")

# 1. Initialize FastMCP Server with Production Metadata
mcp = FastMCP(
 name="Enterprise Core MCP Gateway",
 dependencies=["pydantic", "psycopg", "uvicorn", "starlette"]
)

# ---------------------------------------------------------------------------
# Strict Pydantic Data Models (Type Safety for LLM Tool Invocation)
# ---------------------------------------------------------------------------

class DatabaseQueryRequest(BaseModel):
 query: str = Field(
 ..., 
 description="A read-only SQL query starting strictly with SELECT. Multi-statement or DDL queries are rejected."
 )
 limit: int = Field(
 default=50, 
 ge=1, 
 le=500, 
 description="Maximum number of rows to return (bounded between 1 and 500)."
 )

class TableSchemaResponse(BaseModel):
 table_name: str
 column_count: int
 columns: List[Dict[str, str]]
 primary_keys: List[str]

# ---------------------------------------------------------------------------
# Production MCP Tools
# ---------------------------------------------------------------------------

@mcp.tool()
def inspect_system_telemetry() -> Dict[str, Any]:
 """Retrieve host architecture, memory pressure, and cluster health metrics."""
 return {
 "status": "healthy",
 "cluster_region": os.getenv("CLUSTER_REGION", "us-east-1"),
 "node_id": os.getenv("K8S_NODE_NAME", "worker-node-04"),
 "active_connections": 12,
 "runtime": "Python 3.11.9"
 }

@mcp.tool()
def inspect_table_schema(table_name: str) -> TableSchemaResponse:
 """Retrieve column definitions, data types, and primary key constraints for a database table."""
 logger.info(f"Inspecting table schema for: {table_name}")
 # In production, query information_schema.columns safely
 return TableSchemaResponse(
 table_name=table_name,
 column_count=3,
 columns=[
 {"name": "user_id", "type": "UUID", "nullable": "NO"},
 {"name": "email", "type": "VARCHAR(255)", "nullable": "NO"},
 {"name": "created_at", "type": "TIMESTAMPTZ", "nullable": "NO"}
 ],
 primary_keys=["user_id"]
 )

@mcp.tool()
def execute_readonly_sql(request: DatabaseQueryRequest) -> List[Dict[str, Any]]:
 """Execute a parameterized read-only SELECT query against the enterprise analytical data warehouse."""
 cleaned_query = request.query.strip()

 # Deterministic Ingress Query Firewall
 if not cleaned_query.upper().startswith("SELECT"):
 raise ValueError("Security Policy Violation: Only SELECT queries are permitted on this gateway.")

 dangerous_keywords = ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "TRUNCATE", "GRANT"]
 for keyword in dangerous_keywords:
 if f" {keyword} " in f" {cleaned_query.upper()} ":
 raise ValueError(f"Security Policy Violation: Mutation keyword '{keyword}' is forbidden.")

 logger.info(f"Executing validated query with limit {request.limit}: {cleaned_query[:60]}...")

 # Simulated parameterized result rows
 return [
 {"user_id": "8f1a-4c22", "email": "dev-ops@internal.net", "status": "active"},
 {"user_id": "3b2e-99f1", "email": "billing-admin@internal.net", "status": "verified"}
 ]

# ---------------------------------------------------------------------------
# Production MCP Resources (Dynamic Context Streaming)
# ---------------------------------------------------------------------------

@mcp.resource("schema://enterprise/catalog")
def get_database_catalog() -> str:
 """Live schema catalog resource for contextual retrieval and dynamic prompt caching."""
 return (
 "Enterprise Data Warehouse Catalog:\n"
 "- users: user_id (UUID), email (VARCHAR), plan_tier (VARCHAR), created_at (TIMESTAMPTZ)\n"
 "- transactions: tx_id (UUID), user_id (UUID), amount_usd (NUMERIC), status (VARCHAR)\n"
 "- audit_logs: event_id (BIGINT), actor_id (UUID), action (VARCHAR), timestamp (TIMESTAMPTZ)"
 )

# ---------------------------------------------------------------------------
# Production MCP Prompts (Server-Managed Prompt Templates)
# ---------------------------------------------------------------------------

@mcp.prompt()
def incident_root_cause_prompt(service_name: str, error_signature: str) -> str:
 """Generate a standardized incident investigation template for debugging swarms."""
 return (
 f"You are an enterprise site reliability engineer investigating an outage in service '{service_name}'.\n"
 f"Error Signature: {error_signature}\n\n"
 "Instructions:\n"
 "1. Inspect active table schemas and system telemetry using available MCP tools.\n"
 "2. Formulate targeted, read-only SQL queries to identify recent schema mutations or transaction spikes.\n"
 "3. Provide a deterministic root-cause report containing issue summary, impacted records, and mitigation steps."
 )

# ---------------------------------------------------------------------------
# Server Runtime Entrypoint
# ---------------------------------------------------------------------------

if __name__ == "__main__":
 transport = sys.argv[1].lower() if len(sys.argv) > 1 else "sse"
 port = int(os.getenv("PORT", 8000))

 if transport == "sse":
 logger.info(f"[*] Starting Distributed FastMCP Server over SSE on http://0.0.0.0:{port}/sse")
 mcp.run(transport="sse", port=port)
 else:
 logger.info("[*] Starting Local FastMCP Server over stdio")
 mcp.run(transport="stdio")

"""
Enterprise Ingress Authentication & RBAC Middleware for FastMCP
TechJunkGigs Production Blueprint - Python 3.11+
"""

import hmac
import hashlib
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse

class MCPBearerAuthMiddleware(BaseHTTPMiddleware):
 """Enforces cryptographic token validation on all inbound SSE and message endpoints."""

 def __init__(self, app, expected_api_key: str, required_scope: str = "agent:tools"):
 super().__init__(app)
 self.expected_api_key = expected_api_key
 self.required_scope = required_scope

 async def dispatch(self, request: Request, call_next):
 # Allow internal health checks without auth
 if request.url.path in ["/health", "/metrics"]:
 return await call_next(request)

 # Extract Authorization Header
 auth_header = request.headers.get("Authorization")
 if not auth_header or not auth_header.startswith("Bearer "):
 return JSONResponse(
 status_code=401,
 content={
 "jsonrpc": "2.0",
 "error": {
 "code": -32000,
 "message": "Unauthorized: Missing or malformed Bearer token header."
 },
 "id": None
 }
 )

 token = auth_header.split(" ", 1)[1].strip()

 # Constant-time comparison prevents timing attack vectors
 if not hmac.compare_digest(token, self.expected_api_key):
 return JSONResponse(
 status_code=403,
 content={
 "jsonrpc": "2.0",
 "error": {
 "code": -32001,
 "message": "Forbidden: Invalid credentials or insufficient scope permissions."
 },
 "id": None
 }
 )

 # Token validated: attach identity to request state and proceed
 request.state.caller_identity = "agent_service_account_production"
 return await call_next(request)

"""
Dynamic MCP Client and LangGraph Swarm Binding
TechJunkGigs Production Blueprint - Python 3.11+
"""

import asyncio
from typing import List, Dict, Any
from mcp import ClientSession
from mcp.client.sse import sse_client
from langchain_core.tools import StructuredTool
from pydantic import create_model

async def fetch_dynamic_mcp_tools(server_url: str, bearer_token: str) -> List[StructuredTool]:
 """Connects to a remote MCP gateway via SSE, lists tools, and returns dynamic LangChain tools."""
 headers = {"Authorization": f"Bearer {bearer_token}"}
 langchain_tools = []

 async with sse_client(server_url, headers=headers) as (read_stream, write_stream):
 async with ClientSession(read_stream, write_stream) as session:
 # 1. Initialize Protocol Handshake
 await session.initialize()

 # 2. Query published tools dynamically
 tool_catalog = await session.list_tools()

 for tool_def in tool_catalog.tools:
 tool_name = tool_def.name
 tool_desc = tool_def.description or "No description provided."

 # 3. Create closure capturing tool execution over remote session
 def make_executor(t_name: str):
 async def _execute(**kwargs):
 # Re-open session or multiplex existing connection
 async with sse_client(server_url, headers=headers) as (r, w):
 async with ClientSession(r, w) as sub_session:
 await sub_session.initialize()
 result = await sub_session.call_tool(t_name, arguments=kwargs)
 return "\n".join([item.text for item in result.content if item.type == "text"])
 return _execute

 # 4. Synthesize LangChain StructuredTool dynamically
 compiled_tool = StructuredTool.from_function(
 coroutine=make_executor(tool_name),
 name=tool_name,
 description=tool_desc
 )
 langchain_tools.append(compiled_tool)

 return langchain_tools

"""
Autonomous Multi-Agent LangGraph Swarm powered by Remote MCP Tools
TechJunkGigs Production Blueprint - Python 3.11+
"""

from typing import Annotated, TypedDict, Sequence
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI

# 1. Define Agent State
class AgentState(TypedDict):
 messages: Sequence[BaseMessage]

# 2. Build LangGraph Swarm with Remote MCP Tools
def build_mcp_agent_graph(dynamic_tools: list):
 # Bind dynamically discovered tools to model
 llm = ChatOpenAI(model="gpt-4o", temperature=0.0).bind_tools(dynamic_tools)

 def agent_decision_node(state: AgentState):
 messages = state["messages"]
 response = llm.invoke(messages)
 return {"messages": [response]}

 def should_continue(state: AgentState):
 last_message = state["messages"][-1]
 if hasattr(last_message, "tool_calls") and last_message.tool_calls:
 return "execute_tools"
 return END

 # 3. Assemble Workflow State Machine
 workflow = StateGraph(AgentState)

 workflow.add_node("agent", agent_decision_node)
 workflow.add_node("execute_tools", ToolNode(dynamic_tools))

 workflow.set_entry_point("agent")
 workflow.add_conditional_edges("agent", should_continue)
 workflow.add_edge("execute_tools", "agent")

 return workflow.compile()

"""
Resilience & Circuit Breaker Decorator for Distributed MCP Invocations
TechJunkGigs Production Blueprint - Python 3.11+
"""

import time
import asyncio
from enum import Enum
from functools import wraps

class CircuitState(Enum):
 CLOSED = "CLOSED" # Normal operation
 OPEN = "OPEN" # Failing: reject calls immediately
 HALF_OPEN = "HALF_OPEN" # Testing service recovery

class MCPCircuitBreaker:
 def __init__(self, failure_threshold: int = 3, recovery_timeout_seconds: float = 30.0):
 self.failure_threshold = failure_threshold
 self.recovery_timeout = recovery_timeout_seconds
 self.failure_count = 0
 self.last_failure_time = 0.0
 self.state = CircuitState.CLOSED

 def record_failure(self):
 self.failure_count += 1
 self.last_failure_time = time.time()
 if self.failure_count >= self.failure_threshold:
 self.state = CircuitState.OPEN

 def record_success(self):
 self.failure_count = 0
 self.state = CircuitState.CLOSED

 def can_execute(self) -> bool:
 if self.state == CircuitState.CLOSED:
 return True
 if self.state == CircuitState.OPEN:
 if time.time() - self.last_failure_time > self.recovery_timeout:
 self.state = CircuitState.HALF_OPEN
 return True
 return False
 return True # HALF_OPEN allows a canary request