Build an MCP server in Python

The complete example is in python/ of the examples repository.

Files in the repository

Every snippet on this page comes from these files; open them to see the whole thing.

• python/server.py
• python/shop.py
• python/requirements.txt

Install

python -m venv .venv
.venv/bin/pip install "mcp==2.3.0" httpx

Python 3.10 or later. Version 2 of the SDK renamed the high-level server class to MCPServer (it was FastMCP in version 1; the separate fastmcp package on PyPI is a different project).

The upstream client

shop.py wraps the Sample Shop API with httpx, reads UPSTREAM_URL and UPSTREAM_API_KEY from the environment, and raises one UpstreamError with a kind of invalid_input, not_found, conflict or unavailable. Tools never see headers or status codes.

Tools are typed functions

MCPServer turns a function's signature into the tool's input schema: parameter types become JSON Schema types, defaults make parameters optional, and Annotated[..., Field(...)] adds descriptions and constraints.

from typing import Annotated, Any, Literal

from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import ToolAnnotations
from pydantic import Field

mcp = MCPServer("sample-shop", version="1.0.0", instructions="...")

@mcp.tool(
    title="Search products",
    description="Search the shop catalogue by phrase and optional category. Returns one page of up to `limit` products and a `nextCursor`; when nextCursor is not null, more results exist and you must say so rather than claiming the list is complete. Use get_product for full details of one item. This tool does not place orders.",
    annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=False),
)
def search_products(
    query: Annotated[str, Field(max_length=100, description="Words to match against product names and descriptions.")] = "",
    category: Annotated[Literal["lighting", "furniture", "kitchen"] | None, Field(description="Restrict results to one category.")] = None,
    limit: Annotated[int, Field(ge=1, le=20, description="Page size.")] = 5,
    cursor: Annotated[str | None, Field(description="Opaque token from a previous page's nextCursor. Never invent one.")] = None,
) -> dict[str, Any]:
    try:
        page = shop.search_products(q=query, category=category, limit=limit, cursor=cursor)
    except UpstreamError as error:
        raise _explain(error) from error
    return {"items": [_summary(p) for p in page["items"]], "nextCursor": page["nextCursor"]}

Returning a dict gives the client both a text rendering and structuredContent. Parameter names are camelCase (productId, idempotencyKey, orderId) in the other tools to match the shared interface; that is deliberate, not a style slip.

Errors the assistant can act on

Raising ToolError produces an isError result with your message as its text. _explain turns the four upstream kinds into the four instructions (adjust input, do not retry this id, tell the user about the conflict, wait):

def _explain(error: UpstreamError) -> ToolError:
    if error.kind == "invalid_input":
        return ToolError(f"Invalid input: {error} Adjust the arguments and call again.")
    if error.kind == "not_found":
        return ToolError(f"Not found: {error} Do not retry with the same identifier; ask the user to check it.")
    if error.kind == "conflict":
        return ToolError(f"Conflict: {error} Do not retry automatically; tell the user what happened.")
    return ToolError(f"The shop service is unavailable right now: {error} Wait before retrying, and tell the user.")

The guarded cancel_order uses the same mechanism: without confirm=True it fetches the order and raises a ToolError describing what would be cancelled and asking for confirmation.

Serve it stateless, with a health route

streamable_http_app() returns a Starlette application. The example adds a /health route and runs it with uvicorn, which the SDK already depends on:

if __name__ == "__main__":
    import uvicorn
    from starlette.responses import JSONResponse

    app = mcp.streamable_http_app(stateless_http=True, host=HOST)
    app.add_route("/health", lambda request: JSONResponse({"ok": True}))
    uvicorn.run(app, host=HOST, port=PORT, log_level="warning")

mcp.run("streamable-http", host=..., port=...) does the same without the extra route.

Run and test

node sample-api/server.mjs &
cd python && .venv/bin/python server.py
# in another shell
cd tests && npm ci && MCP_URL=http://127.0.0.1:3002/mcp npm test

Next: connect it to Claude or ChatGPT, or see what production needs.