Build an MCP server in Python with the official SDK

The SDK landscape in 2026

The official Python SDK (mcp on PyPI) reached v2 alongside the 2026-07-28 spec. v2 supports every protocol revision, renames the high-level server class to MCPServer (from mcp.server import MCPServer), and reworks several APIs. v1 (whose high-level class was FastMCP, imported from mcp.server.fastmcp) lives on a maintenance branch with security fixes. pip install mcp now installs 2.x, so pin mcp>=1.28,<2 if you are not ready to migrate, and use the official migration guide when you are. The separate community project also called FastMCP continues independently; this lesson uses the official SDK.

Requirements: Python 3.10+. Install with the CLI extra: uv add "mcp[cli]" or pip install "mcp[cli]".

Anatomy of a high-level server

• MCPServer(name, instructions=..., lifespan=...) creates the server. instructions tell hosts how to use it.
• @mcp.tool() turns a type-hinted function into a tool. Type hints and Pydantic Field metadata become the input schema; the docstring becomes the description; a Pydantic return type becomes an output schema with structured content.
• @mcp.resource("scheme://{param}") registers a resource or template.
• @mcp.prompt() registers a prompt.
• Context (injected when a parameter is typed ctx: Context) gives access to request info, elicitation, progress, reading resources and lifespan state.
• raise ToolError(...) returns a model-readable error (isError: true); MCPError produces a protocol error.
• mcp.run() serves stdio; mcp.run(transport="streamable-http", port=...) serves HTTP (defaults to 127.0.0.1 and path /mcp).

Managing resources with lifespan

Database pools and API clients should be created once, not per call. Use an async context manager as the lifespan; tools read it from ctx.request_context.lifespan_context.

Worked example: a campaign-performance server for a PK/UAE agency

Requirements from the account team:

• Look up campaigns by client and market.
• Get performance for a date range (spend, impressions, clicks, conversions, CPA).
• Expose a glossary of metric definitions as a resource so the model explains metrics consistently.
• Provide a /weekly-summary prompt.
• All read-only; currency shown per market (PKR, AED).

Hands-on: the full server

import os
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
from datetime import date
from typing import Annotated, Literal

import asyncpg                                   # pip install asyncpg (or use sqlite3 for local tests)
from pydantic import BaseModel, Field
from mcp.server import MCPServer
from mcp.server.mcpserver import Context
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import ToolAnnotations

@dataclass
class AppCtx:
    pool: asyncpg.Pool

@asynccontextmanager
async def lifespan(server: MCPServer) -> AsyncIterator[AppCtx]:
    pool = await asyncpg.create_pool(os.environ["ANALYTICS_DSN"], min_size=1, max_size=5)
    try:
        yield AppCtx(pool=pool)
    finally:
        await pool.close()

mcp = MCPServer(
    "campaign-analytics",
    instructions="Read-only campaign analytics. Always state the date range and currency. "
                 "Use find_campaigns before get_performance.",
    lifespan=lifespan,
)

Market = Literal["PK", "AE", "SA", "UK", "US"]
READ_ONLY = ToolAnnotations(read_only_hint=True, open_world_hint=False)

class Campaign(BaseModel):
    campaign_id: str
    name: str
    channel: str
    market: Market

class CampaignList(BaseModel):
    campaigns: list[Campaign]
    total: int

class Performance(BaseModel):
    campaign_id: str
    start: date
    end: date
    currency: str
    spend: float
    impressions: int
    clicks: int
    conversions: int
    cpa: float | None = Field(description="Spend / conversions; null if no conversions")

@mcp.tool(title="Find campaigns", annotations=READ_ONLY)
async def find_campaigns(
    client_id: Annotated[str, Field(description="Client slug, e.g. 'noor-cosmetics'")],
    ctx: Context[AppCtx],
    market: Market | None = None,
    limit: Annotated[int, Field(ge=1, le=50)] = 20,
) -> CampaignList:
    """List a client's campaigns, optionally filtered by market. Returns IDs for get_performance."""
    pool = ctx.request_context.lifespan_context.pool
    rows = await pool.fetch(
        "SELECT id, name, channel, market FROM campaigns WHERE client_id=$1 "
        "AND ($2::text IS NULL OR market=$2) ORDER BY name LIMIT $3", client_id, market, limit)
    return CampaignList(campaigns=[Campaign(campaign_id=r["id"], name=r["name"], channel=r["channel"],
                                            market=r["market"]) for r in rows], total=len(rows))

@mcp.tool(title="Get performance", annotations=READ_ONLY)
async def get_performance(campaign_id: str, start: date, end: date, ctx: Context[AppCtx]) -> Performance:
    """Aggregate performance for one campaign between two dates (inclusive, max 92 days)."""
    if end < start or (end - start).days > 92:
        raise ToolError("Date range must be valid and at most 92 days; split longer ranges.")
    pool = ctx.request_context.lifespan_context.pool
    r = await pool.fetchrow(
        "SELECT c.currency, SUM(spend) spend, SUM(impressions) imp, SUM(clicks) clk, SUM(conversions) conv "
        "FROM daily_stats s JOIN campaigns c ON c.id=s.campaign_id "
        "WHERE s.campaign_id=$1 AND s.day BETWEEN $2 AND $3 GROUP BY c.currency", campaign_id, start, end)
    if r is None:
        raise ToolError(f"No data for {campaign_id} in that range. Check the ID with find_campaigns.")
    conv = int(r["conv"] or 0)
    return Performance(campaign_id=campaign_id, start=start, end=end, currency=r["currency"],
                       spend=float(r["spend"]), impressions=int(r["imp"]), clicks=int(r["clk"]),
                       conversions=conv, cpa=(float(r["spend"]) / conv) if conv else None)

@mcp.resource("glossary://metrics", mime_type="text/markdown")
def metrics_glossary() -> str:
    """Definitions of the metrics this server returns."""
    return ("- **CPA**: spend divided by conversions.\n- **CTR**: clicks divided by impressions.\n"
            "- Currency is the campaign's billing currency; never convert unless asked.")

@mcp.prompt(title="Weekly summary")
def weekly_summary(client_id: str) -> str:
    """Summarize last week's performance for a client."""
    return (f"For client {client_id}: call find_campaigns, then get_performance for last Monday to Sunday "
            "for each active campaign. Use the metrics glossary. Output a table and three insights.")

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

Testing in memory

The v2 Client connects directly to your server object, no subprocess or port:

import pytest
from mcp import Client
from server import mcp

@pytest.fixture
def anyio_backend():
    return "asyncio"

@pytest.mark.anyio
async def test_rejects_long_ranges():
    async with Client(mcp) as client:
        result = await client.call_tool("get_performance",
            {"campaign_id": "x", "start": "2026-01-01", "end": "2026-06-30"})
        assert result.is_error

(For tests that hit the database, provide a test DSN or refactor data access behind an interface you can fake.)

Connecting it

uv run mcp dev server.py opens the Inspector. uv run mcp install server.py writes a Claude Desktop entry. For Claude Code: claude mcp add campaign-analytics -- uv run --with "mcp[cli]" mcp run /absolute/path/server.py. Use absolute paths; hosts launch servers from their own working directory with a minimal environment, so pass secrets such as ANALYTICS_DSN via the host's env configuration (e.g., mcp install ... -v ANALYTICS_DSN=... or -f .env).

Pitfalls

• print() in a stdio server; use logging (stderr).
• Creating a DB connection per call instead of via lifespan.
• Unbounded queries; always cap rows and date ranges.
• Following v1 tutorials with v2 installed (or vice versa); check your installed major version.

Measuring success

Unit-test coverage of tools, p95 latency per tool, error rate, and results size (tokens) per call.