Posted inAI

Building a Custom MCP Server with Python: Standardize Tool and Data Integrations for Claude Desktop and AI Agents

AI tutorial - IT technology blog
AI tutorial - IT technology blog

Context & Why

At 2:15 AM on a Tuesday, our PagerDuty alerts triggered simultaneously across two availability zones. Our authentication cluster had ground to a halt. My teammate was juggling four separate terminal tabs and pasting unformatted JSON dumps into Claude to figure out which microservice threw the first 500 error. Each internal service relied on its own bespoke wrapper script, custom OpenAI schema definition, and brittle glue code. Any minor API change broke the triage workflow completely.

Building one-off API connectors for every AI client does not scale. Anthropic introduced the Model Context Protocol (MCP) to solve this architectural mess. Think of MCP as the Language Server Protocol (LSP) for LLMs. Instead of re-implementing integrations across multiple tools, you implement a standardized interface once. Clients discover capabilities, query contextual resources, and execute actions over standard input/output (stdio) or Server-Sent Events (SSE).

Standardizing on MCP cut our team's mean time to recovery (MTTR) during log triage from 24 minutes down to under 6 minutes. Once your services speak MCP, any compatible host—whether it is Claude Desktop, Cursor, or a custom autonomous CLI agent—can safely query operational metrics, inspect database states, and trigger controlled remediation steps.

Installation

Always isolate server dependencies to prevent package collisions with your system tools. We will use Python 3.10 or higher.

1. Setting Up the Virtual Environment

mkdir mcp-incident-server
cd mcp-incident-server
python3 -m venv .venv
source .venv/bin/activate

2. Installing the Official MCP SDK

Install Anthropic's official Python SDK along with httpx for asynchronous HTTP calls and pydantic for strict schema validation:

pip install --upgrade "mcp[cli]" httpx pydantic

3. Writing the Python MCP Server

Create a file named server.py. We will use the SDK's high-level FastMCP module to expose two operational tools and one real-time metric resource:

import sys
from datetime import datetime, timezone
from typing import Any, Dict, List
from mcp.server.fastmcp import FastMCP

# Initialize the FastMCP server instance
mcp = FastMCP("production-ops-server")

# Mock operational datastore
INCIDENT_LOGS = [
    {
        "id": "INC-101",
        "service": "auth-api",
        "level": "CRITICAL",
        "message": "PostgreSQL connection pool exhausted (500/500 connections active)",
        "timestamp": "2026-10-03T01:58:12Z",
    },
    {
        "id": "INC-102",
        "service": "payment-gw",
        "level": "WARNING",
        "message": "Stripe webhook gateway latency elevated to 1420ms (p99 threshold: 400ms)",
        "timestamp": "2026-10-03T02:05:44Z",
    },
]

@mcp.tool()
def fetch_incident_logs(service: str = "", severity: str = "") -> List[Dict[str, Any]]:
    """Query active incident logs filtered by service name or severity level."""
    results = INCIDENT_LOGS
    if service:
        results = [log for log in results if log["service"].lower() == service.lower()]
    if severity:
        results = [log for log in results if log["level"].lower() == severity.lower()]
    return results

@mcp.tool()
def restart_microservice_pod(service_name: str, force: bool = False) -> str:
    """Trigger a rolling restart for an unhealthy Kubernetes pod during an outage."""
    valid_services = ["auth-api", "payment-gw", "order-processor"]
    if service_name not in valid_services:
        return f"Error: Unknown service '{service_name}'. Allowed targets: {valid_services}"
    
    action = "Forced immediate termination" if force else "Graceful rolling restart"
    now = datetime.now(timezone.utc).isoformat()
    return f"[{now}] {action} initiated for target pod '{service_name}' (status: 202 Accepted)."

@mcp.resource("system://metrics/summary")
def get_system_metrics() -> str:
    """Provide active cluster resource telemetry as a structured context resource."""
    return "Cluster Health: DEGRADED | CPU: 78.4% | Memory: 84.1% | Active DB Connections: 492/500"

if __name__ == "__main__":
    # Listen on standard I/O for client requests
    mcp.run(transport="stdio")

Configuration

With our server implemented, we need to register it with our AI host client. Claude Desktop is the standard environment for local testing and day-to-day operations.

1. Configuring Claude Desktop

Open your platform's configuration file:

  • macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
  • Windows: %APPDATA%\Claude\claude_desktop_config.json
  • Linux: ~/.config/Claude/claude_desktop_config.json

Register your server under the mcpServers section. Always provide the absolute path to the Python binary inside your virtual environment, not the system Python:

{
  "mcpServers": {
    "ops-debugger": {
      "command": "/home/ubuntu/mcp-incident-server/.venv/bin/python",
      "args": [
        "/home/ubuntu/mcp-incident-server/server.py"
      ],
      "env": {
        "LOG_LEVEL": "INFO",
        "DB_TIMEOUT_MS": "5000"
      }
    }
  }
}

2. Security and I/O Isolation

Avoid embedding database passwords or API keys in your Python source files. Inject them via the env block or your operating system environment.

Never print debugging messages directly with print() in stdio mode. FastMCP uses stdout exclusively to serialize JSON-RPC messages. Writing raw text to stdout corrupts message frames and breaks the client connection. Route all internal diagnostics to sys.stderr instead.

Verification & Monitoring

Never deploy infrastructure tools blindly. Validate the server locally using the official test harness before connecting it to a live LLM.

1. Testing with MCP Inspector

Anthropic maintains MCP Inspector, an interactive web UI that simulates client-server handshakes, validates schemas, and captures raw JSON-RPC traffic. Run it with npx:

npx @modelcontextprotocol/inspector /home/ubuntu/mcp-incident-server/.venv/bin/python /home/ubuntu/mcp-incident-server/server.py

Open the provided local URL (typically http://localhost:5173). In the left sidebar, choose fetch_incident_logs, provide {"service": "auth-api"} under parameters, and click Run Tool. Confirm that the tool returns the INC-101 record with a clean 200-style response.

2. Live Client Verification

Fully quit and restart Claude Desktop. Locate the hammer icon at the bottom right of the message bar. You should see ops-debugger active, exposing two tools (fetch_incident_logs, restart_microservice_pod) and your metric resource.

Test the end-to-end loop with this prompt: "Check the active incident logs for auth-api and tell me if we need to restart the pod."

Claude will inspect the schema, ask for permission to run fetch_incident_logs, analyze the database connection bottleneck, and recommend a rolling restart.

3. Troubleshooting Common Runtime Errors

  • JSON-RPC Parse Errors: Stray print statements or unhandled tracebacks polluted stdout. Direct all custom logging to stderr via logging.basicConfig(stream=sys.stderr).
  • Subprocess Exit Code 1 / File Not Found: The binary path specified in claude_desktop_config.json is wrong. Run which python inside your activated .venv and paste that exact string into the config.
  • Client Timeouts: Long-running SQL queries or API calls block the worker loop. Wrap external I/O in asyncio.wait_for(..., timeout=5.0) to fail cleanly before the client disconnects.
Share:
Tags: