Building Python MCP Servers

Building Python MCP Servers

MCP servers expose tools to an LLM client (Claude Desktop, Claude Code, etc.). The LLM is the caller, so the failure modes differ from a normal library: errors must be machine-readable, every input is untrusted, and a green test suite often proves nothing about whether the tools actually work. This skill encodes the patterns that recur when these go wrong.

Quick Start (FastMCP)

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("my-server")

@mcp.tool()
def read_config(path: str) -> dict:
    """Read a config file. `path` must be absolute."""
    p = Path(path)
    if not p.is_absolute():
        return {"error": "path must be absolute"}
    if not p.exists():
        return {"error": f"no such file: {path}"}
    return {"data": p.read_text()}

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

Error Contract: return, don't raise — and stay consistent

An uncaught exception surfaces to the LLM as an opaque protocol error it can't reason about. Return a structured result with a predictable shape instead, and make callers check for it.

• Pick one error shape and use it everywhere. A dict with an "error" key is the common convention. Document that callers must check for it.
• Batch tools must report skips, not swallow them. The most common inconsistency: a single-item tool collects per-item errors, but a sibling "do this across a directory" tool silently continues past files that fail to load. For automation that is invisible data loss. Every batch tool should return both results and a per-item skipped/errors list.

@mcp.tool()
def validate_dir(path: str) -> dict:
    results, skipped = {}, []
    for f in Path(path).glob("*.md"):
        try:
            results[f.name] = _validate(f)
        except Exception as e:
            skipped.append({"file": f.name, "error": str(e)})  # never silently continue
    return {"results": results, "skipped": skipped}

Type footgun: YAML auto-parses an unquoted ISO date (date: 2025-06-15) into a datetime.date, not a str or datetime.datetime. A validator that handles only str/datetime will false-positive on native YAML dates. When validating parsed values, enumerate every type the parser can actually produce.

Wrapping a subprocess / CLI

A huge share of MCP servers shell out to another tool. Three failures recur:

1. Don't discard stdout on a non-zero exit. Many CLIs exit non-zero by design (a linter/mutation-tester reporting findings) and write their real output to stdout with an empty stderr. A wrapper that returns f"Error: {result.stderr}" whenever returncode != 0 reports a successful run to the LLM as an empty "Error: ".

def run_tool(args: list[str]) -> dict:
    r = subprocess.run(args, capture_output=True, text=True)
    return {                       # hand BOTH streams to the model; let it judge
        "returncode": r.returncode,
        "stdout": r.stdout,
        "stderr": r.stderr,
    }

2. Pin to the version you actually wrap, and verify subcommands exist. A server written against a tool's 2.x CLI while the project pins 3.x will call subcommands and flags that no longer exist — every wrapped tool breaks at runtime. Check the installed version's --help, not your memory of it.

3. Parse args safely. Splitting an extra-args string with str.split() breaks quoted, space-containing arguments — use shlex.split(). Never interpolate a client-supplied string into a shell command; pass an argv list to subprocess.run (no shell=True). When a tool accepts a target/path, remember the LLM (or content it read) chose it — validate it.

No module-level global state (it makes the server untestable)

Parsing CLI args at import time and stashing them in module globals (WORKING_DIR, MAKEFILE_PATH, caches…) forces every test to del sys.modules["server"] and re-import under a patched sys.argv just to reset state — brittle and easy to get wrong. Keep configuration in an object or pass it through; construct tools from a factory.

def build_server(config: Config) -> FastMCP:
    mcp = FastMCP("my-server")

    @mcp.tool()
    def do_thing(x: str) -> dict:
        return {"result": _work(x, config)}   # config captured, not global

    return mcp

This also avoids double registration: a module-level "create all tools" loop plus the same loop inside main() registers every tool twice when the file is run directly (uv run server.py, as Claude Desktop does) versus via a console entry point. Register in exactly one place.

Keep blocking work off the protocol event loop

Do not assume a framework moves synchronous tool functions to a worker thread. Some FastMCP runtimes invoke them inline on the protocol event loop. A SQLite query, filesystem walk, dependency traversal, or synchronous HTTP call that takes five seconds can therefore block pings and every unrelated request for the same five seconds.

Make the tool async and move only the blocking boundary to a thread:

import asyncio

@mcp.tool()
async def find_dependents(item_id: int) -> dict:
    rows = await asyncio.to_thread(repository.find_dependents, item_id)
    return {"items": [row.to_dict() for row in rows]}

Keep connection ownership in mind. Do not create a SQLite connection on the event-loop thread and hand that connection to the worker. Open and close it inside repository.find_dependents, or use a pool/driver whose concurrency contract explicitly permits the handoff. A thread wrapper around a shared, thread-affine connection merely trades event-loop starvation for intermittent database errors.

Test responsiveness, not just the slow tool's result. Start a deliberately blocked repository call, invoke a lightweight tool (or protocol ping) before releasing it, and require the lightweight request to finish first:

slow = asyncio.create_task(call_tool("find_dependents", {"item_id": 42}))
await entered_worker.wait()

healthy = await asyncio.wait_for(call_tool("health", {}), timeout=0.2)
assert healthy == {"ok": True}

release_worker.set()
await slow

A timing assertion on the slow call alone cannot detect event-loop starvation; the regression is that independent protocol traffic stops making progress.

Sampling is an optional client capability — contain failures in the tool

ctx.sample(...) is not guaranteed to work just because the tool itself was called successfully. The connected client may not support sampling, or its sampling handler may raise while processing the request. Those are different failure modes at the framework layer, but they are the same tool-level outcome: the requested analysis could not be produced.

Catch the exception around the sampling boundary inside the tool and convert it to the server's normal error shape. Do not rely on the framework's outer exception wrapper; by then the caller receives an opaque protocol/tool error instead of your documented contract.

async def sample_or_error(ctx: Context, prompt: str) -> dict:
    try:
        response = await ctx.sample(prompt)
    except Exception as exc:
        return {"error": f"sampling failed: {exc}"}
    return {"result": response.text or ""}

Keep the try block narrow so unrelated programming errors are not mislabeled as sampling failures. Test both boundaries explicitly: a client with no sampling support, and a configured sampling handler that raises. Also test an empty sampling response if the tool promises an empty-string or other fallback.