diff --git a/.github/workflows/samples.yml b/.github/workflows/samples.yml
new file mode 100644
index 0000000..343b23d
--- /dev/null
+++ b/.github/workflows/samples.yml
@@ -0,0 +1,71 @@
+# Runs the samples' own tests, including their schema conformance suite.
+#
+# `testpaths = ["tests"]` in pyproject.toml scopes the docs deploy gate to the
+# guards it was written for, so a suite under samples/ is never collected there.
+# CONTRIBUTING.md names the failure mode directly: a test written outside tests/
+# "never runs in CI and will pass review looking like coverage it does not
+# provide." This workflow is what stops that from being true here.
+#
+# The conformance suite needs jsonschema, which is not in uv.lock. Installing it
+# in the deploy gate's environment would put a dependency in front of the site
+# build for no benefit to the site, so it is installed here instead, in a job
+# nothing else depends on.
+#
+# No workflow expression appears inside a `run:` block. The two in the
+# concurrency group are an integer pull request number and a ref name GitHub
+# validates, and a concurrency group is not a shell context.
+#
+# Scoped by path. A documentation or schema-only change should not pay for this.
+name: Samples
+
+on:
+  pull_request:
+    paths:
+      - "samples/**"
+      - "specification/v0.1.0/**"
+      - ".github/workflows/samples.yml"
+  push:
+    branches: ["main", "integration"]
+    paths:
+      - "samples/**"
+      - "specification/v0.1.0/**"
+      - ".github/workflows/samples.yml"
+  workflow_dispatch:
+
+permissions: {}
+
+concurrency:
+  group: samples-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  python-guardian-fastmcp:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+    defaults:
+      run:
+        working-directory: samples/python-guardian-fastmcp
+    steps:
+      - name: Check out the repository
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+
+      # 3.11 is the floor the sample's README states. Pinned to a minor rather
+      # than a moving 3.x so a runner image bump cannot quietly change what the
+      # sample is proven against.
+      - name: Set up Python
+        uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
+        with:
+          python-version: "3.11"
+
+      # The sample itself is stdlib-only by design. pytest runs the suite and
+      # jsonschema is what the conformance tests validate with; neither is a
+      # dependency of the sample code.
+      - name: Install the test-only dependencies
+        run: python -m pip install -r requirements-dev.txt
+
+      # test_conformance.py loads specification/v0.1.0/ off disk by relative
+      # path, so this proves the sample still matches the schemas in this
+      # repository rather than a copy of them.
+      - name: Run the tests
+        run: python -m pytest . -q
diff --git a/samples/python-guardian-fastmcp/README.md b/samples/python-guardian-fastmcp/README.md
new file mode 100644
index 0000000..d15b902
--- /dev/null
+++ b/samples/python-guardian-fastmcp/README.md
@@ -0,0 +1,74 @@
+<!-- SPDX-License-Identifier: Apache-2.0 -->
+
+# Python Guardian sample and FastMCP client instrumentation
+
+A minimal ACS v0.1 Guardian Agent (stdlib HTTP, pluggable policy, per-session
+hash chain) plus a FastMCP client wrapper that enforces Guardian verdicts
+before a tool runs. It exists so framework authors can watch the
+Observed-to-Guardian round trip without installing anything beyond Python 3.11.
+
+This is not production code. `POST /acs` has no authentication, so bind
+loopback or put something in front of it, and nothing persists past a JSONL
+log. It is also not a conformance claim: the gaps are listed under
+[Limits](#limits).
+
+## Run it
+
+```bash
+python guardian.py [--port 8787] [--host 127.0.0.1]
+```
+
+The sample code imports nothing outside the standard library. The tests need
+pytest, and the conformance suite needs jsonschema:
+
+```bash
+python -m pip install -r requirements-dev.txt
+python -m pytest . -q
+```
+
+A live demo against a real MCP server needs `pip install fastmcp` and a running
+Guardian:
+
+```bash
+python -c "import asyncio; from fastmcp_instrumentation import demo; asyncio.run(demo())"
+```
+
+## How the tests are split
+
+`test_sample.py` covers behaviour: fail-closed posture, the hash chain, verdict
+handling. `test_conformance.py` covers shape, by loading
+`specification/v0.1.0/` off disk and validating every envelope this sample
+emits against it. Agreeing with yourself about a wire format proves nothing,
+so drift between the sample and the standard fails there rather than in
+somebody else's integration.
+
+Neither suite is collected by the docs deploy gate, because `testpaths` in
+`pyproject.toml` scopes that run to `tests/`. They run from
+`.github/workflows/samples.yml` instead.
+
+## Arguments carry the ACS wrapper
+
+`hooks/tool-call-request.json` requires each argument to be an object with a
+`value` key, so a provenance record can attach per argument:
+
+```json
+"arguments": {"path": {"value": "/tmp/x"}}
+```
+
+`build_tool_call_envelope` takes an ordinary Python mapping and applies that
+wrapper on the way out. The Guardian rejects anything else with `-32600` and
+hands the policy layer plain unwrapped values, so a policy author never has to
+think about the envelope form.
+
+## Limits
+
+- `modify`, `ask` and `defer` are treated as deny. A sample has no
+  modification or approval loop, and downgrading them to allow would be the
+  wrong direction to fail in.
+- Request signatures are accepted but not verified. ACS-Core requires an
+  HMAC-SHA256 over the JCS-canonicalized envelope with an HKDF-derived
+  per-session key, and this sample does not implement it, which is the same
+  gap [#70](https://github.com/GenAI-Security-Project/agent-control-standard/issues/70)
+  tracks against the reference Guardian. Until the wire is authenticated,
+  reachability is the access control.
+- No OpenTelemetry or OCSF trace emission.
diff --git a/samples/python-guardian-fastmcp/fastmcp_instrumentation.py b/samples/python-guardian-fastmcp/fastmcp_instrumentation.py
new file mode 100644
index 0000000..c576290
--- /dev/null
+++ b/samples/python-guardian-fastmcp/fastmcp_instrumentation.py
@@ -0,0 +1,130 @@
+# SPDX-License-Identifier: Apache-2.0
+"""FastMCP client instrumentation for ACS (sample).
+
+Wraps any object with ``await call_tool(name, arguments)`` (a real
+``fastmcp.Client`` or a test double — no fastmcp import required here):
+every call first emits a steps/toolCallRequest envelope to the Guardian
+URL and honors the verdict BEFORE the tool runs.
+
+v0.1 semantics, stated plainly: allow → proceed; deny → raise without
+calling; modify | ask | defer → treated as deny (the sample has no
+modification/approval loop; production clients MUST implement them
+rather than downgrading to allow).
+"""
+
+from __future__ import annotations
+
+import json
+import urllib.request
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Optional
+
+
+class GovernedDenied(RuntimeError):
+    """The Guardian denied this tool call: the tool never ran."""
+
+    def __init__(self, tool: str, reasoning: str) -> None:
+        super().__init__(f"guardian denied {tool!r}: {reasoning}")
+        self.tool = tool
+        self.reasoning = reasoning
+
+
+def build_tool_call_envelope(tool: str, arguments: dict,
+                             agent_id: str = "sample-agent",
+                             session_id: Optional[str] = None,
+                             acs_version: str = "0.1.0") -> dict:
+    """Pure envelope builder (no I/O, fully testable).
+
+    ``arguments`` is taken as a plain mapping and wrapped into the ACS
+    ``{"value": ...}`` form on the way out, so callers keep writing
+    ordinary Python dicts.
+    """
+    return {
+        "jsonrpc": "2.0",
+        "method": "steps/toolCallRequest",
+        "id": 1,
+        "params": {
+            "acs_version": acs_version,
+            "request_id": str(uuid.uuid4()),
+            "timestamp": datetime.now(timezone.utc).isoformat(),
+            "metadata": {
+                "agent_id": agent_id,
+                "session_id": session_id or str(uuid.uuid4()),
+            },
+            "payload": {
+                "tool": {"name": tool},
+                # ACS v0.1 wraps every argument as {"value": ...} so a
+                # provenance record can hang off each one independently.
+                # See specification/v0.1.0/hooks/tool-call-request.json:
+                # a raw scalar here fails validation.
+                "arguments": {k: {"value": v}
+                              for k, v in (arguments or {}).items()},
+            },
+        },
+    }
+
+
+def post_envelope(guardian_url: str, envelope: dict,
+                  timeout: float = 5.0) -> dict:
+    """POST one envelope, return the decoded JSON-RPC response."""
+    data = json.dumps(envelope).encode()
+    req = urllib.request.Request(
+        guardian_url.rstrip("/") + "/acs", data=data,
+        headers={"Content-Type": "application/json"}, method="POST")
+    with urllib.request.urlopen(req, timeout=timeout) as resp:
+        return json.loads(resp.read().decode())
+
+
+class GovernedClient:
+    """fastmcp.Client wrapper enforcing Guardian verdicts pre-execution."""
+
+    def __init__(self, inner: Any, guardian_url: str,
+                 agent_id: str = "sample-agent",
+                 session_id: Optional[str] = None) -> None:
+        self._inner = inner
+        self._guardian_url = guardian_url
+        self._agent_id = agent_id
+        self._session_id = session_id or str(uuid.uuid4())
+
+    async def call_tool(self, name: str, arguments: Optional[dict] = None) -> Any:
+        """Govern, then (only on allow) delegate to the inner client."""
+        envelope = build_tool_call_envelope(
+            name, arguments or {}, self._agent_id, self._session_id)
+        try:
+            response = post_envelope(self._guardian_url, envelope)
+        except Exception as exc:
+            # Guardian unreachable: fail CLOSED (no silent allow).
+            raise GovernedDenied(name, f"guardian unreachable: {exc}"[:200])
+        result = response.get("result") if isinstance(response, dict) else None
+        decision = result.get("decision") if isinstance(result, dict) else None
+        reasoning = (result.get("reasoning", "") if isinstance(result, dict) else "")
+        if decision == "allow":
+            return await self._inner.call_tool(name, arguments or {})
+        if decision in ("modify", "ask", "defer"):
+            raise GovernedDenied(
+                name, f"verdict {decision!r} unsupported by this sample "
+                      f"(treated as deny): {reasoning}")
+        raise GovernedDenied(name, reasoning or f"verdict {decision!r}")
+
+
+def require_fastmcp() -> None:
+    """Import check with a helpful error (real-client path only)."""
+    try:
+        import fastmcp  # noqa: F401
+    except ImportError as exc:
+        raise ImportError(
+            "the live demo needs the 'fastmcp' package "
+            "(pip install fastmcp); unit tests use a fake client"
+        ) from exc
+
+
+# --- live demo (needs fastmcp + a running guardian; not a test) ------------
+async def demo(guardian_url: str = "http://127.0.0.1:8787") -> None:  # pragma: no cover
+    require_fastmcp()
+    from fastmcp import Client
+
+    client = Client("https://example.com/mcp")  # replace with a real server
+    governed = GovernedClient(client, guardian_url)
+    async with client:
+        print(await governed.call_tool("greet", {"name": "acs"}))
diff --git a/samples/python-guardian-fastmcp/guardian.py b/samples/python-guardian-fastmcp/guardian.py
new file mode 100644
index 0000000..319b3ba
--- /dev/null
+++ b/samples/python-guardian-fastmcp/guardian.py
@@ -0,0 +1,261 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Minimal ACS Guardian Agent sample (Python, stdlib only).
+
+Speaks specification/v0.1.0 over POST /acs:
+  handshake/hello      -> ServerHello (fail-closed posture declared)
+  system/ping          -> allow (never advances the chain, per spec)
+  steps/toolCallRequest -> allow | deny via a pluggable policy fn
+
+Every envelope is hash-chained per session into a JSONL log. Policy
+exceptions deny (fail-closed). Unknown methods and version mismatches
+are JSON-RPC errors, never silent allows.
+
+Run:  python guardian.py [--port 8787] [--log .acs/envelopes.jsonl]
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import re
+import uuid
+from datetime import datetime, timezone
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Any, Callable, Optional
+
+ACS_VERSION = "0.1.0"
+
+# method -> namespace allowlist (v0.1 surface of this sample)
+_KNOWN_METHODS = {"handshake/hello", "system/ping", "steps/toolCallRequest"}
+
+
+def _utcnow() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _canonical(obj: Any) -> bytes:
+    return json.dumps(obj, sort_keys=True, separators=(",", ":")).encode()
+
+
+def _chain(prev: Optional[str], envelope: dict) -> str:
+    return hashlib.sha256((prev or "GENESIS").encode() + _canonical(envelope)).hexdigest()
+
+
+PolicyFn = Callable[[str, dict, dict], tuple]
+"""policy(tool_name, arguments, context) -> (decision, reasoning[, reason_codes]).
+
+decision: "allow" | "deny". Anything else is treated as deny. ``arguments``
+arrives unwrapped (plain values), so a policy author never has to think about
+the ACS ``{"value": ...}`` envelope form. Returning a third element is
+optional; when present it becomes ``result.reason_codes``, whose vocabulary
+is free in v0.1."""
+
+
+def arguments_conform(arguments: Any) -> bool:
+    """True when every argument is an object carrying a ``value`` key.
+
+    ACS v0.1 requires this shape so provenance can attach per argument
+    (specification/v0.1.0/hooks/tool-call-request.json). A Guardian that
+    accepted raw scalars here would be validating something the standard
+    does not describe.
+    """
+    return isinstance(arguments, dict) and all(
+        isinstance(v, dict) and "value" in v for v in arguments.values())
+
+
+def unwrap_arguments(arguments: dict) -> dict:
+    """Strip the ACS ``{"value": ...}`` wrapper for the policy layer."""
+    return {k: v.get("value") for k, v in arguments.items()}
+
+
+def sample_policy(tool_name: str, arguments: dict, context: dict) -> tuple:
+    """Demo policy: deny destructive tool names and path escapes."""
+    lowered = tool_name.lower()
+    if any(p in lowered for p in ("delete", "drop", "destroy", "rmtree", "format", "wipe")):
+        return ("deny",
+                f"tool {tool_name!r} matches the destructive-name blocklist",
+                ["destructive_tool"])
+    blob = json.dumps(arguments, default=str)
+    if ".." in blob or re.search(r"ssh-rsa\s+[A-Za-z0-9+/=]+", blob):
+        return ("deny", "arguments carry path-escape or key material",
+                ["path_escape_or_key_material"])
+    return "allow", f"tool {tool_name!r} not on any deny rule"
+
+
+def _error(req_id: Any, code: int, message: str, data: Any = None) -> dict:
+    err: dict[str, Any] = {"code": code, "message": message}
+    if data is not None:
+        err["data"] = data
+    return {"jsonrpc": "2.0", "id": req_id, "error": err}
+
+
+def _result(req_id: Any, request_id: str, decision: str,
+            reasoning: str = "", reason_codes: Optional[list] = None) -> dict:
+    res: dict[str, Any] = {
+        "type": "final",
+        "acs_version": ACS_VERSION,
+        "request_id": request_id,
+        "decision": decision,
+    }
+    if reasoning:
+        res["reasoning"] = reasoning
+    if reason_codes:
+        res["reason_codes"] = reason_codes
+    return {"jsonrpc": "2.0", "id": req_id, "result": res}
+
+
+class Guardian:
+    """Stateful Guardian: handshake, chain log, policy dispatch."""
+
+    def __init__(self, policy: PolicyFn = sample_policy,
+                 log_path: str = ".acs/envelopes.jsonl") -> None:
+        self.policy = policy
+        self.log_path = log_path
+        self._chains: dict[str, str] = {}
+
+    def _record(self, session_id: str, envelope: dict, advance: bool) -> str:
+        head = self._chains.get(session_id)
+        digest = _chain(head, envelope)
+        if advance:
+            self._chains[session_id] = digest
+        try:
+            with open(self.log_path, "a", encoding="utf-8") as fh:
+                fh.write(json.dumps({"chain_hash": digest, "envelope": envelope}) + "\n")
+        except OSError:
+            pass  # logging must never break enforcement
+        return digest
+
+    def handle(self, envelope: Any) -> dict:
+        """Dispatch one decoded JSON-RPC envelope. Never raises."""
+        try:
+            return self._dispatch(envelope)
+        except Exception as exc:  # fail-closed on Guardian bugs too
+            req_id = envelope.get("id") if isinstance(envelope, dict) else None
+            return _error(req_id, -32603, "guardian internal error (fail-closed)",
+                          {"detail": str(exc)[:200]})
+
+    def _dispatch(self, envelope: Any) -> dict:
+        if not isinstance(envelope, dict):
+            return _error(None, -32700, "parse error: envelope must be an object")
+        req_id = envelope.get("id")
+        if envelope.get("jsonrpc") != "2.0" or not isinstance(envelope.get("method"), str):
+            return _error(req_id, -32600, "invalid request envelope")
+        method = envelope["method"]
+        params = envelope.get("params")
+        if not isinstance(params, dict):
+            return _error(req_id, -32600, "params must be an object")
+        if params.get("acs_version") != ACS_VERSION:
+            return _error(req_id, -32001, "UNSUPPORTED_VERSION",
+                          {"supported": [ACS_VERSION]})
+        metadata = params.get("metadata")
+        if not isinstance(metadata, dict) or "agent_id" not in metadata \
+                or "session_id" not in metadata:
+            return _error(req_id, -32600, "metadata.agent_id/session_id required")
+        session_id = str(metadata["session_id"])
+        request_id = str(params.get("request_id", "") or uuid.uuid4())
+
+        if method == "handshake/hello":
+            hello = {
+                "negotiated_version": ACS_VERSION,
+                "methods_evaluated": sorted(_KNOWN_METHODS),
+                "selected_transport": "http",
+                "timeout_config": {"default_ms": 5000},
+                "on_decision_failure": "deny",
+            }
+            self._record(session_id, envelope, advance=True)
+            return {"jsonrpc": "2.0", "id": req_id, "result": hello}
+
+        if method == "system/ping":
+            # Spec: always allow, never advance the chain.
+            self._record(session_id, envelope, advance=False)
+            payload: dict[str, Any] = {"echo": params.get("payload", {}).get("echo")}
+            out = _result(req_id, request_id, "allow")
+            out["result"]["payload"] = payload
+            return out
+
+        if method != "steps/toolCallRequest":
+            return _error(req_id, -32601, f"method not implemented: {method}")
+
+        payload = params.get("payload")
+        if not isinstance(payload, dict):
+            return _error(req_id, -32600, "payload must be an object")
+        tool = payload.get("tool", {})
+        tool_name = tool.get("name") if isinstance(tool, dict) else None
+        arguments = payload.get("arguments")
+        if not tool_name:
+            return _error(req_id, -32600, "payload.tool.name required")
+        if not arguments_conform(arguments):
+            return _error(
+                req_id, -32600,
+                "payload.arguments must map each name to an object carrying a "
+                "'value' key (specification/v0.1.0/hooks/tool-call-request.json)")
+
+        self._record(session_id, envelope, advance=True)
+        codes: Optional[list] = None
+        try:
+            verdict = self.policy(
+                tool_name, unwrap_arguments(arguments),
+                {"agent_id": metadata.get("agent_id"), "session_id": session_id})
+            decision, reasoning = verdict[0], verdict[1]
+            if len(verdict) > 2:
+                codes = list(verdict[2]) or None
+        except Exception as exc:
+            decision = "deny"
+            reasoning = f"policy error (fail-closed): {exc}"[:300]
+            codes = ["policy_error"]
+        if decision not in ("allow", "deny"):
+            reasoning = f"unknown policy verdict {decision!r} treated as deny"
+            decision, codes = "deny", ["unknown_verdict"]
+        return _result(req_id, request_id, decision, reasoning, codes)
+
+
+class _Handler(BaseHTTPRequestHandler):
+    guardian: Guardian = Guardian()  # replaced per serve()
+
+    def log_message(self, *args: Any) -> None:  # quiet stdlib server
+        pass
+
+    def do_POST(self) -> None:
+        if self.path != "/acs":
+            self.send_response(404)
+            self.end_headers()
+            return
+        try:
+            length = int(self.headers.get("Content-Length", "0"))
+        except ValueError:
+            length = 0
+        try:
+            envelope = json.loads(self.rfile.read(length) or b"null")
+        except (ValueError, OSError):
+            envelope = None
+        body = _canonical(self.guardian.handle(envelope))
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+
+def serve(port: int = 8787, host: str = "127.0.0.1",
+          policy: PolicyFn = sample_policy,
+          log_path: str = ".acs/envelopes.jsonl") -> ThreadingHTTPServer:
+    """Start the Guardian. Binds loopback by default (never 0.0.0.0 blindly)."""
+    _Handler.guardian = Guardian(policy=policy, log_path=log_path)
+    server = ThreadingHTTPServer((host, port), _Handler)
+    print(f"Guardian listening at http://{host}:{port}/acs "
+          f"(fail-closed posture, log {log_path})", flush=True)
+    return server
+
+
+def main(argv: Optional[list] = None) -> None:
+    parser = argparse.ArgumentParser(description="Minimal ACS Guardian sample")
+    parser.add_argument("--port", type=int, default=8787)
+    parser.add_argument("--host", default="127.0.0.1")
+    parser.add_argument("--log", default=".acs/envelopes.jsonl")
+    args = parser.parse_args(argv)
+    serve(args.port, args.host, log_path=args.log).serve_forever()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/samples/python-guardian-fastmcp/requirements-dev.txt b/samples/python-guardian-fastmcp/requirements-dev.txt
new file mode 100644
index 0000000..939d0e0
--- /dev/null
+++ b/samples/python-guardian-fastmcp/requirements-dev.txt
@@ -0,0 +1,13 @@
+# Test-only dependencies. The sample code itself imports nothing outside the
+# standard library, which is the point of it.
+#
+# Exact pins rather than ranges: this file decides what the sample is proven
+# against, and a resolver picking a different jsonschema on a Tuesday would
+# change that silently.
+#
+# Deliberately not added to uv.lock. The docs deploy gate installs from that
+# lockfile, and nothing the site build needs should grow because a sample
+# gained a test dependency.
+pytest==8.4.2
+jsonschema==4.26.0
+referencing==0.37.0
diff --git a/samples/python-guardian-fastmcp/test_conformance.py b/samples/python-guardian-fastmcp/test_conformance.py
new file mode 100644
index 0000000..f5d1afb
--- /dev/null
+++ b/samples/python-guardian-fastmcp/test_conformance.py
@@ -0,0 +1,124 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Validates every envelope this sample emits against the repository's own schemas.
+
+The stdlib suite in test_sample.py checks behaviour. It cannot check shape,
+because agreeing with yourself about a wire format proves nothing. These tests
+load `specification/v0.1.0/` straight off disk and validate against it, so a
+drift between the sample and the standard fails here rather than in someone
+else's integration.
+
+Needs `jsonschema`, which is not in `uv.lock`. It runs from
+`.github/workflows/samples.yml`, not from the docs deploy gate.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+jsonschema = pytest.importorskip("jsonschema")
+from referencing import Registry, Resource  # noqa: E402
+
+from fastmcp_instrumentation import build_tool_call_envelope  # noqa: E402
+from guardian import Guardian  # noqa: E402
+
+SPEC = Path(__file__).resolve().parents[2] / "specification" / "v0.1.0"
+SESSION = "11111111-2222-4333-8444-555555555555"
+
+
+@pytest.fixture(scope="module")
+def registry() -> Registry:
+    """Every schema in the tree, keyed by its own $id.
+
+    Relative refs (`handshake.json`, `../provenance.json`) then resolve
+    against that base the way the spec authors intended.
+    """
+    registry = Registry()
+    for path in SPEC.rglob("*.json"):
+        doc = json.loads(path.read_text(encoding="utf-8"))
+        if "$id" in doc:
+            registry = registry.with_resource(
+                doc["$id"], Resource.from_contents(doc))
+    return registry
+
+
+def _validate(instance, schema_name: str, registry: Registry) -> None:
+    schema = json.loads((SPEC / schema_name).read_text(encoding="utf-8"))
+    jsonschema.Draft202012Validator(
+        schema, registry=registry).validate(instance)
+
+
+def _guardian_response(payload: dict, log_path,
+                       method: str = "steps/toolCallRequest") -> dict:
+    envelope = {
+        "jsonrpc": "2.0", "method": method, "id": 1,
+        "params": {
+            "acs_version": "0.1.0",
+            "request_id": "99999999-8888-4777-8666-555555555555",
+            "timestamp": "2026-09-10T06:00:00+00:00",
+            "metadata": {"agent_id": "a", "session_id": SESSION},
+            "payload": payload,
+        },
+    }
+    return Guardian(log_path=str(log_path)).handle(envelope)
+
+
+def test_client_request_envelope_is_conformant(registry):
+    env = build_tool_call_envelope("read_file", {"path": "/tmp/x", "lines": 10})
+    _validate(env, "request-envelope.json", registry)
+
+
+def test_client_payload_is_conformant(registry):
+    env = build_tool_call_envelope("read_file", {"path": "/tmp/x"})
+    _validate(env["params"]["payload"], "hooks/tool-call-request.json", registry)
+
+
+def test_raw_arguments_would_have_failed(registry):
+    """The regression guard.
+
+    This is the exact shape the sample emitted before: arguments mapped to
+    bare scalars. It is invalid, and this test is why nobody can put it back.
+    """
+    bad = {"tool": {"name": "read_file"}, "arguments": {"path": "/tmp/x"}}
+    with pytest.raises(jsonschema.ValidationError):
+        _validate(bad, "hooks/tool-call-request.json", registry)
+
+
+@pytest.mark.parametrize("tool,args,expected", [
+    ("search_web", {"q": {"value": "x"}}, "allow"),
+    ("delete_volume", {}, "deny"),
+    ("read_file", {"path": {"value": "../../etc/passwd"}}, "deny"),
+])
+def test_guardian_results_are_conformant(tool, args, expected, registry, tmp_path):
+    out = _guardian_response(
+        {"tool": {"name": tool}, "arguments": args}, tmp_path / "log.jsonl")
+    # Assert the branch first. response-envelope.json accepts an `error`
+    # response too, so conformance alone would pass on a Guardian that
+    # never reached its policy at all.
+    assert out["result"]["decision"] == expected, out
+    _validate(out, "response-envelope.json", registry)
+
+
+def test_guardian_handshake_is_conformant(registry, tmp_path):
+    out = _guardian_response({}, tmp_path / "log.jsonl", method="handshake/hello")
+    assert "negotiated_version" in out["result"], out
+    _validate(out, "response-envelope.json", registry)
+
+
+def test_guardian_error_is_conformant(registry, tmp_path):
+    out = _guardian_response(
+        {"tool": {"name": "t"}, "arguments": {"a": 1}}, tmp_path / "log.jsonl")
+    assert out["error"]["code"] == -32600, out
+    _validate(out, "response-envelope.json", registry)
+
+
+def test_reasoning_present_on_every_deny(registry, tmp_path):
+    out = _guardian_response(
+        {"tool": {"name": "delete_volume"}, "arguments": {}},
+        tmp_path / "log.jsonl")
+    result = out["result"]
+    assert result["decision"] == "deny"
+    # response-envelope.json states reasoning is REQUIRED on deny.
+    assert result["reasoning"]
diff --git a/samples/python-guardian-fastmcp/test_sample.py b/samples/python-guardian-fastmcp/test_sample.py
new file mode 100644
index 0000000..274a972
--- /dev/null
+++ b/samples/python-guardian-fastmcp/test_sample.py
@@ -0,0 +1,221 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Tests for the Python Guardian + FastMCP sample (stdlib only).
+
+Run:  python -m pytest samples/python-guardian-fastmcp -q
+No third-party dependencies: HTTP via urllib, fake inner client,
+async tests driven by asyncio.run (no plugin needed).
+"""
+
+import asyncio
+import json
+import threading
+import urllib.request
+
+import pytest
+
+from guardian import Guardian, _KNOWN_METHODS, serve
+from fastmcp_instrumentation import (
+    GovernedClient,
+    GovernedDenied,
+    build_tool_call_envelope,
+)
+
+
+def _run(coro):
+    return asyncio.run(coro)
+
+
+@pytest.fixture()
+def live(tmp_path):
+    server = serve(0, log_path=str(tmp_path / "env.jsonl"))
+    port = server.server_address[1]
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    yield f"http://127.0.0.1:{port}"
+    server.shutdown()
+    thread.join(timeout=5)
+
+
+def _post(url, envelope):
+    req = urllib.request.Request(
+        url + "/acs", data=json.dumps(envelope).encode(),
+        headers={"Content-Type": "application/json"}, method="POST")
+    with urllib.request.urlopen(req, timeout=5) as resp:
+        return json.loads(resp.read().decode())
+
+
+def _env(method, payload=None, session="11111111-2222-4333-8444-555555555555"):
+    return {
+        "jsonrpc": "2.0", "method": method, "id": 7,
+        "params": {
+            "acs_version": "0.1.0",
+            "request_id": "22222222-3333-4444-8555-666666666666",
+            "timestamp": "2026-09-10T00:00:00+00:00",
+            "metadata": {"agent_id": "a", "session_id": session},
+            "payload": payload or {},
+        },
+    }
+
+
+def test_handshake(live):
+    out = _post(live, _env("handshake/hello"))
+    hello = out["result"]
+    assert hello["negotiated_version"] == "0.1.0"
+    assert hello["on_decision_failure"] == "deny"
+    assert "steps/toolCallRequest" in hello["methods_evaluated"]
+
+
+def test_ping_allows_without_chain_advance(live, tmp_path):
+    out = _post(live, _env("system/ping", {"echo": "hi"}))
+    assert out["result"]["decision"] == "allow"
+
+
+def test_allow_and_deny(live):
+    ok = _post(live, _env("steps/toolCallRequest",
+                           {"tool": {"name": "search_web"},
+                            "arguments": {"q": {"value": "x"}}}))
+    assert ok["result"]["decision"] == "allow"
+    no = _post(live, _env("steps/toolCallRequest",
+                           {"tool": {"name": "delete_volume"},
+                            "arguments": {}}))
+    assert no["result"]["decision"] == "deny"
+    assert no["result"]["reasoning"]  # reasoning REQUIRED on deny
+
+
+def test_path_escape_denies(live):
+    out = _post(live, _env("steps/toolCallRequest",
+                            {"tool": {"name": "read_file"},
+                             "arguments": {"path": {"value": "../../etc/passwd"}}}))
+    assert out["result"]["decision"] == "deny"
+
+
+def test_unknown_method_is_error_not_allow(live):
+    out = _post(live, _env("steps/mindControl"))
+    assert "error" in out and out["error"]["code"] == -32601
+
+
+def test_version_mismatch_is_error(live):
+    env = _env("system/ping")
+    env["params"]["acs_version"] = "9.9.9"
+    out = _post(live, env)
+    assert "error" in out
+
+
+def test_policy_exception_denies():
+    def boom(tool, args, ctx):
+        raise RuntimeError("policy blew up")
+
+    g = Guardian(policy=boom)
+    out = g.handle(_env("steps/toolCallRequest",
+                        {"tool": {"name": "t"}, "arguments": {}}))
+    assert out["result"]["decision"] == "deny"
+
+
+def test_unknown_verdict_denies():
+    g = Guardian(policy=lambda t, a, c: ("maybe", "shrugging"))
+    out = g.handle(_env("steps/toolCallRequest",
+                        {"tool": {"name": "t"}, "arguments": {}}))
+    assert out["result"]["decision"] == "deny"
+
+
+def test_known_methods_cover_sample_surface():
+    assert {"handshake/hello", "system/ping", "steps/toolCallRequest"} <= _KNOWN_METHODS
+
+
+class _FakeInner:
+    def __init__(self):
+        self.calls = []
+
+    async def call_tool(self, name, arguments):
+        self.calls.append((name, arguments))
+        return {"ok": True}
+
+
+def _governed(inner, monkeypatch, decision, reasoning="nope"):
+    import fastmcp_instrumentation as fmi
+
+    # NOTE: post_envelope is sync in production; the fake stays sync.
+    def fake_post(url, envelope):
+        return {"jsonrpc": "2.0", "id": 1,
+                "result": {"type": "final", "acs_version": "0.1.0",
+                           "request_id": "r", "decision": decision,
+                           "reasoning": reasoning}}
+
+    monkeypatch.setattr(fmi, "post_envelope", fake_post)
+    return GovernedClient(inner, "http://guardian.invalid")
+
+
+def test_governed_allow_calls_through(monkeypatch):
+    inner = _FakeInner()
+    governed = _governed(inner, monkeypatch, "allow")
+    out = _run(governed.call_tool("search", {"q": "x"}))
+    assert out == {"ok": True}
+    assert inner.calls == [("search", {"q": "x"})]
+
+
+def test_governed_deny_never_calls(monkeypatch):
+    inner = _FakeInner()
+    governed = _governed(inner, monkeypatch, "deny")
+    with pytest.raises(GovernedDenied):
+        _run(governed.call_tool("delete_volume", {}))
+    assert inner.calls == []  # tool never ran
+
+
+def test_governed_modify_ask_defer_deny_closed(monkeypatch):
+    for verdict in ("modify", "ask", "defer"):
+        inner = _FakeInner()
+        governed = _governed(inner, monkeypatch, verdict)
+        with pytest.raises(GovernedDenied):
+            _run(governed.call_tool("t", {}))
+        assert inner.calls == []
+
+
+def test_governed_unreachable_denies(monkeypatch):
+    import fastmcp_instrumentation as fmi
+
+    def dead(url, envelope):
+        raise ConnectionError("no route")
+
+    monkeypatch.setattr(fmi, "post_envelope", dead)
+    inner = _FakeInner()
+    with pytest.raises(GovernedDenied, match="unreachable"):
+        _run(GovernedClient(inner, "http://guardian.invalid").call_tool("t", {}))
+
+
+def test_envelope_builder_shape():
+    env = build_tool_call_envelope("t", {"a": 1}, "agent-9", "session-1")
+    assert env["method"] == "steps/toolCallRequest"
+    assert env["params"]["metadata"]["agent_id"] == "agent-9"
+    assert env["params"]["payload"]["tool"] == {"name": "t"}
+    # ACS v0.1 wraps each argument so provenance can attach per argument.
+    assert env["params"]["payload"]["arguments"] == {"a": {"value": 1}}
+
+
+def test_raw_arguments_are_rejected(live):
+    """A scalar argument is not the ACS shape, and must not reach the policy."""
+    out = _post(live, _env("steps/toolCallRequest",
+                            {"tool": {"name": "search_web"},
+                             "arguments": {"q": "x"}}))
+    assert "error" in out and out["error"]["code"] == -32600
+    assert "value" in out["error"]["message"]
+
+
+def test_policy_receives_unwrapped_values():
+    seen = {}
+
+    def spy(tool, args, ctx):
+        seen.update(args)
+        return "allow", "ok"
+
+    g = Guardian(policy=spy)
+    g.handle(_env("steps/toolCallRequest",
+                  {"tool": {"name": "t"},
+                   "arguments": {"path": {"value": "/tmp/x"}}}))
+    assert seen == {"path": "/tmp/x"}
+
+
+def test_reason_codes_reach_the_result(live):
+    out = _post(live, _env("steps/toolCallRequest",
+                            {"tool": {"name": "delete_volume"},
+                             "arguments": {}}))
+    assert out["result"]["reason_codes"] == ["destructive_tool"]