Model Context Protocol (MCP): Building Your First Production MCP Server with Python

┌──────────────────────────────────────────────┐
│                  MCP Client                  │
│       (Claude Desktop, Cursor, Custom)       │
└──────────────────────┬───────────────────────┘
                       │ JSON-RPC 2.0 (stdio / SSE)
                       ▼
┌──────────────────────────────────────────────┐
│                  MCP Server                  │
│  ├── Resources (Data/Files, URI templates)   │
│  ├── Prompts   (Pre-engineered templates)    │
│  └── Tools     (Executable functions)        │
└──────────────────────┬───────────────────────┘
                       │
                       ▼
┌──────────────────────────────────────────────┐
│      Underlying Infrastructure & APIs        │
│   (Postgres, Redis, GitHub API, File System) │
└──────────────────────────────────────────────┘

# server.py
import sqlite3
from typing import List, Dict, Any
from mcp.server.fastmcp import FastMCP

# Initialize FastMCP Server
mcp = FastMCP("SmartBuddy-Internal-Ops")

DB_PATH = "/data/internal_ops.db"

def get_db():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn

@mcp.resource("schema://database")
def get_database_schema() -> str:
    """Returns the SQL schema of the internal production database."""
    with get_db() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT sql FROM sqlite_master WHERE type='table';")
        tables = [row["sql"] for row in cursor.fetchall() if row["sql"]]
        return "\n\n".join(tables)

@mcp.tool()
def query_customer_orders(customer_id: str, limit: int = 10) -> List[Dict[str, Any]]:
    """Fetches recent orders for a specific customer ID from the database.
    
    Args:
        customer_id: The unique customer identifier (e.g., 'cust_98234').
        limit: Maximum number of order records to return (default 10).
    """
    with get_db() as conn:
        cursor = conn.cursor()
        cursor.execute(
            "SELECT order_id, amount, status, created_at FROM orders WHERE customer_id = ? ORDER BY created_at DESC LIMIT ?",
            (customer_id, limit)
        )
        return [dict(row) for row in cursor.fetchall()]

@mcp.tool()
def issue_refund(order_id: str, reason: str, amount_cents: int) -> Dict[str, Any]:
    """Issues a financial refund for an eligible order.
    
    Args:
        order_id: The order to refund.
        reason: Explanation for the customer support audit log.
        amount_cents: Refund amount in USD cents.
    """
    # Defensive input validation
    if amount_cents <= 0:
        raise ValueError("Refund amount must be strictly greater than zero.")

    with get_db() as conn:
        cursor = conn.cursor()
        cursor.execute(
            "UPDATE orders SET status = 'refunded', refund_reason = ? WHERE order_id = ?",
            (reason, order_id)
        )
        conn.commit()

    return {
        "status": "success",
        "order_id": order_id,
        "amount_refunded_cents": amount_cents,
        "timestamp": "2026-09-26T14:30:00Z"
    }

if __name__ == "__main__":
    # Runs the server using stdio transport
    mcp.run(transport="stdio")

{
  "mcpServers": {
    "internal-ops": {
      "command": "python",
      "args": [
        "G:\\My Drive\\2026 Personal Job Finder Agent\\mcp_servers\\server.py"
      ],
      "env": {
        "PYTHONUNBUFFERED": "1"
      }
    }
  }
}

# Launching with SSE transport in FastAPI / Uvicorn
if __name__ == "__main__":
    mcp.run(transport="sse", host="0.0.0.0", port=8000)