warden agent-native surfaces: MCP server, self-proving custody card, proof-of-posture

- walletmcp.py: stdlib stdio JSON-RPC MCP server; 7 outcome-first tools
  with strict schemas; errors are structured (code/missing/remediation),
  never prose; results carry evidence (structuredContent)
- custodycard.py: A2A-style card sharpened to self-proving - embeds each
  member's LTL inclusion proof + STH so a counterparty recomputes trust
  instead of believing it; proof-of-posture nonce challenge returns a
  firewalled, signed posture attestation with the full quorum trail
- cli.py:
- verified end-to-end on the live wallet: all 7 MCP tools, structured
  refusal on bad input, and a counterparty recomputing all 4 inclusion
  proofs + STH signatures from the card alone (no operator trust)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/pacta/cli.py

@@ -197,6 +197,52 @@ def build_parser() -> argparse.ArgumentParser:
     agent.add_argument("--max-sth-age-seconds", type=int, help="Reject signed tree heads older than this.")
     agent.add_argument("--require-verified-verifier", action="store_true", help="Fail closed unless Ed25519 verification ran on the dogfood (certificate-covered) verifier.")
     agent.set_defaults(func=cmd_agent)
+
+    wallet = sub.add_parser("wallet", help="warden: the verified-custody wallet (quorum boundary + signing firewall).")
+    wsub = wallet.add_subparsers(dest="wallet_cmd", required=True)
+
+    w_build = wsub.add_parser("build-quorum", help="Build the quorum verifier members from the pinned proven source workspaces.")
+    w_build.add_argument("--sources-root", required=True, help="Directory holding the pinned fork source checkouts.")
+    w_build.add_argument("--backends", nargs="*", help="Subset of dalek anza risc0 betrusted (default: all).")
+    w_build.add_argument("--timeout", type=int, default=900)
+    w_build.set_defaults(func=cmd_wallet_build_quorum)
+
+    w_init = wsub.add_parser("init", help="Create a wallet - the R4 gate in executable form (refuses below end-to-end coverage).")
+    w_init.add_argument("--wallet", required=True)
+    w_init.add_argument("--evidence", required=True, help="Directory of <component>.attestation.json + .receipt.json (e.g. from `pacta log-fetch`).")
+    w_init.add_argument("--log-public-key", required=True)
+    w_init.add_argument("--trusted-provider", required=True, help="Whose observations you trust (verdicts are always re-derived locally).")
+    w_init.add_argument("--repos-config", default="examples/repos.yaml")
+    w_init.add_argument("--backends", nargs="*")
+    w_init.add_argument("--require-tier", default="R4")
+    w_init.add_argument("--min-members", type=int, default=2)
+    w_init.add_argument("--freshness-days", type=int, default=30)
+    w_init.set_defaults(func=cmd_wallet_init)
+
+    w_status = wsub.add_parser("status", help="Show wallet custody posture.")
+    w_status.add_argument("--wallet", required=True)
+    w_status.set_defaults(func=cmd_wallet_status)
+
+    w_card = wsub.add_parser("card", help="Emit the self-proving custody card (optionally to a .well-known dir).")
+    w_card.add_argument("--wallet", required=True)
+    w_card.add_argument("--out-dir", help="Write .well-known/custody-card.json under this directory.")
+    w_card.add_argument("--log-url", default="https://ltl.zkdefi.org")
+    w_card.set_defaults(func=cmd_wallet_card)
+
+    w_mcp = wsub.add_parser("mcp", help="Serve the agent-native MCP surface over stdio JSON-RPC.")
+    w_mcp.add_argument("--wallet", required=True)
+    w_mcp.add_argument("--log-url", default="https://ltl.zkdefi.org")
+    w_mcp.set_defaults(func=cmd_wallet_mcp)
+
+    w_ledger = wsub.add_parser("verify-ledger", help="Re-check the wallet's hash-chained ledger integrity.")
+    w_ledger.add_argument("--wallet", required=True)
+    w_ledger.set_defaults(func=cmd_wallet_verify_ledger)
+
+    w_unlatch = wsub.add_parser("unlatch", help="Release a tamper latch (deliberate operator act; the note is recorded permanently).")
+    w_unlatch.add_argument("--wallet", required=True)
+    w_unlatch.add_argument("--note", required=True)
+    w_unlatch.set_defaults(func=cmd_wallet_unlatch)
+
     return parser
 
 
@@ -544,6 +590,92 @@ def _log_accountability_checks(
     return diagnostics
 
 
+def cmd_wallet_build_quorum(args: argparse.Namespace) -> int:
+    from .quorum import QUORUM_BACKENDS, build_quorum_member
+
+    names = args.backends or list(QUORUM_BACKENDS)
+    failures = 0
+    for name in names:
+        try:
+            prov = build_quorum_member(name, args.sources_root, timeout=args.timeout)
+            print(f"{name}: built  source={(prov['source_commit'] or '?')[:12]}  sha256={prov['binary_sha256'][:12]}")
+        except (RuntimeError, KeyError) as exc:
+            failures += 1
+            print(f"{name}: FAILED  {exc}", file=sys.stderr)
+    return 1 if failures else 0
+
+
+def cmd_wallet_init(args: argparse.Namespace) -> int:
+    from .wallet import Wallet, WalletError
+
+    try:
+        wallet = Wallet.init(
+            args.wallet,
+            args.evidence,
+            args.log_public_key,
+            repos_config=args.repos_config,
+            trusted_provider=args.trusted_provider,
+            backends=args.backends,
+            require_tier=args.require_tier,
+            min_members=args.min_members,
+            freshness_max_age_days=args.freshness_days,
+        )
+    except WalletError as exc:
+        print(f"R4 gate refused: {exc}", file=sys.stderr)
+        return 1
+    capsule = wallet.capsule()
+    print(f"wallet created: {args.wallet}")
+    for member in capsule["members"]:
+        print(f"  {member['backend']:>10}  {member['risk_tier']}  leaf {member['evidence']['leaf_index']}  src {(member['source_commit'] or '?')[:12]}")
+    print(f"  policy: unanimity of >={capsule['policy']['min_members']} at tier {capsule['policy']['require_tier']}")
+    return 0
+
+
+def cmd_wallet_status(args: argparse.Namespace) -> int:
+    from .wallet import Wallet
+
+    print(json.dumps(Wallet(args.wallet).posture(), indent=2, sort_keys=True))
+    return 0
+
+
+def cmd_wallet_card(args: argparse.Namespace) -> int:
+    from .custodycard import build_custody_card, write_well_known
+    from .wallet import Wallet
+
+    wallet = Wallet(args.wallet)
+    if args.out_dir:
+        path = write_well_known(wallet, args.out_dir, args.log_url)
+        print(f"custody card: {path}")
+    else:
+        print(json.dumps(build_custody_card(wallet, args.log_url), indent=2, sort_keys=True))
+    return 0
+
+
+def cmd_wallet_mcp(args: argparse.Namespace) -> int:
+    from .walletmcp import WalletMCP
+
+    WalletMCP(args.wallet, log_url=args.log_url).serve_stdio()
+    return 0
+
+
+def cmd_wallet_verify_ledger(args: argparse.Namespace) -> int:
+    from .wallet import Wallet
+
+    ok, problems = Wallet(args.wallet).verify_ledger()
+    print("ledger chain: " + ("intact" if ok else "BROKEN"))
+    for problem in problems:
+        print(f"  - {problem}", file=sys.stderr)
+    return 0 if ok else 1
+
+
+def cmd_wallet_unlatch(args: argparse.Namespace) -> int:
+    from .wallet import Wallet
+
+    Wallet(args.wallet).unlatch(args.note)
+    print("latch released; note recorded in the ledger")
+    return 0
+
+
 def cmd_agent(args: argparse.Namespace) -> int:
     card = _card_for_agent(args)
     decision = run_agent_action(

src/pacta/custodycard.py

@@ -0,0 +1,196 @@
+"""The custody card and the proof-of-posture challenge.
+
+A2A publishes agent cards at ``/.well-known/agent-card.json`` and (since
+v1.0) signs them. warden's card goes one step further: it is
+**self-proving**. Alongside the description sits, for every quorum
+member, the transparency-log inclusion proof binding the member's
+attested source to a signed tree head. A counterparty does not have to
+believe the card's adjectives - it can recompute the Merkle roots with
+~40 lines of stdlib code and check the STH signature against the log key
+it already pins.
+
+The proof-of-posture (PoP) challenge is the live counterpart: send a
+nonce, get back a signed posture attestation (capsule hash, ledger head,
+latch state, incident count) whose signature just passed the wallet's
+own outbound firewall - with the quorum trail attached. A heartbeat you
+can audit instead of trust.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from .wallet import Refusal, Wallet
+
+CARD_TYPE = "pacta.wallet.custody_card.v1"
+POP_TYPE = "pacta.wallet.posture_attestation.v1"
+WELL_KNOWN_PATH = ".well-known/custody-card.json"
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def _canonical(document: Any) -> bytes:
+    return json.dumps(document, sort_keys=True, separators=(",", ":")).encode("utf-8")
+
+
+def build_custody_card(wallet: Wallet, log_url: str = "https://ltl.zkdefi.org") -> dict[str, Any]:
+    """The self-proving business card. Everything a counterparty agent needs
+    to evaluate this wallet is embedded or Merkle-bound; nothing requires
+    believing the operator."""
+    capsule = wallet.capsule()
+    posture = wallet.posture()
+    identities = sorted(p.name.removesuffix(".pub.pem") for p in wallet.keys_dir.glob("*.pub.pem"))
+    return {
+        "type": CARD_TYPE,
+        "name": "warden",
+        "description": (
+            "Verified-custody Ed25519 wallet. Inbound acceptance requires unanimity "
+            "of quorum members compiled from sources whose correctness certificates "
+            "are machine-checked in Lean 4 and replay-attested in a public RFC 9162 "
+            "transparency log. Outbound signatures pass the same quorum as a firewall "
+            "before release."
+        ),
+        "protocol": {
+            "mcp": {
+                "transport": "stdio",
+                "command": "pacta wallet mcp --wallet <dir>",
+                "tools": [
+                    "wallet_status",
+                    "verify_inbound",
+                    "request_signature",
+                    "custody_card",
+                    "posture_challenge",
+                    "list_incidents",
+                    "explain_refusal",
+                ],
+            },
+            "refusal_codes": [
+                "EVIDENCE_REQUIRED", "POLICY_DENIED", "CUSTODY_LATCHED",
+                "EVIDENCE_STALE", "MALFORMED_INTENT", "SIGNER_UNAVAILABLE",
+                "FIREWALL_QUARANTINE", "PENDING_AIRGAP",
+            ],
+        },
+        "identities": identities,
+        "quorum": {
+            "policy": capsule["policy"],
+            "members": [
+                {
+                    "backend": m["backend"],
+                    "component": m["component"],
+                    "semantics": m["semantics"],
+                    "entry_point": m["entry_point"],
+                    "source_commit": m["source_commit"],
+                    "binary_sha256": m["binary_sha256"],
+                    "risk_tier": m["risk_tier"],
+                    # The self-proving part: recompute, don't believe.
+                    "transparency_evidence": m["evidence"],
+                }
+                for m in capsule["members"]
+            ],
+        },
+        "signing": capsule["signing"],
+        "state": {
+            "latch": posture["latch"],
+            "ledger_head": posture["ledger"]["head"],
+            "ledger_entries": posture["ledger"]["entries"],
+            "incidents": posture["incidents"],
+        },
+        "evidence_endpoints": {
+            "log": log_url,
+            "log_docs": f"{log_url}/docs",
+            "verify_hint": (
+                "for each member: leaf_hash = SHA256(0x00 || canonical attestation leaf); "
+                "walk inclusion_proof to sth.root_hash; verify sth signature against the "
+                "log public key you pin. Stdlib only; see the log's verify.py."
+            ),
+        },
+        "honesty": [
+            "verification path certificate-covered; signing path trusted base (attested artifact)",
+            "SHA-512 enters the theorems as an opaque oracle",
+            "wire parser outcomes are hypotheses; side channels and reproducible builds are R5, not claimed",
+            "ML-DSA (PQC) slot fail-closed: no proven implementation exists",
+        ],
+        "generated_at": _now(),
+    }
+
+
+def write_well_known(wallet: Wallet, out_dir: str | Path, log_url: str = "https://ltl.zkdefi.org") -> Path:
+    out = Path(out_dir) / WELL_KNOWN_PATH
+    out.parent.mkdir(parents=True, exist_ok=True)
+    card = build_custody_card(wallet, log_url)
+    out.write_text(json.dumps(card, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    return out
+
+
+def posture_challenge(
+    wallet: Wallet,
+    nonce: str,
+    key_name: str = "warden",
+) -> dict[str, Any] | Refusal:
+    """Answer a counterparty's nonce with a firewalled, signed posture.
+
+    The signature on the attestation is produced by the wallet's own
+    outbound path, so it carries the full quorum trail - the counterparty
+    sees not only *that* the wallet signed, but that four provably
+    -equivalent verifiers unanimously accepted the signature before it
+    left the building. If custody is latched, the challenge honestly
+    fails with a CUSTODY_LATCHED refusal instead of a heartbeat.
+    """
+    if not isinstance(nonce, str) or not (8 <= len(nonce) <= 128):
+        raise ValueError("nonce must be a string of 8..128 characters")
+    posture = wallet.posture()
+    body = {
+        "type": POP_TYPE,
+        "nonce": nonce,
+        "posture": posture,
+    }
+    payload = _canonical(body)
+    intent = {
+        "purpose": f"proof-of-posture challenge response (nonce {nonce[:16]}...)",
+        "payload_sha256": hashlib.sha256(payload).hexdigest(),
+    }
+    result = wallet.request_signature(intent, payload, key_name=key_name)
+    if isinstance(result, Refusal):
+        return result
+    return {
+        "type": POP_TYPE,
+        "nonce": nonce,
+        "posture": posture,
+        "payload_hex": payload.hex(),
+        "signature_hex": result["signature_hex"],
+        "public_key_hex": result["public_key_hex"],
+        "identity": result["identity"],
+        "firewall": result["firewall"],
+        "issued_at": result["issued_at"],
+    }
+
+
+def verify_posture_attestation(attestation: dict[str, Any], expected_nonce: str) -> tuple[bool, list[str]]:
+    """Counterparty-side check, stdlib only: nonce echo, payload binding.
+
+    Signature verification is left to the counterparty's own verifier
+    (ideally its own quorum); this helper checks the structure so a
+    counterparty without crypto can still catch replay and splicing.
+    """
+    problems: list[str] = []
+    if attestation.get("type") != POP_TYPE:
+        problems.append(f"unexpected type {attestation.get('type')}")
+    if attestation.get("nonce") != expected_nonce:
+        problems.append("nonce mismatch (replay?)")
+    try:
+        payload = bytes.fromhex(str(attestation.get("payload_hex", "")))
+        body = json.loads(payload)
+    except ValueError:
+        problems.append("payload_hex is not hex/JSON")
+        return False, problems
+    if body.get("nonce") != expected_nonce:
+        problems.append("signed payload does not bind the nonce")
+    if body.get("posture") != attestation.get("posture"):
+        problems.append("posture shown does not match posture signed")
+    return (not problems), problems

src/pacta/walletmcp.py

@@ -0,0 +1,260 @@
+"""warden's Model Context Protocol surface - the agent-native front door.
+
+A dependency-free MCP server over stdio JSON-RPC 2.0. It follows the AX
+canon distilled in docs/agent-native.md: outcome-first tools, strict
+input schemas, and results that carry their own evidence so a calling
+agent never has to trust an adjective. Errors are structured objects
+(``code`` / ``missing`` / ``remediation``), not prose - a refused agent
+gets a machine-actionable receipt it can hand its principal.
+
+This is intentionally tiny and stdlib-only: an agent should be able to
+read the whole trust surface in one sitting. It speaks enough of MCP
+(``initialize``, ``tools/list``, ``tools/call``) to be driven by any MCP
+client, and degrades to a plain JSON-RPC endpoint for scripts.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import sys
+from pathlib import Path
+from typing import Any, Callable
+
+from .custodycard import build_custody_card, posture_challenge
+from .wallet import Refusal, Wallet
+
+PROTOCOL_VERSION = "2025-06-18"
+SERVER_INFO = {"name": "warden", "version": "1.0.0"}
+
+
+def _b64_to_bytes(field: str, value: Any) -> bytes:
+    if not isinstance(value, str):
+        raise _ToolError("MALFORMED_INTENT", f"{field} must be base64 string", [field], "send base64")
+    try:
+        return base64.b64decode(value, validate=True)
+    except Exception as exc:  # noqa: BLE001
+        raise _ToolError("MALFORMED_INTENT", f"{field} is not valid base64: {exc}", [field], "re-encode as base64")
+
+
+class _ToolError(Exception):
+    def __init__(self, code: str, reason: str, missing: list[str], remediation: str) -> None:
+        super().__init__(reason)
+        self.payload = {"code": code, "reason": reason, "missing": missing, "remediation": remediation}
+
+
+TOOLS: list[dict[str, Any]] = [
+    {
+        "name": "wallet_status",
+        "description": "Custody posture: quorum members and tiers, latch state, ledger head and chain integrity, incident and refusal counts. Read this first.",
+        "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
+    },
+    {
+        "name": "verify_inbound",
+        "description": "Run the quorum on an inbound (payload, signature, public_key). Acceptance requires unanimity of the proven verifiers; divergence is classified and, if unexplained, latches custody. All inputs base64.",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "payload_b64": {"type": "string", "description": "message bytes, base64"},
+                "signature_b64": {"type": "string", "description": "64-byte Ed25519 signature, base64"},
+                "public_key_b64": {"type": "string", "description": "32-byte raw Ed25519 public key, base64"},
+                "context": {"type": "string", "description": "free-text label recorded in the ledger"},
+            },
+            "required": ["payload_b64", "signature_b64", "public_key_b64"],
+            "additionalProperties": False,
+        },
+    },
+    {
+        "name": "request_signature",
+        "description": "Outbound signing with intent binding and the quorum firewall. Provide an intent.purpose (recorded as WHY) and the payload; the produced signature is verified by the quorum before release and quarantined if it fails.",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "payload_b64": {"type": "string", "description": "message bytes to sign, base64"},
+                "purpose": {"type": "string", "description": "why this signature is requested (recorded in the ledger)"},
+                "identity": {"type": "string", "description": "wallet identity name (default: warden)"},
+            },
+            "required": ["payload_b64", "purpose"],
+            "additionalProperties": False,
+        },
+    },
+    {
+        "name": "custody_card",
+        "description": "The self-proving business card: quorum membership with embedded transparency-log inclusion proofs a counterparty can recompute, signing provenance, honesty ledger. No trust required.",
+        "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
+    },
+    {
+        "name": "posture_challenge",
+        "description": "Proof-of-posture: send a nonce (8..128 chars), receive a signed posture attestation whose signature passed the outbound firewall, with the full quorum trail attached. An auditable heartbeat.",
+        "inputSchema": {
+            "type": "object",
+            "properties": {"nonce": {"type": "string"}},
+            "required": ["nonce"],
+            "additionalProperties": False,
+        },
+    },
+    {
+        "name": "list_incidents",
+        "description": "Quorum divergences and firewall quarantines recorded by this wallet, newest-first, with severities and trails.",
+        "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
+    },
+    {
+        "name": "explain_refusal",
+        "description": "Fetch a previously issued refusal receipt by index, or the latest. Returns the machine-actionable receipt (code, missing, remediation, signature).",
+        "inputSchema": {
+            "type": "object",
+            "properties": {"index": {"type": "integer", "description": "receipt number; omit for latest"}},
+            "additionalProperties": False,
+        },
+    },
+]
+
+
+class WalletMCP:
+    def __init__(self, wallet_dir: str | Path, log_url: str = "https://ltl.zkdefi.org") -> None:
+        self.wallet = Wallet(wallet_dir)
+        self.log_url = log_url
+        self.handlers: dict[str, Callable[[dict[str, Any]], dict[str, Any]]] = {
+            "wallet_status": self._wallet_status,
+            "verify_inbound": self._verify_inbound,
+            "request_signature": self._request_signature,
+            "custody_card": self._custody_card,
+            "posture_challenge": self._posture_challenge,
+            "list_incidents": self._list_incidents,
+            "explain_refusal": self._explain_refusal,
+        }
+
+    # -- tool implementations ------------------------------------------------
+
+    def _wallet_status(self, _: dict[str, Any]) -> dict[str, Any]:
+        return self.wallet.posture()
+
+    def _verify_inbound(self, args: dict[str, Any]) -> dict[str, Any]:
+        payload = _b64_to_bytes("payload_b64", args.get("payload_b64"))
+        signature = _b64_to_bytes("signature_b64", args.get("signature_b64"))
+        public_key = _b64_to_bytes("public_key_b64", args.get("public_key_b64"))
+        if len(signature) != 64:
+            raise _ToolError("MALFORMED_INTENT", "signature must be 64 bytes", ["signature_b64"], "send a 64-byte Ed25519 signature")
+        if len(public_key) != 32:
+            raise _ToolError("MALFORMED_INTENT", "public key must be 32 bytes", ["public_key_b64"], "send a 32-byte raw Ed25519 key")
+        result = self.wallet.verify_inbound(payload, signature, public_key, context=str(args.get("context", "")))
+        return result.to_dict()
+
+    def _request_signature(self, args: dict[str, Any]) -> dict[str, Any]:
+        import hashlib
+
+        payload = _b64_to_bytes("payload_b64", args.get("payload_b64"))
+        purpose = args.get("purpose")
+        if not isinstance(purpose, str) or not purpose.strip():
+            raise _ToolError("MALFORMED_INTENT", "purpose is required", ["purpose"], "state why the signature is requested")
+        intent = {"purpose": purpose, "payload_sha256": hashlib.sha256(payload).hexdigest()}
+        result = self.wallet.request_signature(
+            intent, payload, key_name=str(args.get("identity", "warden"))
+        )
+        if isinstance(result, Refusal):
+            raise _ToolError(
+                result.code, result.reason, result.missing, result.remediation
+            )
+        return result
+
+    def _custody_card(self, _: dict[str, Any]) -> dict[str, Any]:
+        return build_custody_card(self.wallet, self.log_url)
+
+    def _posture_challenge(self, args: dict[str, Any]) -> dict[str, Any]:
+        try:
+            result = posture_challenge(self.wallet, str(args.get("nonce", "")))
+        except ValueError as exc:
+            raise _ToolError("MALFORMED_INTENT", str(exc), ["nonce"], "send an 8..128 character nonce")
+        if isinstance(result, Refusal):
+            raise _ToolError(result.code, result.reason, result.missing, result.remediation)
+        return result
+
+    def _list_incidents(self, _: dict[str, Any]) -> dict[str, Any]:
+        incidents = []
+        for path in sorted(self.wallet.incidents_dir.glob("*.json"), reverse=True):
+            incidents.append(json.loads(path.read_text(encoding="utf-8")))
+        return {"incidents": incidents, "count": len(incidents)}
+
+    def _explain_refusal(self, args: dict[str, Any]) -> dict[str, Any]:
+        receipts = sorted(self.wallet.receipts_dir.glob("*.json"))
+        if not receipts:
+            raise _ToolError("EVIDENCE_REQUIRED", "no refusal receipts issued yet", [], "there is nothing to explain")
+        idx = args.get("index")
+        if idx is None:
+            path = receipts[-1]
+        else:
+            match = [p for p in receipts if p.stem == f"{int(idx):04d}"]
+            if not match:
+                raise _ToolError("EVIDENCE_REQUIRED", f"no receipt {idx}", [], f"valid indices 0..{len(receipts)-1}")
+            path = match[0]
+        return json.loads(path.read_text(encoding="utf-8"))
+
+    # -- JSON-RPC plumbing ---------------------------------------------------
+
+    def handle(self, message: dict[str, Any]) -> dict[str, Any] | None:
+        method = message.get("method")
+        msg_id = message.get("id")
+        if method == "initialize":
+            return self._ok(msg_id, {
+                "protocolVersion": PROTOCOL_VERSION,
+                "capabilities": {"tools": {"listChanged": False}},
+                "serverInfo": SERVER_INFO,
+            })
+        if method in ("notifications/initialized", "initialized"):
+            return None
+        if method == "tools/list":
+            return self._ok(msg_id, {"tools": TOOLS})
+        if method == "tools/call":
+            params = message.get("params") or {}
+            name = params.get("name")
+            args = params.get("arguments") or {}
+            handler = self.handlers.get(name)
+            if handler is None:
+                return self._err(msg_id, -32602, f"unknown tool {name}")
+            try:
+                payload = handler(args)
+                return self._ok(msg_id, {
+                    "content": [{"type": "text", "text": json.dumps(payload, indent=2, sort_keys=True)}],
+                    "structuredContent": payload,
+                    "isError": False,
+                })
+            except _ToolError as exc:
+                return self._ok(msg_id, {
+                    "content": [{"type": "text", "text": json.dumps(exc.payload, indent=2, sort_keys=True)}],
+                    "structuredContent": exc.payload,
+                    "isError": True,
+                })
+            except Exception as exc:  # noqa: BLE001 - never crash the server
+                err = {"code": "INTERNAL", "reason": f"{type(exc).__name__}: {exc}", "missing": [], "remediation": "inspect the wallet directory"}
+                return self._ok(msg_id, {
+                    "content": [{"type": "text", "text": json.dumps(err, indent=2, sort_keys=True)}],
+                    "structuredContent": err,
+                    "isError": True,
+                })
+        if msg_id is None:
+            return None
+        return self._err(msg_id, -32601, f"method not found: {method}")
+
+    @staticmethod
+    def _ok(msg_id: Any, result: dict[str, Any]) -> dict[str, Any]:
+        return {"jsonrpc": "2.0", "id": msg_id, "result": result}
+
+    @staticmethod
+    def _err(msg_id: Any, code: int, message: str) -> dict[str, Any]:
+        return {"jsonrpc": "2.0", "id": msg_id, "error": {"code": code, "message": message}}
+
+    def serve_stdio(self, stdin: Any = None, stdout: Any = None) -> None:
+        stdin = stdin or sys.stdin
+        stdout = stdout or sys.stdout
+        for line in stdin:
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                message = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+            response = self.handle(message)
+            if response is not None:
+                stdout.write(json.dumps(response) + "\n")
+                stdout.flush()