MCP Apps 实战：接线、协商与测试

#!/usr/bin/env python3
"""Check the MCP Apps wiring of a stdio MCP server (stdlib only).

Needs check_instructions.py (from the Server Instructions article) in the same
directory: it provides the stdio peer, the era probe and the tool listing.

For every tool whose `_meta.ui.resourceUri` is set, the script checks that the
URI uses `ui://`, that `resources/read` returns an HTML document served as
`text/html;profile=mcp-app`, and that external origins loaded by the HTML are
declared in the resource's `_meta.ui.csp`. With `--call NAME=JSON` it also
calls the tool as a client that did NOT negotiate Apps and expects text back.
CSP and fallback checks are heuristics, not requirements of the specification.

Usage: check_mcp_app.py [--timeout S] [--call NAME=JSON]... -- <server command>
Exit code: 0 = no findings, 1 = warnings, 2 = errors.
"""
import argparse
import json
import re
import sys

from check_instructions import Peer, handshake, list_tools, meta

EXT = "io.modelcontextprotocol/ui"
APP_MIME = "text/html;profile=mcp-app"
APPS_CAPS = {"extensions": {EXT: {"mimeTypes": [APP_MIME]}}}
ORIGIN = r"""(https?://[^/"'\s)>]+)"""
LOADS = [r"""\bsrc\s*=\s*["']?""" + ORIGIN, r"""<link[^>]+href\s*=\s*["']?""" + ORIGIN,
         r"""fetch\(\s*["']""" + ORIGIN, r"""url\(\s*["']?""" + ORIGIN]


def read_resource(peer, era, version, uri):
    params = {"uri": uri}
    if era == "modern":
        params["_meta"] = meta(version, APPS_CAPS)
    return peer.call("resources/read", params)


def declared_origins(csp):
    return {o.rstrip("/") for key in ("connectDomains", "resourceDomains", "frameDomains", "baseUriDomains")
            for o in (csp or {}).get(key, [])}


def check_tool(peer, era, version, name, ui):
    findings, uri = [], ui.get("resourceUri")
    if not isinstance(uri, str) or not uri.startswith("ui://"):
        return [("ERROR", "bad-scheme", "%s: resourceUri %r must be a ui:// URI" % (name, uri))]
    bad = [v for v in ui.get("visibility", []) if v not in ("model", "app")]
    if bad:
        findings.append(("WARN", "bad-visibility", "%s: unknown visibility value(s) %s" % (name, bad)))
    reply = read_resource(peer, era, version, uri)
    contents = (reply.get("result") or {}).get("contents") or []
    if not contents:
        return findings + [("ERROR", "unreadable-resource", "%s: resources/read %s failed: %s"
                            % (name, uri, json.dumps(reply.get("error") or reply)[:120]))]
    item = contents[0]
    if item.get("mimeType") != APP_MIME:
        findings.append(("ERROR", "bad-mime", "%s: %s is served as %r, hosts only render %s"
                         % (name, uri, item.get("mimeType"), APP_MIME)))
    html = item.get("text") or ""
    if not html.strip():
        findings.append(("ERROR", "empty-html", "%s: %s has no text content" % (name, uri)))
    csp = ((item.get("_meta") or {}).get("ui") or {}).get("csp")
    used = {o.rstrip("/") for pattern in LOADS for o in re.findall(pattern, html, re.I)}
    for origin in sorted(used - declared_origins(csp)):
        findings.append(("WARN", "origin-not-in-csp", "%s: HTML loads %s but _meta.ui.csp does not declare it"
                         % (name, origin)))
    return findings


def check_fallback(cmd, timeout, name, arguments):
    """Call a UI tool as a client without the Apps extension and expect text."""
    peer = Peer(cmd, timeout)
    try:
        era, version, _ = handshake(peer)
        params = {"name": name, "arguments": arguments}
        if era == "modern":
            params["_meta"] = meta(version)
        result = peer.call("tools/call", params).get("result") or {}
    finally:
        peer.close()
    texts = [c.get("text", "") for c in result.get("content", []) if c.get("type") == "text"]
    if result.get("isError") or not any(t.strip() for t in texts):
        return [("WARN", "no-text-fallback", "%s: no usable text for a client without Apps support" % name)]
    return []


def main():
    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
    ap.add_argument("--timeout", type=float, default=5.0)
    ap.add_argument("--call", action="append", default=[], metavar="NAME=JSON")
    ap.add_argument("cmd", nargs=argparse.REMAINDER)
    args = ap.parse_args()
    cmd = args.cmd[1:] if args.cmd[:1] == ["--"] else args.cmd
    if not cmd:
        ap.error("missing server command after --")
    peer = Peer(cmd, args.timeout)
    findings = []
    try:
        era, version, result = handshake(peer, APPS_CAPS)
        if era is None:
            print("[ERROR] %s" % result)
            return 2
        tools = list_tools(peer, era, version, APPS_CAPS) or {}
        bound = {n: (t.get("_meta") or {}).get("ui") for n, t in tools.items() if (t.get("_meta") or {}).get("ui")}
        print("era: %s (protocol %s)" % (era, version))
        print("tools: %d, bound to a UI: %s" % (len(tools), ", ".join(sorted(bound)) or "none"))
        advertised = EXT in (result.get("capabilities") or {}).get("extensions", {})
        if bound and era == "modern" and not advertised:
            findings.append(("WARN", "extension-not-advertised", "server/discover does not list %s" % EXT))
        for name in sorted(bound):
            findings += check_tool(peer, era, version, name, bound[name])
    finally:
        peer.close()
    for spec in args.call:
        name, _, raw = spec.partition("=")
        findings += check_fallback(cmd, args.timeout, name, json.loads(raw or "{}"))
    for level, code, message in findings:
        print("[%s] %s: %s" % (level, code, message))
    if not findings:
        print("[OK] no findings")
    return 2 if any(f[0] == "ERROR" for f in findings) else 1 if findings else 0


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

"""Demo MCP Apps server on the official Python SDK (pip install mcp).

APPS_DEMO selects a wiring variant:
  ok          tool bound to a ui:// resource served with the app MIME type (default)
  bad-mime    resource served as plain text/html, which hosts will not render
  missing     tool points at a ui:// resource that nothing serves (the SDK's Apps
              helper refuses such a binding, so this variant wires it by hand)
  http-uri    tool points at an https:// URI instead of ui://
  external    HTML loads a CDN script but declares no CSP domains
  csp-ok      same HTML, CSP declares the CDN origin
  no-fallback the tool returns no text, so clients without Apps support see nothing
"""
import os
from datetime import datetime, timezone

from mcp.server.apps import Apps, ResourceCsp
from mcp.server.mcpserver import MCPServer

MODE = os.environ.get("APPS_DEMO", "ok")
URI = "ui://clock/app.html"
HTML = "<!doctype html><title>Clock</title><p id=t>loading</p><script>document.getElementById('t').textContent='clock';</script>"
if MODE in ("external", "csp-ok"):
    HTML = HTML.replace("<script>", '<script src="https://cdn.example.com/chart.js"></script><script>')


def now() -> str:
    """Return the current UTC time as an ISO 8601 string (readable without any UI)."""
    return datetime.now(timezone.utc).isoformat()


if MODE in ("ok", "external", "csp-ok", "no-fallback"):
    apps = Apps()
    handler = (lambda: "") if MODE == "no-fallback" else now
    apps.tool(resource_uri=URI, name="get_time", description="Current UTC time")(handler)
    csp = ResourceCsp(resource_domains=["https://cdn.example.com"]) if MODE == "csp-ok" else None
    apps.add_html_resource(URI, HTML, name="clock", csp=csp)
    server = MCPServer("demo-clock", extensions=[apps])
else:  # hand-wired variants: plain tool metadata plus a plain resource
    server = MCPServer("demo-clock")
    ui_uri = {"missing": "ui://clock/gone.html", "http-uri": "https://example.com/app.html"}.get(MODE, URI)
    server.tool(name="get_time", description="Current UTC time", meta={"ui": {"resourceUri": ui_uri}})(now)
    mime = "text/html" if MODE == "bad-mime" else "text/html;profile=mcp-app"
    server.resource(URI, name="clock", mime_type=mime)(lambda: HTML)

if __name__ == "__main__":
    server.run("stdio")