Model Context Protocol (MCP) in Production: The Definitive Architectural Guide

from mcp.server.fastmcp import FastMCP
import asyncpg

mcp = FastMCP("Technofreaks-Cloud-Telemetry")

@mcp.tool()
async def query_cluster_metrics(cluster_id: str, metric_name: str) -> dict:
    """Query real-time cluster health and latency from production database."""
    conn = await asyncpg.connect("postgresql://readonly:secret@db.technofreaks.internal/telemetry")
    row = await conn.fetchrow(
        "SELECT p99_latency_ms, error_rate, active_nodes FROM cluster_health WHERE cluster_id = $1 ORDER BY timestamp DESC LIMIT 1",
        cluster_id
    )
    await conn.close()
    return dict(row) if row else {"error": "Cluster not found"}

if __name__ == "__main__":
    mcp.run(transport="sse")