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
Run it in a container (Docker)¶
The example also ships a Dockerfile and a standalone docker-compose.yml — an alternative to
running it under systemd (see the Guide):
docker build -t example-mcp .
docker run --rm -p 8900:8900 example-mcp # MCP at http://localhost:8900/mcp
# or: docker compose up --build
curl -s http://localhost:8900/healthz # -> {"ok": true}
python smoke_test.py http://localhost:8900/mcp
The image sets MCP_HOST=0.0.0.0 (so the published port is reachable from outside the container),
declares a HEALTHCHECK on GET /healthz, and uses OUTPUT_DIR for job artifacts. Set
PUBLIC_BASE_URL at run time so the file URLs point at your public origin. There is no app-level
auth — front it with Cloudflare Tunnel + Access for the remote/auth boundary, exactly as the
systemd path does (or apply the Guide's Option B built-in Bearer verifier and front it with your own
reverse proxy instead).
Dockerfile¶
# syntax=docker/dockerfile:1
#
# example-mcp — self-contained reference MCP container (Streamable HTTP).
# The teaching demo for the Remote MCP Server Guide. Build context = this directory.
#
# serve (default): CMD python server.py -> streamable-http MCP at :8900/mcp
#
# NOTE: MCP_HOST defaults to 0.0.0.0 here (not 127.0.0.1) so the server is reachable
# from outside the container (compose net / `docker run -p`). The app has NO auth;
# Cloudflare Access is the boundary when deployed — a plain `docker run` is unauthenticated.
#
FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim
# Only dependency: the official MCP SDK (FastMCP + client). It pulls in
# starlette/uvicorn/httpx, so nothing else is needed at runtime.
RUN uv pip install --system --no-cache "mcp>=1.0"
WORKDIR /app
COPY server.py pyproject.toml /app/
# Non-root (uid 1000, matching the other images); OUTPUT_DIR holds rendered files.
RUN useradd -m -u 1000 -s /bin/bash app && \
mkdir -p /tmp/example-mcp-files && chown -R app:app /app /tmp/example-mcp-files
USER app
# MCP_HOST=0.0.0.0 so the port is reachable from the host / compose network.
ENV MCP_HOST=0.0.0.0 \
MCP_PORT=8900 \
OUTPUT_DIR=/tmp/example-mcp-files
EXPOSE 8900
# Liveness probe via stdlib urllib (no curl in the slim image). /healthz needs no state.
HEALTHCHECK --interval=30s --timeout=5s --retries=5 --start-period=10s \
CMD python -c "import urllib.request,sys; sys.exit(0 if urllib.request.urlopen('http://127.0.0.1:8900/healthz', timeout=3).status==200 else 1)"
CMD ["python", "server.py"]
docker-compose.yml¶
# Standalone teaching demo for example-mcp — kept SEPARATE from the main repo's
# docker-compose.yml on purpose (this one is just for trying the reference server
# locally / behind a tunnel, not part of the BadmintonGPT deploy).
#
# docker compose up --build # build + run; MCP at http://localhost:8900/mcp
# curl localhost:8900/healthz # {"ok": true}
#
# The port is host-published (8900:8900) for the demo. There is NO app-level auth —
# Cloudflare Access is the boundary when deployed; a local run is unauthenticated and
# fine for a demo. When behind a tunnel, set PUBLIC_BASE_URL to the public hostname so
# the file URLs returned by get_render_result point at the tunnel, not 127.0.0.1.
services:
example-mcp:
build: .
image: example-mcp
ports:
- "8900:8900"
environment:
MCP_HOST: 0.0.0.0
MCP_PORT: 8900
OUTPUT_DIR: /tmp/example-mcp-files
# Uncomment + set to your public hostname when serving behind a Cloudflare Tunnel:
# PUBLIC_BASE_URL: https://example-mcp.<your-zone>
volumes:
- ./demo-files:/tmp/example-mcp-files
restart: unless-stopped
healthcheck:
test: ["CMD", "python", "-c", "import urllib.request,sys; sys.exit(0 if urllib.request.urlopen('http://127.0.0.1:8900/healthz', timeout=3).status==200 else 1)"]
interval: 30s
timeout: 5s
retries: 5
start_period: 10s
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 typing import Annotated
from mcp.server.fastmcp import FastMCP
from pydantic import Field
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: Annotated[float, Field(description="First number to add.")],
b: Annotated[float, Field(description="Second number to add.")],
) -> dict:
"""Add two numbers and return {"sum": a+b}."""
return {"sum": a + b}
@mcp.tool()
def echo(
text: Annotated[str, Field(description="Text to echo back; must be a non-empty string.")],
) -> 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 ------------------
# Progress reporting: the worker updates stage/total_stages/message as it goes, and
# get_render_status echoes them. Consuming agents/UIs use these fields to render a
# live progress bar. Derive the total from the actual step list — never hardcode it
# somewhere else, or the bar will desync when you add a step.
RENDER_STEPS = ["preparing assets", "rendering frames", "encoding output"]
def _run_render(job_id: str, spec: dict) -> None:
try:
total = len(RENDER_STEPS)
for i, step in enumerate(RENDER_STEPS, start=1):
JOBS[job_id] |= {
"state": "running",
"stage": i,
"total_stages": total,
"message": f"[{i}/{total}] {step}",
}
time.sleep(1) # pretend each step 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", "stage": total, "message": "done", "path": str(out)}
except Exception as e: # noqa: BLE001
JOBS[job_id] |= {"state": "failed", "error": str(e)}
@mcp.tool()
def start_render(
spec: Annotated[dict | None, Field(
description="Optional free-form render spec (JSON object) passed through "
"to the job; omit for defaults.")] = None,
) -> dict:
"""Start a long render job. Returns {job_id, state} immediately (async)."""
job_id = uuid.uuid4().hex
JOBS[job_id] = {"state": "queued", "stage": 0, "total_stages": len(RENDER_STEPS), "message": "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: Annotated[str, Field(description="The job_id returned by start_render.")],
) -> dict:
"""Return job progress: {job_id, state, stage, total_stages, message, error?}.
state in queued|running|succeeded|failed. stage/total_stages/message let the
calling agent/UI show a live progress bar. Must return immediately (clients poll).
"""
j = JOBS.get(job_id)
if not j:
return {"error": f"no such job: {job_id}"}
return {
"job_id": job_id,
"state": j["state"],
"stage": j.get("stage", 0),
"total_stages": j.get("total_stages", 0),
"message": j.get("message", ""),
"error": j.get("error"),
}
@mcp.tool()
def get_render_result(
job_id: Annotated[str, Field(description="The job_id returned by start_render.")],
) -> 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}))
# stage/total_stages/message are what an agent UI turns into a progress bar
print(f"status -> {st.get('state')} {st.get('stage')}/{st.get('total_stages')} {st.get('message', '')}")
if st.get("state") in ("succeeded", "failed"):
break
await asyncio.sleep(1)
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