Building MCP Servers

Interview answer (say this first). An MCP server is a program that speaks JSON-RPC over a transport and advertises three kinds of capability: tools (callable actions), resources (read-only URIs), and prompts (reusable templates). You register each one, and the framework derives the tool’s input schema from its type hints and its output schema from its return type. You return structured results, raise a clean tool error for expected failures, log for operators, and choose a transport — stdio for local processes, streamable HTTP for remote. You test a server by connecting a client to it in-process as well as over the real transport.

Why this exists

Writing a custom integration for every AI host does not scale. Each host expects a different function signature, a different schema format, and a different error convention. An MCP server writes the capability once and exposes it through a standard protocol, so any compliant host can discover and call it.

The problems a server has to solve are concrete:

1. Discovery. The client must learn what exists, without a hard-coded list. That is the job of tools/list, resources/list, and prompts/list.

2. Contract. The client and the model must know each tool’s arguments and types. That is the job of the generated JSON Schema.

3. Errors. A tool can fail for expected reasons (bad input, not found) and unexpected reasons (a crashed dependency). Expected failures should come back as a clear, model-readable error; unexpected ones should not leak stack traces.

4. Operability. Someone has to see what the server is doing. The server needs logging, progress, and a health path.

A server that ignores these produces the classic failure:

Tool:  def do_stuff(x): return something_maybe
Schema: x is untyped -> the model sends a string where a number was needed.
Error:  a raw traceback is returned, the model retries the same bad call forever.

Nothing here is exotic, but each omission is a production incident waiting to happen. Building a server means making each of these explicit.

Note:

The one-sentence purpose. An MCP server turns your code into a discoverable, typed, observable capability that any compliant client can call safely.

Start from zero

Three ideas to hold:

• The schema is derived, not written. Type hints and the return type are the source of truth. Annotate carefully.
• A tool error is a result, not a crash. Expected failures come back through the protocol so the model can react.
• stdio for local, HTTP for remote. The same server object runs over both.

The core idea

Think of a restaurant. The menu is the advertised tool list. The kitchen is your code. The waiter is the protocol: the order arrives as JSON, the kitchen cooks, and a plate goes back. Good restaurants also explain the dish (description), specify options (schema), and say “we’re out of that” politely (tool error) instead of shouting a stack trace.

flowchart TD
    A["MCPServer(name, version)"] --> B["@mcp.tool()"]
    A --> C["@mcp.resource(uri)"]
    A --> D["@mcp.prompt()"]
    B --> E["input schema<br/>from parameter types"]
    B --> F["output schema<br/>from return type"]
    A --> G["run(transport='stdio' or 'streamable-http')"]
    G --> H["client initializes,<br/>lists, and calls"]
    H --> I["result or ToolError<br/>back to the caller"]

The registration decorators are the menu. The framework does the JSON-RPC plumbing and the schema generation. Your job is to make the contract precise and the behavior safe.

How it works

1. You create the server. A name, a version, and optional instructions for clients.
2. You register tools. Each function’s parameters become the input schema; the docstring becomes the description.
3. You register resources and prompts. Fixed URIs and templates for reads; prompt functions for reusable text.
4. The server starts on a transport. stdio for a local process, streamable HTTP for remote.
5. The client initializes. Capabilities are negotiated from what you registered.
6. The client lists. It receives your tools, resources, and prompts with their schemas.
7. A call arrives. The server validates the arguments against the input schema; invalid arguments produce an error result.
8. Your function runs. It may use ctx for logging and progress.
9. The result is validated. If the return type declares a model, the result is checked and returned as structuredContent.
10. Errors are shaped. A ToolError becomes a clean error result; an unexpected exception is caught and reported without ending the session.

The last step matters most for reliability. A tool that raises an uncontrolled exception should not take down the whole server; the framework isolates it and returns is_error=True, and the agent loop can decide what to do.

The syntax you will use

Examples use the official Python SDK, mcp 2.2.0. In 2.x the high-level class is MCPServer; it was FastMCP in 1.x, and importing the old path now raises a migration error.

Create the server. Version and instructions help clients and operators.

from mcp.server.mcpserver import MCPServer

mcp = MCPServer(
    name="demo-server",
    version="1.0.0",
    instructions="Tools for internal document search.",
)

Register a tool. Types come from the signature; the docstring becomes the description.

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b

Prefer flat parameters for a flat schema. The framework makes one field per parameter.

@mcp.tool()
def search_docs(query: str, top_k: int = 5) -> str:
    """Search internal documents."""
    return run_search(query, top_k)

A single Pydantic parameter nests under its name. Verified in 2.x: the schema exposes one property called args.

from pydantic import BaseModel, Field

class SearchArgs(BaseModel):
    query: str = Field(description="What to search for.")
    top_k: int = Field(default=5, ge=1, le=50)

@mcp.tool()
def search_with_model(args: SearchArgs) -> str:
    """Search with one structured argument object."""
    return run_search(args.query, args.top_k)

# Schema: {"properties": {"args": {"$ref": "#/$defs/SearchArgs"}}, ...}
# Call:   {"args": {"query": "billing", "top_k": 3}}

Register a resource and a template. Resources are reads, addressed by URI.

@mcp.resource("docs://handbook")
def handbook() -> str:
    """The employee handbook."""
    return "Welcome to the handbook."

@mcp.resource("users://{user_id}/profile")
def profile(user_id: str) -> str:
    """Return a user profile by id."""
    return f"Profile for {user_id}"

Register a prompt. Prompts are user-triggered, not model tools.

@mcp.prompt()
def review(code: str, language: str = "python") -> str:
    """Ask for a code review."""
    return f"Review this {language} code:\n{code}"

Return a model for structured output. The return type becomes the output schema.

class User(BaseModel):
    name: str
    plan: str

@mcp.tool()
def get_user(user_id: str) -> User:
    """Look up a user by id."""
    return User(name="Ada", plan="pro")

# output_schema: {name: string, plan: string}
# structured_content: {"name": "Ada", "plan": "pro"}

Raise a clean tool error for expected failures. This becomes is_error=True with a readable message.

from mcp.server.mcpserver.exceptions import ToolError

@mcp.tool()
def divide(a: int, b: int) -> float:
    """Divide a by b."""
    if b == 0:
        raise ToolError("b must not be zero")
    return a / b

Use the context for logging and progress. In 2.x on protocol 2026-07-28 these emit deprecation warnings; prefer server-side logs for new code.

from mcp.server.mcpserver import Context

@mcp.tool()
async def index_docs(path: str, ctx: Context) -> str:
    """Index documents and report progress."""
    await ctx.info(f"indexing {path}")
    await ctx.report_progress(0, 1, "starting")
    # ... work ...
    await ctx.report_progress(1, 1, "done")
    return "indexed"

Share expensive resources with a lifespan.

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(server: MCPServer) -> AsyncIterator[dict]:
    pool = await open_pool()
    try:
        yield {"pool": pool}
    finally:
        await pool.close()

mcp = MCPServer("app", version="1.0.0", lifespan=lifespan)

Run over stdio or HTTP. stdio for local; HTTP for remote. Both are blocking calls.

mcp.run(transport="stdio")                 # local process
mcp.run(transport="streamable-http")       # remote, bound to 127.0.0.1:8000 by default

run(transport="streamable-http", ...) forwards tuning options such as host, port, and stateless_http as keywords. (run_streamable_http_async is async def, so passing it bare to anyio.run raises TypeError; use the sync run, or wrap it.)

mcp.run(transport="streamable-http", host="0.0.0.0", port=8000, stateless_http=True)

# Async form: wrap it, do not pass it to anyio.run bare.
# import anyio
# anyio.run(lambda: mcp.run_streamable_http_async(host="0.0.0.0", port=8000, stateless_http=True))

Test in-process. The client can connect directly to the server object — no subprocess, no port.

import asyncio
from mcp import Client

async def test_add():
    async with Client(mcp) as client:
        result = await client.call_tool("add", {"a": 2, "b": 3})
        assert result.structured_content == {"result": 5}

asyncio.run(test_add())

Examples: simple to real

Example 1 — the smallest useful server. One tool, one transport, and it works.

from mcp.server.mcpserver import MCPServer

mcp = MCPServer("tiny", version="0.1.0")

@mcp.tool()
def ping() -> str:
    """Return pong."""
    return "pong"

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

Everything else in this page is refinement. This is the shape to memorize.

Example 2 — list it with a real client. Discovery proves the registration worked.

async with Client(mcp) as client:
    tools = await client.list_tools()
    print([(t.name, t.input_schema.get("required", [])) for t in tools.tools])
# [('ping', [])]  — a no-parameter tool has no "required" key

If a tool does not appear, it was never registered — check the decorator and that the function is defined at module level.

Example 3 — validation is automatic and safe. Bad arguments become an error result, not an exception in the client.

await client.call_tool("add", {"a": "two", "b": 3})
# is_error = True
# message (abridged): "Error executing tool add: 1 validation error for addArguments\n
#                      a\n  Input should be a valid integer, unable to parse string as an integer ..."

The session stays alive. The agent can read the message, fix the call, and retry. This is the behavior you want.

Example 4 — structured output beats parsing text. A declared return type gives the caller typed JSON.

class Order(BaseModel):
    id: str
    total: float

@mcp.tool()
def get_order(order_id: str) -> Order:
    """Fetch an order."""
    return repo.load(order_id)

# structured_content: {"id": "A-1", "total": 42.5}

The client validates against output_schema instead of scraping prose. Downstream agent code stays type-safe.

Example 5 — a resource template replaces N tool calls. Read many objects through one discoverable pattern.

resources/templates/list -> users://{user_id}/profile
resources/read users://42/profile -> "Profile for 42"
resources/read users://99/profile -> "Profile for 99"

One registration, unbounded instances. Keep reference reads as resources so the tool catalog stays small.

Example 6 — test in-process, then verify over stdio. Fast unit tests, plus one real transport test.

# Fast: no subprocess.
async with Client(mcp) as client:
    assert (await client.call_tool("ping", {})).content[0].text == "pong"

# Real: exercises process startup and the actual wire.
from mcp import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client
params = StdioServerParameters(command="python", args=["server.py"])
async with stdio_client(params) as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()
        assert (await session.call_tool("ping", {})).content[0].text == "pong"

The in-process test catches logic bugs. The stdio test catches packaging, import, and startup bugs that never appear in-process.

In production

• Annotate every parameter and return type. The schema is generated from annotations. An untyped parameter becomes Any, and the model is free to send the wrong thing.
• Write descriptions as instructions. Say what the tool does, when to use it, when not to, and whether it writes. The docstring is what the model reads.
• Prefer flat parameters unless you need one object. In 2.x a single Pydantic parameter nests under its name, which surprises clients. Verified behavior: one model parameter means one object argument.
• Return models, not prose, when the caller will use the data. A declared return type produces output_schema and structuredContent.
• Raise ToolError for expected failures. It becomes a readable, safe error result. Let unexpected exceptions be caught by the framework so the session survives.
• Never leak stack traces or secrets in errors. Error text goes to the model and the logs. Sanitize it.
• Use a lifespan for pools and clients. A shared database pool or HTTP client belongs to the server, not to each request or session.
• Log to stderr, never to stdout, on stdio. stdout is the JSON-RPC channel. A stray print corrupts the protocol.
• Bound long work and report progress. Long calls hit client timeouts. Progress notifications and a sane timeout keep the agent alive.
• Pin the SDK major version. MCP 2.x renamed FastMCP to MCPServer and changed handler APIs. pip install "mcp<2" keeps v1 code running; migrating means changing imports and handler signatures.
• Test both in-process and over the real transport. In-process finds logic bugs; stdio and HTTP find wiring bugs.
• Version your tools and keep names stable. Renaming a tool breaks stored prompts, evals, and callers. Migrate deliberately.

Interview questions

1. What are the three primitives an MCP server can expose?

Answer. Tools, resources, and prompts. Tools are callable actions the model may invoke. Resources are read-only content addressed by URI, and resource templates make one URI pattern serve many instances. Prompts are reusable message templates, usually triggered by the user. A server advertises which primitives it supports during initialize.

Follow-up: “Which one does the model choose automatically?” Tools, when the host offers them for selection. Prompts are typically user-invoked, and resources are fetched by the application.

Trap. Exposing read-only data as a tool. It works, but it adds to the model’s selection burden. Resources are the better home for reference content.

2. Where does the tool’s JSON Schema come from?

Answer. From the function signature. Each parameter becomes a property with its annotated type, defaults become defaults, and required parameters are those without defaults. The docstring becomes the description. The return annotation, if it is a model, becomes the output schema.

Follow-up: “What happens to an untyped parameter?” It becomes a permissive Any-like field, so the model can send anything and validation cannot protect you. Annotations are part of the contract, not decoration.

Trap. Assuming the schema is inferred from the function body. It is inferred from the annotations; the body is never inspected for types.

3. How should a server report an expected failure?

Answer. Raise a ToolError with a clear, safe message. The framework turns it into an error result with is_error=True, the session stays alive, and the model can read the message and retry differently. Unexpected exceptions are caught at the boundary so one bad tool does not kill the connection.

Follow-up: “What should the error text contain?” What went wrong and what the caller can do about it. Never internal paths, stack traces, or secrets, because the text reaches the model and the logs.

Trap. Letting every exception propagate raw, or returning errors as normal success text. The caller cannot distinguish success from failure.

4. Why does structured output matter?

Answer. Because downstream code should not parse prose. If a tool declares a model return type, the server publishes an output_schema and returns structuredContent that the client can validate. That makes results machine-checkable and keeps the agent’s data handling type-safe.

Follow-up: “Are primitives also structured?” Yes — a primitive return is wrapped in a {"result": ...} object. The schema is published either way.

Trap. Returning a formatted string and parsing it later. It works until the format changes, and then it fails silently.

5. What is the lifespan for, and what does it not do?

Answer. It is a server-wide startup and shutdown hook. Open expensive shared resources — database pools, HTTP clients — on enter and close them on exit. It is not per session and not per request, so it must not hold user-specific state.

Follow-up: “Where does per-session state go?” Externalized in a store keyed by session, or in the session lifecycle. A lifespan object is shared by every caller.

Trap. Storing a user’s data in the lifespan. One user’s state then leaks into another user’s requests.

6. Why must a stdio server never log to stdout?

Answer. Because stdout is the JSON-RPC channel. Any non-protocol output — a print, a library banner — is parsed as a message and corrupts the stream. Log to stderr instead, which the host captures separately.

Follow-up: “How do you debug then?” Write to stderr and read the host’s captured logs, or use in-process tests where you can print freely. Never mix diagnostics with the protocol channel.

Trap. Adding a debug print during development and leaving it in. It works locally until the output happens to look like a message, then fails unpredictably.

7. How do you choose between stdio and streamable HTTP?

Answer. Use stdio when the server is local and runs as the user, such as a filesystem or repository tool. It needs no auth and no network. Use streamable HTTP when the server is remote, shared, or needs to survive client restarts. HTTP adds auth, sessions, and scaling concerns, but it is the only option across machines.

Follow-up: “What changes in the code?” Mostly the run(...) call. The tools and resources are identical; the transport changes lifecycle, auth, and session behavior.

Trap. Assuming a stdio server can be exposed over HTTP unchanged and safely. A local tool often has no authentication because it relies on local trust; exposing it remotely without auth is a breach.

8. How do you test an MCP server?

Answer. Two layers. First, connect a client to the server object in-process and assert on structured results and errors — fast and no subprocess. Second, run the server over the real transport (stdio or HTTP) and connect a real client, which catches startup, packaging, and transport bugs. Test the schema too: the model sees it, so it is part of the interface.

Follow-up: “What should the tests assert on?” structured_content for correctness, is_error for failure paths, and the published input_schema for contract stability. Snapshot the schema so an accidental rename fails loudly.

Trap. Testing only list_tools and never calling a tool. Listing proves registration, not behavior.

Remember this

• Register tools, resources, and prompts; the schema is derived from annotations.
• Annotate everything — untyped parameters defeat validation.
• Return models for structured output; raise ToolError for expected failures.
• stdio for local, streamable HTTP for remote; never log to stdout on stdio.
• In 2.x, FastMCP is now MCPServer. Pin the major version and test in-process plus over the wire.