MCP Server in Python: Complete 2026 Tutorial with FastAPI

Short version: Building an MCP server in Python means choosing between stdio transport for local speed or HTTP/SSE for remote deployment. Use the official mcp-python SDK with FastAPI for HTTP servers, implement proper tool registration with JSON schemas, add authentication middleware before going to production, and test with mcp-inspector before connecting to Claude. The complete implementation takes around 200 lines for a basic server, 500+ for production-ready code with error handling and logging.

Why Python for MCP Servers Makes Sense (and When It Does Not)

The Model Context Protocol has implementations in TypeScript, Python, Rust, and Go. Python is not the fastest option, but it is the most practical for teams already running Python backends or data pipelines. If your existing tools are in Python, writing MCP wrappers in Python means you can import your libraries directly instead of shelling out to separate processes.

The official mcp-python SDK provides protocol compliance out of the box. You get the JSON-RPC message handling, capability negotiation, and transport abstractions without implementing the spec yourself. This matters because MCP has subtle requirements around message ordering and capability advertisement that are easy to get wrong.

Here is the honest tradeoff: Python MCP servers are slower than TypeScript or Rust equivalents. Cold start time is higher, memory usage is higher, and throughput under load is lower. For a tool that queries a database and returns results, this rarely matters. For a tool that processes large files or handles many concurrent requests, you will feel the difference.

The FastAPI choice specifically comes from needing HTTP transport with SSE support. Flask can work but requires additional libraries for SSE. Django is overkill. FastAPI gives you async support, automatic request validation, and built-in OpenAPI docs that help during development.

When should you not use Python? If your MCP server needs to process binary data at high speed, consider Rust. If you are already in a Node.js ecosystem and your team knows TypeScript better, use the TypeScript SDK. Do not pick Python just because it is trendy for AI work. Pick it because your existing code is in Python or your team ships faster in Python.

Indian teams building on AWS Mumbai or GCP asia-south1 should note that Python Lambda cold starts are noticeable. If you are deploying MCP servers as serverless functions, the cold start penalty hits every time the function scales from zero. Consider keeping a minimum instance warm or using container-based deployment instead.

For local development, you will want a machine that handles multiple Python processes without thermal throttling during testing.

Top Pick · Devslaptop

MacBook Pro 14" M4 Pro

The 2026 default for AI dev + local model inference up to 30B params.

$2,399+On Amazon

As an Amazon Associate, StackPicks earns from qualifying purchases at no extra cost to you.

Setting Up the Project Structure and Dependencies

A production MCP server needs more than just the SDK. Here is a minimal but complete dependency list:

# Create virtual environment
python -m venv .venv
source .venv/bin/activate

# Install core dependencies
pip install mcp fastapi uvicorn sse-starlette pydantic python-dotenv

# Development dependencies
pip install pytest pytest-asyncio httpx ruff mypy

The project structure should separate transport concerns from tool logic:

mcp-server/
├── src/
│   ├── __init__.py
│   ├── server.py          # MCP server setup
│   ├── tools/
│   │   ├── __init__.py
│   │   ├── database.py    # Database query tools
│   │   └── files.py       # File operation tools
│   ├── transports/
│   │   ├── __init__.py
│   │   ├── stdio.py       # Stdio transport handler
│   │   └── http.py        # FastAPI HTTP transport
│   └── middleware/
│       ├── __init__.py
│       └── auth.py        # Authentication middleware
├── tests/
├── .env
├── pyproject.toml
└── README.md

The separation between tools and transports is important. Your tool logic should not care whether it is being called over stdio or HTTP. This lets you run the same tools locally during development (stdio) and deployed remotely in production (HTTP) without code changes.

For pyproject.toml, use modern Python packaging:

[project]
name = "my-mcp-server"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
    "mcp>=1.0.0",
    "fastapi>=0.110.0",
    "uvicorn[standard]>=0.27.0",
    "sse-starlette>=2.0.0",
    "pydantic>=2.0.0",
    "python-dotenv>=1.0.0",
]

[project.scripts]
mcp-server = "src.server:main"

The entry point script matters for Claude Desktop integration. When you configure Claude to run your server, it needs a clean command that starts the stdio transport and handles shutdown gracefully.

One mistake I see often: developers put configuration in code instead of environment variables. Your database connection strings, API keys, and feature flags should come from .env files or environment variables. This is not just security hygiene — it lets you run the same server binary in development, staging, and production with different configurations.

The mcp-python SDK version matters. Check the official MCP Python SDK repository for the current stable release. Breaking changes between versions can cause subtle protocol mismatches that are hard to debug.

If you are integrating with AWS services, the patterns from the AWS MCP Server GA release apply here too — especially around IAM role assumptions and credential handling.

Implementing Tool Registration with Proper Schemas

MCP tools need JSON schemas that describe their inputs. The schema is not optional decoration — Claude uses it to understand what arguments to pass and what types to expect. Poor schemas lead to malformed tool calls and frustrated users.

Here is a complete tool implementation pattern:

from mcp.server import Server
from mcp.types import Tool, TextContent
from pydantic import BaseModel, Field
from typing import Any
import json

# Define input schema using Pydantic
class QueryDatabaseInput(BaseModel):
    query: str = Field(
        description="SQL SELECT query to execute. Must be read-only."
    )
    database: str = Field(
        default="primary",
        description="Database name: 'primary' or 'analytics'"
    )
    limit: int = Field(
        default=100,
        ge=1,
        le=1000,
        description="Maximum rows to return"
    )

# Create server instance
server = Server("database-tools")

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="query_database",
            description="Execute a read-only SQL query against the database. Returns results as JSON.",
            inputSchema=QueryDatabaseInput.model_json_schema()
        ),
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
    if name == "query_database":
        # Validate input using Pydantic
        try:
            params = QueryDatabaseInput(**arguments)
        except Exception as e:
            return [TextContent(
                type="text",
                text=f"Invalid arguments: {str(e)}"
            )]
        
        # Execute query (simplified - add connection pooling in production)
        result = await execute_query(
            params.query,
            params.database,
            params.limit
        )
        
        return [TextContent(
            type="text",
            text=json.dumps(result, indent=2, default=str)
        )]
    
    return [TextContent(
        type="text",
        text=f"Unknown tool: {name}"
    )]

The Pydantic integration is not accidental. Pydantic's model_json_schema() generates JSON Schema that matches what MCP expects. You get validation on both sides: Claude validates against the schema before calling, and your server validates again when receiving the call.

Notice the Field descriptions. These are not comments for developers — Claude reads them to understand how to use each parameter. Write descriptions as if explaining to a junior developer who has never seen your codebase.

Common schema mistakes:

For complex tools with many parameters, consider splitting into multiple focused tools instead of one tool with many optional parameters. Claude handles multiple simple tools better than one complex tool.

The patterns here align with what we covered in Claude Skills — well-defined tool boundaries make skills more reusable.

HTTP Transport with FastAPI and SSE

Stdio transport works for local tools, but production deployments need HTTP. The MCP spec uses Server-Sent Events for server-to-client streaming and standard POST requests for client-to-server messages.

Here is the FastAPI implementation:

from fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.responses import StreamingResponse
from sse_starlette.sse import EventSourceResponse
from mcp.server import Server
from mcp.server.sse import SseServerTransport
import asyncio
import json

app = FastAPI(title="MCP Server")

# Store active SSE connections
connections: dict[str, SseServerTransport] = {}

# Your MCP server instance
mcp_server = Server("my-tools")

@app.get("/sse")
async def sse_endpoint(request: Request):
    """SSE endpoint for server-to-client messages"""
    
    # Generate connection ID
    connection_id = str(id(request))
    
    async def event_generator():
        transport = SseServerTransport("/messages")
        connections[connection_id] = transport
        
        try:
            async with transport.connect_sse(
                request.scope,
                request.receive,
                request._send
            ) as streams:
                await mcp_server.run(
                    streams[0],
                    streams[1],
                    mcp_server.create_initialization_options()
                )
        finally:
            del connections[connection_id]
    
    return EventSourceResponse(event_generator())

@app.post("/messages")
async def messages_endpoint(request: Request):
    """POST endpoint for client-to-server messages"""
    
    body = await request.json()
    connection_id = request.headers.get("x-connection-id")
    
    if not connection_id or connection_id not in connections:
        raise HTTPException(400, "Invalid or missing connection ID")
    
    transport = connections[connection_id]
    await transport.handle_post_message(request.scope, request.receive, request._send)
    
    return {"status": "ok"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

This implementation has a critical detail: connection tracking. Each SSE connection gets an ID, and POST messages must include that ID to route to the correct connection. Without this, you cannot support multiple concurrent clients.

Dev Favoritemonitor

Dell UltraSharp U2723QE

27" 4K IPS Black, USB-C 90W dock. The current dev-monitor default.

$599+On Amazon

As an Amazon Associate, StackPicks earns from qualifying purchases at no extra cost to you.

The tradeoffs with HTTP transport:

Advantages:

• Deploy once, use from anywhere
• Multiple users can share one server
• Easier to scale horizontally
• Can add load balancing, caching

Disadvantages:

• Network latency on every tool call
• Need to handle connection drops
• SSE connections can timeout
• More complex auth requirements

For Indian deployments specifically, put your MCP server in the same region as your users. A server in AWS Mumbai serving users in Mumbai adds minimal latency. A server in US-East serving Mumbai users adds noticeable delay on every tool call. This matters because MCP tools often make multiple round trips during a conversation.

The HTTP transport also enables the MCP Server security patterns that are essential for production — authentication headers, rate limiting middleware, and request logging.

Adding Authentication and Rate Limiting

A public MCP server without authentication is a security incident waiting to happen. Even internal servers need auth because MCP tools often have access to databases, APIs, and file systems.

Here is a practical auth middleware pattern:

from fastapi import Request, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
import time
from collections import defaultdict
import asyncio

security = HTTPBearer()

# Rate limiting storage (use Redis in production)
rate_limit_store: dict[str, list[float]] = defaultdict(list)
RATE_LIMIT = 100  # requests per minute
RATE_WINDOW = 60  # seconds

async def verify_token(credentials: HTTPAuthorizationCredentials) -> dict:
    """Verify JWT token and return claims"""
    token = credentials.credentials
    
    try:
        # In production, fetch public key from your OIDC provider
        # This example uses symmetric key for simplicity
        payload = jwt.decode(
            token,
            key=os.getenv("JWT_SECRET"),
            algorithms=["HS256"],
            audience="mcp-server"
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(401, "Token expired")
    except jwt.InvalidTokenError as e:
        raise HTTPException(401, f"Invalid token: {str(e)}")

def check_rate_limit(client_id: str) -> bool:
    """Check if client is within rate limit"""
    now = time.time()
    
    # Clean old entries
    rate_limit_store[client_id] = [
        t for t in rate_limit_store[client_id]
        if now - t < RATE_WINDOW
    ]
    
    if len(rate_limit_store[client_id]) >= RATE_LIMIT:
        return False
    
    rate_limit_store[client_id].append(now)
    return True

async def auth_middleware(request: Request):
    """Combined auth and rate limit check"""
    
    # Skip auth for health check
    if request.url.path == "/health":
        return None
    
    # Get and verify token
    auth_header = request.headers.get("Authorization")
    if not auth_header or not auth_header.startswith("Bearer "):
        raise HTTPException(401, "Missing authorization header")
    
    credentials = HTTPAuthorizationCredentials(
        scheme="Bearer",
        credentials=auth_header[7:]
    )
    claims = await verify_token(credentials)
    
    # Check rate limit
    client_id = claims.get("sub", "anonymous")
    if not check_rate_limit(client_id):
        raise HTTPException(429, "Rate limit exceeded")
    
    # Attach claims to request state for tool access control
    request.state.user = claims
    return claims

# Apply to FastAPI app
app = FastAPI(dependencies=[Depends(auth_middleware)])

The rate limit implementation here is in-memory, which breaks with multiple server instances. For production, use Redis with atomic increment operations:

import redis.asyncio as redis

redis_client = redis.from_url(os.getenv("REDIS_URL"))

async def check_rate_limit_redis(client_id: str) -> bool:
    key = f"ratelimit:{client_id}"
    
    pipe = redis_client.pipeline()
    pipe.incr(key)
    pipe.expire(key, RATE_WINDOW)
    results = await pipe.execute()
    
    current_count = results[0]
    return current_count <= RATE_LIMIT

Tool-level authorization is often overlooked. Not every authenticated user should access every tool. Add permission checks inside your tool handlers:

@server.call_tool()
async def call_tool(name: str, arguments: dict, context: dict) -> list[TextContent]:
    user = context.get("user", {})
    user_roles = user.get("roles", [])
    
    # Tool-specific permission check
    tool_permissions = {
        "query_database": ["db_reader", "admin"],
        "write_file": ["admin"],
        "send_email": ["comms", "admin"],
    }
    
    required_roles = tool_permissions.get(name, [])
    if required_roles and not any(r in user_roles for r in required_roles):
        return [TextContent(
            type="text",
            text=f"Permission denied: requires one of {required_roles}"
        )]
    
    # ... rest of tool logic

For teams building SaaS products, these auth patterns integrate with your existing identity provider. See the SaaS building guide for the full authentication stack.

Testing and Debugging MCP Servers

MCP servers fail silently in ways that are hard to debug. The client sends a tool call, the server crashes, and Claude just sees a timeout. Building a testing strategy upfront saves hours of frustration.

Start with the MCP Inspector, the official debugging tool:

# Install inspector globally
npx @modelcontextprotocol/inspector

# Point it at your server
npx @modelcontextprotocol/inspector python src/server.py

The inspector shows you exactly what messages flow between client and server. You can see the initialize handshake, list_tools response, and tool call/response pairs. When something breaks, you see where.

For automated testing, test tools in isolation first:

import pytest
from src.tools.database import QueryDatabaseInput, execute_query

@pytest.mark.asyncio
async def test_query_database_valid_input():
    """Test with valid SQL query"""
    params = QueryDatabaseInput(
        query="SELECT id, name FROM users LIMIT 10",
        database="primary",
        limit=10
    )
    
    result = await execute_query(
        params.query,
        params.database,
        params.limit
    )
    
    assert isinstance(result, list)
    assert len(result) <= 10

@pytest.mark.asyncio
async def test_query_database_injection_attempt():
    """Test that SQL injection is blocked"""
    with pytest.raises(ValueError, match="write operation"):
        params = QueryDatabaseInput(
            query="SELECT 1; DROP TABLE users; --",
            database="primary",
            limit=10
        )
        await execute_query(
            params.query,
            params.database,
            params.limit
        )

Integration tests should exercise the full MCP protocol:

import pytest
from httpx import AsyncClient
from src.transports.http import app

@pytest.mark.asyncio
async def test_full_tool_call_flow():
    """Test complete MCP interaction over HTTP"""
    async with AsyncClient(app=app, base_url="http://test") as client:
        # Initialize connection
        init_response = await client.post("/messages", json={
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2024-11-05",
                "capabilities": {},
                "clientInfo": {"name": "test", "version": "1.0"}
            }
        })
        assert init_response.status_code == 200
        
        # List tools
        tools_response = await client.post("/messages", json={
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/list"
        })
        assert "query_database" in str(tools_response.json())

Universal Pickmouse

Logitech MX Master 3S

The dev productivity mouse. 8K DPI, quiet-click, works everywhere.

$99+On Amazon

As an Amazon Associate, StackPicks earns from qualifying purchases at no extra cost to you.

Logging strategy matters. Log at tool call boundaries with structured data:

import structlog

logger = structlog.get_logger()

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    log = logger.bind(
        tool_name=name,
        argument_keys=list(arguments.keys())
    )
    
    log.info("tool_call_started")
    
    try:
        result = await execute_tool(name, arguments)
        log.info("tool_call_completed", result_length=len(str(result)))
        return result
    except Exception as e:
        log.error("tool_call_failed", error=str(e), error_type=type(e).__name__)
        raise

For rapid iteration during development, a good keyboard makes a difference when you are constantly switching between editor, terminal, and browser.

#1 RTINGS 2026keyboard

Keychron Q1 Max

The #1 mechanical for devs per RTINGS 2026. QMK/VIA, hot-swap.

$219+On Amazon

As an Amazon Associate, StackPicks earns from qualifying purchases at no extra cost to you.

Deployment Patterns and Production Checklist

Deploying an MCP server is not the same as deploying a REST API. The SSE connections are long-lived, which breaks assumptions in typical load balancer configurations.

Docker deployment is the most portable option:

FROM python:3.12-slim

WORKDIR /app

# Install dependencies first for better caching
COPY pyproject.toml .
RUN pip install --no-cache-dir .

# Copy application code
COPY src/ src/

# Non-root user for security
RUN useradd -m -u 1000 appuser
USER appuser

# Health check
HEALTHCHECK --interval=30s --timeout=3s \
  CMD curl -f http://localhost:8000/health || exit 1

EXPOSE 8000

CMD ["uvicorn", "src.transports.http:app", "--host", "0.0.0.0", "--port", "8000"]

Load balancer configuration needs sticky sessions for SSE:

# AWS ALB target group config
Stickiness:
  Type: lb_cookie
  DurationSeconds: 3600

# Or use connection-based stickiness
TargetGroupAttributes:
  - Key: stickiness.enabled
    Value: "true"
  - Key: stickiness.type
    Value: source_ip

Production checklist before going live:

For Indian SaaS products, integrate with Razorpay for billing if you are charging for MCP server access. Metered billing based on tool calls works well for usage-based pricing.

The MVP in 48 hours guide covers the broader deployment stack if you are building a complete product around your MCP server.

What Could Go Wrong: Failure Modes and Mitigations

Every production system fails. Knowing how your MCP server fails helps you build appropriate safeguards.

Connection drops during tool execution. The client disconnects while a long-running tool is still processing. Without cancellation handling, you waste compute and may leave resources in inconsistent states.

Mitigation:

import asyncio

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    # Create cancellation-aware task
    try:
        result = await asyncio.wait_for(
            execute_long_running_tool(name, arguments),
            timeout=300  # 5 minute max
        )
        return result
    except asyncio.CancelledError:
        # Client disconnected, clean up
        await cleanup_partial_work()
        raise
    except asyncio.TimeoutError:
        return [TextContent(
            type="text",
            text="Tool execution timed out after 5 minutes"
        )]

Memory leaks from accumulated connections. SSE connections that disconnect uncleanly leave entries in your connection tracking dict. Over hours or days, memory usage climbs.

Mitigation:

import asyncio
from datetime import datetime, timedelta

# Track connection creation time
connections: dict[str, tuple[SseServerTransport, datetime]] = {}

async def cleanup_stale_connections():
    """Run periodically to remove dead connections"""
    while True:
        await asyncio.sleep(60)
        
        now = datetime.now()
        stale_threshold = timedelta(hours=1)
        
        stale_ids = [
            cid for cid, (_, created_at) in connections.items()
            if now - created_at > stale_threshold
        ]
        
        for cid in stale_ids:
            del connections[cid]
            logger.info("cleaned_stale_connection", connection_id=cid)

Tool returns malformed data. Your tool returns a Python dict, but it contains datetime objects or custom classes that do not serialize to JSON.

Mitigation: Always serialize explicitly with a default handler:

import json
from datetime import datetime, date
from decimal import Decimal

def json_serializer(obj):
    if isinstance(obj, (datetime, date)):
        return obj.isoformat()
    if isinstance(obj, Decimal):
        return float(obj)
    if hasattr(obj, "__dict__"):
        return obj.__dict__
    return str(obj)

return [TextContent(
    type="text",
    text=json.dumps(result, default=json_serializer, indent=2)
)]

Dependency service outages. Your tool calls an external API that goes down. Without proper error handling, the tool hangs or crashes.

Mitigation: Circuit breakers and fallbacks:

from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=60)
async def call_external_api(params):
    async with httpx.AsyncClient(timeout=10) as client:
        response = await client.post(API_URL, json=params)
        response.raise_for_status()
        return response.json()

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    try:
        result = await call_external_api(arguments)
        return [TextContent(type="text", text=json.dumps(result))]
    except Exception as e:
        return [TextContent(
            type="text",
            text=f"External service unavailable: {str(e)}. Try again later."
        )]

These patterns apply to any MCP server, not just Python. The Meta Ads MCP connector deals with similar reliability concerns when calling external advertising APIs.

For managing state across tools, the patterns from Zustand on the frontend have server-side equivalents — explicit state containers that multiple tools can read and write.

The complete implementation in this tutorial gives you a foundation. Real production deployments add monitoring dashboards, alerting rules, automated scaling, and incident response runbooks. But the core — tool registration, transport handling, authentication, and error management — stays the same.