uv add "mcp[cli]"

Requires Python ≥ 3.10. The `[cli]` extra adds the `mcp` CLI for development tooling.

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    "MyServer",
    stateless_http=True,   # recommended for production HTTP
    json_response=True,    # recommended for scalability
)

from pydantic import BaseModel, Field

class WeatherData(BaseModel):
    temperature: float = Field(description="Temperature in Celsius")
    condition: str
    humidity: float

@mcp.tool()
def get_weather(city: str, unit: str = "celsius") -> WeatherData:
    """Get current weather for a city.

    Returns structured weather data validated against WeatherData schema.
    """
    # Implementation calls a real weather API
    return WeatherData(temperature=22.5, condition="sunny", humidity=45.0)

@mcp.resource("file://documents/{name}")
def read_document(name: str) -> str:
    """Read a document by name from the document store."""
    # Read from disk, DB, or cache
    return f"Content of {name}"

@mcp.resource("config://settings")
def get_settings() -> str:
    """Return current application settings as JSON."""
    return '{"theme": "dark", "debug": false}'

from mcp.server.fastmcp.prompts import base

@mcp.prompt(title="Code Review")
def review_code(code: str, language: str = "python") -> list[base.Message]:
    """Generate a structured code review prompt."""
    return [
        base.UserMessage(f"Please review this {language} code:"),
        base.UserMessage(f"```{language}\n{code}\n```"),
        base.AssistantMessage("I'll analyze the code for correctness, style, and potential issues."),
    ]

from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession

@mcp.tool()
async def long_running_task(
    task_name: str,
    steps: int,
    ctx: Context[ServerSession, None],
) -> str:
    """Run a multi-step task with progress reporting."""
    await ctx.info(f"Starting task: {task_name}")

    for i in range(steps):
        await ctx.report_progress(
            progress=(i + 1) / steps,
            total=1.0,
            message=f"Step {i + 1} of {steps}",
        )

    await ctx.info("Task complete")
    return f"Completed {task_name}"

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass

import httpx
from mcp.server.fastmcp import Context, FastMCP

@dataclass
class AppState:
    http_client: httpx.AsyncClient

@asynccontextmanager
async def lifespan(server: FastMCP) -> AsyncIterator[AppState]:
    async with httpx.AsyncClient() as client:
        yield AppState(http_client=client)

mcp = FastMCP("MyServer", lifespan=lifespan)

@mcp.tool()
async def fetch_url(url: str, ctx: Context) -> str:
    """Fetch content from a URL using the shared HTTP client."""
    state: AppState = ctx.request_context.lifespan_context
    response = await state.http_client.get(url)
    return response.text

@mcp.tool()
def divide(a: float, b: float) -> float:
    """Divide a by b."""
    if b == 0:
        raise ValueError("Division by zero is not allowed")
    return a / b

from mcp.types import CallToolResult, TextContent

@mcp.tool()
def safe_parse(data: str) -> CallToolResult:
    """Parse data, returning errors inline rather than raising."""
    try:
        result = parse(data)
        return CallToolResult(
            content=[TextContent(type="text", text=str(result))]
        )
    except ParseError as exc:
        return CallToolResult(
            content=[TextContent(type="text", text=f"Parse failed: {exc}")],
            isError=True,
        )

undefined

| Pattern | Recommendation |
|---------|---------------|
| Transport (production) | Streamable HTTP with `stateless_http=True, json_response=True` |
| Transport (local/stdio) | stdio via `mcp.run()` or `uv run mcp run server.py` |
| I/O tools | Use `async def` |
| Shared state | Use lifespan context |
| Structured output | Return Pydantic `BaseModel` subclass |
| Progress reporting | Use `ctx.report_progress()` |
| Secrets/config | Pass via environment variables, not hardcoded |