Building MCP Servers in Python: Step-by-Step Tutorial

Prerequisites and Environment Setup

Before writing a single line of MCP server code, get your environment right. This matters more than most tutorials admit. Python 3.10 or higher is required — the MCP SDK uses modern type annotation features that break on older versions. We recommend 3.11 or 3.12 for production deployments. Use a virtual environment for every project. This is non-negotiable when you're deploying multiple MCP servers that may have conflicting dependency requirements.

# Create project directory and virtual environment
mkdir my-enterprise-mcp && cd my-enterprise-mcp
python3.11 -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

# Install the MCP SDK and common enterprise dependencies
pip install mcp httpx python-dotenv pydantic

# Verify installation
python -c "import mcp; print(f'MCP SDK version: {mcp.__version__}')"

The mcp package is Anthropic's official Python SDK. httpx is the recommended HTTP client for making requests to backend systems from your server — it handles async natively, which matters for performance. python-dotenv manages environment variables for credentials. pydantic handles data validation, which you'll use heavily when building robust tool parameter validation.

For the rest of this tutorial, we'll build an MCP server for a fictional internal CRM API. The patterns are identical to what you'd use for Salesforce, a PostgreSQL database, SAP, or any other enterprise system. Our MCP Enterprise Guide covers the architectural decisions behind these patterns.

Your First MCP Server

Let's build the simplest possible working MCP server. Three tools, clean structure, running in five minutes. Then we'll build on it.

import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import json

# Initialise the MCP server with a name and version
app = Server("enterprise-crm")

@app.list_tools()
async def list_tools():
    """Declare all tools this server exposes."""
    return [
        Tool(
            name="get_customer",
            description="Retrieve a customer record by ID. Returns name, email, company, account tier, and last contact date.",
            inputSchema={
                "type": "object",
                "properties": {
                    "customer_id": {
                        "type": "string",
                        "description": "The unique customer identifier (e.g. CUST-00123)"
                    }
                },
                "required": ["customer_id"]
            }
        ),
        Tool(
            name="search_customers",
            description="Search customers by company name or email domain. Returns a list of matching customers with IDs.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Company name or email domain to search for"
                    },
                    "limit": {
                        "type": "integer",
                        "description": "Maximum number of results to return (default: 10, max: 50)",
                        "default": 10
                    }
                },
                "required": ["query"]
            }
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    """Route tool calls to their handlers."""
    if name == "get_customer":
        return await handle_get_customer(arguments)
    elif name == "search_customers":
        return await handle_search_customers(arguments)
    else:
        raise ValueError(f"Unknown tool: {name}")

async def handle_get_customer(args: dict):
    customer_id = args["customer_id"]
    # In production: call your CRM API here
    mock_customer = {
        "id": customer_id,
        "name": "Acme Corporation",
        "email": "procurement@acme.com",
        "account_tier": "Enterprise",
        "last_contact": "2026-02-14"
    }
    return [TextContent(type="text", text=json.dumps(mock_customer, indent=2))]

async def handle_search_customers(args: dict):
    query = args["query"]
    limit = args.get("limit", 10)
    # In production: query your CRM search API here
    return [TextContent(type="text", text=json.dumps({
        "query": query, "results": [], "count": 0
    }))]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

That's a complete, working MCP server. Run it with python server.py and it listens on stdio. Claude can now list its tools and call them. The structure is always the same: initialise the server, declare tools via @app.list_tools(), handle calls via @app.call_tool(), implement handler functions that return TextContent objects.

Connecting to a Real Enterprise API

Mocked data gets you nowhere. Here's how to wire your MCP server to an actual backend — with authentication, async HTTP calls, and proper error handling. We'll use an API key authenticated REST service as the example, since that's the pattern for most enterprise SaaS APIs.

import asyncio
import os
import json
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import httpx
from dotenv import load_dotenv

load_dotenv()  # Load credentials from .env file

app = Server("enterprise-crm")

# Initialise HTTP client at module level — reuse across all requests
CRM_BASE_URL = os.getenv("CRM_BASE_URL", "https://api.yourcrm.internal")
CRM_API_KEY = os.getenv("CRM_API_KEY")

if not CRM_API_KEY:
    raise EnvironmentError("CRM_API_KEY environment variable is required")

http_client = httpx.AsyncClient(
    base_url=CRM_BASE_URL,
    headers={
        "Authorization": f"Bearer {CRM_API_KEY}",
        "Content-Type": "application/json",
        "User-Agent": "ClaudeImplementation-MCP/1.0",
    },
    timeout=30.0
)

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="get_customer",
            description="Retrieve full customer record by ID including account tier, ARR, renewal date, and support history.",
            inputSchema={
                "type": "object",
                "properties": {
                    "customer_id": {"type": "string", "description": "Customer ID in format CUST-XXXXX"}
                },
                "required": ["customer_id"]
            }
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "get_customer":
        return await handle_get_customer(arguments)
    raise ValueError(f"Unknown tool: {name}")

async def handle_get_customer(args: dict):
    customer_id = args.get("customer_id")
    if not customer_id:
        raise ValueError("customer_id is required")

    # Validate format before hitting the API
    if not customer_id.startswith("CUST-"):
        return [TextContent(type="text", text=json.dumps({
            "error": "invalid_format",
            "message": "Customer ID must be in format CUST-XXXXX"
        }))]

    try:
        response = await http_client.get(f"/v2/customers/{customer_id}")
        response.raise_for_status()
        return [TextContent(type="text", text=response.text)]

    except httpx.HTTPStatusError as e:
        if e.response.status_code == 404:
            return [TextContent(type="text", text=json.dumps({
                "error": "not_found",
                "message": f"No customer found with ID {customer_id}"
            }))]
        return [TextContent(type="text", text=json.dumps({
            "error": "api_error",
            "status_code": e.response.status_code,
            "message": "CRM API returned an error"
        }))]

    except httpx.TimeoutException:
        return [TextContent(type="text", text=json.dumps({
            "error": "timeout",
            "message": "CRM API timed out. The system may be under load."
        }))]

OAuth Authentication for Enterprise SaaS

Many enterprise systems — Salesforce, Microsoft 365, Google Workspace — require OAuth 2.0 rather than simple API keys. The pattern is to handle the OAuth flow at server startup, store the token in memory, and implement automatic token refresh. Your MCP server should never expose OAuth flows to Claude — the authentication layer is entirely the server's responsibility.

import time
import httpx
from dataclasses import dataclass
from typing import Optional

@dataclass
class TokenCache:
    access_token: str
    expires_at: float

    def is_expired(self) -> bool:
        # Refresh 5 minutes before actual expiry
        return time.time() >= (self.expires_at - 300)

class OAuthClient:
    def __init__(self, token_url: str, client_id: str, client_secret: str, scope: str):
        self.token_url = token_url
        self.client_id = client_id
        self.client_secret = client_secret
        self.scope = scope
        self._token_cache: Optional[TokenCache] = None

    async def get_access_token(self) -> str:
        if self._token_cache and not self._token_cache.is_expired():
            return self._token_cache.access_token

        # Fetch new token via client credentials grant
        async with httpx.AsyncClient() as client:
            response = await client.post(
                self.token_url,
                data={
                    "grant_type": "client_credentials",
                    "client_id": self.client_id,
                    "client_secret": self.client_secret,
                    "scope": self.scope,
                }
            )
            response.raise_for_status()
            token_data = response.json()

            self._token_cache = TokenCache(
                access_token=token_data["access_token"],
                expires_at=time.time() + token_data.get("expires_in", 3600)
            )
            return self._token_cache.access_token

Exposing Resources from Your MCP Server

Resources let Claude read structured data without a tool call — useful for documentation, reference data, and content that doesn't change frequently. A Confluence MCP server exposing individual pages as resources is a good example. Claude can list available resources and read them on demand, grounding its responses in real organisational content.

from mcp.types import Resource, ResourceContents, TextResourceContents
from mcp.types import AnyUrl

@app.list_resources()
async def list_resources():
    """Expose key business documents as readable resources."""
    return [
        Resource(
            uri="crm://docs/pricing-tiers",
            name="Account Pricing Tiers",
            description="Current pricing structure for Starter, Professional, and Enterprise tiers",
            mimeType="text/plain"
        ),
        Resource(
            uri="crm://docs/sla-policy",
            name="SLA Policy",
            description="Service level agreements by account tier",
            mimeType="text/plain"
        ),
    ]

@app.read_resource()
async def read_resource(uri: AnyUrl):
    uri_str = str(uri)
    if uri_str == "crm://docs/pricing-tiers":
        return ResourceContents(
            uri=uri,
            contents=[TextResourceContents(
                uri=uri,
                mimeType="text/plain",
                text=await fetch_pricing_document()
            )]
        )
    raise ValueError(f"Unknown resource URI: {uri_str}")

async def fetch_pricing_document():
    # In production: fetch from your documentation system
    return """
Starter Plan: $49/user/month, up to 10 users, email support only
Professional Plan: $149/user/month, up to 100 users, phone + email support
Enterprise Plan: Custom pricing, unlimited users, dedicated CSM
"""

Testing Your MCP Server

Test your MCP server before connecting it to Claude. The MCP Inspector is Anthropic's official debugging tool — install it with npx @modelcontextprotocol/inspector and point it at your server. It lets you list tools, call tools directly, and inspect the JSON protocol messages flowing between client and server. This is invaluable for debugging tool definitions and catching format errors before Claude encounters them.

For automated testing, write unit tests for each handler function independently. Test normal cases, error cases, edge cases in input validation, and timeout behaviour. Mock the HTTP client so your tests run without hitting real backend systems. Use pytest with pytest-asyncio for async test support.

import pytest
from unittest.mock import AsyncMock, patch
from server import handle_get_customer

@pytest.mark.asyncio
async def test_get_customer_invalid_format():
    """Reject customer IDs that don't match expected format."""
    result = await handle_get_customer({"customer_id": "not-a-valid-id"})
    import json
    data = json.loads(result[0].text)
    assert data["error"] == "invalid_format"

@pytest.mark.asyncio
async def test_get_customer_not_found():
    """Return structured error when customer doesn't exist."""
    with patch("server.http_client") as mock_client:
        mock_response = AsyncMock()
        mock_response.status_code = 404
        import httpx
        mock_client.get.side_effect = httpx.HTTPStatusError(
            "404", request=None, response=mock_response
        )
        result = await handle_get_customer({"customer_id": "CUST-99999"})
        import json
        data = json.loads(result[0].text)
        assert data["error"] == "not_found"

Deploying to Production

A Docker container is the right deployment unit for an enterprise MCP server. Package your server with its dependencies, configure it via environment variables, and deploy it where it can reach your backend systems — typically inside your corporate network or private cloud environment.

# Use explicit version tag — never 'latest' in production
FROM python:3.11-slim

WORKDIR /app

# Install dependencies first (Docker layer caching)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy server code
COPY server.py .
COPY oauth_client.py .

# Run as non-root user for security
RUN adduser --disabled-password --gecos "" mcpuser
USER mcpuser

# Credentials passed via environment variables at runtime
ENV CRM_BASE_URL=""
ENV CRM_API_KEY=""

CMD ["python", "server.py"]

For production deployments that Claude Code or Claude Cowork will connect to over HTTP (rather than stdio), you'll need to wrap your server with an HTTP transport layer. The MCP SDK supports this via the SSE transport. Deploy behind an API gateway that handles TLS termination, authentication of inbound Claude connections, and rate limiting. Your MCP server itself should only accept connections from known, authenticated Claude host applications — never expose it to the open internet.

For enterprise deployments at scale, our MCP Server Development service handles the full production deployment — container build, security hardening, network architecture, monitoring setup, and integration with your existing CI/CD pipeline. We've done this across financial services, healthcare, and manufacturing environments with strict compliance requirements.

Common Mistakes and How to Avoid Them

After reviewing dozens of enterprise MCP implementations, these are the mistakes that cause production incidents:

Synchronous blocking calls inside async handlers. Using the standard requests library instead of httpx, or calling a blocking database driver inside an async function, will block the entire event loop and make your server unresponsive. Every I/O operation in an MCP server handler must be properly async. Use asyncio.to_thread() for truly blocking operations that can't be made async.

Not handling timeouts at the HTTP client level. Backend systems fail. An MCP server that hangs indefinitely waiting for a backend response will cause the Claude workflow calling it to time out and fail unhelpfully. Set explicit timeouts — 30 seconds for most operations, shorter for simple lookups. Return a structured timeout error so Claude can respond intelligently.

Logging sensitive data. Debug logging that captures tool call arguments can inadvertently log customer data, API keys, or PII. Set up structured logging from the start and be deliberate about what gets logged at each level. Never log full API responses unless you've sanitised them.

Not versioning the server. When you change a tool signature, existing Claude agents that call your server will break. Always version your server, maintain backwards compatibility for one major version, and communicate changes to teams that depend on your server before deploying.