	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > Building an MCP Server in Python

Building an MCP Server in Python

Author: Venkata Sudhakar

The Model Context Protocol (MCP) is an open standard that lets AI assistants connect to external tools and data sources through a common interface. Instead of each AI application building custom integrations for every tool, MCP defines a universal protocol: an MCP server exposes tools and resources, and any MCP client (Claude Desktop, Cursor, LangChain, and others) can connect to it and use those tools without any custom glue code. Building an MCP server means your tools become immediately available to any AI that speaks MCP.

The Python mcp library (from Anthropic) makes building a server straightforward. You create a FastMCP server instance, decorate Python functions with @mcp.tool() to expose them as callable tools, and optionally decorate data-fetching functions with @mcp.resource() to expose structured data. The server communicates over stdio (for local tools) or SSE/HTTP (for remote servers). When an AI client calls your tool, MCP handles the JSON serialisation, argument validation, and error propagation automatically.

The below example builds a complete MCP server for a data migration assistant that exposes three tools: row count comparison, CDC lag check, and schema validation.

# pip install mcp
from mcp.server.fastmcp import FastMCP

# Create the MCP server - name appears in the client tool list
mcp = FastMCP("Migration Assistant")

@mcp.tool()
def compare_row_counts(table: str) -> dict:
    """Compare row counts between source and target database for a given table.
    Returns source count, target count, difference, and whether they match.
    """
    # In production: connect to real databases
    source_counts = {"customers": 125000, "orders": 890000, "products": 4500}
    target_counts = {"customers": 125000, "orders": 889997, "products": 4500}

source = source_counts.get(table, 0)
    target = target_counts.get(table, 0)
    diff = target - source

return {
        "table":    table,
        "source":   source,
        "target":   target,
        "diff":     diff,
        "match":    source == target,
        "status":   "OK" if source == target else "MISMATCH"
    }

@mcp.tool()
def check_cdc_lag(consumer_group: str) -> dict:
    """Check CDC replication lag for a Kafka consumer group.
    Returns lag in seconds and number of pending messages.
    """
    # In production: query Kafka consumer group API
    return {
        "consumer_group":  consumer_group,
        "lag_seconds":     2,
        "pending_messages": 150,
        "status":          "CATCHING_UP" if 150 > 0 else "SYNCED"
    }

@mcp.tool()
def validate_schema(table: str) -> dict:
    """Validate that a table schema matches between source and target.
    Returns list of column mismatches if any.
    """
    # Simulate schema comparison result
    known_issues = {
        "orders": [{"column": "amount", "source_type": "DECIMAL(10,2)", "target_type": "NUMERIC(10,2)", "compatible": True}]
    }
    issues = known_issues.get(table, [])
    return {
        "table":   table,
        "issues":  issues,
        "valid":   len([i for i in issues if not i["compatible"]]) == 0
    }

Adding a resource and running the server,

Configuring in Claude Desktop (claude_desktop_config.json),

{
  "mcpServers": {
    "migration-assistant": {
      "command": "python",
      "args": ["/path/to/migration_server.py"]
    }
  }
}

# After restarting Claude Desktop, the tools appear automatically:
# - compare_row_counts
# - check_cdc_lag
# - validate_schema
# - migration://status resource

# Claude can now call these tools in conversation:
# User: "Are we ready to cut over the orders table?"
# Claude: [calls compare_row_counts("orders")] -> 3 rows missing, not ready
# Claude: [calls check_cdc_lag("migration-cdc")] -> 2s lag, still catching up
# Claude: "The orders table has 3 missing rows and CDC is still catching up.
#          Wait for both to resolve before cutover."

MCP transforms your Python functions into universally usable AI tools with minimal code. The same server works with Claude Desktop, Cursor, any LangChain MCP client, and any other MCP-compatible host without modification. For production deployments, use transport="sse" with a proper HTTP server instead of stdio so the server runs as a persistent service rather than a subprocess. Add authentication by passing an auth handler to mcp.run() to control which clients can access your tools.

Send your comments, suggestions or queries regarding this site to [email protected].