Build a Production-Ready MCP Server with FastMCP in Python

Step-by-step tutorial: build an MCP server with FastMCP — exposing tools, resources, and streaming endpoints that Claude, Cursor, or any MCP host can call.

The Model Context Protocol (MCP) solved the N×M integration mess that plagued early agent builders. But knowing what MCP is — covered in our MCP complete guide — is different from actually building one that runs in production.

This tutorial walks through building a complete MCP server with Python and FastMCP, including tools, resources, proper error handling, and streaming support. By the end, you’ll have a server that Claude Desktop, Cursor, or any MCP host can connect to and call.

What FastMCP Is (and Why We Use It)

FastMCP started as a simplified layer on top of the official MCP Python SDK and has since been adopted as the recommended high-level API. FastMCP v3 is currently the stable release (v3.0.0), built on MCP protocol 1.25.0.

The official mcp package gives you low-level list_tools() / call_tool() handlers. FastMCP wraps those with Python decorators, automatic JSON Schema generation, and a clean server lifecycle. For production work, it cuts boilerplate by roughly 60%.

Step 1: Install and Verify

Create a virtual environment and install FastMCP. We recommend using uv for speed, but pip works identically.

mkdir weather-server && cd weather-server
uv venv .venv && source .venv/bin/activate
uv pip install fastmcp httpx

Verify the install:

fastmcp version

You should see FastMCP 3.x and MCP 1.x versions printed. If you’re on pip, run python -m fastmcp version instead.

Step 2: Create Your First MCP Server

Create a file called server.py:

from fastmcp import FastMCP

mcp = FastMCP(
    name="weather-server",
    version="0.1.0",
    description="A lightweight weather MCP server",
)

@mcp.tool()
def get_temperature(city: str, unit: str = "celsius") -> str:
    """Get the current temperature for a city.

    Args:
        city: The city name (e.g., 'London', 'Tokyo').
        unit: Temperature unit, either 'celsius' or 'fahrenheit'.
    """
    temperatures = {
        "london": {"celsius": 15, "fahrenheit": 59},
        "tokyo": {"celsius": 22, "fahrenheit": 72},
        "new york": {"celsius": 18, "fahrenheit": 64},
    }

    city_lower = city.lower()
    if city_lower not in temperatures:
        return f"Unknown city: {city}. Try London, Tokyo, or New York."

    temp = temperatures[city_lower][unit] if unit in temperatures[city_lower] else temperatures[city_lower]["celsius"]
    return f"It is currently {temp}°{unit[0].upper()} in {city}."

if __name__ == "__main__":
    mcp.run()

Run it:

fastmcp run server.py

FastMCP starts an stdio transport by default. You’ll see the server banner and a waiting prompt. This server is now ready to accept MCP protocol messages over stdin/stdout.

Step 3: Connect with Claude Desktop

Create or update your Claude Desktop MCP config file:

• macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
• Windows: %APPDATA%\Claude\claude_desktop_config.json

{
  "mcpServers": {
    "weather": {
      "command": "/home/runner/.local/bin/fastmcp",
      "args": ["run", "/path/to/weather-server/server.py"]
    }
  }
}

Replace the paths with your actual fastmcp binary location (run which fastmcp) and the absolute path to server.py.

After restarting Claude Desktop, you’ll see the weather server listed. Claude will automatically discover the get_temperature tool and call it when you ask about weather in any of the hardcoded cities.

Step 4: Add an API-Backed Tool

Hardcoded lookups are useful for testing, but real servers call external APIs. Let’s wire up Open-Meteo — a free, no-key weather API.

from fastmcp import FastMCP
import httpx
from datetime import datetime

mcp = FastMCP(
    name="weather-server",
    version="0.2.0",
    description="Real-time weather via Open-Meteo API",
)

CITY_COORDS = {
    "london": (51.5074, -0.1278),
    "tokyo": (35.6762, 139.6503),
    "new york": (40.7128, -74.0060),
    "paris": (48.8566, 2.3522),
    "sydney": (-33.8688, 151.2093),
    "mumbai": (19.0760, 72.8777),
    "berlin": (52.5200, 13.4050),
}


@mcp.tool()
async def get_temperature(city: str, unit: str = "celsius") -> str:
    """Get the current temperature for a city.

    Args:
        city: The city name (e.g., 'London', 'Tokyo').
        unit: Temperature unit — 'celsius' or 'fahrenheit'.
    """
    city_lower = city.lower()
    if city_lower not in CITY_COORDS:
        cities = ", ".join(CITY_COORDS.keys())
        return f"Unknown city: {city}. Supported cities: {cities}."

    lat, lon = CITY_COORDS[city_lower]
    url = "https://api.open-meteo.com/v1/forecast"
    params = {
        "latitude": lat,
        "longitude": lon,
        "current": "temperature_2m",
        "temperature_unit": unit,
    }

    async with httpx.AsyncClient() as client:
        response = await client.get(url, params=params, timeout=10.0)
        response.raise_for_status()

    data = response.json()
    temp = data["current"]["temperature_2m"]
    unit_symbol = "°C" if unit == "celsius" else "°F"
    timestamp = data["current"]["time"]

    return (
        f"It is currently {temp}{unit_symbol} in {city.title()} "
        f"(measured at {timestamp}, source: Open-Meteo)."
    )


if __name__ == "__main__":
    mcp.run()

The key change is the async tool function. FastMCP handles the async event loop for you — the MCP client sees no difference between sync and async tools. We use httpx.AsyncClient because it’s the idiomatic async HTTP library for Python, and it avoids blocking the MCP stdio transport.

Step 5: Expose Resources

MCP servers can expose resources — static or dynamic data that clients can read by URI. Think of resources as a read-only data surface the agent can pull from without calling a tool.

Add these to the same server file:

from fastmcp import Resource


@mcp.resource("weather://supported-cities")
def supported_cities() -> str:
    """Returns the list of cities this server supports."""
    lines = [f"- {name.title()}" for name in CITY_COORDS]
    return "Supported cities:\n" + "\n".join(lines)


@mcp.resource("weather://about")
def server_about() -> str:
    """Returns metadata about the weather MCP server."""
    return (
        "Weather MCP Server v0.2.0\n"
        "Source: Open-Meteo API (free, no auth required)\n"
        "Units: celsius (default), fahrenheit\n"
        f"Cities covered: {len(CITY_COORDS)} locations worldwide"
    )

When an MCP host connects, it reads the resource list during the initialization handshake. The agent can then read weather://supported-cities on demand — useful when the host needs to know what cities are available before asking for a temperature.

Resources return plain text by default, but you can also return structured data:

@mcp.resource("weather://config", mime_type="application/json")
def server_config() -> str:
    """Returns the server configuration as JSON."""
    import json
    config = {"version": "0.2.0", "cities": list(CITY_COORDS.keys()), "units": ["celsius", "fahrenheit"]}
    return json.dumps(config)

When clients request this resource at weather://config, the application/json MIME type signals that the content is parseable JSON. MCP hosts use the MIME type to decide whether to display, parse, or store the resource.

Step 6: Add a Streaming Tool

Some operations take time — think database migration status checks, file downloads, or multi-step orchestrations. MCP supports streaming responses through mcp.tool() with the async for pattern.

Here’s a streaming tool that checks weather for a list of cities and yields results incrementally:

@mcp.tool()
async def get_temperatures_batch(cities: str, unit: str = "celsius") -> str:
    """Get temperatures for multiple cities at once.

    Args:
        cities: A comma-separated list of city names (e.g., 'London, Tokyo, Berlin').
        unit: Temperature unit — 'celsius' or 'fahrenheit'.
    """
    city_list = [c.strip() for c in cities.split(",")]
    if not city_list:
        return "No cities specified."

    results = []
    for city in city_list:
        city_lower = city.lower()
        if city_lower not in CITY_COORDS:
            results.append(f"  ⚠ {city}: not supported")
            continue

        lat, lon = CITY_COORDS[city_lower]
        url = "https://api.open-Meteo.com/v1/forecast"
        params = {
            "latitude": lat,
            "longitude": lon,
            "current": "temperature_2m",
            "temperature_unit": unit,
        }

        try:
            async with httpx.AsyncClient() as client:
                response = await client.get(url, params=params, timeout=10.0)
                response.raise_for_status()
            data = response.json()
            temp = data["current"]["temperature_2m"]
            unit_symbol = "°C" if unit == "celsius" else "°F"
            results.append(f"  ✓ {city.title()}: {temp}{unit_symbol}")
        except Exception as e:
            results.append(f"  ✗ {city.title()}: error ({e})")

    return "\n".join(results)


if __name__ == "__main__":
    mcp.run()

For a fully streaming tool that yields intermediate results as they arrive (rather than batching at the end), use FastMCP’s task system with the tasks extra:

uv pip install "fastmcp[tasks]"

Then define a long-running task:

from fastmcp import Task


@mcp.tool()
async def monitor_weather(cities: str, unit: str = "celsius") -> str:
    """Monitor temperatures across cities with intermediate updates.

    Returns a task token that the client can poll for status.

    Args:
        cities: Comma-separated city names.
        unit: 'celsius' or 'fahrenheit'.
    """
    city_list = [c.strip() for c in cities.split(",")]
    return f"Monitoring {len(city_list)} cities in {unit}..."

Tasks let the client poll for progress on a long-running operation — the client receives a task token and can check completion status without holding the stdio connection open. This is essential for operations that exceed standard timeout windows.

Complete Server

Here’s the full, runnable server — no truncation:

from fastmcp import FastMCP
import httpx
import json
from datetime import datetime

mcp = FastMCP(
    name="weather-server",
    version="0.3.0",
    description="Real-time weather via Open-Meteo API with streaming support",
)

CITY_COORDS = {
    "london": (51.5074, -0.1278),
    "tokyo": (35.6762, 139.6503),
    "new york": (40.7128, -74.0060),
    "paris": (48.8566, 2.3522),
    "sydney": (-33.8688, 151.2093),
    "mumbai": (19.0760, 72.8777),
    "berlin": (52.5200, 13.4050),
}


async def _fetch_weather(lat: float, lon: float, unit: str) -> str:
    """Internal helper to call Open-Meteo."""
    url = "https://api.open-meteo.com/v1/forecast"
    params = {
        "latitude": lat,
        "longitude": lon,
        "current": "temperature_2m",
        "temperature_unit": unit,
    }
    async with httpx.AsyncClient() as client:
        response = await client.get(url, params=params, timeout=10.0)
        response.raise_for_status()
    data = response.json()
    temp = data["current"]["temperature_2m"]
    unit_symbol = "°C" if unit == "celsius" else "°F"
    timestamp = data["current"]["time"]
    return f"{temp}{unit_symbol} (as of {timestamp})"


@mcp.tool()
def list_supported_cities() -> str:
    """Returns the list of cities this server supports."""
    return "Supported cities: " + ", ".join(c.title() for c in CITY_COORDS)


@mcp.tool()
async def get_temperature(city: str, unit: str = "celsius") -> str:
    """Get the current temperature for a city.

    Args:
        city: The city name (e.g., 'London', 'Tokyo').
        unit: Temperature unit — 'celsius' or 'fahrenheit'.
    """
    city_lower = city.lower()
    if city_lower not in CITY_COORDS:
        cities = ", ".join(CITY_COORDS.keys())
        return f"Unknown city: {city}. Supported cities: {cities}."

    lat, lon = CITY_COORDS[city_lower]
    temp = await _fetch_weather(lat, lon, unit)
    return f"It is currently {temp} in {city.title()} (source: Open-Meteo)."


@mcp.tool()
async def get_temperatures_batch(cities: str, unit: str = "celsius") -> str:
    """Get temperatures for multiple cities at once.

    Args:
        cities: A comma-separated list of city names (e.g., 'London, Tokyo, Berlin').
        unit: Temperature unit — 'celsius' or 'fahrenheit'.
    """
    city_list = [c.strip() for c in cities.split(",")]
    if not city_list:
        return "No cities specified."

    results = []
    for city in city_list:
        city_lower = city.lower()
        if city_lower not in CITY_COORDS:
            results.append(f"  ⚠ {city}: not supported")
            continue

        lat, lon = CITY_COORDS[city_lower]
        try:
            temp = await _fetch_weather(lat, lon, unit)
            results.append(f"  ✓ {city.title()}: {temp}")
        except Exception as e:
            results.append(f"  ✗ {city.title()}: error ({e})")

    return "\n".join(results)


@mcp.resource("weather://supported-cities")
def supported_cities_resource() -> str:
    """Returns the list of cities as a resource."""
    lines = [f"- {name.title()}" for name in CITY_COORDS]
    return "Supported cities:\n" + "\n".join(lines)


@mcp.resource("weather://config", mime_type="application/json")
def server_config_resource() -> str:
    """Returns the server configuration as JSON."""
    config = {
        "version": "0.3.0",
        "cities": list(CITY_COORDS.keys()),
        "units": ["celsius", "fahrenheit"],
        "source": "Open-Meteo",
    }
    return json.dumps(config)


if __name__ == "__main__":
    mcp.run()

Step 7: Production Considerations

Running fastmcp run with stdio transport works locally, but production deployments need different choices.

HTTP (SSE) Transport

For remote access, run the server over HTTP with Server-Sent Events:

if __name__ == "__main__":
    mcp.run(transport="sse", host="0.0.0.0", port=8000)

This starts an ASGI server on port 8000. Clients connect via http://your-server:8000/sse. This is the transport you want when your MCP server lives behind a load balancer or needs to serve multiple concurrent clients.

Version Pinning

FastMCP follows semantic versioning with the caveat that breaking changes may occur in minor versions when the MCP protocol itself evolves. Pin to exact versions in production (FastMCP versioning policy):

fastmcp==3.0.0  # production — pinned
fastmcp>=3.0.0  # risky — may pull breaking changes

Logging

MCP servers communicate over stdio or SSE. Anything you print goes to the client as protocol data and will break the connection. Use the mcp logger instead:

import logging

logger = logging.getLogger("mcp")

async def _fetch_weather(lat, lon, unit):
    logger.info(f"Fetching weather for ({lat}, {lon}) in {unit}")
    # ... API call

What to Build Next

This server covers the three MCP primitives — tools (the get_temperature and get_temperatures_batch functions), resources (weather://supported-cities, weather://config), and the server lifecycle. From here, the patterns scale naturally:

• Database tools — wrap psycopg/SQLAlchemy queries in @mcp.tool() functions with parameterized inputs.
• File system resources — serve configuration files, documentation, or logs via @mcp.resource() with appropriate MIME types.
• MCP composition — the A2A protocol lets your MCP server call other agents as sub-components, creating a server-of-servers architecture. Once you’re comfortable with single-server patterns, multi-agent composition is the logical next step.

For a broader view of where MCP fits in the agent protocol stack — including how it relates to A2A, Agent Cards, and gateway patterns — see our AI Agent Protocol Stack 2026.