3. Building servers with the Python SDK¶

Intermediate · 16 min read

Page 2 built the protocol by hand. The official SDK does all of that for you: it turns type hints into JSON schemas, validates arguments, speaks both protocol eras, and runs over stdio or HTTP. You write plain Python functions.

pip install "mcp>=2"          # the official Python SDK; add [cli] for the `mcp dev` / `mcp install` commands

mcp 1.x vs 2.x

In mcp 2.x the server class is MCPServer (from mcp.server.mcpserver import MCPServer). In 1.x it was FastMCP (from mcp.server.fastmcp import FastMCP), and most blog posts still use that name. The decorators (@mcp.tool(), @mcp.resource(), @mcp.prompt()) work the same way. The blocks on this page run in order like a notebook; they're marked "no-run" because they need pip install mcp, and the outputs shown are from a real run with mcp 2.3.

3.1 Your first server¶

# no-run — pip install "mcp>=2"
from mcp.server.mcpserver import MCPServer

mcp = MCPServer("orders", instructions="Order lookups and refunds for ShopKart support agents.")

ORDERS = {"A100": {"status": "shipped", "eta": "Friday", "total_inr": 2499},
          "A101": {"status": "processing", "eta": "Monday", "total_inr": 899}}

@mcp.tool()
def get_eta(order_id: str) -> str:
    """Delivery ETA of an order. Use when the user asks when an order will arrive."""
    return ORDERS[order_id]["eta"]

# To serve it, save this as orders_server.py and end the file with:
#     if __name__ == "__main__":
#         mcp.run()                    # stdio by default

That's a complete MCP server. The function name becomes the tool name, the docstring becomes the description, and the type hints become the input schema.

3.2 Test it in-process¶

You don't need Claude Desktop to try a server. mcp.Client accepts the server object directly and connects in memory, which is how you should unit-test servers:

# no-run
import asyncio
import json
from mcp import Client

async def main():
    async with Client(mcp) as client:
        print("protocol:", client.protocol_version)
        for tool in (await client.list_tools()).tools:
            print(tool.name, "-", tool.description)
            print("  input schema:", json.dumps(tool.input_schema))
        result = await client.call_tool("get_eta", {"order_id": "A100"})
        print("result:", result.content[0].text, "| isError:", result.is_error)

asyncio.run(main())

Output

protocol: 2026-07-28
get_eta - Delivery ETA of an order. Use when the user asks when an order will arrive.
  input schema: {"type": "object", "properties": {"order_id": {"title": "Order Id", "type": "string"}}, "required": ["order_id"], "title": "get_etaArguments"}
result: Friday | isError: False

3.3 Structured output¶

Return a Pydantic model (or a TypedDict or dataclass) and the SDK publishes an output schema and fills structuredContent, so client code gets typed JSON instead of parsing text:

# no-run
from typing import Literal
from pydantic import BaseModel

class OrderStatus(BaseModel):
    order_id: str
    status: Literal["processing", "shipped", "delivered", "cancelled"]
    eta: str

@mcp.tool()
def get_order_status(order_id: str) -> OrderStatus:
    """Shipping status of an order. Use when the user gives an order ID like A100."""
    order = ORDERS[order_id]
    return OrderStatus(order_id=order_id, status=order["status"], eta=order["eta"])

async def main():
    async with Client(mcp) as client:
        tool = next(t for t in (await client.list_tools()).tools if t.name == "get_order_status")
        print("output schema fields:", list(tool.output_schema["properties"]))
        result = await client.call_tool("get_order_status", {"order_id": "A101"})
        print("structured:", result.structured_content)
        print("text for the model:", result.content[0].text.replace("\n", " "))

asyncio.run(main())

Output

output schema fields: ['order_id', 'status', 'eta']
structured: {'order_id': 'A101', 'status': 'processing', 'eta': 'Monday'}
text for the model: {   "order_id": "A101",   "status": "processing",   "eta": "Monday" }

A function returning a plain str or int also gets an output schema, wrapped as {"result": ...}.

3.4 Errors the model can use¶

There are three kinds of failure, and the SDK treats them differently:

What happened	What the model sees	Server log
You raise `ToolError("...")`	`isError: true` with your message	INFO, no traceback
Arguments fail the schema (wrong type, `Field` limits)	`isError: true` with the validation message	INFO
Any other exception (a bug, a `KeyError`)	`isError: true` with only `Error executing tool <name>`	ERROR with traceback

The third row is deliberate: a crash's message might contain internal details (SQL, paths, secrets), so it stays on the server. Raise ToolError for failures you expect, with a message that tells the model how to fix the call:

# no-run
from pydantic import Field
from mcp.server.mcpserver.exceptions import ToolError

@mcp.tool()
def refund_order(order_id: str,
                 reason: Literal["damaged", "late", "wrong_item"],
                 amount_inr: int = Field(gt=0, le=50_000, description="Refund amount in rupees")) -> str:
    """Refund an order. Use only after confirming the order exists."""
    if order_id not in ORDERS:
        raise ToolError(f"order {order_id} not found: IDs look like A100")
    if amount_inr > ORDERS[order_id]["total_inr"]:
        raise ToolError(f"refund {amount_inr} is more than the order total {ORDERS[order_id]['total_inr']}")
    return f"refund rf_{order_id} of Rs {amount_inr} created ({reason})"

async def main():
    async with Client(mcp) as client:
        for args in [{"order_id": "A100", "reason": "late", "amount_inr": 500},
                     {"order_id": "Z9", "reason": "late", "amount_inr": 500},
                     {"order_id": "A101", "reason": "late", "amount_inr": 5000},
                     {"order_id": "A100", "reason": "late", "amount_inr": -5}]:
            result = await client.call_tool("refund_order", args)
            print(f"isError={result.is_error}: {result.content[0].text.splitlines()[0]}")
        crash = await client.call_tool("get_eta", {"order_id": "Z9"})      # KeyError inside the tool
        print(f"isError={crash.is_error}: {crash.content[0].text}")

asyncio.run(main())

Output

isError=False: refund rf_A100 of Rs 500 created (late)
isError=True: Error executing tool refund_order: order Z9 not found: IDs look like A100
isError=True: Error executing tool refund_order: refund 5000 is more than the order total 899
isError=True: Error executing tool refund_order: 1 validation error for refund_orderArguments
isError=True: Error executing tool get_eta

3.5 Resources and URI templates¶

Resources are read-only data the host can attach as context. A fixed URI is a resource; a URI with {placeholders} is a resource template, and the placeholders become function arguments:

# no-run
@mcp.resource("policy://refunds")
def refund_policy() -> str:
    """The current refund policy."""
    return "Refunds within 7 days of delivery. Damaged items: full refund."

@mcp.resource("orders://{order_id}")
def order_record(order_id: str) -> dict:
    """The full order record as JSON."""
    return {"order_id": order_id, **ORDERS[order_id]}

async def main():
    async with Client(mcp) as client:
        print("resources:", [r.uri for r in (await client.list_resources()).resources])
        print("templates:", [t.uri_template for t in (await client.list_resource_templates()).resource_templates])
        policy = await client.read_resource("policy://refunds")
        print(policy.contents[0].text)
        record = await client.read_resource("orders://A101")
        print(record.contents[0].text.replace("\n", " "))

asyncio.run(main())

Output

resources: ['policy://refunds']
templates: ['orders://{order_id}']
Refunds within 7 days of delivery. Damaged items: full refund.
{   "order_id": "A101",   "status": "processing",   "eta": "Monday",   "total_inr": 899 }

Tool or resource? If the model should decide when to fetch it, make it a tool. If the app or user picks it (a file, a policy document to pin into context), make it a resource. Many hosts only support tools well, so important data is often exposed as both.

3.6 Prompts¶

Prompts are reusable templates the user triggers, often shown as slash commands in the host:

# no-run
@mcp.prompt()
def apology_email(customer: str, issue: str) -> str:
    """Draft a short apology email to a customer."""
    return f"Write a 3-sentence apology email to {customer} about: {issue}. Offer one concrete next step."

async def main():
    async with Client(mcp) as client:
        prompt = (await client.list_prompts()).prompts[0]
        print(prompt.name, "args:", [a.name for a in prompt.arguments])
        filled = await client.get_prompt("apology_email", {"customer": "Asha", "issue": "late delivery"})
        print(filled.messages[0].role, "->", filled.messages[0].content.text)

asyncio.run(main())

Output

apology_email args: ['customer', 'issue']
user -> Write a 3-sentence apology email to Asha about: late delivery. Offer one concrete next step.

3.7 Progress for slow tools¶

Add a parameter typed Context and the SDK injects it (it doesn't appear in the input schema). Use it to report progress on long jobs, so the host can show a progress bar instead of a frozen spinner:

# no-run
from mcp.server.mcpserver import Context

@mcp.tool()
async def export_orders(month: str, ctx: Context) -> str:
    """Export all orders for a month (YYYY-MM) to CSV."""
    batches = 3
    for i in range(1, batches + 1):
        await asyncio.sleep(0.01)                      # pretend to fetch a batch
        await ctx.report_progress(i, batches, f"batch {i}/{batches}")
    return f"exported orders-{month}.csv"

async def on_progress(progress, total, message):
    print(f"  progress {progress:.0f}/{total:.0f}: {message}")

async def main():
    async with Client(mcp) as client:
        tool = next(t for t in (await client.list_tools()).tools if t.name == "export_orders")
        print("ctx hidden from schema:", list(tool.input_schema["properties"]))
        result = await client.call_tool("export_orders", {"month": "2026-09"}, progress_callback=on_progress)
        print(result.content[0].text)

asyncio.run(main())

Output

ctx hidden from schema: ['month']
  progress 1/3: batch 1/3
  progress 2/3: batch 2/3
  progress 3/3: batch 3/3
exported orders-2026-09.csv

For logs, write to stderr (or OpenTelemetry). MCP's own logging feature is deprecated in the 2026-07-28 spec.

3.8 Asking the user: Resolve and Elicit¶

Some tools need the user's input mid-call: a confirmation, a missing detail. In mcp 2.x you declare that need as a resolver. The SDK runs it before your tool body, asks the client, and injects the answer. It uses the multi round-trip pattern on 2026-07-28 connections and the older mid-call request on older ones, so the same code works with both.

# no-run
from typing import Annotated
from mcp import types
from mcp.server.mcpserver import Elicit, Resolve, ElicitationResult, AcceptedElicitation

class Confirm(BaseModel):
    confirm: bool

def ask_to_confirm(order_id: str) -> Elicit[Confirm]:
    return Elicit(f"Really cancel order {order_id}? This can't be undone.", Confirm)

@mcp.tool()
def cancel_order(order_id: str,
                 answer: Annotated[ElicitationResult[Confirm], Resolve(ask_to_confirm)]) -> str:
    """Cancel an order. The user is asked to confirm first."""
    if isinstance(answer, AcceptedElicitation) and answer.data.confirm:
        ORDERS[order_id]["status"] = "cancelled"
        return f"{order_id} cancelled"
    return f"{order_id} kept: the user did not confirm"

async def ask_user(context, params):                    # the host side: show the form, return the answer
    print("  host asks:", params.message)
    return types.ElicitResult(action="accept", content={"confirm": True})

async def main():
    tool = next(t for t in await mcp.list_tools() if t.name == "cancel_order")
    print("model only sees:", list(tool.input_schema["properties"]))
    for mode in ["auto", "legacy"]:                     # 2026-07-28, then the 2025 handshake
        async with Client(mcp, mode=mode, elicitation_callback=ask_user) as client:
            result = await client.call_tool("cancel_order", {"order_id": "A100"})
            print(f"{client.protocol_version}: {result.content[0].text}")

asyncio.run(main())

Output

model only sees: ['order_id']
  host asks: Really cancel order A100? This can't be undone.
2026-07-28: A100 cancelled
  host asks: Really cancel order A100? This can't be undone.
2025-11-25: A100 cancelled

The answer parameter never appears in the schema, so the model can't fake the user's confirmation. That makes elicitation a good place for approval gates on destructive tools. Only ask for simple values (booleans, strings, numbers, enums), and never for passwords or API keys: the spec forbids requesting secrets through form elicitation.

3.9 Shared resources with lifespan¶

Database pools and HTTP clients should be created once when the server starts, not on every call. Pass a lifespan context manager; what it yields is available on every request:

# no-run
from contextlib import asynccontextmanager
from dataclasses import dataclass

@dataclass
class AppState:
    db: dict                                            # stand-in for an asyncpg pool or httpx client

@asynccontextmanager
async def lifespan(server):
    print("  startup: open DB pool")
    try:
        yield AppState(db={"customers": 1204})
    finally:
        print("  shutdown: close DB pool")

stats = MCPServer("stats", lifespan=lifespan)

@stats.tool()
def customer_count(ctx: Context) -> int:
    """Number of customers in the database."""
    state: AppState = ctx.request_context.lifespan_context
    return state.db["customers"]

async def main():
    async with Client(stats) as client:
        result = await client.call_tool("customer_count", {})
        print("customers:", result.structured_content["result"])

asyncio.run(main())

Output

  startup: open DB pool
customers: 1204
  shutdown: close DB pool

3.10 Run and inspect it¶

Save the server as orders_server.py with mcp.run() at the bottom, then:

python orders_server.py                       # stdio: a host starts this command for you
mcp dev orders_server.py                      # opens the MCP Inspector in your browser (needs Node.js)
npx @modelcontextprotocol/inspector python orders_server.py    # the same Inspector, without the mcp CLI

The MCP Inspector is the Postman of MCP: it lists tools, resources and prompts, lets you call them with your own arguments, and shows the raw JSON-RPC messages. Use it before connecting a real host, so you know whether a bug is in your server or in the host.

Tool design checklist¶

Do	Why
Name tools `verb_noun` and write the docstring as instructions: what, when, returns	It's the only thing the model reads
Use `Literal`, `Field(ge=, le=)` and Pydantic models for arguments	Bad calls fail before your code runs
Return compact structured output	Saves context tokens; clients can use it without parsing
Raise `ToolError` with a fix-it message	The model can retry correctly
Keep tools few and distinct	Overlapping tools confuse the model (tool selection)
Make write tools idempotent and confirm destructive ones	Retries and hallucinated calls are safe

Interview questions¶

How does an MCP server SDK build a tool's schema?

From the function signature: the name becomes the tool name, the docstring the description, and the type hints (via Pydantic) the JSON input schema, including Literal enums and Field limits. Parameters typed as Context, or filled by resolvers, are left out of the schema. The return type becomes the output schema.

Why does the SDK hide the message of unexpected exceptions from the client?

Exception text can leak internals (SQL, file paths, tokens) into the model's context and possibly to the user. Expected failures should be raised as ToolError with a safe, actionable message; anything else returns a generic error to the client and logs the full traceback on the server.

When would you expose something as a resource instead of a tool?

When it's read-only data the application or user chooses to attach (a file, a policy, a record), rather than something the model decides to fetch mid-reasoning. Tools are model-controlled; resources are application-controlled. Because host support for resources varies, important data is often exposed as both.

How do you get a human confirmation inside an MCP tool?

Use elicitation: the server asks the client to show a small form ("Really cancel A100?"). In the Python SDK you declare it with a Resolve(...) parameter returning Elicit(message, schema); the SDK handles the protocol (multi round-trip on 2026-07-28, a mid-call request on older versions). The answer isn't in the input schema, so the model can't fabricate it.

How do you test an MCP server?

Unit-test it in-process with Client(server): list tools, check the schemas, call tools and assert on structured_content and is_error. Use the MCP Inspector for manual exploration, then test end to end with a real host and a few scripted conversations.

Practice¶

Add a list_orders(status) tool that returns a list of OrderStatus and check its output schema.
Make ask_user in 3.8 return action="decline" and check the order is kept.
Add a resource template customers://{customer_id}/orders and read it with the client.

Next: 4. Clients and hosts Back to: MCP overview · Notes overview