Model Context Protocol (MCP) Explained — The $4k/Month Token Waste We Fixed by Ditching Custom Integrations

from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types

# This is a minimal MCP server. In production, you'd add connection pooling and pagination.
server = Server('demo')

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name='get_weather',
            description='Get current weather for a city',
            inputSchema={
                'type': 'object',
                'properties': {
                    'city': {'type': 'string', 'description': 'City name'}
                },
                'required': ['city']
            }
        )
    ]

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == 'get_weather':
        city = arguments['city']
        # In production, call a real weather API with retries and circuit breaker
        return [types.TextContent(type='text', text=f'Weather in {city}: sunny, 22°C')]
    raise ValueError(f'Unknown tool: {name}')

async def run():
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name='demo',
                server_version='0.1.0',
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )

if __name__ == '__main__':
    import asyncio
    asyncio.run(run())

import asyncpg
from mcp.server import Server
from pydantic import BaseModel, Field

# Pydantic model validates tool arguments before they reach the DB
class UserQuery(BaseModel):
    user_id: int = Field(..., description='User ID to look up')

server = Server('postgres-demo')
pool = None  # initialized in run()

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list:
    if name == 'get_user':
        # Validate arguments with Pydantic — prevents injection
        args = UserQuery(**arguments)
        async with pool.acquire() as conn:
            # Parameterized query — never use f-strings
            row = await conn.fetchrow('SELECT id, name, email FROM users WHERE id = $1', args.user_id)
            if row:
                return [types.TextContent(type='text', text=f'User: {row["name"]} ({row["email"]})')]
            return [types.TextContent(type='text', text='User not found')]
    raise ValueError(f'Unknown tool: {name}')

async def run():
    global pool
    pool = await asyncpg.create_pool(dsn='postgres://user:pass@localhost/db', min_size=5, max_size=20)
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, ...)

if __name__ == '__main__':
    import asyncio
    asyncio.run(run())

import asyncio
import time

# Simulate a direct API call vs MCP tool call
async def direct_api():
    # Direct HTTP call to your service
    await asyncio.sleep(0.015)  # 15ms
    return 'result'

async def mcp_tool_call():
    # MCP serialization + transport + deserialization
    await asyncio.sleep(0.015)  # actual work
    await asyncio.sleep(0.070)  # MCP overhead: JSON serialization, transport, deserialization
    return 'result'

async def benchmark():
    # Run 100 calls each
    direct_start = time.time()
    for _ in range(100):
        await direct_api()
    direct_elapsed = time.time() - direct_start

    mcp_start = time.time()
    for _ in range(100):
        await mcp_tool_call()
    mcp_elapsed = time.time() - mcp_start

    print(f'Direct API: {direct_elapsed*10:.1f}ms avg')  # ~15ms
    print(f'MCP tool: {mcp_elapsed*10:.1f}ms avg')      # ~85ms

asyncio.run(benchmark())

import pybreaker
from mcp.server import Server

# Circuit breaker for external API calls
breaker = pybreaker.CircuitBreaker(fail_max=5, reset_timeout=60)

server = Server('resilient-demo')

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list:
    if name == 'call_external_api':
        try:
            # If the external API fails 5 times in a row, the circuit opens for 60 seconds
            result = await breaker.call(external_api_call, arguments)
            return [types.TextContent(type='text', text=result)]
        except pybreaker.CircuitBreakerError:
            # Return a fallback response instead of failing
            return [types.TextContent(type='text', text='Service temporarily unavailable. Please try again later.')]
    raise ValueError(f'Unknown tool: {name}')

async def external_api_call(args):
    # Simulate an external API call that may fail
    import aiohttp
    async with aiohttp.ClientSession() as session:
        async with session.get('https://api.example.com/data', params=args) as resp:
            resp.raise_for_status()
            return await resp.text()

from mcp.server import Server
import mcp.types as types

server = Server('error-handling-demo')

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list:
    try:
        if name == 'divide':
            a = arguments['a']
            b = arguments['b']
            result = a / b  # May raise ZeroDivisionError
            return [types.TextContent(type='text', text=f'Result: {result}')]
    except ZeroDivisionError:
        # Return a structured error instead of crashing
        return [types.TextContent(type='text', text='Error: Division by zero is not allowed.')]
    except KeyError as e:
        return [types.TextContent(type='text', text=f'Error: Missing argument {e}')]
    except Exception as e:
        # Log the full exception for debugging
        import logging
        logging.exception('Unhandled error in tool call')
        return [types.TextContent(type='text', text=f'Internal server error.')]
    raise ValueError(f'Unknown tool: {name}')

# OpenAI function calling example (simpler but OpenAI-only)
import openai

# Define tools inline in the API request
tools = [
    {
        'type': 'function',
        'function': {
            'name': 'get_weather',
            'description': 'Get current weather for a city',
            'parameters': {
                'type': 'object',
                'properties': {
                    'city': {'type': 'string'}
                },
                'required': ['city']
            }
        }
    }
]

# Single API call — no separate server needed
response = openai.chat.completions.create(
    model='gpt-4',
    messages=[{'role': 'user', 'content': 'What\'s the weather in Paris?'}],
    tools=tools
)

# MCP requires a separate server process and JSON-RPC transport
# More complex but works with any MCP-compatible client

import logging
from prometheus_client import Counter, Histogram, start_http_server
from mcp.server import Server

# Prometheus metrics
TOOL_CALL_COUNT = Counter('mcp_tool_call_count', 'Number of tool calls', ['tool_name', 'status'])
TOOL_CALL_DURATION = Histogram('mcp_tool_call_duration_seconds', 'Duration of tool calls', ['tool_name'])
RESOURCE_SIZE = Histogram('mcp_resource_size_bytes', 'Size of resource responses', ['resource_name'])

server = Server('monitored-demo')

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list:
    with TOOL_CALL_DURATION.labels(tool_name=name).time():
        try:
            result = await actual_tool_handler(name, arguments)
            TOOL_CALL_COUNT.labels(tool_name=name, status='success').inc()
            return result
        except Exception as e:
            TOOL_CALL_COUNT.labels(tool_name=name, status='error').inc()
            raise

async def run():
    # Start Prometheus metrics server on port 8001
    start_http_server(8001)
    logging.basicConfig(level=logging.DEBUG)
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, ...)