"""Reference solution — MCP server/client and the bank's tool estate. Two halves that must be built together: * the PROTOCOL — JSON-RPC 2.0, capability negotiation at initialize, tools/list, tools/call, resources, prompts, and change notifications; * the ESTATE — a registry that versions tools, classifies their side effects, declares the scopes they need, and answers discovery QUERIES THAT ARE AUTHORIZATION-AWARE. MCP solved the M x N integration problem. It did NOT solve authorization, versioning or governance — those are the platform's job, and they are the second half of this file. Deterministic: no clock, no randomness, no network. ``python solution.py`` runs a worked session end to end. """ from __future__ import annotations import json import re from dataclasses import dataclass, field from enum import Enum from typing import Any, Callable, Dict, List, Mapping, Optional, Sequence, Tuple PROTOCOL_VERSION = "2025-06-18" # ====================================================================================== # 1. JSON-RPC 2.0 # ====================================================================================== PARSE_ERROR = -32700 INVALID_REQUEST = -32600 METHOD_NOT_FOUND = -32601 INVALID_PARAMS = -32602 INTERNAL_ERROR = -32603 class JsonRpcError(Exception): """A protocol-level failure. Distinct from a TOOL failure, which is a successful JSON-RPC response carrying ``isError: true`` — see :meth:`MCPServer._tools_call`.""" def __init__(self, code: int, message: str, data: Any = None) -> None: super().__init__(message) self.code = code self.message = message self.data = data def to_payload(self) -> Dict[str, Any]: error: Dict[str, Any] = {"code": self.code, "message": self.message} if self.data is not None: error["data"] = self.data return error def make_request(request_id: int, method: str, params: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: message: Dict[str, Any] = {"jsonrpc": "2.0", "id": request_id, "method": method} if params is not None: message["params"] = dict(params) return message def make_notification(method: str, params: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: """A notification has NO id and MUST NOT be answered. Getting this wrong is the most common JSON-RPC implementation bug.""" message: Dict[str, Any] = {"jsonrpc": "2.0", "method": method} if params is not None: message["params"] = dict(params) return message def validate_envelope(message: Mapping[str, Any]) -> None: if message.get("jsonrpc") != "2.0": raise JsonRpcError(INVALID_REQUEST, "jsonrpc must be '2.0'") method = message.get("method") if not isinstance(method, str) or not method: raise JsonRpcError(INVALID_REQUEST, "method must be a non-empty string") if "id" in message and not isinstance(message["id"], (int, str)): raise JsonRpcError(INVALID_REQUEST, "id must be a string or number") if "params" in message and not isinstance(message["params"], dict): raise JsonRpcError(INVALID_REQUEST, "params must be an object") # ====================================================================================== # 2. A JSON Schema subset — the contract enforcer # ====================================================================================== @dataclass(frozen=True) class ValidationError: path: str message: str def __str__(self) -> str: return f"{self.path}: {self.message}" def validate_schema(value: Any, schema: Mapping[str, Any], path: str = "$") -> List[ValidationError]: """Validate ``value`` against a useful subset of JSON Schema. Supported: type (incl. "integer" distinct from "number"), enum, const, required, properties, additionalProperties, items, minimum/maximum/exclusive*, minLength/ maxLength, pattern, minItems/maxItems, uniqueItems. Returns ALL errors, sorted by path, so a repair loop can fix everything in one turn instead of discovering problems one at a time. """ errors: List[ValidationError] = [] _validate(value, schema, path, errors) errors.sort(key=lambda e: (e.path, e.message)) return errors _TYPE_CHECKS: Mapping[str, Callable[[Any], bool]] = { "object": lambda v: isinstance(v, dict), "array": lambda v: isinstance(v, list), "string": lambda v: isinstance(v, str), # bool is a subclass of int in Python; a schema saying "integer" must reject True. "integer": lambda v: isinstance(v, int) and not isinstance(v, bool), "number": lambda v: isinstance(v, (int, float)) and not isinstance(v, bool), "boolean": lambda v: isinstance(v, bool), "null": lambda v: v is None, } def _validate(value: Any, schema: Mapping[str, Any], path: str, errors: List[ValidationError]) -> None: expected = schema.get("type") if expected is not None: types = [expected] if isinstance(expected, str) else list(expected) for t in types: if t not in _TYPE_CHECKS: raise ValueError(f"unsupported schema type: {t!r}") if not any(_TYPE_CHECKS[t](value) for t in types): errors.append(ValidationError(path, f"expected type {'|'.join(types)}, got {_type_name(value)}")) return # every further check would be noise if "const" in schema and value != schema["const"]: errors.append(ValidationError(path, f"must be {schema['const']!r}")) if "enum" in schema and value not in schema["enum"]: errors.append(ValidationError(path, f"must be one of {schema['enum']!r}")) if isinstance(value, str): if "minLength" in schema and len(value) < schema["minLength"]: errors.append(ValidationError(path, f"shorter than minLength {schema['minLength']}")) if "maxLength" in schema and len(value) > schema["maxLength"]: errors.append(ValidationError(path, f"longer than maxLength {schema['maxLength']}")) if "pattern" in schema and re.search(schema["pattern"], value) is None: errors.append(ValidationError(path, f"does not match pattern {schema['pattern']!r}")) if isinstance(value, (int, float)) and not isinstance(value, bool): if "minimum" in schema and value < schema["minimum"]: errors.append(ValidationError(path, f"below minimum {schema['minimum']}")) if "maximum" in schema and value > schema["maximum"]: errors.append(ValidationError(path, f"above maximum {schema['maximum']}")) if "exclusiveMinimum" in schema and value <= schema["exclusiveMinimum"]: errors.append(ValidationError(path, f"must be > {schema['exclusiveMinimum']}")) if "exclusiveMaximum" in schema and value >= schema["exclusiveMaximum"]: errors.append(ValidationError(path, f"must be < {schema['exclusiveMaximum']}")) if isinstance(value, dict): for name in schema.get("required", []): if name not in value: errors.append(ValidationError(f"{path}.{name}", "required property is missing")) properties = schema.get("properties", {}) for name, sub in properties.items(): if name in value: _validate(value[name], sub, f"{path}.{name}", errors) if schema.get("additionalProperties") is False: for name in sorted(value): if name not in properties: errors.append(ValidationError(f"{path}.{name}", "additional property is not allowed")) if isinstance(value, list): if "minItems" in schema and len(value) < schema["minItems"]: errors.append(ValidationError(path, f"fewer than minItems {schema['minItems']}")) if "maxItems" in schema and len(value) > schema["maxItems"]: errors.append(ValidationError(path, f"more than maxItems {schema['maxItems']}")) if schema.get("uniqueItems") and len(value) != len({json.dumps(v, sort_keys=True) for v in value}): errors.append(ValidationError(path, "items must be unique")) item_schema = schema.get("items") if item_schema is not None: for i, item in enumerate(value): _validate(item, item_schema, f"{path}[{i}]", errors) def _type_name(value: Any) -> str: if value is None: return "null" if isinstance(value, bool): return "boolean" if isinstance(value, int): return "integer" if isinstance(value, float): return "number" if isinstance(value, str): return "string" if isinstance(value, list): return "array" if isinstance(value, dict): return "object" return type(value).__name__ # ====================================================================================== # 3. Semantic versioning # ====================================================================================== @dataclass(frozen=True, order=True) class Version: major: int minor: int patch: int @classmethod def parse(cls, text: str) -> "Version": match = re.fullmatch(r"(\d+)\.(\d+)\.(\d+)", text.strip()) if not match: raise ValueError(f"not a semantic version: {text!r}") return cls(int(match.group(1)), int(match.group(2)), int(match.group(3))) def __str__(self) -> str: return f"{self.major}.{self.minor}.{self.patch}" def satisfies(self, constraint: str) -> bool: """Support ``^X.Y.Z`` (compatible-within-major), ``~X.Y.Z`` (within-minor), ``X.Y.Z`` (exact) and ``*``.""" constraint = constraint.strip() if constraint == "*": return True if constraint.startswith("^"): base = Version.parse(constraint[1:]) return self.major == base.major and (self.minor, self.patch) >= (base.minor, base.patch) if constraint.startswith("~"): base = Version.parse(constraint[1:]) return (self.major, self.minor) == (base.major, base.minor) and self.patch >= base.patch return self == Version.parse(constraint) class SchemaChange(str, Enum): """How a schema edit must be released. The rule is about who breaks.""" PATCH = "patch" # descriptions only — nobody breaks MINOR = "minor" # relaxation: new optional field, widened enum — old callers keep working MAJOR = "major" # restriction: new required field, removed field, narrowed enum, type change def classify_schema_change(old: Mapping[str, Any], new: Mapping[str, Any]) -> SchemaChange: """Decide the release type from two input schemas. The asymmetry is the lesson: ADDING a required property is breaking, REMOVING one is not; NARROWING an enum is breaking, WIDENING one is not. Callers break when the contract gets *stricter*, never when it gets looser. """ old_props = old.get("properties", {}) new_props = new.get("properties", {}) old_required = set(old.get("required", [])) new_required = set(new.get("required", [])) if new_required - old_required: return SchemaChange.MAJOR if set(old_props) - set(new_props): return SchemaChange.MAJOR for name in set(old_props) & set(new_props): before, after = old_props[name], new_props[name] if before.get("type") != after.get("type"): return SchemaChange.MAJOR if "enum" in before and "enum" in after and set(after["enum"]) < set(before["enum"]): return SchemaChange.MAJOR for key in ("minimum", "minLength", "minItems"): if key in after and after.get(key, float("-inf")) > before.get(key, float("-inf")): return SchemaChange.MAJOR for key in ("maximum", "maxLength", "maxItems"): if key in after and after.get(key, float("inf")) < before.get(key, float("inf")): return SchemaChange.MAJOR if old.get("additionalProperties") is not False and new.get("additionalProperties") is False: return SchemaChange.MAJOR if set(new_props) - set(old_props): return SchemaChange.MINOR if old_required - new_required: return SchemaChange.MINOR for name in set(old_props) & set(new_props): before, after = old_props[name], new_props[name] if "enum" in before and "enum" in after and set(after["enum"]) > set(before["enum"]): return SchemaChange.MINOR return SchemaChange.PATCH # ====================================================================================== # 4. The tool estate # ====================================================================================== class SideEffect(str, Enum): READ = "read" WRITE_IDEMPOTENT = "write_idempotent" WRITE_NON_IDEMPOTENT = "write_non_idempotent" IRREVERSIBLE = "irreversible" #: Retry is derived from the side-effect class by the PLATFORM, never chosen by the #: agent author. This mapping is the whole reason the class is a required field. RETRYABLE: Mapping[SideEffect, bool] = { SideEffect.READ: True, SideEffect.WRITE_IDEMPOTENT: True, SideEffect.WRITE_NON_IDEMPOTENT: False, # only with an idempotency key (Phase 10) SideEffect.IRREVERSIBLE: False, } class Lifecycle(str, Enum): ACTIVE = "active" DEPRECATED = "deprecated" RETIRED = "retired" @dataclass(frozen=True) class ToolSpec: """One version of one tool. Immutable: a change is a new version, never an edit.""" name: str version: Version title: str description: str # the model reads this — it is PROMPT SURFACE, not docs input_schema: Mapping[str, Any] side_effect: SideEffect required_scopes: Tuple[str, ...] = () data_classification: str = "internal" # public | internal | confidential | restricted owner: str = "unknown" tenants: Tuple[str, ...] = () # empty means "all tenants" lifecycle: Lifecycle = Lifecycle.ACTIVE superseded_by: Optional[str] = None # "name@X.Y.Z" @property def key(self) -> str: return f"{self.name}@{self.version}" def to_mcp(self) -> Dict[str, Any]: """The wire shape a client sees from ``tools/list``. Note what is NOT here: scopes, data classification, owner, tenants. Those are PLATFORM metadata. Sending them to the model would waste context and leak the shape of your control model into prompt text. """ return { "name": self.name, "title": self.title, "description": self.description, "inputSchema": dict(self.input_schema), "_meta": {"version": str(self.version), "sideEffect": self.side_effect.value}, } @dataclass(frozen=True) class Principal: """Who is asking. Discovery is answered relative to this, never in the abstract.""" agent_id: str tenant: str scopes: Tuple[str, ...] = () max_classification: str = "internal" _CLASSIFICATION_ORDER = ("public", "internal", "confidential", "restricted") def classification_rank(name: str) -> int: try: return _CLASSIFICATION_ORDER.index(name) except ValueError: raise ValueError(f"unknown data classification: {name!r}") from None class ToolRegistry: """The bank's tool estate: many tools, many versions, one authorization-aware view.""" def __init__(self) -> None: self._by_name: Dict[str, Dict[Version, ToolSpec]] = {} # -- publishing --------------------------------------------------------------- def publish(self, spec: ToolSpec) -> None: """Publish a new version. Republishing an existing (name, version) is refused — an immutable registry is what makes a version pin meaningful.""" versions = self._by_name.setdefault(spec.name, {}) if spec.version in versions: raise ValueError(f"{spec.key} already published; publish a new version instead") previous = self.latest(spec.name) if previous is not None: if spec.version <= previous.version: raise ValueError( f"{spec.key} is not newer than the current latest {previous.key}") change = classify_schema_change(previous.input_schema, spec.input_schema) required = { SchemaChange.MAJOR: previous.version.major + 1, SchemaChange.MINOR: previous.version.major, SchemaChange.PATCH: previous.version.major, }[change] if change is SchemaChange.MAJOR and spec.version.major != required: raise ValueError( f"{spec.key}: schema change is {change.value}; " f"major version must be {required}") if change is SchemaChange.MINOR and ( spec.version.major != previous.version.major or spec.version.minor <= previous.version.minor ): raise ValueError( f"{spec.key}: schema change is minor; expected " f"{previous.version.major}.{previous.version.minor + 1}.0") versions[spec.version] = spec def deprecate(self, name: str, version: str, *, superseded_by: Optional[str] = None) -> ToolSpec: return self._set_lifecycle(name, version, Lifecycle.DEPRECATED, superseded_by) def retire(self, name: str, version: str) -> ToolSpec: return self._set_lifecycle(name, version, Lifecycle.RETIRED, None) def _set_lifecycle(self, name, version, lifecycle, superseded_by) -> ToolSpec: parsed = Version.parse(version) spec = self._by_name.get(name, {}).get(parsed) if spec is None: raise KeyError(f"no such tool version: {name}@{version}") updated = ToolSpec( name=spec.name, version=spec.version, title=spec.title, description=spec.description, input_schema=spec.input_schema, side_effect=spec.side_effect, required_scopes=spec.required_scopes, data_classification=spec.data_classification, owner=spec.owner, tenants=spec.tenants, lifecycle=lifecycle, superseded_by=superseded_by if superseded_by is not None else spec.superseded_by, ) self._by_name[name][parsed] = updated return updated # -- reading ------------------------------------------------------------------ def versions(self, name: str) -> List[ToolSpec]: return [self._by_name.get(name, {})[v] for v in sorted(self._by_name.get(name, {}))] def latest(self, name: str) -> Optional[ToolSpec]: """Latest ACTIVE version, or None. A deprecated tool is still resolvable by an explicit pin, but it is never what 'latest' means.""" candidates = [s for s in self.versions(name) if s.lifecycle is Lifecycle.ACTIVE] return candidates[-1] if candidates else None def resolve(self, name: str, constraint: str = "*") -> Optional[ToolSpec]: """Highest non-retired version satisfying the constraint.""" candidates = [ s for s in self.versions(name) if s.lifecycle is not Lifecycle.RETIRED and s.version.satisfies(constraint) ] return candidates[-1] if candidates else None # -- authorization-aware discovery -------------------------------------------- def discover(self, principal: Principal, *, include_deprecated: bool = False) -> List[ToolSpec]: """What this principal may see, RIGHT NOW. Filtering happens here, not after ranking and not in the client. A tool the principal may not call must not appear in ``tools/list`` at all: its existence, its name and its description are themselves information, and a model that can see a tool will eventually try to call it. """ allowed_rank = classification_rank(principal.max_classification) held = set(principal.scopes) out: List[ToolSpec] = [] for name in sorted(self._by_name): for spec in reversed(self.versions(name)): if spec.lifecycle is Lifecycle.RETIRED: continue if spec.lifecycle is Lifecycle.DEPRECATED and not include_deprecated: continue if spec.tenants and principal.tenant not in spec.tenants: continue if classification_rank(spec.data_classification) > allowed_rank: continue if not set(spec.required_scopes) <= held: continue out.append(spec) break # newest visible version of each tool only return out # ====================================================================================== # 5. The MCP server # ====================================================================================== @dataclass(frozen=True) class ToolCallResult: content: str is_error: bool = False ToolHandler = Callable[[Mapping[str, Any]], ToolCallResult] @dataclass(frozen=True) class Resource: uri: str name: str mime_type: str text: str tenants: Tuple[str, ...] = () @dataclass(frozen=True) class Prompt: name: str description: str arguments: Tuple[str, ...] template: str class MCPServer: """A Model Context Protocol server over JSON-RPC 2.0. Implements: initialize, tools/list, tools/call, resources/list, resources/read, prompts/list, prompts/get, ping — plus notifications/tools/list_changed emitted when the estate changes. """ def __init__( self, *, name: str, version: str, registry: ToolRegistry, handlers: Mapping[str, ToolHandler], principal: Principal, resources: Sequence[Resource] = (), prompts: Sequence[Prompt] = (), protocol_versions: Sequence[str] = (PROTOCOL_VERSION,), ) -> None: self.name = name self.version = version self.registry = registry self.handlers = dict(handlers) self.principal = principal self.resources = {r.uri: r for r in resources} self.prompts = {p.name: p for p in prompts} self.protocol_versions = list(protocol_versions) self.initialized = False self.negotiated_version: Optional[str] = None self.outbound: List[Dict[str, Any]] = [] # notifications the server emitted self.call_log: List[Tuple[str, Mapping[str, Any]]] = [] # -- dispatch ----------------------------------------------------------------- def handle(self, message: Mapping[str, Any]) -> Optional[Dict[str, Any]]: """Handle one message. Returns a response, or None for a notification.""" try: validate_envelope(message) except JsonRpcError as exc: return {"jsonrpc": "2.0", "id": message.get("id"), "error": exc.to_payload()} method = message["method"] params = message.get("params", {}) is_notification = "id" not in message try: if method.startswith("notifications/"): self._handle_notification(method, params) return None if is_notification: # A non-notification method sent without an id: nothing to answer to. return None if method != "initialize" and not self.initialized: raise JsonRpcError(INVALID_REQUEST, "server not initialized") result = self._dispatch(method, params) return {"jsonrpc": "2.0", "id": message["id"], "result": result} except JsonRpcError as exc: if is_notification: return None return {"jsonrpc": "2.0", "id": message["id"], "error": exc.to_payload()} def _dispatch(self, method: str, params: Mapping[str, Any]) -> Dict[str, Any]: routes: Mapping[str, Callable[[Mapping[str, Any]], Dict[str, Any]]] = { "initialize": self._initialize, "ping": lambda p: {}, "tools/list": self._tools_list, "tools/call": self._tools_call, "resources/list": self._resources_list, "resources/read": self._resources_read, "prompts/list": self._prompts_list, "prompts/get": self._prompts_get, } handler = routes.get(method) if handler is None: raise JsonRpcError(METHOD_NOT_FOUND, f"unknown method: {method}") return handler(params) def _handle_notification(self, method: str, params: Mapping[str, Any]) -> None: if method == "notifications/initialized": self.initialized = True # -- methods ------------------------------------------------------------------ def _initialize(self, params: Mapping[str, Any]) -> Dict[str, Any]: requested = params.get("protocolVersion") if not isinstance(requested, str): raise JsonRpcError(INVALID_PARAMS, "protocolVersion is required") # Version negotiation: honour the request if supported, else offer our newest. # NEVER fail the handshake outright — that turns a version skew into an outage. chosen = requested if requested in self.protocol_versions else self.protocol_versions[-1] self.negotiated_version = chosen return { "protocolVersion": chosen, "capabilities": { "tools": {"listChanged": True}, "resources": {"subscribe": False, "listChanged": False}, "prompts": {"listChanged": False}, }, "serverInfo": {"name": self.name, "version": self.version}, } def _tools_list(self, params: Mapping[str, Any]) -> Dict[str, Any]: specs = self.registry.discover(self.principal) return {"tools": [s.to_mcp() for s in specs]} def _tools_call(self, params: Mapping[str, Any]) -> Dict[str, Any]: name = params.get("name") if not isinstance(name, str): raise JsonRpcError(INVALID_PARAMS, "name is required") arguments = params.get("arguments", {}) if not isinstance(arguments, dict): raise JsonRpcError(INVALID_PARAMS, "arguments must be an object") visible = {s.name: s for s in self.registry.discover(self.principal)} spec = visible.get(name) if spec is None: # Deliberately indistinguishable from "does not exist": a caller must not be # able to probe the estate for tools it is not entitled to see. raise JsonRpcError(METHOD_NOT_FOUND, f"unknown tool: {name}") errors = validate_schema(arguments, spec.input_schema) if errors: # A CONTRACT violation is a protocol error: the call never reaches the tool. raise JsonRpcError(INVALID_PARAMS, "arguments failed schema validation", data={"errors": [str(e) for e in errors]}) handler = self.handlers.get(name) if handler is None: raise JsonRpcError(INTERNAL_ERROR, f"no handler registered for {name}") self.call_log.append((name, dict(arguments))) result = handler(arguments) # A TOOL failure is a successful JSON-RPC response with isError -- so the model # can see it and correct itself, instead of the client raising. return {"content": [{"type": "text", "text": result.content}], "isError": result.is_error} def _resources_list(self, params: Mapping[str, Any]) -> Dict[str, Any]: visible = [ r for r in self.resources.values() if not r.tenants or self.principal.tenant in r.tenants ] return {"resources": [ {"uri": r.uri, "name": r.name, "mimeType": r.mime_type} for r in sorted(visible, key=lambda r: r.uri) ]} def _resources_read(self, params: Mapping[str, Any]) -> Dict[str, Any]: uri = params.get("uri") if not isinstance(uri, str): raise JsonRpcError(INVALID_PARAMS, "uri is required") resource = self.resources.get(uri) if resource is None or (resource.tenants and self.principal.tenant not in resource.tenants): raise JsonRpcError(INVALID_PARAMS, f"unknown resource: {uri}") return {"contents": [{"uri": uri, "mimeType": resource.mime_type, "text": resource.text}]} def _prompts_list(self, params: Mapping[str, Any]) -> Dict[str, Any]: return {"prompts": [ {"name": p.name, "description": p.description, "arguments": [{"name": a, "required": True} for a in p.arguments]} for p in sorted(self.prompts.values(), key=lambda p: p.name) ]} def _prompts_get(self, params: Mapping[str, Any]) -> Dict[str, Any]: name = params.get("name") prompt = self.prompts.get(name) if isinstance(name, str) else None if prompt is None: raise JsonRpcError(INVALID_PARAMS, f"unknown prompt: {name}") arguments = params.get("arguments", {}) missing = [a for a in prompt.arguments if a not in arguments] if missing: raise JsonRpcError(INVALID_PARAMS, f"missing arguments: {missing}") text = prompt.template for key, value in arguments.items(): text = text.replace("{" + key + "}", str(value)) return {"messages": [{"role": "user", "content": {"type": "text", "text": text}}]} # -- estate changes ----------------------------------------------------------- def notify_tools_changed(self) -> Dict[str, Any]: """Emit ``notifications/tools/list_changed``. Without this, a client caches a tool list and calls a tool that was retired an hour ago. With it, the client re-lists. This is the mechanism that makes a deprecation window actually work. """ message = make_notification("notifications/tools/list_changed") self.outbound.append(message) return message # ====================================================================================== # 6. The MCP client # ====================================================================================== class MCPClient: """One client per server connection, as the protocol intends. Caches the tool list and invalidates it on ``notifications/tools/list_changed`` — the cache is why the notification exists. """ def __init__(self, server: MCPServer, *, name: str = "bank-agent-kernel", version: str = "1.0.0") -> None: self.server = server self.name = name self.version = version self._next_id = 0 self.server_capabilities: Dict[str, Any] = {} self.protocol_version: Optional[str] = None self._tool_cache: Optional[List[Dict[str, Any]]] = None self.list_calls = 0 def _send(self, method: str, params: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: self._next_id += 1 response = self.server.handle(make_request(self._next_id, method, params)) if response is None: raise JsonRpcError(INTERNAL_ERROR, "server returned no response to a request") if response.get("id") != self._next_id: raise JsonRpcError(INTERNAL_ERROR, "response id did not match request id") if "error" in response: error = response["error"] raise JsonRpcError(error["code"], error["message"], error.get("data")) return response["result"] def initialize(self, protocol_version: str = PROTOCOL_VERSION) -> Dict[str, Any]: result = self._send("initialize", { "protocolVersion": protocol_version, "capabilities": {"sampling": {}, "elicitation": {}}, "clientInfo": {"name": self.name, "version": self.version}, }) self.protocol_version = result["protocolVersion"] self.server_capabilities = result["capabilities"] self.server.handle(make_notification("notifications/initialized")) return result def supports(self, capability: str) -> bool: return capability in self.server_capabilities def list_tools(self, *, force: bool = False) -> List[Dict[str, Any]]: if self._tool_cache is None or force: self.list_calls += 1 self._tool_cache = self._send("tools/list")["tools"] return self._tool_cache def on_notification(self, message: Mapping[str, Any]) -> None: if message.get("method") == "notifications/tools/list_changed": self._tool_cache = None def call_tool(self, name: str, arguments: Mapping[str, Any]) -> Dict[str, Any]: return self._send("tools/call", {"name": name, "arguments": arguments}) def read_resource(self, uri: str) -> str: return self._send("resources/read", {"uri": uri})["contents"][0]["text"] def get_prompt(self, name: str, arguments: Mapping[str, Any]) -> str: result = self._send("prompts/get", {"name": name, "arguments": arguments}) return result["messages"][0]["content"]["text"] # ====================================================================================== # 7. The repair loop # ====================================================================================== def repair_arguments( arguments: Mapping[str, Any], schema: Mapping[str, Any], *, defaults: Optional[Mapping[str, Any]] = None, ) -> Tuple[Dict[str, Any], List[ValidationError]]: """Deterministic repairs before asking the model to try again. Fixes only what is UNAMBIGUOUS: coerce numeric strings to numbers and numbers to strings where the schema says so, drop disallowed additional properties, and fill missing required fields from ``defaults``. Everything else is returned as remaining errors for the model to fix — guessing a missing account number is not a repair, it is a fabrication. """ repaired = dict(arguments) properties = schema.get("properties", {}) for name, sub in properties.items(): if name not in repaired: continue value = repaired[name] expected = sub.get("type") if expected == "number" and isinstance(value, str): try: repaired[name] = float(value) except ValueError: pass elif expected == "integer" and isinstance(value, str): try: repaired[name] = int(value) except ValueError: pass elif expected == "string" and isinstance(value, (int, float)) and not isinstance(value, bool): repaired[name] = str(value) elif expected == "boolean" and isinstance(value, str) and value.lower() in ("true", "false"): repaired[name] = value.lower() == "true" if schema.get("additionalProperties") is False: for name in list(repaired): if name not in properties: del repaired[name] for name in schema.get("required", []): if name not in repaired and defaults and name in defaults: repaired[name] = defaults[name] return repaired, validate_schema(repaired, schema) # ====================================================================================== # Worked example # ====================================================================================== PAYMENT_LOOKUP_SCHEMA = { "type": "object", "properties": { "reference": {"type": "string", "pattern": r"^PMT-\d{3,}$"}, "include_legs": {"type": "boolean"}, }, "required": ["reference"], "additionalProperties": False, } PAYMENT_RELEASE_SCHEMA = { "type": "object", "properties": { "reference": {"type": "string", "pattern": r"^PMT-\d{3,}$"}, "amount_minor": {"type": "integer", "minimum": 1}, "currency": {"type": "string", "enum": ["AED", "USD", "EUR"]}, }, "required": ["reference", "amount_minor", "currency"], "additionalProperties": False, } def _build_registry() -> ToolRegistry: registry = ToolRegistry() registry.publish(ToolSpec( name="payments.lookup", version=Version(1, 0, 0), title="Look up a payment", description="Return the status, amount and counterparties of a payment by its reference.", input_schema=PAYMENT_LOOKUP_SCHEMA, side_effect=SideEffect.READ, required_scopes=("payments.read",), data_classification="confidential", owner="wholesale-payments")) registry.publish(ToolSpec( name="payments.release", version=Version(1, 0, 0), title="Release a held payment", description="Release a payment currently on hold. Requires dual control above threshold.", input_schema=PAYMENT_RELEASE_SCHEMA, side_effect=SideEffect.IRREVERSIBLE, required_scopes=("payments.release",), data_classification="restricted", owner="wholesale-payments", tenants=("wholesale",))) registry.publish(ToolSpec( name="crm.notes.append", version=Version(1, 0, 0), title="Append a CRM note", description="Append a note to a customer record.", input_schema={"type": "object", "properties": {"customer_id": {"type": "string"}, "note": {"type": "string"}}, "required": ["customer_id", "note"], "additionalProperties": False}, side_effect=SideEffect.WRITE_IDEMPOTENT, required_scopes=("crm.write",), owner="retail-crm")) return registry def main() -> None: # pragma: no cover - narrative output registry = _build_registry() print("=" * 78) print("1. THE HANDSHAKE") print("=" * 78) principal = Principal(agent_id="payments-investigator", tenant="wholesale", scopes=("payments.read", "crm.write"), max_classification="confidential") handlers = { "payments.lookup": lambda a: ToolCallResult( f"{a['reference']}: HELD, 250000 AED, beneficiary 'Acme Trading FZE'"), "crm.notes.append": lambda a: ToolCallResult("note appended"), "payments.release": lambda a: ToolCallResult("released"), } server = MCPServer(name="bank-payments-mcp", version="2.3.1", registry=registry, handlers=handlers, principal=principal, resources=[Resource("policy://sanctions/hold-release", "Hold release policy", "text/markdown", "Releases above 100k need dual control.", tenants=("wholesale",))], prompts=[Prompt("investigate", "Investigate a held payment", ("reference",), "Investigate payment {reference}. State the hold reason.")]) client = MCPClient(server) info = client.initialize() print(f" negotiated protocol : {info['protocolVersion']}") print(f" server : {info['serverInfo']['name']} {info['serverInfo']['version']}") print(f" capabilities : {sorted(info['capabilities'])}") print(f" tools.listChanged : {info['capabilities']['tools']['listChanged']}") print() print("=" * 78) print("2. AUTHORIZATION-AWARE DISCOVERY") print("=" * 78) tools = client.list_tools() print(f" this principal sees : {[t['name'] for t in tools]}") print(" payments.release is ABSENT — not filtered later, never listed:") print(" - it needs scope 'payments.release', which this agent does not hold") print(" - it is classified 'restricted'; the agent is cleared to 'confidential'") try: client.call_tool("payments.release", {"reference": "PMT-771", "amount_minor": 1, "currency": "AED"}) except JsonRpcError as exc: print(f" calling it anyway : [{exc.code}] {exc.message} " f"(indistinguishable from 'does not exist')") print() print("=" * 78) print("3. CONTRACT ENFORCEMENT AND THE REPAIR LOOP") print("=" * 78) bad = {"reference": "PMT771", "include_legs": "yes", "urgent": True} errors = validate_schema(bad, PAYMENT_LOOKUP_SCHEMA) print(f" model proposed : {bad}") for error in errors: print(f" ✗ {error}") repaired, remaining = repair_arguments(bad, PAYMENT_LOOKUP_SCHEMA) print(f" after deterministic repair: {repaired}") print(f" remaining for the model to fix: {[str(e) for e in remaining]}") good = dict(repaired, reference="PMT-771", include_legs=True) result = client.call_tool("payments.lookup", good) print(f" valid call : isError={result['isError']} " f"content={result['content'][0]['text']!r}") print() print("=" * 78) print("4. VERSIONING: WHO BREAKS?") print("=" * 78) v2_schema = dict(PAYMENT_LOOKUP_SCHEMA, properties=dict(PAYMENT_LOOKUP_SCHEMA["properties"], as_of={"type": "string"})) print(f" add an OPTIONAL field -> {classify_schema_change(PAYMENT_LOOKUP_SCHEMA, v2_schema).value}") v3_schema = dict(v2_schema, required=["reference", "as_of"]) print(f" make it REQUIRED -> {classify_schema_change(v2_schema, v3_schema).value}") narrowed = dict(PAYMENT_RELEASE_SCHEMA, properties=dict(PAYMENT_RELEASE_SCHEMA["properties"], currency={"type": "string", "enum": ["AED"]})) print(f" NARROW an enum -> {classify_schema_change(PAYMENT_RELEASE_SCHEMA, narrowed).value}") widened = dict(PAYMENT_RELEASE_SCHEMA, properties=dict(PAYMENT_RELEASE_SCHEMA["properties"], currency={"type": "string", "enum": ["AED", "USD", "EUR", "GBP"]})) print(f" WIDEN an enum -> {classify_schema_change(PAYMENT_RELEASE_SCHEMA, widened).value}") try: registry.publish(ToolSpec( name="payments.lookup", version=Version(1, 1, 0), title="x", description="x", input_schema=v3_schema, side_effect=SideEffect.READ, required_scopes=("payments.read",), data_classification="confidential")) except ValueError as exc: print(f" registry refuses a breaking change as a minor bump:\n {exc}") print() print("=" * 78) print("5. DEPRECATION AND THE CHANGE NOTIFICATION") print("=" * 78) registry.publish(ToolSpec( name="payments.lookup", version=Version(2, 0, 0), title="Look up a payment", description="Return the status of a payment as of a point in time.", input_schema=v3_schema, side_effect=SideEffect.READ, required_scopes=("payments.read",), data_classification="confidential", owner="wholesale-payments")) registry.deprecate("payments.lookup", "1.0.0", superseded_by="payments.lookup@2.0.0") print(f" latest active : {registry.latest('payments.lookup').key}") print(f" pinned to ^1.0.0 : {registry.resolve('payments.lookup', '^1.0.0').key} " f"(deprecated, still resolvable)") print(f" client cache before : {[t['_meta']['version'] for t in client.list_tools()]}" f" (list_tools calls so far: {client.list_calls})") client.on_notification(server.notify_tools_changed()) print(f" after list_changed : {[t['_meta']['version'] for t in client.list_tools()]}" f" (list_tools calls so far: {client.list_calls})") print() print("=" * 78) print("6. RESOURCES AND PROMPTS") print("=" * 78) print(f" resource : {client.read_resource('policy://sanctions/hold-release')!r}") print(f" prompt : {client.get_prompt('investigate', {'reference': 'PMT-771'})!r}") print() print("=" * 78) print("7. PROTOCOL ERRORS VS TOOL ERRORS") print("=" * 78) raw = server.handle(make_request(99, "tools/call", {"name": "payments.lookup", "arguments": {}})) print(f" missing required arg -> JSON-RPC error {raw['error']['code']}: " f"{raw['error']['data']['errors']}") server.handlers["payments.lookup"] = lambda a: ToolCallResult("core banking timeout", is_error=True) raw = server.handle(make_request(100, "tools/call", {"name": "payments.lookup", "arguments": {"reference": "PMT-771", "as_of": "2026-03-12"}})) print(f" tool failure -> HTTP-200-shaped result, isError=" f"{raw['result']['isError']}, text={raw['result']['content'][0]['text']!r}") print(" ... so the model sees it and can react. A protocol error never reaches the model.") print() print("=" * 78) print("8. THE ESTATE, AS THE PLATFORM SEES IT") print("=" * 78) for name in ("payments.lookup", "payments.release", "crm.notes.append"): for spec in registry.versions(name): print(f" {spec.key:<28} {spec.lifecycle.value:<11} " f"{spec.side_effect.value:<21} retryable={RETRYABLE[spec.side_effect]!s:<5} " f"scopes={list(spec.required_scopes)} class={spec.data_classification}") if __name__ == "__main__": # pragma: no cover main()