trending/ai-engineering-from-scratch
1
0
Fork 0
mirror of https://github.com/rohitg00/ai-engineering-from-scratch.git synced 2026-10-02 01:54:39 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(phase-19/13): modernize the MCP registry capstone

Browse Source 
The production capstone must separate Registry publication from stateless runtime discovery.
This commit is contained in:
Rohit Ghumare
2026-08-21 15:59:34 +01:00
parent 860c4f8086
commit 67db9192c3
13 changed files with 1744 additions and 529 deletions
Download Patch File Download Diff File Expand all files Collapse all files
phases/19-capstone-projects/13-mcp-server-with-registry/assets/mcp-topology.svg
+20 -20
View File
@@ -14,25 +14,25 @@
</style>
</defs>
<text x="480" y="24" text-anchor="middle" class="title">MCP 2026 — StreamableHTTP + scopes + OPA + registry</text>
<text x="480" y="24" text-anchor="middle" class="title">MCP 2026-07-28: stateless requests + policy + registry</text>
<rect x="40" y="50" width="260" height="80" class="cool"/>
<text x="170" y="72" text-anchor="middle" class="head">MCP clients</text>
<text x="170" y="94" text-anchor="middle" class="small">Claude Code, Cursor 3, Amp,</text>
<text x="170" y="112" text-anchor="middle" class="small">OpenCode, Gemini CLI</text>
<text x="170" y="94" text-anchor="middle" class="small">version + capabilities</text>
<text x="170" y="112" text-anchor="middle" class="small">on every request</text>
<rect x="340" y="50" width="260" height="80" class="dsk"/>
<text x="470" y="72" text-anchor="middle" class="head">StreamableHTTP</text>
<text x="470" y="94" text-anchor="middle" class="small">JSON-RPC + streaming</text>
<text x="470" y="112" text-anchor="middle" class="small">stateless, horizontally scalable</text>
<text x="470" y="72" text-anchor="middle" class="head">Streamable HTTP</text>
<text x="470" y="94" text-anchor="middle" class="small">one POST per JSON-RPC message</text>
<text x="470" y="112" text-anchor="middle" class="small">no protocol session</text>
<rect x="640" y="50" width="280" height="80" class="cold"/>
<text x="780" y="72" text-anchor="middle" class="head">OAuth 2.1 scopes</text>
<text x="780" y="94" text-anchor="middle" class="small">SPIFFE workload identity</text>
<text x="780" y="112" text-anchor="middle" class="small">per-tool scope enforcement</text>
<text x="780" y="94" text-anchor="middle" class="small">issuer + audience + expiry</text>
<text x="780" y="112" text-anchor="middle" class="small">scope checked per tool call</text>
<rect x="40" y="160" width="560" height="200" class="box"/>
<text x="320" y="182" text-anchor="middle" class="head">read-only MCP server (FastMCP)</text>
<text x="320" y="182" text-anchor="middle" class="head">read-only MCP replicas</text>
<rect x="60" y="200" width="170" height="40" class="cool"/>
<text x="145" y="220" text-anchor="middle" class="step">postgres.readonly</text>
<text x="145" y="234" text-anchor="middle" class="small">scope: postgres:query:readonly</text>
@@ -55,28 +55,28 @@
<text x="780" y="182" text-anchor="middle" class="head">destructive MCP server</text>
<rect x="660" y="200" width="240" height="44" class="hot"/>
<text x="780" y="220" text-anchor="middle" class="step">jira.create / linear.create</text>
<text x="780" y="234" text-anchor="middle" class="small">scope: +approved:by:human</text>
<text x="780" y="234" text-anchor="middle" class="small">write scope + separate approval</text>
<rect x="660" y="254" width="240" height="44" class="hot"/>
<text x="780" y="274" text-anchor="middle" class="step">postgres.write</text>
<text x="780" y="288" text-anchor="middle" class="small">fresh Slack approval required</text>
<text x="780" y="288" text-anchor="middle" class="small">action-bound approval required</text>
<rect x="660" y="310" width="240" height="40" class="dsk"/>
<text x="780" y="330" text-anchor="middle" class="step">Slack approval card</text>
<text x="780" y="344" text-anchor="middle" class="small">elevates scope for 15 min</text>
<text x="780" y="330" text-anchor="middle" class="step">approval record</text>
<text x="780" y="344" text-anchor="middle" class="small">actor + tool + args + target + expiry</text>
<rect x="40" y="380" width="420" height="130" class="box"/>
<text x="250" y="402" text-anchor="middle" class="head">registry service</text>
<text x="250" y="422" text-anchor="middle" class="small">polls .well-known/mcp-capabilities</text>
<text x="250" y="440" text-anchor="middle" class="small">validates JSON Schema</text>
<text x="250" y="458" text-anchor="middle" class="small">UI: list / search / validate / enable</text>
<text x="250" y="476" text-anchor="middle" class="small">team ownership + SLO per server</text>
<text x="250" y="498" text-anchor="middle" class="caption">AAIF Registry spec</text>
<text x="250" y="422" text-anchor="middle" class="small">publishes and indexes server.json</text>
<text x="250" y="440" text-anchor="middle" class="small">validates Registry schema</text>
<text x="250" y="458" text-anchor="middle" class="small">probes live server/discover</text>
<text x="250" y="476" text-anchor="middle" class="small">reports metadata/runtime drift</text>
<text x="250" y="498" text-anchor="middle" class="caption">Registry and MCP versions are separate</text>
<rect x="500" y="380" width="420" height="130" class="box"/>
<text x="710" y="402" text-anchor="middle" class="head">audit + load</text>
<text x="710" y="422" text-anchor="middle" class="small">per-tenant JSONL audit log</text>
<text x="710" y="440" text-anchor="middle" class="small">Presidio PII redaction before write</text>
<text x="710" y="458" text-anchor="middle" class="small">load test: 100 concurrent clients</text>
<text x="710" y="476" text-anchor="middle" class="small">horizontal scale via LB</text>
<text x="710" y="498" text-anchor="middle" class="caption">MCP conformance tests green</text>
<text x="710" y="476" text-anchor="middle" class="small">two replicas without affinity</text>
<text x="710" y="498" text-anchor="middle" class="caption">receiver-side wire evidence</text>
</svg>
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 5.3 KiB

After

Width:  |  Height:  |  Size: 5.4 KiB

phases/19-capstone-projects/13-mcp-server-with-registry/code/main.py
+486 -140
View File
@@ -1,28 +1,71 @@
"""MCP server + registry + OPA policy gate scaffold.
"""Stateless MCP server, registry metadata, policy, and audit simulation.
The hard architectural primitives are: (a) a stateless StreamableHTTP-style
dispatch that looks up a tool, checks scopes through an OPA-style policy,
and executes with audit log enrichment; (b) a registry that pulls
.well-known/mcp-capabilities from each server and validates. This scaffold
implements a minimal in-memory version of both so the handshakes are visible.
This stdlib-only model keeps two discovery layers separate:
Run: python main.py
* ``server.json`` describes installation and remote transport metadata to a
registry.
* ``server/discover`` reports live protocol versions and capabilities.
It does not open a network listener, validate a real OAuth token, call OPA, or
publish to a registry. Run with ``python3 code/main.py`` from the lesson root.
"""
from __future__ import annotations
import hashlib
import json
import re
import time
from copy import deepcopy
from dataclasses import asdict, dataclass, field
from typing import Callable
# ---------------------------------------------------------------------------
# tool schema -- typed input + required scope
# ---------------------------------------------------------------------------
PROTOCOL_VERSION = "2026-07-28"
REGISTRY_SCHEMA = "https://static.modelcontextprotocol.io/schemas/2025-12-11/server.schema.json"
PUBLISHER_DOMAIN = "example.com"
SERVER_NAME_RE = re.compile(r"^[a-zA-Z0-9.-]+/[a-zA-Z0-9._-]+$")
DOMAIN_LABEL_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$")
REMOTE_URL_RE = re.compile(r"^https?://\S+$")
VERSION_RANGE_RE = re.compile(r"(?:^[~^<>=]|[*]|(?:^|\.)x(?:$|\.))", re.IGNORECASE)
@dataclass
def request_meta() -> dict:
return {
"io.modelcontextprotocol/protocolVersion": PROTOCOL_VERSION,
"io.modelcontextprotocol/clientCapabilities": {},
"io.modelcontextprotocol/clientInfo": {
"name": "registry-capstone-client",
"version": "1.0.0",
},
}
def error(code: int, message: str, data: dict | None = None) -> dict:
detail = {"code": code, "message": message}
if data is not None:
detail["data"] = data
return {"error": detail}
def validate_meta(meta: object) -> dict | None:
if not isinstance(meta, dict):
return error(-32602, "params._meta must be an object")
requested = meta.get("io.modelcontextprotocol/protocolVersion")
if not isinstance(requested, str):
return error(-32602, "protocolVersion must be a string")
if requested != PROTOCOL_VERSION:
return error(
-32022,
"Unsupported protocol version",
{"supported": [PROTOCOL_VERSION], "requested": requested},
)
if not isinstance(meta.get("io.modelcontextprotocol/clientCapabilities"), dict):
return error(-32602, "clientCapabilities must be an object")
return None
@dataclass(frozen=True)
class ToolSchema:
name: str
required_scope: str
@@ -37,200 +80,503 @@ Handler = Callable[[dict], dict]
@dataclass
class MCPServer:
name: str
title: str
description: str
version: str
url: str
trusted_issuer: str
tools: dict[str, ToolSchema] = field(default_factory=dict)
handlers: dict[str, Handler] = field(default_factory=dict)
@property
def server_info(self) -> dict:
return {"name": self.name, "version": self.version}
def register(self, schema: ToolSchema, handler: Handler) -> None:
if schema.name in self.tools:
raise ValueError(f"duplicate tool: {schema.name}")
self.tools[schema.name] = schema
self.handlers[schema.name] = handler
def capabilities(self) -> dict:
"""The .well-known/mcp-capabilities document."""
def result(self, **fields: object) -> dict:
return {
"server": self.name,
"transport": "streamable_http",
"url": self.url,
"tools": [
{"name": t.name, "scope": t.required_scope,
"destructive": t.destructive,
"description": t.description,
"input_schema": t.input_schema}
for t in self.tools.values()
],
"resultType": "complete",
**fields,
"_meta": {"io.modelcontextprotocol/serverInfo": self.server_info},
}
def discover(self, meta: dict) -> dict:
invalid = validate_meta(meta)
if invalid:
return invalid
return self.result(
supportedVersions=[PROTOCOL_VERSION],
capabilities={"tools": {"listChanged": False}},
ttlMs=3_600_000,
cacheScope="public",
)
def tools_list(self, meta: dict) -> dict:
invalid = validate_meta(meta)
if invalid:
return invalid
tools = [
{
"name": tool.name,
"description": tool.description,
"inputSchema": tool.input_schema,
"annotations": {
"readOnlyHint": not tool.destructive,
"destructiveHint": tool.destructive,
},
}
for tool in sorted(self.tools.values(), key=lambda item: item.name)
]
return self.result(tools=tools, ttlMs=60_000, cacheScope="private")
def registry_document(self) -> dict:
return {
"$schema": REGISTRY_SCHEMA,
"name": self.name,
"title": self.title,
"description": self.description,
"version": self.version,
"remotes": [{"type": "streamable-http", "url": self.url}],
}
# ---------------------------------------------------------------------------
# OAuth-style scope set
# ---------------------------------------------------------------------------
@dataclass
@dataclass(frozen=True)
class Token:
user: str
scopes: set[str]
approved_at: float = 0.0 # epoch; scope_elevation freshness for destructive tools
issuer: str
audience: str
scopes: frozenset[str]
expires_at: float
def has_scope(self, s: str) -> bool:
return s in self.scopes
def has_scope(self, scope: str) -> bool:
return scope in self.scopes
def fresh_approval(self, now: float, window_s: int = 900) -> bool:
return "approved:by:human" in self.scopes and (now - self.approved_at) <= window_s
def is_expired(self, now: float) -> bool:
return now >= self.expires_at
# ---------------------------------------------------------------------------
# OPA-style policy -- Rego-like function over (tool, token, args)
# ---------------------------------------------------------------------------
def arguments_digest(args: dict) -> str:
normalized = json.dumps(
args,
ensure_ascii=False,
separators=(",", ":"),
sort_keys=True,
)
return hashlib.sha256(normalized.encode("utf-8")).hexdigest()
def policy_decide(server: MCPServer, tool: str, token: Token, args: dict,
now: float) -> tuple[bool, str]:
if tool not in server.tools:
@dataclass(frozen=True)
class ApprovalRecord:
actor: str
tool: str
arguments_digest: str
target: str
expires_at: float
@classmethod
def for_action(
cls,
actor: str,
tool: str,
args: dict,
target: str,
expires_at: float,
) -> ApprovalRecord:
return cls(actor, tool, arguments_digest(args), target, expires_at)
def authorize(
self,
actor: str,
tool: str,
args: dict,
target: str,
now: float,
) -> tuple[bool, str]:
if self.actor != actor:
return False, "approval actor does not match token subject"
if self.tool != tool:
return False, "approval tool does not match requested tool"
if self.target != target:
return False, "approval target does not match this server"
if self.arguments_digest != arguments_digest(args):
return False, "approval arguments do not match requested action"
if now >= self.expires_at:
return False, "approval has expired"
return True, "ok"
def policy_decide(
server: MCPServer,
tool: str,
token: Token,
args: dict,
now: float,
approval: ApprovalRecord | None = None,
) -> tuple[bool, str]:
if token.issuer != server.trusted_issuer:
return False, "token issuer is not trusted by this server"
if token.audience != server.url:
return False, "token audience does not match this server"
if token.is_expired(now):
return False, "token has expired"
schema = server.tools.get(tool)
if schema is None:
return False, f"no such tool: {tool}"
schema = server.tools[tool]
if not token.has_scope(schema.required_scope):
return False, f"missing scope: {schema.required_scope}"
if schema.destructive and not token.fresh_approval(now):
return False, "destructive tool requires fresh human approval (Slack card)"
# payload size cap example
if len(json.dumps(args)) > 8192:
return False, "payload too large (> 8 KB)"
return False, "payload too large"
if schema.destructive:
if approval is None:
return False, "destructive tool requires an action-bound approval"
approved, reason = approval.authorize(
token.user,
tool,
args,
server.url,
now,
)
if not approved:
return False, reason
return True, "ok"
# ---------------------------------------------------------------------------
# audit log -- structured JSONL with PII redaction
# ---------------------------------------------------------------------------
def redact(payload: dict) -> dict:
"""Presidio-style redaction stand-in: email, SSN, phone."""
s = json.dumps(payload)
s = re.sub(r"[\w.+-]+@[\w-]+\.[\w.-]+", "[email]", s)
s = re.sub(r"\b\d{3}-\d{2}-\d{4}\b", "[ssn]", s)
return json.loads(s)
text = json.dumps(payload)
text = re.sub(r"[\w.+-]+@[\w-]+\.[\w.-]+", "[email]", text)
text = re.sub(r"\b\d{3}-\d{2}-\d{4}\b", "[ssn]", text)
return json.loads(text)
@dataclass
@dataclass(frozen=True)
class AuditEntry:
ts: float
user: str
issuer: str
tool: str
outcome: str
args_redacted: dict
response_redacted: dict
# ---------------------------------------------------------------------------
# dispatch -- policy-gated tool invocation
# ---------------------------------------------------------------------------
def dispatch(server: MCPServer, token: Token, tool: str, args: dict,
audit: list[AuditEntry]) -> dict:
def dispatch(
server: MCPServer,
token: Token,
tool: str,
args: dict,
meta: dict,
audit: list[AuditEntry],
approval: ApprovalRecord | None = None,
) -> dict:
invalid = validate_meta(meta)
if invalid:
return invalid
now = time.time()
ok, reason = policy_decide(server, tool, token, args, now)
if not ok:
audit.append(AuditEntry(now, token.user, tool, f"denied:{reason}",
redact(args), {}))
return {"error": {"code": 403, "message": reason}}
handler = server.handlers[tool]
allowed, reason = policy_decide(server, tool, token, args, now, approval)
if not allowed:
audit.append(
AuditEntry(now, token.user, token.issuer, tool, f"denied:{reason}", redact(args), {})
)
return error(-32000, reason)
try:
result = handler(args)
audit.append(AuditEntry(now, token.user, tool, "ok",
redact(args), redact(result)))
return {"result": result}
response = server.handlers[tool](args)
except Exception as exc:
audit.append(AuditEntry(now, token.user, tool, f"error:{exc}",
redact(args), {}))
return {"error": {"code": 500, "message": str(exc)}}
audit.append(
AuditEntry(now, token.user, token.issuer, tool, "handler_error", redact(args), {})
)
return server.result(
content=[{"type": "text", "text": str(exc)}],
isError=True,
)
audit.append(
AuditEntry(
now,
token.user,
token.issuer,
tool,
"allowed",
redact(args),
redact(response),
)
)
return server.result(
content=[{"type": "text", "text": json.dumps(response)}],
structuredContent=response,
isError=False,
)
# ---------------------------------------------------------------------------
# registry -- polls capabilities and validates
# ---------------------------------------------------------------------------
def validate_registry_document(document: object) -> list[str]:
"""Validate the official fields used by this lesson's remote-only profile.
This dependency-free subset is not a replacement for validation against
the complete pinned Registry JSON Schema before publication.
"""
if not isinstance(document, dict):
return ["server.json must be an object"]
issues: list[str] = []
for key in ("name", "description", "version"):
if key not in document:
issues.append(f"missing {key}")
schema_uri = document.get("$schema")
if "$schema" in document and schema_uri != REGISTRY_SCHEMA:
issues.append("unsupported registry schema")
name = document.get("name")
if "name" in document and (
not isinstance(name, str)
or not 3 <= len(name) <= 200
or SERVER_NAME_RE.fullmatch(name) is None
):
issues.append("name must match namespace/server and be 3-200 characters")
description = document.get("description")
if "description" in document and (
not isinstance(description, str) or not 1 <= len(description) <= 100
):
issues.append("description must be a 1-100 character string")
title = document.get("title")
if "title" in document and (
not isinstance(title, str) or not 1 <= len(title) <= 100
):
issues.append("title must be a 1-100 character string")
version = document.get("version")
if "version" in document and (
not isinstance(version, str)
or not 1 <= len(version) <= 255
or VERSION_RANGE_RE.search(version) is not None
or version.casefold() == "latest"
):
issues.append("version must be one concrete 1-255 character version")
remotes = document.get("remotes")
if not isinstance(remotes, list) or not remotes:
issues.append("remote profile requires a non-empty remotes list")
else:
for index, remote in enumerate(remotes):
if not isinstance(remote, dict):
issues.append(f"remotes[{index}] must be an object")
continue
if remote.get("type") not in {"streamable-http", "sse"}:
issues.append(f"remotes[{index}].type must be streamable-http or sse")
remote_url = remote.get("url")
if not isinstance(remote_url, str) or REMOTE_URL_RE.fullmatch(remote_url) is None:
issues.append(f"remotes[{index}].url must be an http(s) URL template")
return issues
def reverse_dns_namespace(domain: str) -> str:
normalized = domain.casefold().rstrip(".")
labels = normalized.split(".")
if len(labels) < 2 or any(DOMAIN_LABEL_RE.fullmatch(label) is None for label in labels):
raise ValueError("publisher domain must be a valid multi-label DNS name")
return ".".join(reversed(labels))
def validate_publisher_namespace(document: object, verified_domain: str) -> list[str]:
"""Check domain ownership outside the server.json shape contract."""
if not isinstance(document, dict) or not isinstance(document.get("name"), str):
return []
namespace = document["name"].partition("/")[0]
expected = reverse_dns_namespace(verified_domain)
if namespace != expected and not namespace.startswith(f"{expected}."):
return [
f"name namespace must be {expected} or its child for verified domain {verified_domain}"
]
return []
def validate_runtime_alignment(document: dict, discovery: object) -> list[str]:
"""Compare publication identity with the live server/discover identity."""
if not isinstance(discovery, dict):
return ["server/discover result must be an object"]
meta = discovery.get("_meta")
server_info = (
meta.get("io.modelcontextprotocol/serverInfo") if isinstance(meta, dict) else None
)
if not isinstance(server_info, dict):
return ["server/discover must include serverInfo for registry drift checks"]
issues: list[str] = []
if server_info.get("name") != document.get("name"):
issues.append("runtime serverInfo.name does not match server.json name")
if server_info.get("version") != document.get("version"):
issues.append("runtime serverInfo.version does not match server.json version")
return issues
@dataclass
class Registry:
publisher_domain: str = PUBLISHER_DOMAIN
entries: dict[str, dict] = field(default_factory=dict)
runtime_discovery: dict[str, dict] = field(default_factory=dict)
def register(self, server: MCPServer) -> None:
self.entries[server.name] = server.capabilities()
document = server.registry_document()
issues = validate_registry_document(document)
issues.extend(validate_publisher_namespace(document, self.publisher_domain))
if issues:
raise ValueError("; ".join(issues))
discovery = server.discover(request_meta())
if "error" in discovery:
raise ValueError("runtime discovery failed")
alignment_issues = validate_runtime_alignment(document, discovery)
if alignment_issues:
raise ValueError("; ".join(alignment_issues))
self.entries[server.name] = deepcopy(document)
self.runtime_discovery[server.name] = deepcopy(discovery)
def search(self, query: str) -> list[tuple[str, str]]:
out: list[tuple[str, str]] = []
q = query.lower()
for server_name, cap in self.entries.items():
for t in cap["tools"]:
if q in t["name"].lower() or q in t["description"].lower():
out.append((server_name, t["name"]))
return out
def search(self, query: str) -> list[str]:
needle = query.casefold()
return sorted(
name
for name, entry in self.entries.items()
if needle in name.casefold()
or needle in entry["title"].casefold()
or needle in entry["description"].casefold()
)
# ---------------------------------------------------------------------------
# demo servers -- read-only and destructive
# ---------------------------------------------------------------------------
def build_readonly_server() -> MCPServer:
s = MCPServer(name="internal-readonly-mcp", url="https://mcp.internal/readonly")
s.register(ToolSchema("postgres.readonly", "postgres:query:readonly", False,
"Read-only Postgres query",
{"type": "object", "properties": {"sql": {"type": "string"}}}),
lambda a: {"rows": [[1]], "sql_echo": a.get("sql", "")})
s.register(ToolSchema("s3.list", "s3:list", False, "List S3 objects",
{"type": "object", "properties": {"bucket": {"type": "string"}}}),
lambda a: {"objects": [{"key": "a/b.txt", "size": 128}]})
s.register(ToolSchema("jira.search", "jira:read", False, "Search Jira issues",
{"type": "object", "properties": {"jql": {"type": "string"}}}),
lambda a: {"issues": [{"id": "PROJ-42", "title": "fix widget"}]})
return s
server = MCPServer(
name="com.example/internal-readonly",
title="Internal Read-Only Tools",
description="Read-only incident and data lookup tools.",
version="1.0.0",
url="https://mcp.internal.example.com/readonly",
trusted_issuer="https://auth.internal.example.com",
)
server.register(
ToolSchema(
"postgres.readonly",
"postgres:query:readonly",
False,
"Run an approved read-only query.",
{
"type": "object",
"properties": {"sql": {"type": "string"}},
"required": ["sql"],
"additionalProperties": False,
},
),
lambda args: {"rows": [[1]], "sql": args["sql"]},
)
server.register(
ToolSchema(
"s3.list",
"s3:list",
False,
"List objects in one approved bucket.",
{
"type": "object",
"properties": {"bucket": {"type": "string"}},
"required": ["bucket"],
"additionalProperties": False,
},
),
lambda args: {"bucket": args["bucket"], "objects": ["a/b.txt"]},
)
return server
def build_destructive_server() -> MCPServer:
s = MCPServer(name="internal-destructive-mcp", url="https://mcp.internal/destructive")
s.register(ToolSchema("jira.create", "jira:write", True, "Create Jira issue",
{"type": "object", "properties": {"title": {"type": "string"}}}),
lambda a: {"id": "PROJ-99", "created": True})
return s
server = MCPServer(
name="com.example/internal-destructive",
title="Internal Destructive Tools",
description="State-changing tools behind explicit approval.",
version="1.0.0",
url="https://mcp.internal.example.com/destructive",
trusted_issuer="https://auth.internal.example.com",
)
server.register(
ToolSchema(
"jira.create",
"jira:write",
True,
"Create one Jira issue after explicit approval.",
{
"type": "object",
"properties": {"title": {"type": "string"}},
"required": ["title"],
"additionalProperties": False,
},
),
lambda args: {"id": "PROJ-99", "title": args["title"], "created": True},
)
return server
def main() -> None:
ro = build_readonly_server()
rw = build_destructive_server()
readonly = build_readonly_server()
destructive = build_destructive_server()
registry = Registry()
registry.register(ro)
registry.register(rw)
registry.register(readonly)
registry.register(destructive)
audit: list[AuditEntry] = []
# token with read-only scopes
readonly_token = Token(user="u42", scopes={"postgres:query:readonly",
"s3:list",
"jira:read"})
# token with write scope but no fresh human approval
write_token_no_approval = Token(user="u42", scopes={"jira:write"})
# token with write scope AND approval fresh
write_token_approved = Token(user="u42",
scopes={"jira:write", "approved:by:human"},
approved_at=time.time() - 60)
readonly_token = Token(
"u42",
readonly.trusted_issuer,
readonly.url,
frozenset({"postgres:query:readonly", "s3:list"}),
time.time() + 3_600,
)
approved_token = Token(
"u42",
destructive.trusted_issuer,
destructive.url,
frozenset({"jira:write"}),
time.time() + 3_600,
)
approved_args = {"title": "new bug"}
approval = ApprovalRecord.for_action(
approved_token.user,
"jira.create",
approved_args,
destructive.url,
time.time() + 900,
)
print("=== registry search ===")
print(" 'jira' ->", registry.search("jira"))
print(" 'postgres' ->", registry.search("postgres"))
print("=== registry metadata and runtime discovery ===")
print(json.dumps(registry.entries[readonly.name], indent=2))
print(json.dumps(registry.runtime_discovery[readonly.name], indent=2))
print("tools:", json.dumps(readonly.tools_list(request_meta()), indent=2))
print("\n=== dispatch: postgres.readonly (read scope) ===")
r = dispatch(ro, readonly_token, "postgres.readonly",
{"sql": "SELECT email FROM users LIMIT 1"}, audit)
print(" ", r)
print("\n=== policy-gated calls ===")
print(
dispatch(
readonly,
readonly_token,
"postgres.readonly",
{"sql": "SELECT 1"},
request_meta(),
audit,
)
)
print(
dispatch(
destructive,
approved_token,
"jira.create",
approved_args,
request_meta(),
audit,
approval,
)
)
print("\n=== dispatch: jira.create without approval (expect deny) ===")
r = dispatch(rw, write_token_no_approval, "jira.create", {"title": "new bug"}, audit)
print(" ", r)
print("\n=== dispatch: jira.create with fresh approval ===")
r = dispatch(rw, write_token_approved, "jira.create", {"title": "new bug"}, audit)
print(" ", r)
print("\n=== audit log (redacted) ===")
for e in audit:
print(" ", json.dumps(asdict(e), default=str))
print("\n=== audit log ===")
for entry in audit:
print(json.dumps(asdict(entry), sort_keys=True))
if __name__ == "__main__":
phases/19-capstone-projects/13-mcp-server-with-registry/code/tests/test_main.py
+377
View File
@@ -0,0 +1,377 @@
"""Tests for the stateless MCP and registry boundary model."""
from __future__ import annotations
import sys
import time
import unittest
from pathlib import Path
from unittest.mock import patch
LESSON_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(LESSON_ROOT / "code"))
import main
class MCPRegistryCapstoneTests(unittest.TestCase):
def setUp(self) -> None:
self.readonly = main.build_readonly_server()
self.destructive = main.build_destructive_server()
def token(
self,
server: main.MCPServer,
*scopes: str,
issuer: str | None = None,
audience: str | None = None,
expires_at: float | None = None,
) -> main.Token:
return main.Token(
"learner",
issuer or server.trusted_issuer,
audience or server.url,
frozenset(scopes),
expires_at if expires_at is not None else time.time() + 3_600,
)
def approval(
self,
server: main.MCPServer,
args: dict,
*,
actor: str = "learner",
tool: str = "jira.create",
target: str | None = None,
expires_at: float | None = None,
) -> main.ApprovalRecord:
return main.ApprovalRecord.for_action(
actor,
tool,
args,
target or server.url,
expires_at if expires_at is not None else time.time() + 900,
)
def test_discover_advertises_current_revision_without_session_state(self) -> None:
result = self.readonly.discover(main.request_meta())
self.assertEqual(result["supportedVersions"], ["2026-07-28"])
self.assertEqual(result["ttlMs"], 3_600_000)
self.assertEqual(result["cacheScope"], "public")
self.assertEqual(result["resultType"], "complete")
self.assertNotIn("session", result)
self.assertEqual(
result["_meta"]["io.modelcontextprotocol/serverInfo"]["name"],
self.readonly.name,
)
def test_unsupported_version_uses_reserved_error_and_exact_data(self) -> None:
meta = main.request_meta()
meta["io.modelcontextprotocol/protocolVersion"] = "2025-11-25"
result = self.readonly.discover(meta)
self.assertEqual(result["error"]["code"], -32022)
self.assertEqual(
result["error"]["data"],
{"supported": ["2026-07-28"], "requested": "2025-11-25"},
)
def test_missing_or_non_string_version_is_invalid_params(self) -> None:
for requested in (None, 20260728, ["2026-07-28"]):
with self.subTest(requested=requested):
meta = main.request_meta()
if requested is None:
del meta["io.modelcontextprotocol/protocolVersion"]
else:
meta["io.modelcontextprotocol/protocolVersion"] = requested
result = self.readonly.discover(meta)
self.assertEqual(result["error"]["code"], -32602)
self.assertNotIn("data", result["error"])
def test_tools_list_is_deterministic_cacheable_and_typed(self) -> None:
result = self.readonly.tools_list(main.request_meta())
names = [tool["name"] for tool in result["tools"]]
self.assertEqual(names, sorted(names))
self.assertEqual(result["ttlMs"], 60_000)
self.assertEqual(result["cacheScope"], "private")
self.assertTrue(all("inputSchema" in tool for tool in result["tools"]))
def test_registry_document_and_runtime_discovery_remain_separate(self) -> None:
registry = main.Registry()
registry.register(self.readonly)
metadata = registry.entries[self.readonly.name]
runtime = registry.runtime_discovery[self.readonly.name]
self.assertEqual(metadata["$schema"], main.REGISTRY_SCHEMA)
self.assertEqual(metadata["remotes"][0]["type"], "streamable-http")
self.assertNotIn("tools", metadata)
self.assertIn("capabilities", runtime)
def test_example_domain_identity_uses_reverse_dns_in_both_layers(self) -> None:
self.assertEqual(main.reverse_dns_namespace(main.PUBLISHER_DOMAIN), "com.example")
for server in (self.readonly, self.destructive):
with self.subTest(server=server.name):
document = server.registry_document()
discovery = server.discover(main.request_meta())
server_info = discovery["_meta"]["io.modelcontextprotocol/serverInfo"]
self.assertTrue(document["name"].startswith("com.example/"))
self.assertEqual(server_info["name"], document["name"])
self.assertEqual(server_info["version"], document["version"])
self.assertEqual(
main.validate_publisher_namespace(document, main.PUBLISHER_DOMAIN),
[],
)
def test_registry_rejects_name_outside_verified_domain_namespace(self) -> None:
wrong_namespace = ".".join(("io", "example"))
self.readonly.name = f"{wrong_namespace}/internal-readonly"
with self.assertRaisesRegex(ValueError, "name namespace must be com.example"):
main.Registry().register(self.readonly)
def test_registry_rejects_publication_runtime_identity_drift(self) -> None:
mismatches = (
("name", "com.example/different-server", "runtime serverInfo.name"),
("version", "2.0.0", "runtime serverInfo.version"),
)
for field, value, expected in mismatches:
with self.subTest(field=field):
discovery = self.readonly.discover(main.request_meta())
discovery["_meta"]["io.modelcontextprotocol/serverInfo"][field] = value
with patch.object(self.readonly, "discover", return_value=discovery):
with self.assertRaisesRegex(ValueError, expected):
main.Registry().register(self.readonly)
def test_invalid_registry_document_is_rejected(self) -> None:
issues = main.validate_registry_document({"name": "missing-fields"})
self.assertIn("missing description", issues)
self.assertIn("missing version", issues)
self.assertIn("remote profile requires a non-empty remotes list", issues)
self.assertIn("name must match namespace/server and be 3-200 characters", issues)
def test_registry_subset_validates_official_field_shapes(self) -> None:
mutations = [
("name", "missing-slash", "name must match"),
("description", "", "description must be"),
("title", "x" * 101, "title must be"),
("title", None, "title must be"),
("version", "^1.2.3", "version must be"),
]
for key, value, expected in mutations:
with self.subTest(key=key):
document = self.readonly.registry_document()
document[key] = value
self.assertTrue(
any(expected in issue for issue in main.validate_registry_document(document))
)
for remote, expected in (
({"type": "stdio", "url": "https://example.com/mcp"}, ".type must be"),
({"type": "streamable-http", "url": "file:///tmp/mcp"}, ".url must be"),
({"type": "streamable-http"}, ".url must be"),
):
with self.subTest(remote=remote):
document = self.readonly.registry_document()
document["remotes"] = [remote]
self.assertTrue(
any(expected in issue for issue in main.validate_registry_document(document))
)
def test_registry_subset_accepts_schema_optional_and_official_sse_remote(self) -> None:
document = self.readonly.registry_document()
del document["$schema"]
document["remotes"] = [
{"type": "sse", "url": "https://mcp.internal.example.com/events"}
]
self.assertEqual(main.validate_registry_document(document), [])
def test_dispatch_requires_metadata_on_every_call(self) -> None:
audit: list[main.AuditEntry] = []
token = self.token(self.readonly, "postgres:query:readonly")
result = main.dispatch(
self.readonly,
token,
"postgres.readonly",
{"sql": "SELECT 1"},
{},
audit,
)
self.assertEqual(result["error"]["code"], -32602)
self.assertEqual(audit, [])
def test_token_audience_is_bound_to_one_server(self) -> None:
audit: list[main.AuditEntry] = []
wrong_audience = self.token(
self.destructive,
"jira:write",
audience=self.readonly.url,
)
result = main.dispatch(
self.destructive,
wrong_audience,
"jira.create",
{"title": "bad audience"},
main.request_meta(),
audit,
)
self.assertIn("audience", result["error"]["message"])
self.assertEqual(len(audit), 1)
def test_token_issuer_must_be_trusted(self) -> None:
audit: list[main.AuditEntry] = []
token = self.token(
self.readonly,
"postgres:query:readonly",
issuer="https://attacker.example.com",
)
result = main.dispatch(
self.readonly,
token,
"postgres.readonly",
{"sql": "SELECT 1"},
main.request_meta(),
audit,
)
self.assertIn("issuer", result["error"]["message"])
self.assertEqual(len(audit), 1)
def test_expired_token_is_rejected(self) -> None:
audit: list[main.AuditEntry] = []
token = self.token(
self.readonly,
"postgres:query:readonly",
expires_at=time.time() - 1,
)
result = main.dispatch(
self.readonly,
token,
"postgres.readonly",
{"sql": "SELECT 1"},
main.request_meta(),
audit,
)
self.assertIn("expired", result["error"]["message"])
def test_destructive_tool_requires_an_approval_record(self) -> None:
audit: list[main.AuditEntry] = []
token = self.token(self.destructive, "jira:write")
result = main.dispatch(
self.destructive,
token,
"jira.create",
{"title": "unsafe"},
main.request_meta(),
audit,
)
self.assertIn("action-bound approval", result["error"]["message"])
def test_approval_is_bound_to_actor_tool_target_and_expiry(self) -> None:
args = {"title": "approved change"}
token = self.token(self.destructive, "jira:write")
approvals = (
(self.approval(self.destructive, args, actor="other"), "actor"),
(self.approval(self.destructive, args, tool="other.tool"), "tool"),
(
self.approval(self.destructive, args, target="https://other.example.com/mcp"),
"target",
),
(
self.approval(self.destructive, args, expires_at=time.time() - 1),
"expired",
),
)
for approval, expected in approvals:
with self.subTest(expected=expected):
result = main.dispatch(
self.destructive,
token,
"jira.create",
args,
main.request_meta(),
[],
approval,
)
self.assertIn(expected, result["error"]["message"])
def test_approval_cannot_be_replayed_with_changed_arguments(self) -> None:
approved_args = {"title": "approved change"}
token = self.token(self.destructive, "jira:write")
approval = self.approval(self.destructive, approved_args)
result = main.dispatch(
self.destructive,
token,
"jira.create",
{"title": "different change"},
main.request_meta(),
[],
approval,
)
self.assertIn("arguments", result["error"]["message"])
def test_exact_action_approval_allows_destructive_call_without_magic_scope(self) -> None:
args = {"title": "approved change"}
token = self.token(self.destructive, "jira:write")
result = main.dispatch(
self.destructive,
token,
"jira.create",
args,
main.request_meta(),
[],
self.approval(self.destructive, args),
)
self.assertFalse(result["isError"])
self.assertTrue(result["structuredContent"]["created"])
def test_allowed_call_returns_complete_result_and_audit_record(self) -> None:
audit: list[main.AuditEntry] = []
token = self.token(self.readonly, "postgres:query:readonly")
result = main.dispatch(
self.readonly,
token,
"postgres.readonly",
{"sql": "SELECT 1"},
main.request_meta(),
audit,
)
self.assertEqual(result["resultType"], "complete")
self.assertFalse(result["isError"])
self.assertEqual(result["structuredContent"]["rows"], [[1]])
self.assertEqual(audit[0].outcome, "allowed")
def test_redaction_happens_before_audit_persistence(self) -> None:
redacted = main.redact({"email": "learner@example.com", "ssn": "123-45-6789"})
self.assertEqual(redacted, {"email": "[email]", "ssn": "[ssn]"})
if __name__ == "__main__":
unittest.main()
phases/19-capstone-projects/13-mcp-server-with-registry/code/ts/README.md
+23 -5
View File
@@ -1,9 +1,24 @@
# Lesson 13 - Internal MCP Server (TypeScript)
# Lesson 13 - Stateless MCP Server (TypeScript)
TypeScript half of the capstone. The Python side (`code/main.py`) ships the
registry and policy gate; this project is the MCP transport: hand-rolled
newline-delimited JSON-RPC 2.0 over stdio with three mock incident tools. No
`@modelcontextprotocol/sdk`; you get to see every byte on the wire.
newline-delimited JSON-RPC 2.0 over stdio with three mock incident tools. It
implements MCP `2026-07-28` directly, without `@modelcontextprotocol/sdk`, so
you can inspect every byte on the wire.
The protocol is stateless even though the mock incident store persists data.
Every request repeats its protocol version and client capabilities in
`params._meta`; no connection, process, or earlier request establishes a
session. The server exposes mandatory `server/discover`, identifies itself in
every successful result, and publishes deterministic, cacheable tool listings.
`tools/call` validates arguments against the same bounded schemas returned by
`tools/list`; malformed arguments for a known tool return a complete tool result
with `isError: true` and never reach its executor.
The runtime identity is `com.example/internal-incidents`. It uses the reverse-DNS
namespace for the verified `example.com` publisher. A matching published
`server.json` must use that same name even though the local npm package has its
own private project name.
## Layout
@@ -11,11 +26,11 @@ newline-delimited JSON-RPC 2.0 over stdio with three mock incident tools. No
src/
index.ts entry: fixture demo (default) or stdio loop (--serve)
transport.ts stdin readline + fixture replay
protocol.ts initialize / tools/list / tools/call / shutdown
protocol.ts request validation / server/discover / tools/list / tools/call
tools.ts three incident tools + executors
types.ts JSON-RPC + tool shapes
tests/
protocol.test.ts roundtrip, list shape, dispatch, parse error
protocol.test.ts stateless metadata, discovery, tools, errors, roundtrip
```
## Run
@@ -27,3 +42,6 @@ npm test
npm start # self-terminating fixture demo
npm run serve # real stdio loop (waits on stdin)
```
The demo is self-terminating. The real stdio server stays alive until its input
stream closes; there is no MCP shutdown request or initialization handshake.
phases/19-capstone-projects/13-mcp-server-with-registry/code/ts/src/index.ts
+28 -52
View File
@@ -1,79 +1,55 @@
// Internal MCP server: TypeScript skeleton, hand-rolled stdio JSON-RPC.
// Python side ships the registry and policy gate; this project is the MCP
// transport with three mock incident tools.
// Refs: docs/en.md (this lesson),
// MCP 2025-11-25 spec: https://modelcontextprotocol.io/specification/2025-11-25
// JSON-RPC 2.0: https://www.jsonrpc.org/specification
// MCP registry 2026: https://github.com/modelcontextprotocol/registry
import type { JsonRpcRequest } from "./types.js";
import { makeState, PROTOCOL_VERSION } from "./protocol.js";
import { makeContext, makeRequest } from "./protocol.js";
import { replayFixture, serveStdio } from "./transport.js";
import { TOOL_DESCRIPTORS, makeExecutors, makeIncidents } from "./tools.js";
function demoFixture(): JsonRpcRequest[] {
return [
{ jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: PROTOCOL_VERSION } },
{ jsonrpc: "2.0", id: 2, method: "tools/list" },
{
jsonrpc: "2.0",
id: 3,
method: "tools/call",
params: { name: "incidents_list", arguments: { severity: "p0" } },
},
{
jsonrpc: "2.0",
id: 4,
method: "tools/call",
params: { name: "incidents_get", arguments: { id: "INC-101" } },
},
{
jsonrpc: "2.0",
id: 5,
method: "tools/call",
params: { name: "incidents_ack", arguments: { id: "INC-101" } },
},
{
jsonrpc: "2.0",
id: 6,
method: "tools/call",
params: { name: "incidents_get", arguments: { id: "INC-101" } },
},
{
jsonrpc: "2.0",
id: 7,
method: "tools/call",
params: { name: "no_such_tool", arguments: {} },
},
{ jsonrpc: "2.0", id: 8, method: "shutdown" },
{ jsonrpc: "2.0", method: "notifications/initialized" },
makeRequest(1, "server/discover"),
makeRequest(2, "tools/list"),
makeRequest(3, "tools/call", {
name: "incidents_list",
arguments: { severity: "p0" },
}),
makeRequest(4, "tools/call", {
name: "incidents_get",
arguments: { id: "INC-101" },
}),
makeRequest(5, "tools/call", {
name: "incidents_ack",
arguments: { id: "INC-101" },
}),
makeRequest(6, "tools/call", {
name: "incidents_get",
arguments: { id: "INC-101" },
}),
makeRequest(7, "tools/call", { name: "no_such_tool", arguments: {} }),
makeRequest(8, "tools/list", {}, "2027-01-01"),
];
}
function runDemo(): void {
const state = makeState(TOOL_DESCRIPTORS, makeExecutors(makeIncidents()));
const context = makeContext(TOOL_DESCRIPTORS, makeExecutors(makeIncidents()));
process.stdout.write("=".repeat(72) + "\n");
process.stdout.write("PHASE 19 LESSON 13 - internal MCP server (TypeScript, no SDK)\n");
process.stdout.write("PHASE 19 LESSON 13 - stateless MCP server (TypeScript, no SDK)\n");
process.stdout.write("=".repeat(72) + "\n");
const messages = demoFixture();
const replies = replayFixture(state, messages);
const responders = messages.filter((m) => m.id !== undefined);
for (let i = 0; i < responders.length; i += 1) {
const req = responders[i];
const replies = replayFixture(context, messages);
for (let i = 0; i < messages.length; i += 1) {
const req = messages[i];
const rep = replies[i];
if (!req || !rep) continue;
process.stdout.write("\n>>> " + JSON.stringify(req) + "\n");
process.stdout.write("<<< " + JSON.stringify(rep) + "\n");
}
process.stdout.write("\nnotification (no response) processed for notifications/initialized\n");
}
function main(): void {
if (process.argv.includes("--serve")) {
const state = makeState(TOOL_DESCRIPTORS, makeExecutors(makeIncidents()));
serveStdio(state);
const context = makeContext(TOOL_DESCRIPTORS, makeExecutors(makeIncidents()));
serveStdio(context);
return;
}
runDemo();
phases/19-capstone-projects/13-mcp-server-with-registry/code/ts/src/protocol.ts
+203 -45
View File
@@ -1,84 +1,236 @@
import type {
JsonSchema,
JsonRpcRequest,
JsonRpcRequestId,
JsonRpcResponse,
ToolArgs,
ToolDescriptor,
ToolExecutor,
} from "./types.js";
export const PROTOCOL_VERSION = "2025-11-25";
export const SERVER_INFO = { name: "lesson-13-internal-mcp", version: "1.0.0" };
export const PROTOCOL_VERSION = "2026-07-28";
export const SUPPORTED_VERSIONS = [PROTOCOL_VERSION] as const;
export const PROTOCOL_VERSION_KEY = "io.modelcontextprotocol/protocolVersion";
export const CLIENT_CAPABILITIES_KEY = "io.modelcontextprotocol/clientCapabilities";
export const CLIENT_INFO_KEY = "io.modelcontextprotocol/clientInfo";
export const SERVER_INFO_KEY = "io.modelcontextprotocol/serverInfo";
export const SERVER_NAME = "com.example/internal-incidents";
export const SERVER_INFO = { name: SERVER_NAME, version: "1.0.0" };
export const SERVER_CAPABILITIES = { tools: { listChanged: false } };
export type ProtocolState = {
export type ServerContext = {
descriptors: ToolDescriptor[];
executors: Record<string, ToolExecutor>;
shutdownRequested: boolean;
};
export function makeState(
export function makeContext(
descriptors: ToolDescriptor[],
executors: Record<string, ToolExecutor>,
): ProtocolState {
return { descriptors, executors, shutdownRequested: false };
): ServerContext {
return { descriptors, executors };
}
function handleInitialize(): unknown {
class RpcProblem extends Error {
constructor(
readonly code: number,
message: string,
readonly data?: unknown,
) {
super(message);
}
}
function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value);
}
function validateSchema(value: unknown, schema: JsonSchema, path: string): string[] {
if (schema.type === "object") {
if (!isRecord(value)) return [`${path} must be an object`];
const issues: string[] = [];
const properties = schema.properties ?? {};
for (const required of schema.required ?? []) {
if (!Object.hasOwn(value, required)) issues.push(`${path}.${required} is required`);
}
if (schema.additionalProperties === false) {
for (const key of Object.keys(value)) {
if (!Object.hasOwn(properties, key)) issues.push(`${path}.${key} is not allowed`);
}
}
for (const [key, propertySchema] of Object.entries(properties)) {
if (Object.hasOwn(value, key)) {
issues.push(...validateSchema(value[key], propertySchema, `${path}.${key}`));
}
}
return issues;
}
if (schema.type === "string" && typeof value !== "string") {
return [`${path} must be a string`];
}
if (schema.enum && !schema.enum.includes(value as string)) {
return [`${path} must be one of ${schema.enum.join(", ")}`];
}
return [];
}
export function requestMeta(
version = PROTOCOL_VERSION,
clientCapabilities: Record<string, unknown> = {},
): Record<string, unknown> {
return {
protocolVersion: PROTOCOL_VERSION,
capabilities: { tools: { listChanged: false } },
serverInfo: SERVER_INFO,
[PROTOCOL_VERSION_KEY]: version,
[CLIENT_CAPABILITIES_KEY]: clientCapabilities,
[CLIENT_INFO_KEY]: { name: "lesson-13-demo-client", version: "1.0.0" },
};
}
function handleToolsList(state: ProtocolState): unknown {
return { tools: state.descriptors };
export function makeRequest(
id: JsonRpcRequestId,
method: string,
params: Record<string, unknown> = {},
version = PROTOCOL_VERSION,
): JsonRpcRequest {
return {
jsonrpc: "2.0",
id,
method,
params: { ...params, _meta: requestMeta(version) },
};
}
function handleToolsCall(state: ProtocolState, params: Record<string, unknown>): unknown {
const name = String(params.name ?? "");
const args = (params.arguments as ToolArgs | undefined) ?? {};
const fn = state.executors[name];
if (!fn) {
return { content: [{ type: "text", text: `unknown tool: ${name}` }], isError: true };
function complete(
payload: Record<string, unknown>,
cache?: { ttlMs: number; cacheScope: "public" | "private" },
): Record<string, unknown> {
const result: Record<string, unknown> = {
resultType: "complete",
...payload,
_meta: { [SERVER_INFO_KEY]: { ...SERVER_INFO } },
};
if (cache) {
result.ttlMs = cache.ttlMs;
result.cacheScope = cache.cacheScope;
}
return result;
}
function validateRequest(msg: JsonRpcRequest): Record<string, unknown> {
if (!isRecord(msg.params)) {
throw new RpcProblem(-32602, "params must be an object");
}
const meta = msg.params._meta;
if (!isRecord(meta)) {
throw new RpcProblem(-32602, "params._meta is required");
}
const requested = meta[PROTOCOL_VERSION_KEY];
if (typeof requested !== "string") {
throw new RpcProblem(-32602, `${PROTOCOL_VERSION_KEY} is required`);
}
if (!(SUPPORTED_VERSIONS as readonly string[]).includes(requested)) {
throw new RpcProblem(-32022, "Unsupported protocol version", {
supported: [...SUPPORTED_VERSIONS],
requested,
});
}
if (!isRecord(meta[CLIENT_CAPABILITIES_KEY])) {
throw new RpcProblem(-32602, `${CLIENT_CAPABILITIES_KEY} is required`);
}
const clientInfo = meta[CLIENT_INFO_KEY];
if (
clientInfo !== undefined &&
(!isRecord(clientInfo) ||
typeof clientInfo.name !== "string" ||
typeof clientInfo.version !== "string")
) {
throw new RpcProblem(-32602, `${CLIENT_INFO_KEY} is malformed`);
}
return msg.params;
}
function handleDiscover(): Record<string, unknown> {
return complete(
{
supportedVersions: [...SUPPORTED_VERSIONS],
capabilities: SERVER_CAPABILITIES,
instructions: "Use incident tools to list, inspect, and acknowledge incidents.",
},
{ ttlMs: 3_600_000, cacheScope: "public" },
);
}
function handleToolsList(context: ServerContext): Record<string, unknown> {
const tools = [...context.descriptors].sort((left, right) =>
left.name < right.name ? -1 : left.name > right.name ? 1 : 0,
);
return complete({ tools }, { ttlMs: 300_000, cacheScope: "public" });
}
function handleToolsCall(
context: ServerContext,
params: Record<string, unknown>,
): Record<string, unknown> {
const name = params.name;
const rawArgs = params.arguments ?? {};
if (typeof name !== "string" || !name) {
throw new RpcProblem(-32602, "tools/call requires a non-empty string name");
}
const executor = context.executors[name];
const descriptor = context.descriptors.find((candidate) => candidate.name === name);
if (!executor || !descriptor) {
throw new RpcProblem(-32602, `Unknown tool: ${name}`);
}
const schemaIssues = validateSchema(rawArgs, descriptor.inputSchema, "arguments");
if (schemaIssues.length > 0) {
return complete({
content: [
{
type: "text",
text: `Invalid arguments for ${name}: ${schemaIssues.join("; ")}`,
},
],
isError: true,
});
}
try {
return { content: fn(args), isError: false };
return complete({ content: executor(rawArgs as ToolArgs), isError: false });
} catch (err) {
return { content: [{ type: "text", text: String(err) }], isError: true };
return complete({ content: [{ type: "text", text: String(err) }], isError: true });
}
}
function handleShutdown(state: ProtocolState): unknown {
state.shutdownRequested = true;
return {};
function rpcError(
id: JsonRpcRequestId,
code: number,
message: string,
data?: unknown,
): JsonRpcResponse {
return {
jsonrpc: "2.0",
id,
error: data === undefined ? { code, message } : { code, message, data },
};
}
export function dispatch(state: ProtocolState, msg: JsonRpcRequest): JsonRpcResponse | null {
if (msg.id === undefined) {
return null;
}
export function dispatch(context: ServerContext, msg: JsonRpcRequest): JsonRpcResponse | null {
if (msg.id === undefined) return null;
const id = msg.id;
const params = msg.params ?? {};
try {
if (msg.method === "initialize") {
return { jsonrpc: "2.0", id, result: handleInitialize() };
const params = validateRequest(msg);
if (msg.method === "server/discover") {
return { jsonrpc: "2.0", id, result: handleDiscover() };
}
if (msg.method === "tools/list") {
return { jsonrpc: "2.0", id, result: handleToolsList(state) };
return { jsonrpc: "2.0", id, result: handleToolsList(context) };
}
if (msg.method === "tools/call") {
return { jsonrpc: "2.0", id, result: handleToolsCall(state, params) };
return { jsonrpc: "2.0", id, result: handleToolsCall(context, params) };
}
if (msg.method === "shutdown") {
return { jsonrpc: "2.0", id, result: handleShutdown(state) };
}
return {
jsonrpc: "2.0",
id,
error: { code: -32601, message: `Method not found: ${msg.method}` },
};
return rpcError(id, -32601, `Method not found: ${msg.method}`);
} catch (err) {
return { jsonrpc: "2.0", id, error: { code: -32603, message: String(err) } };
if (err instanceof RpcProblem) {
return rpcError(id, err.code, err.message, err.data);
}
return rpcError(id, -32603, "Internal error", { detail: String(err) });
}
}
@@ -91,9 +243,15 @@ export function parseRpc(
} catch (err) {
return { ok: false, err: String(err), code: -32700 };
}
const m = raw as JsonRpcRequest;
if (!m || typeof m !== "object" || m.jsonrpc !== "2.0" || typeof m.method !== "string") {
if (!isRecord(raw) || raw.jsonrpc !== "2.0" || typeof raw.method !== "string") {
return { ok: false, err: "invalid JSON-RPC envelope", code: -32600 };
}
return { ok: true, msg: m };
if (
"id" in raw &&
typeof raw.id !== "string" &&
typeof raw.id !== "number"
) {
return { ok: false, err: "invalid JSON-RPC envelope", code: -32600 };
}
return { ok: true, msg: raw as JsonRpcRequest };
}
phases/19-capstone-projects/13-mcp-server-with-registry/code/ts/src/tools.ts
+3
View File
@@ -17,6 +17,7 @@ export const TOOL_DESCRIPTORS: ToolDescriptor[] = [
type: "object",
properties: { severity: { type: "string", enum: ["p0", "p1", "p2"] } },
required: [],
additionalProperties: false,
},
annotations: { readOnlyHint: true },
},
@@ -27,6 +28,7 @@ export const TOOL_DESCRIPTORS: ToolDescriptor[] = [
type: "object",
properties: { id: { type: "string" } },
required: ["id"],
additionalProperties: false,
},
annotations: { readOnlyHint: true },
},
@@ -37,6 +39,7 @@ export const TOOL_DESCRIPTORS: ToolDescriptor[] = [
type: "object",
properties: { id: { type: "string" } },
required: ["id"],
additionalProperties: false,
},
annotations: { destructiveHint: true, readOnlyHint: false },
},
phases/19-capstone-projects/13-mcp-server-with-registry/code/ts/src/transport.ts
+7 -9
View File
@@ -1,10 +1,10 @@
import { createInterface } from "node:readline";
import type { JsonRpcRequest, JsonRpcResponse } from "./types.js";
import { dispatch, parseRpc, type ProtocolState } from "./protocol.js";
import { dispatch, parseRpc, type ServerContext } from "./protocol.js";
export type LineSink = (line: string) => void;
export function processLine(state: ProtocolState, line: string, sink: LineSink): void {
export function processLine(context: ServerContext, line: string, sink: LineSink): void {
const trimmed = line.trim();
if (!trimmed) return;
const parsed = parseRpc(trimmed);
@@ -18,28 +18,26 @@ export function processLine(state: ProtocolState, line: string, sink: LineSink):
sink(JSON.stringify(err));
return;
}
const resp = dispatch(state, parsed.msg);
const resp = dispatch(context, parsed.msg);
if (resp) sink(JSON.stringify(resp));
}
export function replayFixture(
state: ProtocolState,
context: ServerContext,
messages: JsonRpcRequest[],
): JsonRpcResponse[] {
const out: JsonRpcResponse[] = [];
for (const msg of messages) {
const reply = dispatch(state, msg);
const reply = dispatch(context, msg);
if (reply) out.push(reply);
}
return out;
}
export function serveStdio(state: ProtocolState): void {
export function serveStdio(context: ServerContext): void {
const rl = createInterface({ input: process.stdin, terminal: false });
const sink: LineSink = (line) => process.stdout.write(line + "\n");
rl.on("line", (line) => {
processLine(state, line, sink);
if (state.shutdownRequested) rl.close();
processLine(context, line, sink);
});
rl.on("close", () => process.exit(0));
}
phases/19-capstone-projects/13-mcp-server-with-registry/code/ts/src/types.ts
+5 -3
View File
@@ -1,8 +1,9 @@
export type JsonRpcId = number | string | null;
export type JsonRpcRequestId = number | string;
export type JsonRpcResponseId = JsonRpcRequestId | null;
export type JsonRpcRequest = {
jsonrpc: "2.0";
id?: JsonRpcId;
id?: JsonRpcRequestId;
method: string;
params?: Record<string, unknown>;
};
@@ -15,7 +16,7 @@ export type JsonRpcError = {
export type JsonRpcResponse = {
jsonrpc: "2.0";
id: JsonRpcId;
id: JsonRpcResponseId;
result?: unknown;
error?: JsonRpcError;
};
@@ -25,6 +26,7 @@ export type JsonSchema = {
properties?: Record<string, JsonSchema>;
required?: string[];
enum?: string[];
additionalProperties?: boolean;
};
export type ToolAnnotations = {
phases/19-capstone-projects/13-mcp-server-with-registry/code/ts/tests/protocol.test.ts
+229 -63
View File
@@ -1,30 +1,131 @@
import { test } from "node:test";
import { strict as assert } from "node:assert";
import { dispatch, makeState, parseRpc, PROTOCOL_VERSION } from "../src/protocol.js";
import {
CLIENT_CAPABILITIES_KEY,
CLIENT_INFO_KEY,
dispatch,
makeContext,
makeRequest,
parseRpc,
PROTOCOL_VERSION,
PROTOCOL_VERSION_KEY,
SERVER_INFO_KEY,
SERVER_NAME,
} from "../src/protocol.js";
import { TOOL_DESCRIPTORS, makeExecutors, makeIncidents } from "../src/tools.js";
import { processLine, replayFixture } from "../src/transport.js";
import type { JsonRpcRequest } from "../src/types.js";
function freshState() {
return makeState(TOOL_DESCRIPTORS, makeExecutors(makeIncidents()));
function freshContext() {
return makeContext(TOOL_DESCRIPTORS, makeExecutors(makeIncidents()));
}
test("initialize returns protocol version and server info", () => {
const state = freshState();
const resp = dispatch(state, { jsonrpc: "2.0", id: 1, method: "initialize" });
test("server/discover returns the current contract and public cache policy", () => {
const resp = dispatch(freshContext(), makeRequest(1, "server/discover"));
assert.ok(resp);
assert.equal(resp.id, 1);
const result = resp.result as { protocolVersion: string; serverInfo: { name: string } };
assert.equal(result.protocolVersion, PROTOCOL_VERSION);
assert.equal(result.serverInfo.name, "lesson-13-internal-mcp");
const result = resp.result as {
resultType: string;
supportedVersions: string[];
capabilities: { tools: { listChanged: boolean } };
ttlMs: number;
cacheScope: string;
_meta: Record<string, { name: string }>;
};
assert.equal(result.resultType, "complete");
assert.deepEqual(result.supportedVersions, [PROTOCOL_VERSION]);
assert.equal(result.capabilities.tools.listChanged, false);
assert.equal(result.ttlMs, 3_600_000);
assert.equal(result.cacheScope, "public");
assert.equal(SERVER_NAME, "com.example/internal-incidents");
assert.equal(result._meta[SERVER_INFO_KEY]?.name, SERVER_NAME);
});
test("tools/list shape includes name + inputSchema for each tool", () => {
const state = freshState();
const resp = dispatch(state, { jsonrpc: "2.0", id: 2, method: "tools/list" });
test("every request requires params._meta", () => {
const resp = dispatch(freshContext(), {
jsonrpc: "2.0",
id: 2,
method: "tools/list",
params: {},
});
assert.ok(resp);
const result = resp.result as { tools: Array<{ name: string; inputSchema: unknown }> };
assert.equal(resp.error?.code, -32602);
});
test("every request requires client capabilities", () => {
const resp = dispatch(freshContext(), {
jsonrpc: "2.0",
id: 3,
method: "tools/list",
params: { _meta: { [PROTOCOL_VERSION_KEY]: PROTOCOL_VERSION } },
});
assert.ok(resp);
assert.equal(resp.error?.code, -32602);
});
test("unsupported versions return exact -32022 negotiation data", () => {
const resp = dispatch(freshContext(), makeRequest(4, "tools/list", {}, "2027-01-01"));
assert.ok(resp);
assert.equal(resp.error?.code, -32022);
assert.deepEqual(resp.error?.data, {
supported: [PROTOCOL_VERSION],
requested: "2027-01-01",
});
});
test("missing and non-string protocol versions return -32602 without negotiation data", () => {
const metas: Array<Record<string, unknown>> = [
{ [CLIENT_CAPABILITIES_KEY]: {} },
{ [PROTOCOL_VERSION_KEY]: 20260728, [CLIENT_CAPABILITIES_KEY]: {} },
{ [PROTOCOL_VERSION_KEY]: [PROTOCOL_VERSION], [CLIENT_CAPABILITIES_KEY]: {} },
];
for (const [index, meta] of metas.entries()) {
const resp = dispatch(freshContext(), {
jsonrpc: "2.0",
id: 40 + index,
method: "tools/list",
params: { _meta: meta },
});
assert.ok(resp);
assert.equal(resp.error?.code, -32602);
assert.equal(resp.error?.data, undefined);
}
});
test("malformed optional clientInfo is rejected", () => {
const resp = dispatch(freshContext(), {
jsonrpc: "2.0",
id: 5,
method: "tools/list",
params: {
_meta: {
[PROTOCOL_VERSION_KEY]: PROTOCOL_VERSION,
[CLIENT_CAPABILITIES_KEY]: {},
[CLIENT_INFO_KEY]: { name: "missing-version" },
},
},
});
assert.ok(resp);
assert.equal(resp.error?.code, -32602);
});
test("tools/list is complete, deterministic, and cacheable", () => {
const resp = dispatch(freshContext(), makeRequest(6, "tools/list"));
assert.ok(resp);
const result = resp.result as {
resultType: string;
tools: Array<{ name: string; inputSchema: unknown }>;
ttlMs: number;
cacheScope: string;
};
assert.equal(result.resultType, "complete");
assert.equal(result.tools.length, 3);
assert.deepEqual(
result.tools.map((tool) => tool.name),
["incidents_ack", "incidents_get", "incidents_list"],
);
assert.equal(result.ttlMs, 300_000);
assert.equal(result.cacheScope, "public");
for (const t of result.tools) {
assert.equal(typeof t.name, "string");
assert.ok(t.inputSchema);
@@ -32,67 +133,122 @@ test("tools/list shape includes name + inputSchema for each tool", () => {
});
test("tools/call dispatches to incidents_get", () => {
const state = freshState();
const resp = dispatch(state, {
jsonrpc: "2.0",
id: 3,
method: "tools/call",
params: { name: "incidents_get", arguments: { id: "INC-101" } },
});
const resp = dispatch(
freshContext(),
makeRequest(7, "tools/call", {
name: "incidents_get",
arguments: { id: "INC-101" },
}),
);
assert.ok(resp);
const result = resp.result as { isError: boolean; content: Array<{ text: string }> };
const result = resp.result as {
resultType: string;
isError: boolean;
content: Array<{ text: string }>;
_meta: Record<string, { version: string }>;
};
assert.equal(result.resultType, "complete");
assert.equal(result.isError, false);
assert.equal(result._meta[SERVER_INFO_KEY]?.version, "1.0.0");
const text = result.content[0]?.text ?? "";
assert.ok(text.includes("INC-101"));
});
test("tools/call unknown tool returns isError=true", () => {
const state = freshState();
const resp = dispatch(state, {
jsonrpc: "2.0",
id: 4,
method: "tools/call",
params: { name: "nope", arguments: {} },
});
test("tools/call unknown tool returns protocol error -32602", () => {
const resp = dispatch(
freshContext(),
makeRequest(8, "tools/call", { name: "nope", arguments: {} }),
);
assert.ok(resp);
const result = resp.result as { isError: boolean };
assert.equal(result.isError, true);
assert.equal(resp.error?.code, -32602);
assert.match(resp.error?.message ?? "", /Unknown tool/);
});
test("tools/call rejects a malformed tool name", () => {
const resp = dispatch(
freshContext(),
makeRequest(9, "tools/call", { name: "", arguments: {} }),
);
assert.ok(resp);
assert.equal(resp.error?.code, -32602);
});
test("known tools enforce every advertised input schema as complete tool errors", () => {
const cases = [
{ name: "incidents_list", arguments: { severity: "p3" } },
{ name: "incidents_get", arguments: {} },
{ name: "incidents_ack", arguments: { id: 101 } },
{ name: "incidents_list", arguments: { extra: true } },
{ name: "incidents_get", arguments: [] },
];
for (const [index, params] of cases.entries()) {
const resp = dispatch(
freshContext(),
makeRequest(50 + index, "tools/call", params),
);
assert.ok(resp);
assert.equal(resp.error, undefined);
const result = resp.result as {
resultType: string;
isError: boolean;
content: Array<{ text: string }>;
};
assert.equal(result.resultType, "complete");
assert.equal(result.isError, true);
assert.match(result.content[0]?.text ?? "", /Invalid arguments/);
}
});
test("incidents_ack flips acked state", () => {
const state = freshState();
dispatch(state, {
jsonrpc: "2.0",
id: 5,
method: "tools/call",
params: { name: "incidents_ack", arguments: { id: "INC-103" } },
});
const resp = dispatch(state, {
jsonrpc: "2.0",
id: 6,
method: "tools/call",
params: { name: "incidents_get", arguments: { id: "INC-103" } },
});
const context = freshContext();
dispatch(
context,
makeRequest(10, "tools/call", {
name: "incidents_ack",
arguments: { id: "INC-103" },
}),
);
const resp = dispatch(
context,
makeRequest(11, "tools/call", {
name: "incidents_get",
arguments: { id: "INC-103" },
}),
);
assert.ok(resp);
const text = (resp.result as { content: Array<{ text: string }> }).content[0]?.text ?? "";
assert.ok(text.includes('"acked":true'));
});
test("shutdown sets flag", () => {
const state = freshState();
dispatch(state, { jsonrpc: "2.0", id: 7, method: "shutdown" });
assert.equal(state.shutdownRequested, true);
test("requests never inherit metadata from a prior call", () => {
const context = freshContext();
assert.ok(dispatch(context, makeRequest(12, "tools/list"))?.result);
const missingMeta = dispatch(context, {
jsonrpc: "2.0",
id: 13,
method: "tools/list",
params: {},
});
assert.equal(missingMeta?.error?.code, -32602);
});
test("notification (no id) returns null", () => {
const state = freshState();
const resp = dispatch(state, { jsonrpc: "2.0", method: "notifications/initialized" });
test("JSON-RPC notifications return no response", () => {
const resp = dispatch(freshContext(), {
jsonrpc: "2.0",
method: "notifications/tools/list_changed",
});
assert.equal(resp, null);
});
test("unknown method returns -32601", () => {
const state = freshState();
const resp = dispatch(state, { jsonrpc: "2.0", id: 8, method: "no/such" });
test("legacy lifecycle methods are not implemented", () => {
const initialize = dispatch(freshContext(), makeRequest(14, "initialize"));
const shutdown = dispatch(freshContext(), makeRequest(15, "shutdown"));
assert.equal(initialize?.error?.code, -32601);
assert.equal(shutdown?.error?.code, -32601);
});
test("unknown modern method returns -32601", () => {
const resp = dispatch(freshContext(), makeRequest(16, "no/such"));
assert.ok(resp);
assert.equal(resp.error?.code, -32601);
});
@@ -103,21 +259,31 @@ test("parseRpc rejects malformed JSON", () => {
});
test("processLine emits -32700 envelope on parse failure", () => {
const state = freshState();
const lines: string[] = [];
processLine(state, "not json", (line) => lines.push(line));
processLine(freshContext(), "not json", (line) => lines.push(line));
assert.equal(lines.length, 1);
const parsed = JSON.parse(lines[0]!) as { error?: { code: number } };
assert.equal(parsed.error?.code, -32700);
});
test("replayFixture roundtrip drives full fixture sequence", () => {
const state = freshState();
test("parseRpc rejects a null MCP request id", () => {
const result = parseRpc('{"jsonrpc":"2.0","id":null,"method":"tools/list"}');
assert.equal(result.ok, false);
if (!result.ok) assert.equal(result.code, -32600);
});
test("replayFixture roundtrip drives only modern requests", () => {
const msgs: JsonRpcRequest[] = [
{ jsonrpc: "2.0", id: 1, method: "initialize" },
{ jsonrpc: "2.0", id: 2, method: "tools/list" },
{ jsonrpc: "2.0", method: "notifications/initialized" },
makeRequest(17, "server/discover"),
makeRequest(18, "tools/list"),
makeRequest(19, "tools/call", {
name: "incidents_list",
arguments: { severity: "p1" },
}),
];
const replies = replayFixture(state, msgs);
assert.equal(replies.length, 2);
const replies = replayFixture(freshContext(), msgs);
assert.equal(replies.length, 3);
for (const reply of replies) {
assert.equal((reply.result as { resultType: string }).resultType, "complete");
}
});
phases/19-capstone-projects/13-mcp-server-with-registry/docs/en.md
+281 -103
View File
@@ -1,152 +1,330 @@
# Capstone 13 — MCP Server with Registry and Governance
# Capstone 13: Stateless MCP Server with Registry and Governance
> The Model Context Protocol stopped being the future and became the default tool-use spec in 2026. Anthropic, OpenAI, Google, and every major IDE ship MCP clients. Pinterest published its internal ecosystem of MCP servers. The AAIF Registry formalized capability metadata at `.well-known`. AWS ECS published the reference stateless deployment. Block's goose-agent put the same protocol inside a hosted assistant. The 2026 production shape is: StreamableHTTP transport, OAuth 2.1 scopes, OPA policy gating, and a registry that lets platform teams discover, validate, and enable servers. Build that end to end.
> Production MCP is not one server process. It is a chain of contracts: publishable metadata, live discovery, a stateless request envelope, authorization, policy, audit, and deployment evidence.
**Type:** Capstone
**Languages:** Python (server, via FastMCP) or TypeScript (@modelcontextprotocol/sdk), Go (registry service)
**Prerequisites:** Phase 11 (LLM engineering), Phase 13 (tools and MCP), Phase 14 (agents), Phase 17 (infrastructure), Phase 18 (safety)
**Phases exercised:** P11 · P13 · P14 · P17 · P18
**Time:** 25 hours
**Languages:** Python and TypeScript reference models; any production language
**Prerequisites:** Phase 11, Phase 13, Phase 14, Phase 17, and Phase 18
**Protocol target:** MCP `2026-07-28`
**Time:** ~25 hours
## Problem
## Learning Objectives
MCP became the tool-use lingua franca. Claude Code, Cursor 3, Amp, OpenCode, Gemini CLI, and every managed agent now consume MCP servers. The production challenges are not authoring servers (FastMCP makes that easy) but deploying them at scale with enterprise requirements: per-tenant OAuth scopes, OPA policy on destructive tools, StreamableHTTP stateless scaling, a registry for discovery, audit logs per tool call. Pinterest's internal MCP ecosystem and the AAIF Registry spec set the 2026 bar.
- Implement the stateless MCP request and result envelope.
- Keep Registry metadata separate from live protocol discovery.
- Build deterministic, cache-aware tool discovery.
- Enforce issuer, audience, scope, and approval policy for every tool call.
- Deploy Streamable HTTP without session affinity.
- Prove behavior at the wire, authorization, policy, registry, and audit boundaries.
You will build an MCP server exposing 10 internal tools (Postgres read-only, S3 listing, Jira, Linear, Datadog, etc.), a registry UI for platform discovery, and a human-approval gate for destructive tools. The load test demonstrates StreamableHTTP horizontal scaling. The audit trail satisfies an enterprise security review.
## The Problem
## Concept
An internal platform needs read-only data tools and a small set of state-changing tools. Developers must be able to discover the server, understand how to connect, inspect its live capabilities, and call only the operations they are authorized to use.
MCP 2026 revision mandates StreamableHTTP as the default transport. Unlike the earlier stdio-and-SSE shape, StreamableHTTP is stateless by default: a single HTTP endpoint accepts JSON-RPC requests, streams responses, and supports long-lived connections for notifications. Stateless means horizontally scalable behind a load balancer.
The difficult part is not registering a function. The difficult part is keeping six different truths aligned:
Authorization is OAuth 2.1 with per-tool scopes. A token carries scopes like `jira:read`, `s3:list`, `postgres:query:readonly`. The MCP server checks scopes at tool-call time, not just session start. For high-risk tools, the server rejects any call whose scope is not elevated to `approved:by:human` within the last N minutes — that elevation comes from a Slack review card.
1. `server.json` says where the server can be installed or reached.
2. `server/discover` says what the live process supports now.
3. Every request says which protocol revision and client capabilities it uses.
4. Authorization binds a caller to the correct issuer, resource, and scopes.
5. Policy decides whether this specific action may run.
6. Audit evidence records what crossed the boundary without leaking secrets or sensitive payloads.
The registry is a separate service. Every MCP server exposes a `.well-known/mcp-capabilities` document with its tool manifest, transport URL, auth requirements. The registry polls, validates, and indexes. Platform teams use the registry UI to see what tools are available, what scopes they need, and which teams own them.
If any one of these drifts, the platform may list a server that cannot be reached, route an incompatible client, accept a token minted for another resource, or expose a destructive action without the expected review.
## Architecture
## The Two Discovery Layers
```
MCP client (Claude Code, Cursor 3, ...)
|
v
StreamableHTTP over HTTPS (JSON-RPC + streaming)
|
v
MCP server (FastMCP) behind load balancer
|
+------+------+---------+----------+------------+
v v v v v
Postgres S3 listing Jira Linear Datadog
(read-only) (paged) (read) (read) (query)
|
+------+-------------+
v v
OPA policy gate destructive tool MCP (separate server)
|
v
human approval via Slack
|
v
audit log (append-only, per-tenant)
The Registry and the live MCP server answer different questions.
registry service
|
v GET /.well-known/mcp-capabilities from each server
v
UI: search / validate / enable-disable / ownership
| Layer | Contract | Question it answers |
|---|---|---|
| Publication | `server.json` and Registry API | What is this server, where is its package or remote endpoint, and how is it configured? |
| Runtime | `server/discover` | Which protocol versions, capabilities, extensions, and server identity does this process support? |
The official Registry uses a versioned `server.json` schema. A remote entry can name a Streamable HTTP URL:
```json
{
"$schema": "https://static.modelcontextprotocol.io/schemas/2025-12-11/server.schema.json",
"name": "com.example/internal-readonly",
"title": "Internal Read-Only Tools",
"description": "Read-only incident and data lookup tools.",
"version": "1.0.0",
"remotes": [
{
"type": "streamable-http",
"url": "https://mcp.internal.example.com/readonly"
}
]
}
```
## Stack
The Registry schema version and the MCP protocol revision are independent. Do not rewrite one date to match the other. Validate each document against its own contract.
- Server framework: FastMCP (Python) or `@modelcontextprotocol/sdk` (TypeScript)
- Transport: StreamableHTTP over HTTPS (stateless)
- Auth: OAuth 2.1 with workload identity via SPIFFE / SPIRE
- Policy: OPA / Rego rules per tool; policy decision service per request
- Registry: self-hosted, consumes `.well-known/mcp-capabilities` manifests
- Human approval: Slack interactive message for destructive tools
- Deployment: AWS ECS Fargate or Fly.io, one server per tenant or shared with tenant scoping
- Audit: structured JSONL per-tenant bucket with per-call lineage
Schema validity does not prove namespace ownership. A publisher verified for `example.com` uses the reverse-DNS namespace `com.example/*` or one of its child namespaces. The Registry authentication flow proves that ownership. Keeping the domain labels in their ordinary order names a different namespace.
The stdlib model's `validate_registry_document` function is intentionally a partial remote-profile validator. It checks the official required `name`, `description`, and `version` fields; the optional `title`; the published name and length constraints; concrete-version shape; and each `streamable-http` or `sse` remote's HTTP(S) URL shape. It additionally requires a non-empty `remotes` list because this capstone always live-probes a remote. `validate_publisher_namespace` separately checks the name against the verified publisher domain, while `validate_runtime_alignment` compares the publication name and version with live `serverInfo`. The official schema also supports package-only records and more remote fields. Before publication, validate the entire document with the pinned official JSON Schema or `mcp-publisher`; do not present this dependency-free subset as full schema validation.
The server must implement `server/discover`; a client may call it before other methods. This capstone client does so after resolving the endpoint, and receives the current protocol revision and live capabilities:
```json
{
"resultType": "complete",
"supportedVersions": ["2026-07-28"],
"capabilities": {
"tools": {
"listChanged": false
}
},
"_meta": {
"io.modelcontextprotocol/serverInfo": {
"name": "com.example/internal-readonly",
"version": "1.0.0"
}
},
"ttlMs": 3600000,
"cacheScope": "public"
}
```
A private catalog may index extra ownership, review, or lifecycle data, but it must not invent that data as MCP wire fields or root `server.json` fields. Store organizational policy beside the published record. When public custom metadata is necessary, use the Registry's `_meta.io.modelcontextprotocol.registry/publisher-provided` extension and stay within its 4 KB limit.
## Stateless MCP Core
MCP revision `2026-07-28` removes protocol sessions and the `initialize` / `notifications/initialized` handshake. It also removes `Mcp-Session-Id`.
Every request carries protocol context in `params._meta`:
```json
{
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
"io.modelcontextprotocol/clientCapabilities": {},
"io.modelcontextprotocol/clientInfo": {
"name": "internal-platform-client",
"version": "1.0.0"
}
}
```
The version and capabilities are request facts, not connection facts. A load balancer may send consecutive requests to different healthy replicas because either replica can validate the request from the message itself.
Ordinary results include `resultType: "complete"`. Servers should place their identity in `_meta.io.modelcontextprotocol/serverInfo` on each result. A missing or non-string protocol version is invalid params `-32602`. Error `-32022` is only for a supplied string that is unsupported, with exactly `{"supported": ["2026-07-28"], "requested": "..."}` as its data.
### Cacheable discovery
`tools/list` must be deterministic for the same effective tool set. The result includes:
- `ttlMs`, a freshness hint for the client;
- `cacheScope`, either `public` or `private`;
- a stable tool order so identical lists can reuse prompt caches;
- `resultType: "complete"` and server identity metadata.
Per-user authorization should normally produce `cacheScope: "private"`. Do not put user-specific tool visibility behind a shared public cache.
## Streamable HTTP
A network server exposes one MCP endpoint that accepts POST. Each JSON-RPC request or notification gets its own POST.
For a request, the server returns either one JSON object or an SSE stream scoped to that request. A long-lived `subscriptions/listen` request carries opted-in change notifications. There is no standalone GET stream, session DELETE, session header, or `Last-Event-ID` replay in the current transport.
Each request includes:
- `MCP-Protocol-Version`, matching the body metadata;
- `Mcp-Method`, matching the JSON-RPC method;
- `Mcp-Name` for `tools/call`, `resources/read`, and `prompts/get`;
- `Accept: application/json, text/event-stream`.
Reject mismatched mirrored headers with the specified `-32020` error. Validate `Origin`, bind local development servers to loopback, authenticate remote clients, and treat a closed request-scoped SSE response as cancellation.
```mermaid
flowchart LR
R[Registry API] --> J[server.json]
J --> C[MCP client]
C --> D[server/discover]
C --> L[tools/list]
C --> G[Authorization and policy gateway]
G --> RO[Read-only MCP replicas]
G --> RW[State-changing MCP replicas]
RO --> A[Audit sink]
RW --> H[Approval record]
RW --> A
```
```figure
cf-mcp-gate
```
## Authorization and Policy
Transport metadata is not authorization. Validate authorization on every call.
For remote servers:
1. Discover protected-resource metadata.
2. Select the authorization server for that resource.
3. Prefer Client ID Metadata Documents for client registration. Treat Dynamic Client Registration as compatibility support.
4. Send the resource indicator during authorization.
5. Validate a returned `iss` value against the authorization server recorded for the flow.
6. Key client credentials by issuer. Never reuse registration data across issuers.
7. Validate token issuer, audience or resource, expiry, and scopes at the MCP server.
8. Apply a second policy decision to the concrete tool and arguments.
Tool annotations such as `readOnlyHint` and `destructiveHint` help clients present risk. They are not trusted authorization controls.
### Approval is a record, not a magic scope
A state-changing call needs an approval record bound to the actor, tool, normalized arguments or digest, target environment, expiry, and one-time or repeat-use policy. A chat message alone is not proof of approval.
The Python model hashes canonical JSON with sorted keys, then binds that digest with the token subject, tool name, server URL, and expiry. Replaying the record after changing even one argument fails before the handler runs. Approval is separate evidence, not a scope added to the access token.
Keep high-risk tools on a separately reviewable surface when that materially reduces blast radius. Separation is useful only if credentials, policy, deployment identity, and audit controls are also separate.
## Build It
1. **Tool surface.** Expose 10 internal tools: Postgres read-only query, S3 list objects, Jira search/fetch, Linear search/fetch, Datadog metric query, PagerDuty on-call lookup, GitHub read-only, Notion search, Slack search, Salesforce read. Each tool has a typed schema and a scope label.
### 1. Model publication metadata
2. **FastMCP server.** Mount the tools. Configure StreamableHTTP transport. Add a middleware for OAuth token introspection and scope enforcement.
Create and schema-validate `server.json`. Include a stable name inside the namespace authenticated for the publisher, plus version, description, official `repository` or `packages` metadata when applicable, and a remote or stdio transport. Keep secrets as declared environment-variable inputs, never literal values.
3. **OPA policy.** Rego policy per tool: what scopes permit invocation, what PII redaction applies, what payload-size caps apply. Decision service called on every tool call.
### 2. Implement live discovery
4. **Registry service.** Separate Go or TS service that polls `.well-known/mcp-capabilities` from registered servers, validates with JSON Schema, and exposes a list / search / validate / enable-disable UI.
Implement `server/discover` before any feature RPC. Advertise supported protocol versions, capabilities, extensions, and server identity. Add a version rejection case using `-32022`.
5. **Capability manifest.** Each server exposes `.well-known/mcp-capabilities` with: tool list, auth requirements, transport URL, owner team, SLO.
### 3. Implement the stateless envelope
6. **Destructive tool separation.** Tools that mutate state (Jira create, Linear create, Postgres write) live on a second MCP server with a stricter auth flow: tokens must have a `approved:by:human` scope elevated via Slack card within 15 minutes.
Require protocol version and client capabilities in every request. Return `resultType` and server identity in every result. Remove initialization state, connection-scoped capability caches, and session identifiers.
7. **Audit log.** Append-only JSONL per tenant: `{timestamp, user, tool, args_redacted, response_redacted, outcome}`. PII redaction via Presidio before write.
### 4. Build the tool surface
8. **Load test.** 100 concurrent clients on StreamableHTTP. Demonstrate horizontal scaling by adding a second replica; show the load balancer redistributing without session stickiness.
Start with two read-only tools and one state-changing tool. Give each a bounded JSON Schema, precise description, deterministic result shape, and honest annotations. Add output schemas when clients rely on structured results.
9. **Conformance tests.** Run the official MCP conformance suite against both servers. Pass all mandatory sections.
### 5. Add cache-aware listing
## Use It
Return tools in stable order with `ttlMs` and `cacheScope`. Exercise cache expiry and list-change notification behavior separately.
### 6. Add authorization and policy
Validate issuer, audience, expiry, and scope. Run a policy decision for every tool call. Bind approvals to exact high-risk actions. Deny missing or stale approvals before executing a handler.
### 7. Separate registry and runtime validation
Validate the static `server.json` record, then probe the remote endpoint with `server/discover`. Report drift when the published remote, identity, version, or required capabilities disagree with the live process.
### 8. Add audit evidence
Record actor, issuer, resource, tool, policy decision, request identifier, trace context, latency, and outcome. Redact or digest sensitive arguments and results before persistence. Keep the audit sink outside model-visible context.
### 9. Exercise horizontal scaling
Place two stateless replicas behind a load balancer. Send at least 100 concurrent requests. Demonstrate that correctness does not depend on affinity. If a tool needs cross-call state, mint an explicit opaque handle and store it in a shared durable system.
### 10. Cross the real wire
Run conformance checks against the actual server binary. Capture request headers and JSON bodies, not only SDK objects. Exercise wrong version, header mismatch, missing scope, wrong audience, malformed arguments, handler failure, cancellation, and cache expiry.
## Local Reference Models
The Python model demonstrates registry metadata, reverse-DNS publisher namespace validation, publication-to-runtime identity checks, live discovery, deterministic tool listing, per-request metadata, trusted-issuer, audience, expiry, and scope checks, action-bound approvals, a documented partial Registry validator, policy, and audit without opening a network socket:
```bash
cd phases/19-capstone-projects/13-mcp-server-with-registry
python3 code/main.py
python3 -m unittest discover -s code/tests -v
```
$ curl -H "Authorization: Bearer eyJhbGc..." \
-X POST https://mcp.internal.example.com/ \
-d '{"jsonrpc":"2.0","method":"tools/call",
"params":{"name":"postgres.readonly","arguments":{"sql":"SELECT 1"}}}'
[registry] capability validated: postgres.readonly v1.2
[policy] scope postgres:query:readonly present; allowed
[audit] logged: user=u42 tool=postgres.readonly outcome=ok
response: { "result": { "rows": [[1]] } }
The TypeScript project exposes the stateless JSON-RPC shape over stdio without an MCP SDK. Its `tools/call` path enforces the same bounded input schemas advertised by `tools/list`; invalid arguments for a known tool return a complete result with `isError: true` without invoking the executor:
```bash
cd phases/19-capstone-projects/13-mcp-server-with-registry/code/ts
npm install
npm run typecheck
npm test
npm run demo
```
These models prove local contract logic. They do not prove HTTP headers, OAuth exchange, Registry publication, OPA integration, load balancing, or collector receipt.
## Wire Example
```http
POST /mcp HTTP/1.1
Host: mcp.internal.example.com
Content-Type: application/json
Accept: application/json, text/event-stream
MCP-Protocol-Version: 2026-07-28
Mcp-Method: tools/call
Mcp-Name: postgres.readonly
Authorization: Bearer REDACTED
{
"jsonrpc": "2.0",
"id": 42,
"method": "tools/call",
"params": {
"name": "postgres.readonly",
"arguments": {"sql": "SELECT 1"},
"_meta": {
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
"io.modelcontextprotocol/clientCapabilities": {},
"io.modelcontextprotocol/clientInfo": {
"name": "internal-platform-client",
"version": "1.0.0"
}
}
}
}
```
## Ship It
`outputs/skill-mcp-server.md` describes the deliverable. A production-grade MCP server + registry + audit layer for internal tools with OAuth 2.1 scopes and OPA gating.
Ship a repository containing:
| Weight | Criterion | How it is measured |
|:-:|---|---|
| 25 | Spec conformance | StreamableHTTP + capability manifest passes MCP conformance tests |
| 20 | Security | Scope enforcement, OPA coverage across every tool, secret hygiene |
| 20 | Observability | Per-tool-call audit log with PII redaction |
| 20 | Scale | 100-client load test horizontal scale demonstration |
| 15 | Registry UX | Discover / validate / enable-disable workflow |
| **100** | | |
- a schema-valid `server.json`;
- read-only and state-changing server surfaces;
- `server/discover`, deterministic `tools/list`, and policy-gated `tools/call`;
- a Streamable HTTP deployment with two interchangeable replicas;
- authorization and approval integration;
- a Registry publisher or private Registry API adapter;
- policy definitions and action-bound approval records;
- redacted audit output and trace propagation;
- wire-level failure evidence and a rollback plan.
| Weight | Criterion | Evidence |
|---:|---|---|
| 25 | Protocol correctness | Stateless request metadata, discovery, results, headers, and negative cases |
| 20 | Authorization | Issuer, audience, expiry, scope, and action-bound approval cases |
| 15 | Registry integrity | Valid `server.json`, publication record, live discovery probe, and drift report |
| 15 | Policy and safety | Allow, deny, malformed, stale approval, and sensitive-data cases |
| 15 | Scale and reliability | Two replicas, no affinity dependency, cancellation, timeout, and recovery |
| 10 | Auditability | Redacted receiver-side audit and trace evidence |
## Exercises
1. Add a new tool (Confluence search). Ship it through the registry validation flow without touching the core server.
2. Write an OPA policy that redacts Postgres query results containing columns named `email`, `ssn`, or `phone`. Exercise with a probe query.
3. Benchmark StreamableHTTP vs stdio on local latency. Report per-call p50/p95.
4. Implement per-tenant quota: maximum N calls per minute per tool per tenant. Enforce via a second OPA rule.
5. Run the MCP conformance suite from [mcp-conformance-tests](https://github.com/modelcontextprotocol/conformance) and fix every failure.
1. Change the published remote URL while leaving the live server unchanged. Make the registry validation report the exact drift.
2. Send `tools/list` twice with identical inputs and prove byte-stable tool order. Then expire `ttlMs` and refresh.
3. Send a valid body with a different `MCP-Protocol-Version` header. Return `-32020` and do not invoke policy or the tool.
4. Mint a token for the read-only server and present it to the state-changing server. Prove audience validation fails before the handler runs.
5. Bind an approval to one normalized argument digest. Change one field and prove the approval cannot be replayed.
6. Route consecutive calls to alternating replicas. Replace hidden process memory with an explicit shared handle wherever the workflow needs persistence.
7. Break a request-scoped SSE connection and retry with a new JSON-RPC request ID. Verify that no `Last-Event-ID` recovery path is used.
## Key Terms
| Term | What people say | What it actually means |
|------|-----------------|------------------------|
| StreamableHTTP | "2026 MCP transport" | Stateless HTTP + streaming; replaces SSE + stdio for networked servers |
| Capability manifest | "Well-known doc" | `.well-known/mcp-capabilities` with tool list, auth, transport URL |
| OPA / Rego | "Policy engine" | Open Policy Agent for authorizing tool calls against external rules |
| Scope elevation | "Approved-by-human" | Short-lived scope granted via Slack approval, required for destructive tools |
| Registry | "Tool discovery" | Service that indexes MCP servers from their capability manifests |
| Workload identity | "SPIFFE / SPIRE" | Cryptographic service identity for OAuth token issuance |
| Conformance suite | "Spec tests" | Official MCP test battery for StreamableHTTP + tool manifest correctness |
|---|---|---|
| Stateless MCP | "No state anywhere" | No protocol session; cross-call state is explicit and server-managed |
| `server.json` | "The tool manifest" | Registry metadata for naming, packaging, configuration, and transports |
| `server/discover` | "The handshake" | A normal mandatory RPC for live versions and capabilities, not a session initializer |
| Cache scope | "Can I cache it?" | Whether a cacheable result is safe for shared or private reuse |
| Policy decision | "The token allows it" | A separate decision over actor, tool, target, arguments, and context |
| Approval record | "A human clicked yes" | Evidence bound to one actor and consequential action under an expiry policy |
| Explicit handle | "A session ID" | Ordinary application data for named server-managed state, not protocol connection state |
## Further Reading
- [Model Context Protocol 2026 Roadmap](https://blog.modelcontextprotocol.io/posts/2026-mcp-roadmap/) — StreamableHTTP, capability metadata, registry
- [AAIF MCP Registry spec](https://github.com/modelcontextprotocol/registry) — the 2026 registry spec
- [AWS ECS reference deployment](https://aws.amazon.com/blogs/containers/deploying-model-context-protocol-mcp-servers-on-amazon-ecs/) — reference production deployment
- [Pinterest internal MCP ecosystem](https://www.infoq.com/news/2026/04/pinterest-mcp-ecosystem/) — the reference internal deployment
- [Block `goose` MCP usage](https://block.github.io/goose/) — reference agent consumption pattern
- [FastMCP](https://github.com/jlowin/fastmcp) — Python server framework
- [Open Policy Agent](https://www.openpolicyagent.org/) — policy engine reference
- [SPIFFE / SPIRE](https://spiffe.io) — workload identity reference
- [MCP specification 2026-07-28](https://modelcontextprotocol.io/specification/2026-07-28)
- [MCP 2026-07-28 key changes](https://modelcontextprotocol.io/specification/2026-07-28/changelog)
- [Streamable HTTP](https://modelcontextprotocol.io/specification/2026-07-28/basic/transports/streamable-http)
- [Server discovery](https://modelcontextprotocol.io/specification/2026-07-28/server/discover)
- [MCP authorization](https://modelcontextprotocol.io/specification/2026-07-28/basic/authorization)
- [MCP Registry overview](https://github.com/modelcontextprotocol/registry/blob/main/docs/modelcontextprotocol-io/about.mdx)
- [Registry publishing quickstart](https://github.com/modelcontextprotocol/registry/blob/main/docs/modelcontextprotocol-io/quickstart.mdx)
phases/19-capstone-projects/13-mcp-server-with-registry/outputs/skill-mcp-server.md
+30 -25
View File
@@ -1,46 +1,51 @@
---
name: mcp-server-platform
description: Deploy a production MCP server with StreamableHTTP, OAuth 2.1 scopes, OPA policy, human-approval gate for destructive tools, and a registry for discovery.
version: 1.0.0
description: Design a stateless MCP 2026-07-28 server with registry metadata, live discovery, authorization, policy, audit, and scale evidence.
version: 2.0.0
phase: 19
lesson: 13
tags: [capstone, mcp, fastmcp, streamablehttp, oauth, opa, registry, governance]
tags: [capstone, mcp, stateless, streamable-http, oauth, registry, governance]
---
Given an enterprise environment, ship an MCP server with 10 internal tools, a registry service for discovery, and a governance layer that gates destructive tools via Slack approval.
Given an internal platform need, design a stateless MCP server and governance boundary targeting protocol revision `2026-07-28`.
Build plan:
1. FastMCP server exposing 10 read-only tools (Postgres, S3, Jira, Linear, Datadog, PagerDuty, GitHub, Notion, Slack, Salesforce), each with typed schema and required scope.
2. StreamableHTTP transport, stateless behind a load balancer.
3. OAuth 2.1 token introspection middleware; workload identity via SPIFFE / SPIRE.
4. OPA / Rego policy decisions on every tool call: scope enforcement, PII redaction, payload size caps.
5. Destructive tools (Jira create, Linear create, Postgres write) on a separate MCP server requiring scope `approved:by:human` elevated via Slack card within 15 minutes.
6. Registry service that polls `.well-known/mcp-capabilities` from each server, validates with JSON Schema, and exposes a list/search/validate/enable UI.
7. Per-tenant JSONL audit log with Presidio PII redaction before write.
8. 100-client load test demonstrating horizontal scale; pass MCP conformance suite.
1. A schema-valid `server.json` whose reverse-DNS name matches the publisher's authenticated namespace.
2. Mandatory `server/discover` for live versions, capabilities, extensions, and server identity.
3. Version and client capabilities in every request `_meta`; `resultType` and server identity in every result.
4. Deterministic `tools/list` with `ttlMs` and `cacheScope`.
5. POST-only Streamable HTTP with required version, method, and name headers; no protocol sessions, GET stream, session DELETE, or replay header.
6. Authorization that validates issuer, audience, expiry, and scopes on every call.
7. Policy over actor, tool, target, and normalized arguments. Bind high-risk approvals to the exact action and expiry, then prove that changing one argument rejects replay.
8. Redacted audit and trace evidence outside model-visible context.
9. A registry adapter that validates `server.json`, probes `server/discover`, and reports metadata/runtime drift.
10. Two interchangeable replicas and a concurrent load probe with no session affinity.
Assessment rubric:
| Weight | Criterion | Measurement |
|:-:|---|---|
| 25 | Spec conformance | StreamableHTTP + capability manifest passes MCP conformance tests |
| 20 | Security | Scope enforcement, OPA coverage across every tool, secret hygiene |
| 20 | Observability | Per-tool-call audit log with PII redaction on write |
| 20 | Scale | 100-client load test with horizontal scale demonstration |
| 15 | Registry UX | Discover / validate / enable-disable workflow exercised |
| 25 | Protocol correctness | Stateless envelopes, discovery, results, headers, and negative cases |
| 20 | Authorization | Issuer, audience, expiry, scope, and exact-action approval cases |
| 15 | Registry integrity | Valid `server.json`, live probe, and drift report |
| 15 | Policy and safety | Allow, deny, malformed, stale approval, and sensitive-data cases |
| 15 | Scale | Two replicas with no affinity dependency plus cancellation and recovery |
| 10 | Auditability | Redacted receiver-side audit and trace evidence |
Hard rejects:
- Servers that require stateful sessions (violates 2026 StreamableHTTP stateless contract).
- Single-server topology where destructive tools share the same auth surface as read-only.
- Audit logs that persist raw PII.
- Ignoring the capability manifest; registry integration is a hard requirement.
- A current MCP design that uses `initialize`, `notifications/initialized`, or `Mcp-Session-Id`.
- Treating `server.json` as live capability discovery or inventing `.well-known/mcp-capabilities` as an MCP requirement.
- Publishing a server name outside the namespace authenticated for that publisher.
- Accepting a token without issuer and audience or resource validation.
- Treating tool annotations or a chat approval as authorization.
- Audit records that persist secrets or raw sensitive data.
Refusal rules:
- Refuse to deploy without OAuth; anonymous access is disqualifying.
- Refuse to ship destructive tools without the Slack approval flow.
- Refuse to expose a tool whose scope or description is not in the capability manifest.
- Refuse to claim production readiness from the local simulations alone.
- Refuse to expose a state-changing tool without policy and action-bound approval evidence.
- Refuse to publish metadata that points to an endpoint whose live discovery cannot be verified.
Output: a repo containing the two MCP servers (read-only + destructive), the registry service, the Slack approval integration, the OPA policies, the 100-client load-test harness, conformance-test results, and a write-up describing which tools you considered exposing but did not (and why) plus the top three OPA rules that caught near-misses during dry-run.
Output: a build plan and evidence matrix covering publication metadata, live discovery, stateless transport, tool schemas, authorization, policy, approval, audit, and scale. End with the highest-risk boundary and the exact failure test that proves it fails closed.
phases/19-capstone-projects/13-mcp-server-with-registry/quiz.json
+52 -64
View File
@@ -1,90 +1,78 @@
{
"lesson": "13-mcp-server-with-registry",
"title": "Capstone 13 — MCP Server with Registry and Governance",
"title": "Capstone 13: Stateless MCP Server with Registry and Governance",
"questions": [
{
"stage": "pre",
"question": "Why does the 2026 MCP revision favor StreamableHTTP for production servers?",
"question": "Why does this capstone use both server.json and server/discover?",
"options": [
"It is the only transport that supports tool calls",
"It is stateless by default, so a single endpoint behind a load balancer can scale horizontally",
"It encrypts payloads automatically",
"It removes the need for authentication"
],
"correct": 1,
"explanation": ""
},
{
"stage": "pre",
"question": "Which authorization model gates tool calls in this capstone?",
"options": [
"OAuth 2.1 tokens carrying per-tool scopes, checked at tool-call time",
"API keys per IP address",
"Mutual TLS without scopes",
"A shared admin password"
],
"correct": 0,
"explanation": ""
},
{
"stage": "check",
"question": "What is the role of the .well-known/mcp-capabilities document?",
"options": [
"Lists outbound IP ranges",
"Holds the OPA policy bundle",
"Exposes the tool manifest, transport URL, and auth requirements so the registry can validate and index the server",
"Stores audit logs for the server"
],
"correct": 2,
"explanation": ""
},
{
"stage": "check",
"question": "Why do destructive tools live on a separate MCP server in this design?",
"options": [
"They run on slower hardware",
"They require a different programming language",
"They cannot be exposed over StreamableHTTP",
"They are gated behind an approval token elevated via a Slack card within a short window"
"server.json stores sessions, while server/discover deletes them",
"The Registry requires both files to contain identical version dates",
"They are two names for the same tool manifest",
"server.json publishes install and endpoint metadata, while server/discover reports live protocol capabilities"
],
"correct": 3,
"explanation": ""
"explanation": "The Registry record helps clients locate and configure a server. The live RPC reports what the running process supports. Their schemas and version domains are independent, but the published identity must use the authenticated namespace and match live serverInfo."
},
{
"stage": "check",
"question": "What does OPA / Rego decide on every tool call?",
"question": "What replaces the initialize handshake in MCP revision 2026-07-28?",
"options": [
"Whether the caller's scopes permit invocation, plus PII redaction and payload caps",
"How the server should re-rank the response",
"Which model to invoke",
"Where to write the audit log"
"Mcp-Session-Id plus notifications/initialized",
"Every request carries protocol version and client capabilities, and clients can call mandatory server/discover",
"A GET request that opens a permanent server stream",
"A session cookie sent once per TCP connection"
],
"correct": 0,
"explanation": ""
"correct": 1,
"explanation": "The protocol is stateless. Version and capabilities are request metadata, while server/discover provides live version and capability discovery without creating a session."
},
{
"stage": "check",
"question": "Which tools/list behavior supports safe and efficient discovery?",
"options": [
"A response cached forever without a scope",
"A random tool order on each replica",
"A connection-local list created during initialize",
"Deterministic order plus ttlMs and a correct public or private cacheScope"
],
"correct": 3,
"explanation": "Stable ordering improves reproducibility and prompt caching. ttlMs communicates freshness, and cacheScope prevents user-specific results from leaking through shared caches."
},
{
"stage": "check",
"question": "Which Streamable HTTP design matches the current stateless transport?",
"options": [
"All clients require sticky load-balancer routing",
"POST creates a session, GET resumes it, and DELETE closes it",
"A standalone GET stream uses Last-Event-ID replay",
"Each JSON-RPC message uses its own POST, with JSON or request-scoped SSE for a request"
],
"correct": 3,
"explanation": "Current Streamable HTTP is POST-only for MCP messages and has no protocol session, standalone GET stream, session DELETE, or Last-Event-ID resumability."
},
{
"stage": "post",
"question": "Which evidence demonstrates StreamableHTTP horizontal scaling in the load test?",
"question": "What must a production server validate before a policy-approved tool handler runs?",
"options": [
"Adding a second replica and showing the load balancer redistributing without session stickiness",
"Running everything in a single process",
"Reducing concurrency to one client",
"Switching to stdio transport"
"Only that server.json lists the remote URL",
"Only the tool's destructiveHint annotation",
"Only that a human posted an approval message",
"Protocol metadata plus token issuer, audience or resource, expiry, scope, and any exact-action approval"
],
"correct": 0,
"explanation": ""
"correct": 3,
"explanation": "Transport metadata, token validation, and policy are separate gates. Annotations and publication metadata do not grant authority."
},
{
"stage": "post",
"question": "Why does the audit log scrub PII via Presidio before being persisted per tenant?",
"question": "Which evidence best supports the claim that the server scales without protocol sessions?",
"options": [
"To meet enterprise security requirements while keeping per-call lineage queryable",
"PII improves search performance",
"To make logs shorter",
"Presidio reduces ClickHouse write amplification"
"A server.json remote entry",
"A wire-level concurrent run across two interchangeable replicas with no affinity dependency",
"A unit test that constructs one response dictionary",
"A diagram showing a load balancer"
],
"correct": 0,
"explanation": ""
"correct": 1,
"explanation": "The real transport and deployment boundary must be exercised. Alternating requests across replicas proves correctness is not hidden in one process or sticky session."
}
]
}