MCP Resources Explained: URIs, Templates, Reads, Subscriptions and Errors

"""Official Python SDK (mcp==2.3.0) resources server at http://127.0.0.1:8230/mcp, no auth.

Resources: a text resource with annotations, a binary resource, three templates (a closed set of notes, files under
`sample-docs/` served through `safe_join`, a status that fails on purpose) and the tool `touch`, which announces a
resource update. The templates keep the SDK's default `ResourceSecurity` policy: traversal, absolute paths and
NUL bytes in a template value are rejected before the handler runs.
"""
import base64
from pathlib import Path

from mcp.server.mcpserver import Context, MCPServer
from mcp.server.mcpserver.exceptions import ResourceError, ResourceNotFoundError
from mcp.shared.path_security import safe_join
from mcp.types import Annotations

ROOT = Path(__file__).resolve().parent / "sample-docs"
NOTES = {"standup": "Standup at 09:30.", "release": "Release freeze on Friday."}
# A 1x1 transparent PNG.
PIXEL = base64.b64decode("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==")

server = MCPServer("resources-ref")


@server.resource("config://app/settings", name="settings", title="Application settings",
                 description="Current settings as JSON", mime_type="application/json",
                 annotations=Annotations(audience=["user"], priority=0.8))
def settings() -> str:
    return '{"theme": "dark", "retries": 3}'


@server.resource("image://logo", name="logo", mime_type="image/png")
def logo() -> bytes:
    return PIXEL


@server.resource("notes://{topic}", name="notes", description="A note by topic")
def note(topic: str) -> str:
    if topic not in NOTES:
        raise ResourceNotFoundError("no note %r" % topic)
    return NOTES[topic]


@server.resource("docs://files/{+path}", name="docs", description="Files under sample-docs")
def doc(path: str) -> str:
    try:
        return safe_join(ROOT, path).read_text(encoding="utf-8")
    except (FileNotFoundError, IsADirectoryError):
        raise ResourceNotFoundError("no file %r" % path) from None


@server.resource("status://{kind}", name="status", description="Service status; busy and crash fail on purpose")
def status(kind: str) -> str:
    if kind == "ok":
        return "all systems normal"
    if kind == "busy":
        raise ResourceError("upstream quota exhausted, retry later")  # an anticipated failure
    if kind == "crash":
        raise RuntimeError("db password is hunter2")  # a crash: the SDK withholds this text from the client
    raise ResourceNotFoundError("no status %r" % kind)


@server.tool()
async def touch(uri: str, ctx: Context) -> str:
    """Announce that a resource changed, so `subscriptions/listen` streams deliver an update."""
    await ctx.notify_resource_updated(uri)
    return "announced " + uri


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

list             HTTP 200 result   ttlMs=0 cacheScope=private resultType=complete resources=2 | ListResourcesResultResponse valid
templates        HTTP 200 result   ttlMs=0 cacheScope=private resultType=complete templates=3 | ListResourceTemplatesResultResponse valid
read-text        HTTP 200 result   ttlMs=0 cacheScope=private resultType=complete contents: application/json text '{"theme": "dark", "retries": 3}' | ReadResourceResultResponse valid
read-blob        HTTP 200 result   ttlMs=0 cacheScope=private resultType=complete contents: image/png blob 70 bytes | ReadResourceResultResponse valid
read-template    HTTP 200 result   ttlMs=0 cacheScope=private resultType=complete contents: text/plain text 'Standup at 09:30.' | ReadResourceResultResponse valid
read-file        HTTP 200 result   ttlMs=0 cacheScope=private resultType=complete contents: text/plain text '# Sample docs\n\nServed through the `docs:' | ReadResourceResultResponse valid
unknown-uri      HTTP 400 error    code -32602 'Unknown resource: config://nope' | JSONRPCErrorResponse+InvalidParamsError valid
unknown-topic    HTTP 400 error    code -32602 "no note 'nope'" | JSONRPCErrorResponse+InvalidParamsError valid
missing-file     HTTP 400 error    code -32602 "no file 'missing.md'" | JSONRPCErrorResponse+InvalidParamsError valid
traversal        HTTP 400 error    code -32602 'Unknown resource: docs://files/../resources_ref_server.py' | JSONRPCErrorResponse+InvalidParamsError valid
absolute-path    HTTP 400 error    code -32602 'Unknown resource: docs://files//etc/passwd' | JSONRPCErrorResponse+InvalidParamsError valid
bad-uri          HTTP 400 error    code -32602 'Unknown resource: not a uri' | JSONRPCErrorResponse+InvalidParamsError valid
non-ascii-uri    HTTP 400 error    code -32602 "no note 'café'" | JSONRPCErrorResponse+InvalidParamsError valid
status-busy      HTTP 200 error    code -32603 'upstream quota exhausted, retry later' | JSONRPCErrorResponse+InternalError valid
status-crash     HTTP 200 error    code -32603 'Error creating resource from template status://crash' | JSONRPCErrorResponse+InternalError valid
header-mismatch  HTTP 400 error    code -32020 "mcp-name header does not match the request body's 'uri' parameter" | JSONRPCErrorResponse+HeaderMismatchError valid
missing-uri      HTTP 400 error    code -32602 'Invalid request parameters' | JSONRPCErrorResponse+InvalidParamsError valid
subscribe-old    HTTP 404 error    code -32601 'Method not found' | JSONRPCErrorResponse+MethodNotFoundError valid
listen-ack       HTTP 200 stream   ack(resourceSubscriptions=1) | SubscriptionsAcknowledgedNotification valid
listen-update    HTTP 200 stream   ack(resourceSubscriptions=1) -> updated(config://app/settings) | ResourceUpdatedNotification+SubscriptionsAcknowledgedNotification valid

#!/usr/bin/env python3
"""Lint the resources of an MCP server against spec 2026-07-28 over Streamable HTTP.

Usage: lint_resources.py URL [--token T] [--max-reads N]
URL: the Streamable HTTP endpoint. The linter lists resources and templates, reads up to N listed resources
(default 5) and sends a few probe reads that no healthy server may satisfy: a missing URI, a missing template
value, path traversal, a mismatching `Mcp-Name` header and a non-ASCII URI. The probes are plain reads and change
nothing; do not point them at an endpoint where reading has side effects.
Levels: FAIL = a MUST of the spec is violated, WARN = a SHOULD or a recommendation is not followed.
Exit code: 0 no FAIL, 1 at least one FAIL, 2 the endpoint cannot be read.
"""
import argparse
import base64
import binascii
import http.client
import json
import re
import sys
from urllib.parse import urlsplit

VERSION = "2026-07-28"
SCHEME_RE = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*:")
VARIABLE_RE = re.compile(r"\{[+#./;?&]?[^}]+\}")
MISSING_URI = "lint-probe://missing/resource"
MISSING_VALUE = "x-lint-missing"
TRAVERSAL_VALUES = ["../x-lint-traversal", "%2e%2e/x-lint-traversal", "/etc/passwd"]
counts = {"PASS": 0, "WARN": 0, "FAIL": 0}


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


def mcp_name(value):
    """Mcp-Name header value: plain ASCII as is, anything else in the Base64 sentinel format."""
    plain = value.isascii() and value.isprintable() and value == value.strip() \
        and not (value.startswith("=?base64?") and value.endswith("?="))
    return value if plain else "=?base64?%s?=" % base64.b64encode(value.encode()).decode()


def request(url, method, params=None, token=None, name=None):
    """Build one 2026-07-28 request: (parsed URL, body, headers).

    `Mcp-Name` is `params.name` or `params.uri`; pass `name` to send another value."""
    meta = {"io.modelcontextprotocol/protocolVersion": VERSION,
            "io.modelcontextprotocol/clientInfo": {"name": "lint-resources", "version": "0"},
            "io.modelcontextprotocol/clientCapabilities": {}}
    headers = {"Accept": "application/json, text/event-stream", "Content-Type": "application/json",
               "MCP-Protocol-Version": VERSION, "Mcp-Method": method}
    source = name if name is not None else (params or {}).get("name", (params or {}).get("uri"))
    if isinstance(source, str):
        headers["Mcp-Name"] = mcp_name(source)
    if token:
        headers["Authorization"] = "Bearer " + token
    body = json.dumps({"jsonrpc": "2.0", "id": 1, "method": method, "params": {**(params or {}), "_meta": meta}})
    return urlsplit(url), body, headers


def rpc(url, method, params=None, token=None, name=None):
    """POST one request and return (HTTP status, JSON-RPC message or None)."""
    parts, body, headers = request(url, method, params, token, name)
    cls = http.client.HTTPSConnection if parts.scheme == "https" else http.client.HTTPConnection
    conn = cls(parts.netloc, timeout=15)
    conn.request("POST", parts.path or "/", body, headers)
    resp = conn.getresponse()
    raw = resp.read().decode("utf-8", "replace")
    conn.close()
    try:
        return resp.status, json.loads(raw)
    except ValueError:
        for line in reversed(raw.splitlines()):  # SSE: the last data line holding a response
            if line.startswith("data:"):
                try:
                    return resp.status, json.loads(line[5:].strip())
                except ValueError:
                    continue
    return resp.status, None


def list_all(url, method, key, token):
    """Return (items, first result) of a paginated list method."""
    items, cursor, first = [], None, None
    for _ in range(50):
        status, msg = rpc(url, method, {"cursor": cursor} if cursor else None, token)
        result = (msg or {}).get("result")
        if status != 200 or not isinstance(result, dict) or not isinstance(result.get(key), list):
            raise ValueError("%s failed: HTTP %s, %s" % (method, status, json.dumps(msg)[:200]))
        items += result[key]
        first = first or result
        cursor = result.get("nextCursor")
        if not cursor:
            return items, first
    raise ValueError("%s still paginating after 50 pages" % method)


def envelope_problems(result):
    """Results of 2026-07-28 MUST carry resultType and the caching hints, with ttlMs >= 0."""
    ttl, scope = result.get("ttlMs"), result.get("cacheScope")
    problems = []
    if not isinstance(result.get("resultType"), str):
        problems.append("resultType is missing")
    if not isinstance(ttl, int) or isinstance(ttl, bool) or ttl < 0:
        problems.append("ttlMs is %r, want an integer >= 0" % ttl)
    if scope not in ("public", "private"):
        problems.append("cacheScope is %r, want 'public' or 'private'" % scope)
    return problems


def envelope_check(result):
    """(ok, detail) of the envelope of a list result."""
    problems = envelope_problems(result)
    return not problems, "; ".join(problems) if problems else "resultType=%s ttlMs=%d cacheScope=%s" % (
        result["resultType"], result["ttlMs"], result["cacheScope"])


def lint_capability(url, token):
    status, msg = rpc(url, "server/discover", None, token)
    result = (msg or {}).get("result")
    caps = result.get("capabilities") if isinstance(result, dict) else None
    declared = isinstance(caps, dict) and isinstance(caps.get("resources"), dict)
    report("capability-declared", declared, "the `resources` capability is %s (server/discover, HTTP %s)"
           % ("declared" if declared else "missing", status))
    if declared:
        print("INFO resources capability: listChanged=%s subscribe=%s" % (
            caps["resources"].get("listChanged", False), caps["resources"].get("subscribe", False)))


def lint_listed(resources, templates):
    named = [r for r in resources if isinstance(r, dict) and isinstance(r.get("uri"), str)
             and SCHEME_RE.match(r["uri"]) and isinstance(r.get("name"), str)]
    report("resource-fields", len(named) == len(resources),
           "%d of %d resources have a name and an absolute URI with a scheme" % (len(named), len(resources)))
    uris = [r["uri"] for r in named]
    dup = sorted({u for u in uris if uris.count(u) > 1})
    report("uri-unique", not dup, "duplicates: %s" % dup if dup else "no duplicate URIs", "WARN")
    plain = [r["uri"] for r in named if not isinstance(r.get("mimeType"), str)]
    report("resource-mime", not plain, "no mimeType: %s" % plain if plain
           else "every listed resource states a mimeType", "WARN")
    good = [t for t in templates if isinstance(t, dict) and isinstance(t.get("name"), str)
            and isinstance(t.get("uriTemplate"), str) and VARIABLE_RE.search(t["uriTemplate"])]
    report("template-fields", len(good) == len(templates),
           "%d of %d templates have a name and a uriTemplate with a {variable}" % (len(good), len(templates)))


def decode_blob(blob):
    try:
        base64.b64decode(blob, validate=True)
        return True
    except (binascii.Error, ValueError, TypeError):
        return False


def lint_reads(url, uris, token):
    """Read each URI and check envelope, contents, blob encoding and mimeType."""
    envelope, content, blob, mime, answered = [], [], [], [], 0
    for uri in uris:
        status, msg = rpc(url, "resources/read", {"uri": uri}, token)
        result = (msg or {}).get("result") if isinstance(msg, dict) else None
        if not isinstance(result, dict):
            content.append("%s: HTTP %s without a result" % (uri, status))
            continue
        answered += 1
        if envelope_problems(result):
            envelope.append(uri)
        items = result.get("contents")
        if not isinstance(items, list) or not items:
            content.append("%s: contents is empty or not an array" % uri)
            continue
        for item in items:
            kinds = [k for k in ("text", "blob") if isinstance(item, dict) and isinstance(item.get(k), str)]
            if not isinstance(item, dict) or not isinstance(item.get("uri"), str) or len(kinds) != 1:
                content.append("%s: an item needs a uri and exactly one of text or blob" % uri)
                continue
            if "blob" in item and not decode_blob(item["blob"]):
                blob.append(uri)
            if not isinstance(item.get("mimeType"), str):
                mime.append(uri)
    report("read-envelope", not envelope, "resultType, ttlMs or cacheScope is wrong for %s" % envelope if envelope
           else "%d reads carry resultType, ttlMs and cacheScope" % answered)
    report("read-contents", not content, "; ".join(content) if content
           else "%d reads return non-empty contents, each item with a uri and exactly one of text or blob" % answered)
    report("read-blob-base64", not blob, "blob is not valid base64: %s" % blob if blob
           else "every blob is valid base64")
    report("read-mime", not mime, "no mimeType on contents of %s" % mime if mime
           else "every content item states a mimeType", "WARN")


def probe_not_found(url, uri, token, answered="FAIL"):
    """Read something that cannot exist. Return (PASS|WARN|FAIL, detail): -32602 is required, the legacy -32002
    only a WARN, an empty contents array is a FAIL, contents for a made-up URI get the level `answered`."""
    status, msg = rpc(url, "resources/read", {"uri": uri}, token)
    if not isinstance(msg, dict):
        return "FAIL", "HTTP %s without a JSON-RPC message for %s" % (status, uri)
    error = msg.get("error")
    if isinstance(error, dict):
        code = error.get("code")
        return ("PASS" if code == -32602 else "WARN" if code == -32002 else "FAIL",
                "%s answered with code %s (HTTP %s)" % (uri, code, status))
    if (msg.get("result") or {}).get("contents") == []:
        return "FAIL", "%s answered with an empty contents array, which hides the missing resource" % uri
    return answered, "%s was answered with contents, so the server cannot tell what is missing" % uri


def worst(outcomes):
    """The most severe (level, detail) of several outcomes; the details of that level are joined."""
    level = max((o[0] for o in outcomes), key=("PASS", "WARN", "FAIL").index)
    return level, "; ".join(o[1] for o in outcomes if o[0] == level)


def fill(template, value):
    return VARIABLE_RE.sub(value, template)


def lint_probes(url, templates, token):
    level, detail = probe_not_found(url, MISSING_URI, token)
    report("not-found-code", level == "PASS", detail, level)
    templates = [t["uriTemplate"] for t in templates if isinstance(t, dict) and isinstance(t.get("uriTemplate"), str)]
    templates = templates[:5]
    if templates:
        level, detail = worst([probe_not_found(url, fill(t, MISSING_VALUE), token, "WARN") for t in templates])
        report("template-not-found", level == "PASS", detail if level != "PASS"
               else "%d template probes answered with -32602" % len(templates), level)
    leaked = []
    for template in templates:
        for value in TRAVERSAL_VALUES:
            uri = fill(template, value)
            status, msg = rpc(url, "resources/read", {"uri": uri}, token)
            if isinstance(msg, dict) and (msg.get("result") or {}).get("contents"):
                leaked.append(uri)
    if templates:
        report("traversal-rejected", not leaked, "read succeeded for %s" % leaked if leaked
               else "%d traversal probes were rejected" % (len(templates) * len(TRAVERSAL_VALUES)))
    else:
        print("INFO no templates, the template and traversal probes are skipped")
    status, msg = rpc(url, "resources/read", {"uri": MISSING_URI}, token, name="lint-probe://other")
    code = ((msg or {}).get("error") or {}).get("code") if isinstance(msg, dict) else None
    report("header-mismatch", status == 400 and code == -32020,
           "a mismatching Mcp-Name was answered with HTTP %s, code %s (want 400, -32020)" % (status, code))
    status, msg = rpc(url, "resources/read", {"uri": "lint-probe://caf\u00e9"}, token)
    code = ((msg or {}).get("error") or {}).get("code") if isinstance(msg, dict) else None
    report("header-base64", code != -32020, "a non-ASCII URI sent as a Base64 sentinel was answered with code %s"
           % code)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("url")
    ap.add_argument("--token", help="bearer token for the endpoint")
    ap.add_argument("--max-reads", type=int, default=5, help="how many listed resources to read")
    args = ap.parse_args()
    if not args.url.startswith(("http://", "https://")):
        ap.error("URL must start with http:// or https://")
    try:
        resources, listing = list_all(args.url, "resources/list", "resources", args.token)
        templates, template_listing = list_all(args.url, "resources/templates/list", "resourceTemplates", args.token)
        print("INFO %d resources, %d templates" % (len(resources), len(templates)))
        lint_capability(args.url, args.token)
        report("list-envelope", *envelope_check(listing))
        report("templates-envelope", *envelope_check(template_listing))
        lint_listed(resources, templates)
        uris = [r["uri"] for r in resources if isinstance(r, dict) and isinstance(r.get("uri"), str)]
        if uris:
            lint_reads(args.url, uris[:args.max_reads], args.token)
        else:
            print("INFO no listed resources, the read checks are skipped")
        lint_probes(args.url, templates, args.token)
    except (OSError, ValueError, http.client.HTTPException) as exc:
        print("ERROR %s" % exc)
        return 2
    print("%d pass, %d warn, %d fail" % (counts["PASS"], counts["WARN"], counts["FAIL"]))
    return 1 if counts["FAIL"] else 0


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