Build MCP Servers in Python: Complete Step-by-Step Tutorial

# macOS / Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# Or with Homebrew
brew install uv

# Verify installation
uv --version

# Create project with uv
uv init weather-mcp-server
cd weather-mcp-server

# Add the MCP SDK with CLI tools
uv add "mcp[cli]"

mkdir weather-mcp-server
cd weather-mcp-server
python3 -m venv .venv
source .venv/bin/activate
pip install "mcp[cli]"

[project]
name = "weather-mcp-server"
version = "1.0.0"
description = "MCP server providing weather data to AI assistants"
requires-python = ">=3.10"
dependencies = [
    "mcp[cli]>=1.2.0",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

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

mkdir -p src/weather_mcp_server
touch src/weather_mcp_server/__init__.py
touch src/weather_mcp_server/server.py

weather-mcp-server/
├── pyproject.toml
├── src/
│   └── weather_mcp_server/
│       ├── __init__.py
│       └── server.py       # Main server code
└── README.md

"""Weather MCP Server — provides weather data to AI assistants."""

from mcp.server.fastmcp import FastMCP

# Create the FastMCP server instance
mcp = FastMCP("weather-server")

# Simulated weather data (replace with a real API in production)
WEATHER_DATA = {
    "New York": {"temp_c": 2, "condition": "Cloudy", "humidity": 65, "wind_kmh": 20},
    "San Francisco": {"temp_c": 14, "condition": "Sunny", "humidity": 70, "wind_kmh": 15},
    "London": {"temp_c": 8, "condition": "Rainy", "humidity": 85, "wind_kmh": 25},
    "Tokyo": {"temp_c": 10, "condition": "Clear", "humidity": 55, "wind_kmh": 10},
    "Sydney": {"temp_c": 25, "condition": "Sunny", "humidity": 60, "wind_kmh": 18},
    "Berlin": {"temp_c": 5, "condition": "Overcast", "humidity": 75, "wind_kmh": 22},
}


@mcp.tool()
def get_weather(city: str) -> str:
    """Get current weather for a city.

    Args:
        city: City name, e.g. "New York", "London", "Tokyo"

    Returns:
        Weather report including temperature, condition, humidity, and wind speed.
    """
    data = WEATHER_DATA.get(city)
    if not data:
        return f"No weather data available for '{city}'. Available cities: {', '.join(WEATHER_DATA.keys())}"

    return (
        f"Weather in {city}:\n"
        f"  Temperature: {data['temp_c']}°C\n"
        f"  Condition: {data['condition']}\n"
        f"  Humidity: {data['humidity']}%\n"
        f"  Wind: {data['wind_kmh']} km/h"
    )


@mcp.tool()
def compare_weather(city1: str, city2: str) -> str:
    """Compare weather between two cities.

    Args:
        city1: First city name
        city2: Second city name

    Returns:
        Side-by-side weather comparison with temperature difference.
    """
    data1 = WEATHER_DATA.get(city1)
    data2 = WEATHER_DATA.get(city2)

    if not data1:
        return f"No data for '{city1}'. Available: {', '.join(WEATHER_DATA.keys())}"
    if not data2:
        return f"No data for '{city2}'. Available: {', '.join(WEATHER_DATA.keys())}"

    diff = data1["temp_c"] - data2["temp_c"]
    warmer = city1 if diff > 0 else city2

    return (
        f"Weather Comparison: {city1} vs {city2}\n"
        f"{'─' * 40}\n"
        f"{city1}: {data1['condition']}, {data1['temp_c']}°C, "
        f"humidity {data1['humidity']}%, wind {data1['wind_kmh']} km/h\n"
        f"{city2}: {data2['condition']}, {data2['temp_c']}°C, "
        f"humidity {data2['humidity']}%, wind {data2['wind_kmh']} km/h\n"
        f"{'─' * 40}\n"
        f"Temperature difference: {abs(diff)}°C ({warmer} is warmer)"
    )


def main():
    """Entry point for the server."""
    mcp.run()


if __name__ == "__main__":
    main()

@mcp.resource("weather://cities")
def list_cities() -> str:
    """List all cities with available weather data.

    Returns a JSON array of supported city names.
    """
    import json
    return json.dumps(list(WEATHER_DATA.keys()), indent=2)


@mcp.resource("weather://city/{city_name}")
def get_city_details(city_name: str) -> str:
    """Get detailed weather data for a specific city as JSON.

    This resource template accepts any city name in the URL.
    """
    import json
    data = WEATHER_DATA.get(city_name)
    if not data:
        return json.dumps({"error": f"City '{city_name}' not found"})
    return json.dumps({"city": city_name, **data}, indent=2)

@mcp.prompt()
def travel_advisory(destination: str) -> str:
    """Get a travel advisory based on current weather at the destination.

    Args:
        destination: The city you plan to travel to
    """
    data = WEATHER_DATA.get(destination)
    if not data:
        return f"Check weather conditions for {destination} and suggest appropriate clothing and travel tips."

    return (
        f"The current weather in {destination} is {data['condition']} "
        f"at {data['temp_c']}°C with {data['humidity']}% humidity "
        f"and winds of {data['wind_kmh']} km/h.\n\n"
        f"Based on these conditions, please suggest:\n"
        f"1. Appropriate clothing to pack\n"
        f"2. Weather-related travel tips\n"
        f"3. Best time of day for outdoor activities"
    )


@mcp.prompt()
def weather_report(cities: str = "all") -> str:
    """Generate a comprehensive weather report.

    Args:
        cities: Comma-separated city names, or "all" for every city
    """
    if cities == "all":
        city_list = list(WEATHER_DATA.keys())
    else:
        city_list = [c.strip() for c in cities.split(",")]

    city_str = ", ".join(city_list)
    return (
        f"Generate a professional weather report for the following cities: {city_str}.\n\n"
        f"Include:\n"
        f"- Current conditions summary\n"
        f"- Temperature trends\n"
        f"- Travel recommendations\n"
        f"- Any weather warnings"
    )

import httpx

@mcp.tool()
async def fetch_weather_api(city: str) -> str:
    """Fetch real weather data from an external API.

    Args:
        city: City name to look up
    """
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(
                "https://api.weatherapi.com/v1/current.json",
                params={"key": "YOUR_API_KEY", "q": city},
                timeout=10.0,
            )
            response.raise_for_status()
            data = response.json()

            current = data["current"]
            return (
                f"Weather in {city}:\n"
                f"  Temperature: {current['temp_c']}°C\n"
                f"  Condition: {current['condition']['text']}\n"
                f"  Humidity: {current['humidity']}%\n"
                f"  Wind: {current['wind_kph']} km/h"
            )

    except httpx.TimeoutException:
        return f"Error: Request timed out while fetching weather for '{city}'. Try again later."
    except httpx.HTTPStatusError as e:
        return f"Error: API returned status {e.response.status_code} for '{city}'."
    except Exception as e:
        return f"Error: Failed to fetch weather for '{city}': {str(e)}"

from mcp.server.fastmcp import Context

@mcp.tool()
async def analyze_weather_trends(city: str, ctx: Context) -> str:
    """Analyze weather trends for a city with progress reporting.

    Args:
        city: City name to analyze
    """
    # Log messages (sent to the MCP client)
    await ctx.info(f"Starting weather analysis for {city}")

    # Report progress (useful for long-running operations)
    await ctx.report_progress(0, 3)

    await ctx.debug("Fetching current data...")
    data = WEATHER_DATA.get(city)
    if not data:
        await ctx.warning(f"No data found for {city}")
        return f"No data available for '{city}'"

    await ctx.report_progress(1, 3)

    await ctx.debug("Computing analysis...")
    # Simulate analysis work
    analysis = f"Weather analysis for {city}: {data['condition']} conditions, {data['temp_c']}°C"

    await ctx.report_progress(2, 3)

    await ctx.info(f"Analysis complete for {city}")
    await ctx.report_progress(3, 3)

    return analysis

import sys

# WRONG — breaks STDIO protocol
print("Debug: processing request")

# CORRECT — uses stderr, does not interfere
print("Debug: processing request", file=sys.stderr)

# Start the Inspector
mcp dev src/weather_mcp_server/server.py

# Or with uv (if not in the activated venv)
uv run mcp dev src/weather_mcp_server/server.py

# Run with stdio transport (default)
mcp run src/weather_mcp_server/server.py

# Using uv (recommended — handles virtual env automatically)
claude mcp add weather-server -- uv run --directory /absolute/path/to/weather-mcp-server mcp run src/weather_mcp_server/server.py

# Or if using a script entry point defined in pyproject.toml
claude mcp add weather-server -- uv run --directory /absolute/path/to/weather-mcp-server weather-mcp-server

claude mcp add weather-server /absolute/path/to/weather-mcp-server/.venv/bin/python -m weather_mcp_server.server

# Check that the server is registered
claude mcp list

> What's the weather in Tokyo?

> Compare the weather in New York and London

> List all available cities in the weather server

claude mcp remove weather-server
claude mcp add weather-server -- uv run --directory /absolute/path/to/weather-mcp-server mcp run src/weather_mcp_server/server.py

"""SQLite MCP Server — gives AI assistants read access to a SQLite database."""

import sqlite3
import json
from pathlib import Path
from contextlib import closing

from mcp.server.fastmcp import FastMCP, Context

# Create the server
mcp = FastMCP("sqlite-server")

# Database path — configure this for your project
DB_PATH = Path("./data/example.db")


def get_connection() -> sqlite3.Connection:
    """Create a database connection with safety settings."""
    conn = sqlite3.connect(str(DB_PATH))
    conn.row_factory = sqlite3.Row  # Return rows as dictionaries
    conn.execute("PRAGMA journal_mode=WAL")  # Better concurrent read performance
    return conn


def init_sample_db():
    """Create a sample database for demonstration."""
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(DB_PATH))
    cursor = conn.cursor()

    cursor.executescript("""
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            email TEXT UNIQUE NOT NULL,
            role TEXT DEFAULT 'user',
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );

        CREATE TABLE IF NOT EXISTS projects (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            description TEXT,
            owner_id INTEGER REFERENCES users(id),
            status TEXT DEFAULT 'active',
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );

        CREATE TABLE IF NOT EXISTS tasks (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            project_id INTEGER REFERENCES projects(id),
            assignee_id INTEGER REFERENCES users(id),
            status TEXT DEFAULT 'todo',
            priority TEXT DEFAULT 'medium',
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );

        -- Insert sample data if tables are empty
        INSERT OR IGNORE INTO users (id, name, email, role) VALUES
            (1, 'Alice Chen', '[email protected]', 'admin'),
            (2, 'Bob Smith', '[email protected]', 'developer'),
            (3, 'Carol Wu', '[email protected]', 'developer'),
            (4, 'Dave Brown', '[email protected]', 'designer');

        INSERT OR IGNORE INTO projects (id, name, description, owner_id, status) VALUES
            (1, 'MCP Server', 'Build custom MCP servers', 1, 'active'),
            (2, 'Web Dashboard', 'Admin dashboard redesign', 4, 'active'),
            (3, 'API Migration', 'Migrate REST to GraphQL', 2, 'completed');

        INSERT OR IGNORE INTO tasks (id, title, project_id, assignee_id, status, priority) VALUES
            (1, 'Set up project structure', 1, 1, 'done', 'high'),
            (2, 'Implement weather tool', 1, 2, 'in_progress', 'high'),
            (3, 'Write unit tests', 1, 3, 'todo', 'medium'),
            (4, 'Design dashboard layout', 2, 4, 'in_progress', 'high'),
            (5, 'Implement auth flow', 2, 2, 'todo', 'high'),
            (6, 'Write API docs', 3, 3, 'done', 'low'),
            (7, 'Performance optimization', 1, 2, 'todo', 'medium');
    """)

    conn.commit()
    conn.close()


@mcp.tool()
def query_database(sql: str) -> str:
    """Execute a read-only SQL query against the database.

    Only SELECT statements are allowed. The query runs in a read-only
    transaction to prevent any data modifications.

    Args:
        sql: A SELECT SQL query to execute

    Returns:
        Query results as a formatted JSON string, or an error message.
    """
    # Security: only allow SELECT statements
    normalized = sql.strip().upper()
    if not normalized.startswith("SELECT"):
        return "Error: Only SELECT queries are allowed. This server provides read-only access."

    # Block dangerous keywords even in subqueries
    dangerous_keywords = ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "CREATE", "EXEC"]
    for keyword in dangerous_keywords:
        if keyword in normalized:
            return f"Error: '{keyword}' statements are not allowed."

    try:
        with closing(get_connection()) as conn:
            cursor = conn.execute(sql)
            columns = [description[0] for description in cursor.description]
            rows = [dict(zip(columns, row)) for row in cursor.fetchall()]

            return json.dumps({
                "columns": columns,
                "row_count": len(rows),
                "rows": rows,
            }, indent=2, default=str)

    except sqlite3.OperationalError as e:
        return f"SQL Error: {str(e)}"
    except Exception as e:
        return f"Error executing query: {str(e)}"


@mcp.tool()
def list_tables() -> str:
    """List all tables in the database with their column definitions.

    Returns:
        A formatted description of every table and its columns.
    """
    try:
        with closing(get_connection()) as conn:
            # Get all table names
            cursor = conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
            )
            tables = [row[0] for row in cursor.fetchall()]

            result = []
            for table in tables:
                # Get column info for each table
                cursor = conn.execute(f"PRAGMA table_info({table})")
                columns = []
                for col in cursor.fetchall():
                    col_name = col[1]
                    col_type = col[2]
                    not_null = "NOT NULL" if col[3] else ""
                    pk = "PRIMARY KEY" if col[5] else ""
                    columns.append(f"    {col_name} {col_type} {pk} {not_null}".strip())

                # Get row count
                cursor = conn.execute(f"SELECT COUNT(*) FROM {table}")
                row_count = cursor.fetchone()[0]

                result.append(
                    f"Table: {table} ({row_count} rows)\n" +
                    "\n".join(columns)
                )

            return "\n\n".join(result)

    except Exception as e:
        return f"Error listing tables: {str(e)}"


@mcp.tool()
async def describe_table(table_name: str, ctx: Context) -> str:
    """Get detailed schema and sample data for a specific table.

    Args:
        table_name: Name of the table to describe

    Returns:
        Table schema, sample rows, and basic statistics.
    """
    await ctx.info(f"Describing table: {table_name}")

    try:
        with closing(get_connection()) as conn:
            # Verify table exists
            cursor = conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table' AND name=?",
                (table_name,)
            )
            if not cursor.fetchone():
                return f"Error: Table '{table_name}' does not exist."

            # Get schema
            cursor = conn.execute(f"PRAGMA table_info({table_name})")
            columns = cursor.fetchall()

            schema_lines = [f"Schema for '{table_name}':"]
            for col in columns:
                schema_lines.append(
                    f"  {col[1]} ({col[2]})"
                    + (" PRIMARY KEY" if col[5] else "")
                    + (" NOT NULL" if col[3] else "")
                    + (f" DEFAULT {col[4]}" if col[4] else "")
                )

            # Get row count
            cursor = conn.execute(f"SELECT COUNT(*) FROM {table_name}")
            count = cursor.fetchone()[0]

            # Get sample rows (first 5)
            cursor = conn.execute(f"SELECT * FROM {table_name} LIMIT 5")
            col_names = [desc[0] for desc in cursor.description]
            sample_rows = [dict(zip(col_names, row)) for row in cursor.fetchall()]

            result = "\n".join(schema_lines)
            result += f"\n\nTotal rows: {count}"
            result += f"\n\nSample data (first 5 rows):\n{json.dumps(sample_rows, indent=2, default=str)}"

            return result

    except Exception as e:
        return f"Error describing table: {str(e)}"


@mcp.resource("db://schema")
def get_full_schema() -> str:
    """Get the complete database schema as a resource.

    This is useful for AI assistants that need to understand the
    database structure before writing queries.
    """
    try:
        with closing(get_connection()) as conn:
            cursor = conn.execute(
                "SELECT sql FROM sqlite_master WHERE type='table' AND sql IS NOT NULL"
            )
            statements = [row[0] for row in cursor.fetchall()]
            return "\n\n".join(statements)
    except Exception as e:
        return f"Error: {str(e)}"


@mcp.prompt()
def data_analysis(question: str) -> str:
    """Generate a data analysis prompt based on the database schema.

    Args:
        question: The data question you want to answer
    """
    schema = get_full_schema()
    return (
        f"You have access to a SQLite database with the following schema:\n\n"
        f"{schema}\n\n"
        f"The user wants to know: {question}\n\n"
        f"Write and execute the appropriate SQL query using the query_database tool. "
        f"Then explain the results clearly."
    )


def main():
    """Initialize the sample database and start the server."""
    init_sample_db()
    mcp.run()


if __name__ == "__main__":
    main()

# Test with the MCP Inspector
mcp dev src/weather_mcp_server/db_server.py

claude mcp add sqlite-server -- uv run --directory /absolute/path/to/weather-mcp-server python src/weather_mcp_server/db_server.py

> What tables are in the database?

> Show me all active projects with their owners

> How many tasks does each developer have? Group by status

> Find all high-priority tasks that are not done yet

# This is the default when you call mcp.run()
mcp.run()

# Explicitly specifying stdio transport
mcp.run(transport="stdio")

mcp.run(transport="sse", host="0.0.0.0", port=8000)

claude mcp add remote-weather --url http://your-server:8000/sse

mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

FROM python:3.12-slim

WORKDIR /app

# Install uv for fast dependency resolution
RUN pip install uv

# Copy project files
COPY pyproject.toml .
COPY src/ ./src/

# Install dependencies
RUN uv pip install --system -e .

# Run with streamable-http transport for remote access
EXPOSE 8000
CMD ["python", "-m", "weather_mcp_server.server"]

import os

def main():
    transport = os.environ.get("MCP_TRANSPORT", "stdio")
    if transport == "stdio":
        mcp.run()
    else:
        mcp.run(
            transport="streamable-http",
            host="0.0.0.0",
            port=int(os.environ.get("MCP_PORT", "8000")),
        )

# Build the package
uv build

# Publish to PyPI
uv publish

# Users install and run with uvx (uv's tool runner)
claude mcp add weather-server -- uvx weather-mcp-server

# Or with pip
pip install weather-mcp-server
claude mcp add weather-server weather-mcp-server