Example server

A complete, runnable reference implementation of a remote MCP server. It demonstrates every pattern in the Guide: sync tools (add, echo), an async job (start_render → get_render_status → get_render_result), file delivery over HTTP (GET /files/{id}), and a health check (GET /healthz).

:material-download: Download example-mcp-server.zip

Run it

cd example-mcp-server
uv sync                      # or: pip install mcp
python server.py             # serves MCP at http://127.0.0.1:8900/mcp

curl -s http://127.0.0.1:8900/healthz            # -> {"ok": true}
python smoke_test.py http://127.0.0.1:8900/mcp   # lists tools, runs the job, downloads a file

server.py

#!/usr/bin/env python3
"""example-mcp — a minimal but complete remote MCP server (Streamable HTTP).

Demonstrates every pattern in the guide:
  - sync tools:  add, echo
  - async job:   start_render -> get_render_status -> get_render_result
  - file output: get_render_result returns a URL served by GET /files/{job_id}
  - health:      GET /healthz

Run:  python server.py      (serves MCP at http://127.0.0.1:8900/mcp)
Env:  MCP_HOST (default 127.0.0.1), MCP_PORT (8900), PUBLIC_BASE_URL, OUTPUT_DIR
"""
from __future__ import annotations

import os
import threading
import time
import uuid
from pathlib import Path

from mcp.server.fastmcp import FastMCP
from starlette.requests import Request
from starlette.responses import FileResponse, JSONResponse

MCP_HOST = os.getenv("MCP_HOST", "127.0.0.1")   # bind localhost; expose via Cloudflare Tunnel
MCP_PORT = int(os.getenv("MCP_PORT", "8900"))
PUBLIC_BASE_URL = os.getenv("PUBLIC_BASE_URL", f"http://{MCP_HOST}:{MCP_PORT}")
OUTPUT_DIR = Path(os.getenv("OUTPUT_DIR", "/tmp/example-mcp-files"))
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

mcp = FastMCP("example-mcp", host=MCP_HOST, port=MCP_PORT)

# In-memory job store. DEMO ONLY — use a DB/redis/file for production (survives restarts).
JOBS: dict[str, dict] = {}


# ---- sync tools ---------------------------------------------------------------
@mcp.tool()
def add(a: float, b: float) -> dict:
    """Add two numbers and return {"sum": a+b}."""
    return {"sum": a + b}


@mcp.tool()
def echo(text: str) -> dict:
    """Echo text back as {"text": ...}. Validates input is a non-empty string."""
    if not isinstance(text, str) or not text:
        return {"error": "text must be a non-empty string"}
    return {"text": text}


# ---- async job (start -> status -> result), with file output ------------------
def _run_render(job_id: str, spec: dict) -> None:
    try:
        JOBS[job_id] |= {"state": "running"}
        time.sleep(2)  # pretend this is heavy work (video/render/GPU/etc.)
        out = OUTPUT_DIR / f"{job_id}.txt"
        out.write_text(f"rendered with spec={spec}\n", encoding="utf-8")
        JOBS[job_id] |= {"state": "succeeded", "path": str(out)}
    except Exception as e:  # noqa: BLE001
        JOBS[job_id] |= {"state": "failed", "error": str(e)}


@mcp.tool()
def start_render(spec: dict | None = None) -> dict:
    """Start a long render job. Returns {job_id, state} immediately (async)."""
    job_id = uuid.uuid4().hex
    JOBS[job_id] = {"state": "queued"}
    threading.Thread(target=_run_render, args=(job_id, spec or {}), daemon=True).start()
    return {"job_id": job_id, "state": "queued"}


@mcp.tool()
def get_render_status(job_id: str) -> dict:
    """Return {state, error?} for a render job. state in queued|running|succeeded|failed."""
    j = JOBS.get(job_id)
    if not j:
        return {"error": f"no such job: {job_id}"}
    return {"job_id": job_id, "state": j["state"], "error": j.get("error")}


@mcp.tool()
def get_render_result(job_id: str) -> dict:
    """Return {ready, url} when succeeded. url is downloadable over HTTP (not a local path)."""
    j = JOBS.get(job_id) or {}
    if j.get("state") != "succeeded":
        return {"ready": False, "state": j.get("state")}
    return {"ready": True, "url": f"{PUBLIC_BASE_URL}/files/{job_id}"}


# ---- plain HTTP routes (not MCP tools) ----------------------------------------
@mcp.custom_route("/healthz", methods=["GET"])
async def healthz(request: Request):
    return JSONResponse({"ok": True})


@mcp.custom_route("/files/{job_id}", methods=["GET"])
async def serve_file(request: Request):
    job_id = request.path_params["job_id"]
    j = JOBS.get(job_id)
    if not j or j.get("state") != "succeeded" or not j.get("path"):
        return JSONResponse({"error": "not ready"}, status_code=404)
    return FileResponse(j["path"], media_type="text/plain", filename=f"{job_id}.txt")


def main() -> None:
    mcp.run(transport="streamable-http")  # MCP endpoint mounted at /mcp


if __name__ == "__main__":
    main()

smoke_test.py

#!/usr/bin/env python3
"""Minimal self-contained smoke test for a remote MCP server.

Connects over Streamable HTTP, lists tools, calls `add`, then runs the async render
job (start -> poll status -> get result URL) and downloads the file.

Usage:
  python smoke_test.py http://127.0.0.1:8900/mcp
  python smoke_test.py https://example-mcp.<zone>/mcp \
      --header "CF-Access-Client-Id: $ID" --header "CF-Access-Client-Secret: $SECRET"

Deps: mcp  (pip install mcp)
"""
from __future__ import annotations

import argparse
import asyncio
import json
import sys

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client


def _payload(result):
    """Extract a tool result as Python data (structured content or JSON text)."""
    if getattr(result, "structuredContent", None):
        return result.structuredContent
    text = "\n".join(c.text for c in result.content if getattr(c, "type", "") == "text")
    try:
        return json.loads(text)
    except Exception:
        return text


def _headers(items: list[str]) -> dict[str, str]:
    out = {}
    for it in items or []:
        k, _, v = it.partition(":")
        out[k.strip()] = v.strip()
    return out


async def run(url: str, headers: dict[str, str]) -> None:
    async with streamablehttp_client(url, headers=headers or None) as (r, w, _):
        async with ClientSession(r, w) as s:
            await s.initialize()
            tools = [t.name for t in (await s.list_tools()).tools]
            print("tools:", tools)

            print("add(2,3) ->", _payload(await s.call_tool("add", {"a": 2, "b": 3})))

            start = _payload(await s.call_tool("start_render", {"spec": {"demo": True}}))
            job_id = start["job_id"]
            print("start_render ->", start)
            for _ in range(20):
                st = _payload(await s.call_tool("get_render_status", {"job_id": job_id}))
                if st.get("state") in ("succeeded", "failed"):
                    break
                await asyncio.sleep(1)
            print("status ->", st)
            res = _payload(await s.call_tool("get_render_result", {"job_id": job_id}))
            print("result ->", res)
            assert res.get("ready") and res.get("url"), "render did not produce a URL"
            print("OK ✅")


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("url", help="MCP endpoint, e.g. http://127.0.0.1:8900/mcp")
    ap.add_argument("--header", action="append", default=[], help='HTTP header "K: V" (repeatable)')
    args = ap.parse_args()
    asyncio.run(run(args.url, _headers(args.header)))
    return 0


if __name__ == "__main__":
    sys.exit(main())

Deploy templates

deploy/cloudflared.config.example.yml (tunnel ingress):

# ~/.cloudflared/config.yml  (one tunnel can host many hostnames)
# Create the tunnel first:  cloudflared tunnel create example-mcp
# Route DNS:                cloudflared tunnel route dns example-mcp example-mcp.<your-zone>
tunnel: <TUNNEL_ID>
credentials-file: /home/<user>/.cloudflared/<TUNNEL_ID>.json
ingress:
  - hostname: example-mcp.<your-zone>
    service: http://127.0.0.1:8900     # the MCP server; /mcp, /healthz, /files all go here
  - service: http_status:404           # catch-all (must be last)

systemd user units — deploy/example-mcp.service and deploy/cloudflared-example-mcp.service:

# ~/.config/systemd/user/example-mcp.service
# systemctl --user daemon-reload && systemctl --user enable --now example-mcp
[Unit]
Description=example-mcp MCP server
After=network-online.target

[Service]
WorkingDirectory=/path/to/example-mcp-server
EnvironmentFile=/path/to/example-mcp-server/.env
ExecStart=/path/to/example-mcp-server/.venv/bin/python server.py
Restart=always
RestartSec=5

[Install]
WantedBy=default.target

# ~/.config/systemd/user/cloudflared-example-mcp.service
# systemctl --user daemon-reload && systemctl --user enable --now cloudflared-example-mcp
# loginctl enable-linger "$USER"   # survive logout / reboot
[Unit]
Description=cloudflared tunnel for example-mcp
After=network-online.target

[Service]
ExecStart=%h/.local/bin/cloudflared tunnel run example-mcp
Restart=always
RestartSec=5

[Install]
WantedBy=default.target