MCP Server Tutorial: Python Complete Guide

Build, test, and deploy a production MCP server using Python.

Python is one of the most popular languages for building MCP (Model Context Protocol) servers. This tutorial covers everything from setup to production deployment, with real working code you can use today.

Why Python for MCP Servers?

Python is ideal for MCP servers because:

• Rich AI ecosystem — LangChain, LlamaIndex, transformers all Python-first
• Simple syntax — Clear, readable code for tool definitions
• Strong typing — Pydantic models for input validation
• Async support — Native asyncio for concurrent operations
• Vast libraries — Every API and service has a Python SDK

Prerequisites

• Python 3.11 or higher
• pip or uv package manager
• Basic familiarity with async Python

# Verify your Python version
python --version
# Python 3.11.x or higher required

Step 1: Project Setup

# Create project directory
mkdir my-mcp-server
cd my-mcp-server

# Create virtual environment
python -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

# Install MCP SDK
pip install mcp pydantic

# Or using uv (faster)
uv init
uv add mcp pydantic

Step 2: Define Your First Tool

Create server.py:

from mcp import Server, Tool
from pydantic import BaseModel, Field
from typing import Optional
import json

# Initialize the MCP server
server = Server(
    name="my-tools",
    version="1.0.0",
    description="A collection of utility tools for AI agents",
)

# Define input schema with Pydantic
class WeatherInput(BaseModel):
    city: str = Field(..., description="Name of the city")
    country: Optional[str] = Field(None, description="Country code (ISO 3166)")
    units: str = Field("metric", description="'metric' or 'imperial'")

# Register the tool
@server.tool("get_weather")
async def get_weather(input: WeatherInput) -> dict:
    """Get current weather conditions for a specified city."""
    import aiohttp

    location = f"{input.city},{input.country}" if input.country else input.city
    url = f"https://api.openweathermap.org/data/2.5/weather"
    params = {
        "q": location,
        "units": input.units,
        "appid": WEATHER_API_KEY,
    }

    async with aiohttp.ClientSession() as session:
        async with session.get(url, params=params) as response:
            data = await response.json()

    return {
        "city": data["name"],
        "temperature": data["main"]["temp"],
        "condition": data["weather"][0]["main"],
        "humidity": data["main"]["humidity"],
        "wind_speed": data["wind"]["speed"],
    }

Step 3: Add More Tools

class SearchInput(BaseModel):
    query: str = Field(..., description="Search query")
    max_results: int = Field(10, ge=1, le=50)

@server.tool("web_search")
async def web_search(input: SearchInput) -> dict:
    """Search the web and return relevant results."""
    # Use your preferred search API
    results = await search_api.search(input.query, limit=input.max_results)
    return {
        "results": [
            {
                "title": r.title,
                "url": r.url,
                "snippet": r.snippet,
            }
            for r in results
        ],
        "total_found": len(results),
    }


class CodeAnalysisInput(BaseModel):
    code: str = Field(..., description="Source code to analyze")
    language: str = Field("python", description="Programming language")

@server.tool("analyze_code")
async def analyze_code(input: CodeAnalysisInput) -> dict:
    """Analyze code for bugs, style issues, and improvements."""
    issues = []

    # Check for common issues
    if "eval(" in input.code:
        issues.append({
            "type": "security",
            "severity": "critical",
            "message": "Use of eval() detected — security risk",
        })

    if len(input.code.split("\n")) > 500:
        issues.append({
            "type": "maintainability",
            "severity": "warning",
            "message": "File exceeds 500 lines — consider splitting",
        })

    return {
        "language": input.language,
        "issues": issues,
        "quality_score": max(0, 100 - len(issues) * 10),
    }

Step 4: Add Resources (Data Sources)

MCP resources let agents read data from your server:

@server.resource("docs://api-reference")
async def get_api_docs():
    """Provide API documentation to agents."""
    return {
        "content": open("docs/api.md").read(),
        "mime_type": "text/markdown",
    }

@server.resource("data://product-catalog")
async def get_product_catalog():
    """Provide product catalog data."""
    return {
        "content": json.dumps(load_catalog()),
        "mime_type": "application/json",
    }

Step 5: Add Middleware

Middleware handles authentication, logging, and rate limiting:

from mcp.middleware import Middleware

class AuthMiddleware(Middleware):
    async def before_tool_call(self, request):
        api_key = request.headers.get("x-api-key")
        if not api_key or not await self.validate_key(api_key):
            raise PermissionError("Invalid or missing API key")

    async def validate_key(self, key: str) -> bool:
        # Check against your database or auth service
        return key in await self.get_valid_keys()

class LoggingMiddleware(Middleware):
    async def before_tool_call(self, request):
        logger.info(f"Tool call: {request.tool_name} by {request.client_id}")

    async def after_tool_call(self, request, response):
        duration = (time.time() - request.start_time) * 1000
        logger.info(
            f"Tool {request.tool_name} completed in {duration:.0f}ms"
        )

# Register middleware
server.add_middleware(AuthMiddleware())
server.add_middleware(LoggingMiddleware())

Step 6: Run the Server

Option A: stdio Transport (Local)

For local development and testing:

if __name__ == "__main__":
    server.run(transport="stdio")

Test with the MCP Inspector:

mcp inspect python server.py

Option B: HTTP Transport (Production)

For production deployment:

if __name__ == "__main__":
    server.run(
        transport="http",
        host="0.0.0.0",
        port=8000,
        ssl_certfile="/path/to/cert.pem",
        ssl_keyfile="/path/to/key.pem",
    )

Step 7: Deploy to Production

Deploy on Railway

# Dockerfile
FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
EXPOSE 8000

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

# Deploy
railway up

Deploy on AWS Lambda

# lambda_handler.py
from mcp import Server
import asyncio

server = Server("my-tools")

def lambda_handler(event, context):
    loop = asyncio.new_event_loop()
    result = loop.run_until_complete(
        server.handle_request(event)
    )
    return result

Deploy as systemd Service

# /etc/systemd/system/mcp-server.service
[Unit]
Description=MCP Server
After=network.target

[Service]
Type=simple
User=mcp
WorkingDirectory=/opt/mcp-server
ExecStart=/opt/mcp-server/.venv/bin/python server.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

Testing Your MCP Server

Unit Testing

import pytest
from my_server import server

@pytest.mark.asyncio
async def test_weather_tool():
    result = await server.call_tool(
        "get_weather",
        {"city": "Berlin", "units": "metric"}
    )
    assert "temperature" in result
    assert isinstance(result["temperature"], (int, float))

@pytest.mark.asyncio
async def test_code_analysis():
    result = await server.call_tool(
        "analyze_code",
        {"code": "eval('test')", "language": "python"}
    )
    assert any(i["type"] == "security" for i in result["issues"])

Integration Testing

@pytest.mark.asyncio
async def test_full_workflow():
    # Search for something
    search_result = await server.call_tool(
        "web_search",
        {"query": "Python MCP tutorial"}
    )

    # Analyze the first result's code
    analysis = await server.call_tool(
        "analyze_code",
        {"code": search_result["results"][0]["snippet"]}
    )

    assert analysis["quality_score"] >= 0

Publishing on SkillExchange

Once your server is deployed:

1. Go to skillexchange.market/publish
2. Enter your MCP server URL
3. Select the tools you want to list
4. Set pricing:
   • Free for exposure
   • €0.05-€0.50 for simple tools
   • €1-€5 for complex operations
5. Add documentation and examples
6. Publish

Your Python MCP server is now available to every AI agent on the platform.

Best Practices Summary

1. Always validate inputs with Pydantic models
2. Use async for all I/O operations
3. Set timeouts — never let tools hang indefinitely
4. Log everything — structured logging with structlog
5. Handle errors gracefully — return useful error messages
6. Version your tools — use semantic versioning
7. Document clearly — tool descriptions matter for agent selection
8. Rate limit — protect your server from abuse
9. Monitor costs — track API calls and resource usage
10. Test in production — use feature flags and canary deployments

Next Steps

• How to Build an MCP Server — TypeScript version
• MCP Server Security — Hardening guide
• MCP Server Hosting Options — Deployment comparison
• Enterprise MCP Deployment — Scale strategies

Built a Python MCP server? Publish it on SkillExchange and start earning.