MCP Server Best Practices: Transport, Security and Tool Design

#!/usr/bin/env python3
"""Probe a Streamable HTTP MCP endpoint against the HTTP rules of spec 2026-07-28 (stdlib only).

Usage: probe_http.py URL [--tool NAME --bad-args JSON] [--token T [--expect-auth] [--bad-audience-token T]]
Levels: FAIL = a MUST in the spec is not met, WARN = a SHOULD or the tools page's error guidance is not followed
(version-header-missing is WARN because a server may serve pre-2025-06-18 clients and treat the request as 2025-03-26).
Exit code: 0 no FAIL, 1 at least one FAIL, 2 endpoint unreachable.
"""
import argparse
import http.client
import json
import sys
from urllib.parse import urlsplit

VERSION = "2026-07-28"
EVIL_ORIGIN = "https://evil.example"
counts = {"PASS": 0, "WARN": 0, "FAIL": 0}


def meta(version):
    return {
        "io.modelcontextprotocol/protocolVersion": version,
        "io.modelcontextprotocol/clientInfo": {"name": "probe", "version": "0"},
        "io.modelcontextprotocol/clientCapabilities": {},
    }


def parse(raw):
    """Return the JSON-RPC response from a JSON body or an SSE stream, else None."""
    try:
        return json.loads(raw)
    except ValueError:
        pass
    found = None
    for line in raw.splitlines():
        if line.startswith("data:"):
            try:
                msg = json.loads(line[5:].strip())
            except ValueError:
                continue
            if isinstance(msg, dict) and ("result" in msg or "error" in msg):
                found = msg
    return found


def request(url, verb="POST", method=None, params=None, headers=None, version=VERSION):
    """Send one request; a header whose value is None is removed. Returns (status, headers, body)."""
    parts = urlsplit(url)
    cls = http.client.HTTPSConnection if parts.scheme == "https" else http.client.HTTPConnection
    conn = cls(parts.netloc, timeout=10)
    hdrs = {"Accept": "application/json, text/event-stream", "MCP-Protocol-Version": version}
    payload = None
    if verb == "POST":
        hdrs.update({"Content-Type": "application/json", "Mcp-Method": method})
        if params and "name" in params:
            hdrs["Mcp-Name"] = params["name"]
        payload = json.dumps({"jsonrpc": "2.0", "id": 1, "method": method,
                              "params": {**(params or {}), "_meta": meta(version)}})
    hdrs.update(headers or {})
    conn.request(verb, parts.path or "/", payload, {k: v for k, v in hdrs.items() if v is not None})
    resp = conn.getresponse()
    raw = resp.read().decode("utf-8", "replace")
    conn.close()
    return resp.status, resp, parse(raw)


def code(body):
    err = body.get("error") if isinstance(body, dict) else None
    return err.get("code") if isinstance(err, dict) else None


def show(status, rpc_code):
    return "%s/%s" % (status, rpc_code) if rpc_code is not None else str(status)


def report(name, ok, detail, level="FAIL"):
    status = "PASS" if ok else level
    counts[status] += 1
    print("%s %s: %s" % (status, name, detail))


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("url")
    ap.add_argument("--tool", help="name of an existing tool, enables the tool-error check")
    ap.add_argument("--bad-args", default="{}", help="JSON arguments that violate the tool's input schema")
    ap.add_argument("--token", help="bearer token that the server accepts")
    ap.add_argument("--expect-auth", action="store_true", help="the endpoint is protected: run the auth checks")
    ap.add_argument("--bad-audience-token", help="a valid token issued for a different resource")
    args = ap.parse_args()
    if args.expect_auth and not args.token:
        ap.error("--expect-auth needs --token (the other checks must get past authentication)")
    auth = {"Authorization": "Bearer " + args.token} if args.token else {}
    url = args.url

    try:
        status, _, body = request(url, method="server/discover", headers=auth)
    except (OSError, http.client.HTTPException) as exc:
        print("ERROR endpoint unreachable: %s" % exc)
        return 2
    result = body.get("result") if isinstance(body, dict) else None
    ok = status == 200 and isinstance(result, dict) and VERSION in result.get("supportedVersions", [])
    report("discover", ok, "HTTP %s, supportedVersions %s" % (status, (result or {}).get("supportedVersions")))
    tools = "tools" in (result or {}).get("capabilities", {})

    def expect(name, want_status, want_code, **kw):
        st, _, b = request(url, **{"headers": auth, **kw})
        report(name, st == want_status and (want_code is None or code(b) == want_code),
               "want %s, got %s" % (show(want_status, want_code), show(st, code(b))))
        return b

    expect("origin", 403, None, method="server/discover", headers={**auth, "Origin": EVIL_ORIGIN})
    expect("method-header-missing", 400, -32020, method="server/discover", headers={**auth, "Mcp-Method": None})
    expect("method-header-mismatch", 400, -32020, method="server/discover", headers={**auth, "Mcp-Method": "tools/list"})
    b = expect("unsupported-version", 400, -32022, method="server/discover", version="2099-01-01",
               headers={**auth})
    supported = ((b or {}).get("error") or {}).get("data", {}).get("supported") if isinstance(b, dict) else None
    report("unsupported-version-lists-supported", bool(supported) and VERSION in supported, "data.supported = %s" % supported)
    expect("unknown-method", 404, -32601, method="nope/nothing")

    st, _, b = request(url, method="server/discover", headers={**auth, "MCP-Protocol-Version": None})
    report("version-header-missing", st == 400, "want 400, got %s" % show(st, code(b)), "WARN")
    st, _, _ = request(url, verb="GET", headers=auth)
    report("get-not-allowed", st == 405, "want 405, got %s" % st, "WARN")
    st, resp, _ = request(url, method="server/discover", headers={**auth, "Mcp-Session-Id": "probe-session"})
    report("no-session-id", resp.getheader("Mcp-Session-Id") is None, "Mcp-Session-Id in response: %s" % resp.getheader("Mcp-Session-Id"), "WARN")

    if args.tool:
        st, _, b = request(url, method="tools/call", params={"name": args.tool, "arguments": json.loads(args.bad_args)}, headers=auth)
        is_error = isinstance(b, dict) and (b.get("result") or {}).get("isError") is True
        report("tool-input-error", st == 200 and is_error, "want 200 with result.isError=true, got %s" % show(st, "isError" if is_error else code(b)), "WARN")
        st, _, b = request(url, method="tools/call", params={"name": "no_such_tool", "arguments": {}}, headers=auth)
        report("unknown-tool", code(b) == -32602, "want JSON-RPC error -32602, got %s" % show(st, code(b) or "result"), "WARN")
    elif tools:
        print("SKIP tool checks: pass --tool NAME --bad-args JSON")

    if args.expect_auth:
        st, resp, b = request(url, method="server/discover")
        challenge = resp.getheader("WWW-Authenticate") or ""
        report("auth-required", st == 401, "no token: want 401, got %s" % st)
        report("auth-invalid-token", request(url, method="server/discover", headers={"Authorization": "Bearer invalid"})[0] == 401,
               "garbage token must get 401")
        if args.bad_audience_token:
            st = request(url, method="server/discover", headers={"Authorization": "Bearer " + args.bad_audience_token})[0]
            report("auth-audience", st == 401, "token for another resource: want 401, got %s" % st)
        meta_url = challenge.split('resource_metadata="')[1].split('"')[0] if 'resource_metadata="' in challenge else None
        origin = "%s://%s" % (urlsplit(url).scheme, urlsplit(url).netloc)
        well_known = "/.well-known/oauth-protected-resource"
        found = None
        for candidate in [meta_url, origin + well_known + urlsplit(url).path, origin + well_known]:
            if not candidate:
                continue
            st, _, doc = request(candidate, verb="GET", headers={})
            if st == 200 and isinstance(doc, dict):
                found = doc
                break
        report("auth-resource-metadata", bool(found and found.get("authorization_servers")),
               "challenge resource_metadata=%s; document authorization_servers=%s" % (meta_url, (found or {}).get("authorization_servers")))

    print("%d pass, %d warn, %d fail" % (counts["PASS"], counts["WARN"], counts["FAIL"]))
    return 1 if counts["FAIL"] else 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except (OSError, http.client.HTTPException) as exc:
        print("ERROR connection failed: %s" % exc)
        sys.exit(2)

"""Official Python SDK (mcp==2.3.0) server on Streamable HTTP, http://127.0.0.1:8000/mcp, no auth."""
from mcp.server.mcpserver import MCPServer

server = MCPServer("ref")


@server.tool()
def echo(text: str) -> str:
    """Echo text."""
    return text


if __name__ == "__main__":
    server.run("streamable-http")

"""Same server behind bearer-token auth, http://127.0.0.1:8001/mcp.

Tokens: good-token is issued for this server; other-audience-token is valid but issued for another resource.
"""
from mcp.server.auth.provider import AccessToken
from mcp.server.auth.settings import AuthSettings
from mcp.server.mcpserver import MCPServer

RESOURCE = "http://127.0.0.1:8001/mcp"


class Verifier:
    async def verify_token(self, token: str) -> AccessToken | None:
        if token == "good-token":
            return AccessToken(token=token, client_id="probe", scopes=[], resource=RESOURCE)
        if token == "other-audience-token":
            return AccessToken(token=token, client_id="probe", scopes=[], resource="https://other.example/mcp")
        return None


server = MCPServer(
    "ref-auth",
    token_verifier=Verifier(),
    auth=AuthSettings(
        issuer_url="https://auth.example.com",
        resource_server_url=RESOURCE,
        validate_token_resource=True,
    ),
)


@server.tool()
def echo(text: str) -> str:
    """Echo text."""
    return text


if __name__ == "__main__":
    server.run("streamable-http", port=8001)