Model Context Protocol (MCP)

import asyncio
import json
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import psycopg2

# Initialize the MCP server
server = Server("customer-db")

# Register a tool Claude can call
@server.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="query_customers",
            description="""Search customer records by various criteria.
            Use this when you need to find customers, check order history,
            or analyze purchase patterns. Returns structured customer data.""",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "SQL SELECT query (read-only)"
                    },
                    "limit": {
                        "type": "integer",
                        "description": "Max rows to return (default: 100, max: 1000)",
                        "default": 100
                    }
                },
                "required": ["query"]
            }
        )
    ]

# Handle tool calls from Claude
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "query_customers":
        try:
            # Safety: only allow SELECT queries
            query = arguments["query"].strip()
            if not query.upper().startswith("SELECT"):
                return [types.TextContent(
                    type="text",
                    text='{"error": "Only SELECT queries are permitted"}'
                )]
            
            conn = psycopg2.connect("postgresql://localhost/customers")
            cursor = conn.cursor()
            limit = min(arguments.get("limit", 100), 1000)
            cursor.execute(query + f" LIMIT {limit}")
            
            columns = [desc[0] for desc in cursor.description]
            rows = [dict(zip(columns, row)) for row in cursor.fetchall()]
            
            return [types.TextContent(
                type="text",
                text=json.dumps({"rows": rows, "count": len(rows)})
            )]
        except Exception as e:
            # Structured error — Claude can understand and retry
            return [types.TextContent(
                type="text",
                text=json.dumps({
                    "error": str(e),
                    "error_type": type(e).__name__,
                    "suggestion": "Check SQL syntax or table names"
                })
            )]

# Start the server
if __name__ == "__main__":
    asyncio.run(stdio_server(server))

{
  "mcpServers": {
    "customer-db": {
      "command": "python",
      "args": ["/path/to/mcp_server.py"],
      "env": {
        "DATABASE_URL": "postgresql://localhost/customers"
      }
    },
    "github": {
      "command": "npx",
      "args": ["@anthropic/mcp-server-github"],
      "env": {
        "GITHUB_TOKEN": "ghp_..."
      }
    }
  }
}

def make_error(error_type: str, message: str, retryable: bool, suggestion: str = ""):
    """
    Create a structured error response that Claude can act on intelligently.
    retryable=True: Claude can try a different approach automatically
    retryable=False: Human intervention needed
    """
    return json.dumps({
        "error": True,
        "error_type": error_type,
        "message": message,
        "retryable": retryable,
        "suggestion": suggestion
    })

# Usage examples:

# Claude can fix this by correcting the query
make_error("INVALID_SQL", "Column 'user_id' doesn't exist", 
           retryable=True,
           suggestion="Available columns: id, email, name, created_at")

# Claude cannot fix this — needs human
make_error("AUTH_FAILED", "Database credentials invalid",
           retryable=False,
           suggestion="Admin must update DATABASE_URL in server config")

# Claude can retry with smaller scope
make_error("TIMEOUT", "Query exceeded 30s limit",
           retryable=True,
           suggestion="Add a date filter to reduce result set size")