On the current Python SDK (mcp 2.3.0) a working MCP server is one decorated function: type hints become the schema, the docstring becomes the description the model reads. Since spec revision 2026-07-28 there is no handshake and no session, so the work that is left is the tool contract: descriptions, errors, auth and safe retries.
Numbers follow one tool from definition to result. Blue is the request, amber the model, green the result. Pans sideways on a phone.
Install the SDK that speaks 2026-07-28
from mcp.server.fastmcp import FastMCP- Handshake revisions only, up to 2025-11-25
- What most older tutorials show
from mcp.server import MCPServer- Serves 2026-07-28 and the older handshake on one server
- 2.3.0 is current (PyPI, 2026-10-02)
Rule of thumb: build new servers on 2.x. On 2.x the old import fails with ModuleNotFoundError: No module named 'mcp.server.fastmcp'; pin mcp<2 only to keep old code running.
FastMCP is now three different things: the 1.x class, its 2.x successor MCPServer, and fastmcp, Prefect's separate framework built on mcp 2. This post uses the official SDK (PyPI, Python 3.10+). Building MCP Servers covers the TypeScript twin.
uv init orders && cd orders
uv add "mcp[cli]" # 2.x: the line that speaks 2026-07-28
uv run server.py # stdio: waits for a client on stdin
The smallest server: one function, one decorator
from mcp.server import MCPServer
mcp = MCPServer("orders")
@mcp.tool()
def get_order(order_id: str) -> str:
"""Look up one order by id: status, total, shipping state.
Do NOT use for refunds; call create_refund instead.
"""
return f"{order_id}: shipped, 42.00 USD"
if __name__ == "__main__":
mcp.run()
Save it as server.py. mcp.run() serves over stdio: the host starts the script and speaks JSON-RPC on stdin and stdout, so never print() to stdout at import time (MCP Transport). This is what a client gets from tools/list, trimmed:
{
"name": "get_order",
"description": "Look up one order by id: status, total, …",
"inputSchema": {
"type": "object",
"properties": {"order_id": {"type": "string"}},
"required": ["order_id"]}}
The function name is the tool name, the docstring is the description, and order_id: str without a default is a required string. The host hands those three fields to the model as a tool definition; it never sees JSON-RPC (Inside a Tool Call).
The docstring is the product
| Weak | Strong | Why |
|---|---|---|
| “Gets order data.” | “Look up one order by id: status, total, shipping.” | Says when, not only what |
| No negative case | “Do NOT use for refunds; call create_refund.” | Often the most valuable line |
data: dict | order_id: OrderId | A loose dict invites invented keys |
| One tool per endpoint | One tool per job | Fewer, clearer choices |
The model reads the description out of context and decides from it alone, so treat it as a prompt. Selection degrades as the catalogue grows; Anthropic's docs put the knee near 30–50 tools, and Tool Surface Design owns those numbers. A reworded description is a behaviour change even when the schema is identical: re-run your selection eval (step 07).
Types that reject bad input, errors the model can act on
from typing import Annotated, Literal
from pydantic import BaseModel, Field
from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import ToolAnnotations
mcp = MCPServer("orders")
OrderId = Annotated[str, Field(pattern=r"^A-\d{2,6}$",
description="e.g. A-91")]
class Order(BaseModel):
id: str
status: Literal["placed", "shipped", "delivered"]
total_cents: int
ORDERS = {"A-91": Order(id="A-91", status="shipped", total_cents=4_200)}
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
def get_order(order_id: OrderId) -> Order:
"""Look up one order by id: status, total, shipping state.
Do NOT use for refunds; call create_refund instead.
"""
if order_id not in ORDERS:
raise ToolError("No such order. Ask the user to check the id.")
return ORDERS[order_id]
The pattern and the Literal land in the schema, and the SDK checks every call against it before your function runs. A typed return (-> Order) adds an outputSchema, and results carry structuredContent beside the text. Now a write:
REFUNDS: dict[str, str] = {} # per process: use your database in production
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=False,
idempotentHint=True))
def create_refund(
order_id: OrderId,
amount_cents: Annotated[int, Field(gt=0)],
idempotency_key: Annotated[str, Field(min_length=8)],
) -> str:
"""Refund part or all of one order. On a retry, reuse the same key."""
if idempotency_key in REFUNDS:
return REFUNDS[idempotency_key]
order = ORDERS.get(order_id)
if order is None or amount_cents > order.total_cents:
raise ToolError("Unknown order, or more than its total. Not retryable.")
REFUNDS[idempotency_key] = f"refund rf_{len(REFUNDS) + 1} for {order_id}"
return REFUNDS[idempotency_key]
{"name": "get_order", "input": {"order_id": "91"}}isError: true
Error executing tool get_order: 1 validation error for get_orderArguments
order_id
String should match pattern '^A-\d{2,6}$' [type=string_pattern_mismatch, input_value='91', input_type=str]pydantic's own text, first three lines: it names the rule{"name": "create_refund",
"input": {"order_id": "A-91", "amount_cents": 9000, …}}isError: true
Error executing tool create_refund: Unknown order, or more than its total. Not retryable.your ToolError text, written for the modelBoth failures come back as results with isError: true, not protocol errors, so the model can read them and correct itself. Raise ToolError for failures you expect and say what to do next; any other exception reaches the model as a bare Error executing tool. JSON-RPC errors are for an unknown tool or a malformed request (MCP Architecture).
The pattern catches typos, not another customer's well-formed order id. Check every id against the caller's identity on the server, never against an argument the model fills in (step 08).
Serve it over HTTP: every request stands alone
Change the last line, and the same server answers Streamable HTTP on one endpoint, POST only:
if __name__ == "__main__":
mcp.run(transport="streamable-http") # http://127.0.0.1:8000/mcp
tools/call on the wire, captured from the server abovePOST /mcp
MCP-Protocol-Version: 2026-07-28
Mcp-Method: tools/call
Mcp-Name: get_order
{"jsonrpc": "2.0", "id": 7, "method": "tools/call",
"params": {"name": "get_order",
"arguments": {"order_id": "A-91"},
"_meta": {
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
"io.modelcontextprotocol/clientCapabilities": {}}}}the first and only message: no initialize before itHTTP/1.1 200 OK (no Mcp-Session-Id header)
{"jsonrpc": "2.0", "id": 7, "result": {
"resultType": "complete", "isError": false,
"structuredContent": {"id": "A-91", "status": "shipped", "total_cents": 4200},
"content": [{"type": "text", "text": "…same, as JSON text…"}]}}any replica could have answeredHTTP 400 {"code": -32020, "message": "mcp-name header does not match the request body's 'name' parameter"}same body, but the header said Mcp-Name: create_refundThat request is the whole conversation. Since 2026-07-28 there is no initialize handshake and no Mcp-Session-Id: each request names its protocol version and client capabilities in _meta, and the headers repeat the method and tool name so a gateway can route without parsing the body. The body is the truth; a header that disagrees is a 400 (MCP Architecture). A client, pinned to the new revision:
import asyncio
from mcp import Client
async def main():
url = "http://127.0.0.1:8000/mcp"
async with Client(url, mode="2026-07-28") as c:
print(c.protocol_version)
r = await c.call_tool("get_order", {"order_id": "A-91"})
print(r.is_error, r.structured_content)
asyncio.run(main())
It prints 2026-07-28, then False {'id': 'A-91', 'status': 'shipped', 'total_cents': 4200}. The default mode="auto" probes server/discover and falls back to the handshake for older servers; the pin refuses them.
Retries land on any replica, so make writes idempotent
Blue is the request, purple the shared state. The in-process REFUNDS dict above only works with one replica.
Statelessness is what lets you run three replicas behind a round-robin load balancer. It also means a dropped stream is a lost request: the client sends it again with a new id, possibly to another replica. Two POSTs of the same create_refund above both returned refund rf_1, because the key, not the connection, identifies the write. Keep that table in the database every replica reads (MCP in Production).
A stateless protocol is not a stateless application. To confirm a write with the user mid-call, the server returns an input_required result and the client retries with the answer; requestState carries the context, sealed with a key every replica shares (State Without Sessions).
Test it without a model, then with one
# test_server.py - uv add --dev pytest anyio; uv run pytest
import pytest
from mcp import Client
from server import mcp
@pytest.mark.anyio
async def test_contract():
async with Client(mcp) as c: # in memory: no process, no network
names = [t.name for t in (await c.list_tools()).tools]
bad = await c.call_tool("get_order", {"order_id": "91"})
big = await c.call_tool("create_refund", {
"order_id": "A-91", "amount_cents": 9_000,
"idempotency_key": "rf-a91-002"})
assert names == ["get_order", "create_refund"]
assert bad.is_error and big.is_error
Client(mcp) connects in memory: no subprocess, no port, no model. Assert what you promise, the tool list and that bad input is an error result. Then add what unit tests cannot cover, a selection eval: fixed requests, each with the tool a model should pick first, run on every description change (Grading Agents). For a manual look, npx @modelcontextprotocol/inspector drives the server by hand.
Point Claude at it
# local: Claude Code starts the process and talks over stdio
claude mcp add --transport stdio orders -- uv run server.py
# remote: the same server over Streamable HTTP
claude mcp add --transport http orders https://mcp.example.com/mcp
Claude Code warns when one tool result passes 10,000 tokens and caps it at 25,000 by default (Claude Code docs), so page or summarise large results. The Messages API connector is the other route: Anthropic's API becomes the MCP client, it carries tool calls only, and it reaches only a public HTTPS URL (MCP with Claude).
From works to deployable
| Gap | What breaks | Fix |
|---|---|---|
| Auth | Any token, or the user's token forwarded upstream | Accept only tokens issued for this server; never pass them on |
| Records | One tenant reads another's order | Authorize each id from the validated token |
| Origin | A web page drives a local server | Bind 127.0.0.1, set allowed origins |
| Writes | A retry refunds twice | Idempotency key in shared storage |
| Supply chain | A changed description steers the model | Pin and review what you install |
A remote server is an OAuth resource server: it answers an unauthenticated request with a 401 that points to its metadata, and accepts only tokens whose audience is itself. Token passthrough is forbidden by the spec; the payments call uses the server's own credential (MCP Authorization). Origin and host checks: MCP Transport. Poisoned descriptions: MCP Security.
“Build an MCP server for a customer's order system. What changes between the demo and production?” Tools by job, typed so bad input fails as a readable error, descriptions that say when not to call. Then the four things the SDK does not do: authorize each record from the token, never forward that token, idempotent writes because a retry can land anywhere, and a selection eval on every description change.
A tutorial says from mcp.server.fastmcp import FastMCP. You run pip install mcp and it fails. Why, and what are your two options?
pip installs mcp 2.x, where FastMCP was renamed MCPServer and the old module raises ModuleNotFoundError. Port it to from mcp.server import MCPServer (and get 2026-07-28), or pin mcp<2 to keep the old code on the handshake revisions.
The stream drops halfway through create_refund and the client retries on another replica. What stops a double refund, and why is the REFUNDS dict not enough?
The idempotency key: the second call finds the first result and returns it. The dict lives in one process, so replica B cannot see replica A's entry; the keyed table has to be in storage every replica reads.
Your HTTP server receives the user's bearer token and sends the same token to the payments API. What is wrong?
That is token passthrough, which the spec forbids. The server must accept only tokens issued for itself and call payments with its own credential, scoped to the user it validated.
- Use mcp 2.x:
from mcp.server import MCPServer.mcp.server.fastmcpis gone. - Type hints and the docstring become the tool; the description is a prompt, so write when not to call.
- Expected failures are
ToolErrorresults the model can act on, not exceptions. - On 2026-07-28 every request stands alone: any replica answers, so writes need an idempotency key in shared storage.
- Auth, record checks and Origin rules are yours; never forward the caller's token.