Building Production-Grade Model Context Protocol (MCP) Servers with FastMCP in Python

"""
Minimal Working FastMCP Server Example
Demonstrating clean Pydantic schema validation, LLM-optimized docstrings,
and typed tool execution for Claude Desktop and Cursor IDE.
"""
from typing import Optional
from pydantic import BaseModel, Field
from mcp.server.fastmcp import FastMCP

# 1. Initialize FastMCP Server with descriptive application name
mcp = FastMCP(
    name="customer-intelligence-server",
    instructions="Production MCP server exposing enterprise customer search and telemetry lookups."
)

# 2. Define strictly typed Pydantic input schema for LLM function calling
class CustomerSearchInput(BaseModel):
    account_id: str = Field(
        ...,
        pattern=r"^acc_[a-zA-Z0-9]{8,16}$",
        description="Unique customer account identifier prefixed with 'acc_' (e.g., 'acc_9842aefc')."
    )
    include_billing: bool = Field(
        default=False,
        description="Whether to fetch associated active subscriptions, ARR, and payment status."
    )
    max_records: int = Field(
        default=10,
        ge=1,
        le=50,
        description="Maximum activity log records to return for token efficiency."
    )

# 3. Expose Tool with clean decorator and LLM-targeted docstrings
@mcp.tool()
async def search_customer_profile(params: CustomerSearchInput) -> dict:
    """
    Search customer account records and return operational metadata.
    
    Use this tool whenever an operator or workflow needs to inspect customer
    subscription tiers, organization domain, health scores, or recent audit events.
    Never invent fake account IDs; validate format before calling.
    """
    # Simulated backend service lookup with typed response
    return {
        "status": "success",
        "account_id": params.account_id,
        "organization": "Acme Global Enterprise",
        "tier": "enterprise",
        "health_score": 98.4,
        "subscription_active": True,
        "recent_audit_count": min(params.max_records, 4)
    }

# 4. Expose Resource for ambient context injection
@mcp.resource("schema://customer-tiers")
def get_customer_tiers_spec() -> str:
    """Expose standard enterprise customer tier definitions and SLA policies."""
    return "Tiers: Starter (99.5% SLA), Professional (99.9% SLA), Enterprise (99.99% SLA, 24/7 dedicated lead)."

# 5. Local stdio transport entrypoint
if __name__ == "__main__":
    mcp.run(transport="stdio")

"""
Enterprise Database Integration: PostgreSQL with Asyncpg Connection Pooling
Features non-blocking connection acquisition, strict parameterized SQL,
statement timeouts, read-replica routing, and connection leak prevention.
"""
import os
import sys
import asyncio
from typing import Optional, List, Dict, Any
from contextlib import asynccontextmanager
from pydantic import BaseModel, Field
import asyncpg
from mcp.server.fastmcp import FastMCP

# 1. Global connection pool container (Singleton Lifecycle Pattern)
class DatabaseManager:
    _pool: Optional[asyncpg.Pool] = None

    @classmethod
    async def get_pool(cls) -> asyncpg.Pool:
        if cls._pool is None:
            db_url = os.environ.get("DATABASE_READ_REPLICA_URL")
            if not db_url:
                raise ValueError("DATABASE_READ_REPLICA_URL environment variable is not configured.")
            
            # Initialize async connection pool with tight enterprise bounds
            cls._pool = await asyncpg.create_pool(
                dsn=db_url,
                min_size=2,
                max_size=10,
                max_inactive_connection_lifetime=300.0,
                command_timeout=5.0,  # 5-second socket timeout
                server_settings={
                    "application_name": "fastmcp_enterprise_server",
                    "statement_timeout": "3000ms"  # Prevent runaway table scans
                }
            )
            # Log to stderr only (never stdout to avoid corrupting stdio transport)
            print("PostgreSQL async connection pool initialized successfully.", file=sys.stderr)
        return cls._pool

    @classmethod
    async def close(cls):
        if cls._pool is not None:
            await cls._pool.close()
            cls._pool = None
            print("PostgreSQL async connection pool closed cleanly.", file=sys.stderr)

# 2. FastMCP Server Definition
mcp = FastMCP("enterprise-financial-warehouse")

# 3. Pydantic Filter Schema: Enforce parameterized constraints
class FinancialLedgerQuery(BaseModel):
    fiscal_year: int = Field(
        ...,
        ge=2020,
        le=2026,
        description="Four-digit fiscal year to filter transactions (2020-2026)."
    )
    department: str = Field(
        ...,
        description="Corporate department code: 'engineering', 'sales', 'infrastructure', or 'operations'."
    )
    min_amount_usd: float = Field(
        default=0.0,
        ge=0.0,
        description="Minimum expenditure threshold in USD."
    )
    limit: int = Field(
        default=25,
        ge=1,
        le=50,
        description="Number of ledger entries to retrieve (max 50 for token budget control)."
    )

# 4. Tool implementation with zero connection leakage and parameterized SQL
@mcp.tool()
async def query_financial_ledger(params: FinancialLedgerQuery) -> Dict[str, Any]:
    """
    Retrieve audited expenditure records from the enterprise financial warehouse.
    
    All queries run against an isolated read-replica. Parameters are validated
    through Pydantic and executed via strict positional placeholders ($1, $2, $3)
    to completely eliminate SQL injection vulnerabilities.
    """
    # Strict parameterized SQL query with forced LIMIT
    query = """
        SELECT 
            transaction_id,
            posting_date,
            vendor_name,
            department_code,
            amount_usd,
            approval_status
        FROM corporate_general_ledger
        WHERE fiscal_year = $1
          AND department_code = $2
          AND amount_usd >= $3
        ORDER BY amount_usd DESC
        LIMIT $4;
    """

    try:
        pool = await DatabaseManager.get_pool()
        # Acquire connection from async pool with context manager
        async with pool.acquire() as conn:
            # Execute parameterized query safely
            records = await conn.fetch(
                query,
                params.fiscal_year,
                params.department.lower(),
                params.min_amount_usd,
                params.limit
            )
            
            # Serialize asyncpg Records to clean JSON-serializable dictionaries
            results: List[Dict[str, Any]] = [
                {
                    "transaction_id": row["transaction_id"],
                    "posting_date": row["posting_date"].isoformat(),
                    "vendor_name": row["vendor_name"],
                    "department": row["department_code"],
                    "amount_usd": float(row["amount_usd"]),
                    "status": row["approval_status"]
                }
                for row in records
            ]

            return {
                "status": "success",
                "count": len(results),
                "filter_applied": {
                    "fiscal_year": params.fiscal_year,
                    "department": params.department,
                    "min_amount": params.min_amount_usd
                },
                "data": results
            }

    except asyncpg.PostgresError as pg_err:
        # Structured diagnostic response that LLM can interpret
        print(f"PostgreSQL Query Error: {pg_err}", file=sys.stderr)
        return {
            "status": "error",
            "error_type": "DATABASE_QUERY_FAILED",
            "message": "The analytical database rejected the query or exceeded the 3000ms timeout.",
            "suggestion": "Narrow the fiscal year or increase the min_amount_usd filter to reduce scan range."
        }
    except Exception as err:
        print(f"Unexpected Error: {err}", file=sys.stderr)
        return {
            "status": "error",
            "error_type": "INTERNAL_EXCEPTION",
            "message": str(err)
        }

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

"""
Production Error Boundary Pattern for FastMCP Tools
Wraps asynchronous tool functions to catch exceptions, prevent stdio crashes,
log diagnostic stack traces to stderr, and return LLM-actionable self-correcting JSON.
"""
import sys
import functools
import traceback
from typing import Callable, Any, Dict
from pydantic import ValidationError

def mcp_error_boundary(tool_name: str) -> Callable:
    """
    Decorator providing an unbreakable fault-tolerance envelope around an MCP tool.
    Guarantees that unhandled exceptions never crash the stdio or SSE transport stream.
    """
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        async def wrapper(*args: Any, **kwargs: Any) -> Dict[str, Any]:
            try:
                return await func(*args, **kwargs)
            except ValidationError as val_err:
                # LLM sent parameters violating Pydantic schema
                print(f"[{tool_name}] Pydantic validation failure: {val_err}", file=sys.stderr)
                return {
                    "status": "error",
                    "code": "SCHEMA_VALIDATION_ERROR",
                    "message": "The arguments provided do not conform to the required parameter schema.",
                    "details": [
                        {"field": err["loc"][-1], "issue": err["msg"]}
                        for err in val_err.errors()
                    ],
                    "actionable_fix": "Inspect the required fields and parameter types before re-invoking."
                }
            except TimeoutError:
                print(f"[{tool_name}] Execution exceeded timeout threshold.", file=sys.stderr)
                return {
                    "status": "error",
                    "code": "TIMEOUT_EXCEEDED",
                    "message": "The upstream query or API call exceeded the maximum allowable execution window.",
                    "actionable_fix": "Reduce query batch limits or broaden filter granularity to lower query complexity."
                }
            except Exception as unhandled_err:
                # Catch-all: Log full traceback to sys.stderr for engineering telemetry
                print(f"[{tool_name}] CRITICAL UNHANDLED ERROR:\n{traceback.format_exc()}", file=sys.stderr)
                return {
                    "status": "error",
                    "code": "INTERNAL_TOOL_EXCEPTION",
                    "message": f"Execution failed due to an unexpected backend condition: {str(unhandled_err)}",
                    "actionable_fix": "Retry with alternative parameters or inform the user that this resource is momentarily degraded."
                }
        return wrapper
    return decorator

{
  "status": "error",
  "code": "SCHEMA_VALIDATION_ERROR",
  "message": "The arguments provided do not conform to the required parameter schema.",
  "details": [{"field": "fiscal_year", "issue": "Input should be greater than or equal to 2020"}],
  "actionable_fix": "Inspect the required fields and parameter types before re-invoking."
}

{
  "mcpServers": {
    "enterprise-finance-server": {
      "command": "/Users/engineering/projects/enterprise-mcp/.venv/bin/python",
      "args": [
        "-u",
        "/Users/engineering/projects/enterprise-mcp/server.py"
      ],
      "env": {
        "PYTHONUNBUFFERED": "1",
        "DATABASE_READ_REPLICA_URL": "postgresql://mcp_readonly:Secr3tToken@db-replica.internal.company.com:5432/finance_warehouse",
        "STATEMENT_TIMEOUT_MS": "3000",
        "MAX_POOL_SIZE": "10",
        "LOG_LEVEL": "INFO"
      }
    }
  }
}