samples: Python Guardian agent plus FastMCP client instrumentation

.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

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.

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"}))