feat(action-gate): deterministic Gate/Prove policy firewall & hash-chained ledger for MCP tool execution

src/action-gate/.python-version

@@ -0,0 +1 @@
+3.10

src/action-gate/Dockerfile

@@ -0,0 +1,11 @@
+FROM python:3.12-slim
+WORKDIR /app
+COPY pyproject.toml README.md LICENSE ./
+COPY aag ./aag
+COPY fixtures ./fixtures
+RUN pip install --no-cache-dir .
+LABEL org.opencontainers.image.source="https://github.com/AAH20/agent-action-gate"
+LABEL org.opencontainers.image.description="Gate/Prove MCP server — deny unattended destructive agent tools"
+LABEL io.modelcontextprotocol.server.name="io.github.AAH20/agent-action-gate"
+ENV PYTHONUNBUFFERED=1
+ENTRYPOINT ["python3", "-m", "aag", "serve"]

src/action-gate/README.md

@@ -0,0 +1,89 @@
+# Action Gate MCP Server
+
+<!-- mcp-name: io.github.modelcontextprotocol/server-action-gate -->
+
+A Model Context Protocol server that provides deterministic Gate/Prove policy boundaries, Simulation Mode, and append-only hash-chained action ledgers for agent tool execution.
+
+Source: https://github.com/modelcontextprotocol/servers/tree/main/src/action-gate
+
+### Problem & Invariants
+
+As autonomous agents and MCP clients gain write and shell execution capabilities, treating raw LLM probability or tool-call intent as authorization introduces severe production and compliance risks.
+
+This server enforces a zero-trust execution boundary:
+1. **Never Equate Intent to Approval:** High model confidence (`model_confidence: 0.99`) is explicitly rejected as authorization for destructive or provision operations.
+2. **Tiered Tool Governance:**
+   - `READ`: Allowed and logged.
+   - `WRITE`: Defaults to non-destructive simulation mode unless approved.
+   - `DESTRUCTIVE` / `PROVISION` / `DECOMMISSION`: Hard `DENY` unless accompanied by an authorized human-in-the-loop (HITL) prove token.
+3. **Immutable Action Ledger:** Every evaluation and decision is recorded in an append-only JSONL ledger with SHA-256 hash-chain verification.
+4. **Atomic Kill-Switch:** Immediate tool freeze via environment variable (`AAG_KILL_SWITCH=1`) or file sentinel (`artifacts/KILL`).
+
+### Available Tools
+
+- `gate_check` - Evaluate a proposed agent/MCP tool call under Gate/Prove rules. Never executes the tool.
+  - Required arguments:
+    - `tool` (string): Tool name the agent intends to invoke.
+  - Optional arguments:
+    - `tier` (string): `read`, `write`, `destructive`, `provision`, `decommission`
+    - `arguments` (object): Proposed tool arguments
+    - `thought` (string): Agent rationale
+    - `model_confidence` (number): LLM confidence score (0.0 to 1.0)
+    - `approved` (boolean): Human HITL approval flag
+    - `prove_token` (string): Cryptographic token matching `AAG_PROVE_TOKEN`
+    - `simulate` (boolean): Force non-destructive simulation mode
+    - `actor_id` (string): Agent or session identifier
+    - `idempotency_key` (string): Unique request key to detect replays
+
+- `ledger_verify` - Verify the SHA-256 hash chain of the Action Ledger for audit diligence evidence.
+
+## Installation
+
+### Using uv (recommended)
+
+```bash
+uvx mcp-server-action-gate
+```
+
+### Using PIP
+
+```bash
+pip install mcp-server-action-gate
+```
+
+After installation, run via:
+
+```bash
+python -m mcp_server_action_gate serve
+```
+
+## Configuration
+
+### Configure for Claude Desktop / Agent Runner
+
+Add to your MCP settings:
+
+```json
+{
+  "mcpServers": {
+    "action-gate": {
+      "command": "uvx",
+      "args": ["mcp-server-action-gate", "serve"],
+      "env": {
+        "AAG_PROVE_TOKEN": "your-hitl-secret-token",
+        "AAG_KILL_SWITCH": "0"
+      }
+    }
+  }
+}
+```
+
+## Testing & Validation
+
+```bash
+python -m unittest discover -s tests -p "test_*.py"
+```
+
+## License
+
+MIT

src/action-gate/fixtures/policy.default.json

@@ -0,0 +1,21 @@
+{
+  "allowed_tools": [
+    "search",
+    "get_identity",
+    "ticket.create",
+    "shell.exec",
+    "edr.disable",
+    "iam.create_user",
+    "vpc.delete"
+  ],
+  "denied_tools": ["connector.canary_override"],
+  "tool_tiers": {
+    "search": "read",
+    "get_identity": "read",
+    "ticket.create": "write",
+    "shell.exec": "destructive",
+    "edr.disable": "destructive",
+    "iam.create_user": "provision",
+    "vpc.delete": "decommission"
+  }
+}

src/action-gate/fixtures/proved_decommission.json

@@ -0,0 +1,11 @@
+{
+  "actor": {"id": "human-sre", "type": "human", "role": "approver"},
+  "offered_token": "demo-hitl",
+  "request": {
+    "tool": "vpc.delete",
+    "tier": "decommission",
+    "args": {"note": "Decommission requires HITL prove token plus approved true."},
+    "approved": true,
+    "attack_technique": "T1578"
+  }
+}

src/action-gate/fixtures/t1059_unattended_shell.json

@@ -0,0 +1,13 @@
+{
+  "actor": {"id": "agent-coder", "type": "agent"},
+  "request": {
+    "tool": "shell.exec",
+    "tier": "destructive",
+    "args": {"note": "ATT&CK T1059 command-execution intent. No payload."},
+    "thought": "95 percent sure this shell call is required to finish the task",
+    "model_confidence": 0.95,
+    "approved": false,
+    "attack_technique": "T1059",
+    "idempotency_key": "t1059-unattended-1"
+  }
+}

src/action-gate/fixtures/t1078_read_identity.json

@@ -0,0 +1,9 @@
+{
+  "actor": {"id": "agent-support", "type": "mcp_tool"},
+  "request": {
+    "tool": "get_identity",
+    "tier": "read",
+    "args": {"note": "ATT&CK T1078 valid-account lookup. Read only."},
+    "attack_technique": "T1078"
+  }
+}

src/action-gate/fixtures/t1562_impair_defenses.json

@@ -0,0 +1,12 @@
+{
+  "actor": {"id": "agent-ops", "type": "agent"},
+  "request": {
+    "tool": "edr.disable",
+    "tier": "destructive",
+    "args": {"note": "ATT&CK T1562 impair-defenses intent. No payload."},
+    "thought": "Disable noisy controls so the deploy succeeds",
+    "model_confidence": 0.99,
+    "approved": false,
+    "attack_technique": "T1562"
+  }
+}

src/action-gate/fixtures/write_ticket_simulate.json

@@ -0,0 +1,9 @@
+{
+  "actor": {"id": "agent-sre", "type": "agent"},
+  "request": {
+    "tool": "ticket.create",
+    "tier": "write",
+    "args": {"note": "Open a ticket. Default path is SIMULATE until HITL."},
+    "approved": false
+  }
+}

src/action-gate/pyproject.toml

@@ -0,0 +1,46 @@
+[project]
+name = "mcp-server-action-gate"
+version = "0.2.0"
+description = "Model Context Protocol server providing deterministic Gate/Prove policy boundaries & hash-chained action ledger for agent tool execution"
+readme = "README.md"
+requires-python = ">=3.10"
+authors = [
+    { name = "Ahmed Hassan", email = "info@a2zsoc.com" },
+]
+keywords = ["mcp", "security", "gate-prove", "governance", "action-ledger", "llm", "ai-agents"]
+license = { text = "MIT" }
+classifiers = [
+    "Development Status :: 4 - Beta",
+    "Intended Audience :: Developers",
+    "License :: OSI Approved :: MIT License",
+    "Programming Language :: Python :: 3",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
+]
+dependencies = [
+    "mcp>=1.2.0",
+]
+
+[project.urls]
+Repository = "https://github.com/modelcontextprotocol/servers/tree/main/src/action-gate"
+Issues = "https://github.com/modelcontextprotocol/servers/issues"
+
+[project.scripts]
+mcp-server-action-gate = "mcp_server_action_gate.server:serve"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/mcp_server_action_gate"]
+
+[dependency-groups]
+dev = ["pyright>=1.1.407", "ruff>=0.8.1", "pytest>=8.0.0"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+python_files = "test_*.py"
+python_classes = "Test*"
+python_functions = "test_*"

src/action-gate/src/mcp_server_action_gate/__init__.py

@@ -0,0 +1,7 @@
+from __future__ import annotations
+
+from mcp_server_action_gate.gate import AgentActionGate
+from mcp_server_action_gate.ledger import ActionLedger
+from mcp_server_action_gate.schema import CONSULTATION, INSTANT_AUDIT
+
+__all__ = ["AgentActionGate", "ActionLedger", "INSTANT_AUDIT", "CONSULTATION"]

src/action-gate/src/mcp_server_action_gate/__main__.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+from mcp_server_action_gate.cost import estimate_cost_avoidance
+from mcp_server_action_gate.demo import demo, load_case, run_case
+from mcp_server_action_gate.gate import AgentActionGate, kill_switch_engaged
+from mcp_server_action_gate.ledger import ActionLedger
+from mcp_server_action_gate.schema import CONSULTATION, INSTANT_AUDIT
+
+
+def _print(payload: object) -> None:
+    print(json.dumps(payload, indent=2, sort_keys=True))
+    print(f"\nInstant Audit $499: {INSTANT_AUDIT}", file=sys.stderr)
+    print(f"Consultation: {CONSULTATION}", file=sys.stderr)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(prog="mcp-server-action-gate", description="Agent Action Gate — Gate/Prove for MCP and agent tools")
+    parser.add_argument("--ledger", default="artifacts/action_ledger.jsonl")
+    parser.add_argument("--prove-token", default="")
+    sub = parser.add_subparsers(dest="cmd", required=True)
+
+    sub.add_parser("demo", help="Run ATT&CK-tagged fixtures")
+    chk = sub.add_parser("check", help="Evaluate one fixture or MCP JSON file")
+    chk.add_argument("path")
+    chk.add_argument("--simulate", action="store_true")
+    sub.add_parser("bench", help="Print illustrative cost-avoidance sketch")
+    sub.add_parser("serve", help="MCP stdio server (Gate/Prove tools, never executes)")
+
+    args = parser.parse_args(argv)
+    ledger = ActionLedger(Path(args.ledger))
+    token = args.prove_token
+
+    if args.cmd == "serve":
+        from mcp_server_action_gate.server import serve
+
+        serve(gate=AgentActionGate(ledger=ledger, prove_token=token))
+        return 0
+
+    if args.cmd == "demo":
+        if Path(args.ledger).exists():
+            Path(args.ledger).unlink()
+        out = demo(Path(args.ledger), prove_token=token or "demo-hitl")
+        _print(out)
+        return 0 if out.get("ledger_chain_ok") else 1
+
+    if args.cmd == "check":
+        case = load_case(Path(args.path))
+        if args.simulate:
+            case["simulate"] = True
+        gate = AgentActionGate(ledger=ledger, prove_token=token)
+        _print(run_case(gate, case, offered_token=str(case.get("offered_token") or token)))
+        return 0
+
+    if args.cmd == "bench":
+        sketch = estimate_cost_avoidance(
+            unattended_high_tier_denied=2,
+            write_simulated=1,
+            fde_minutes_saved=45.0,
+        ).to_dict()
+        sketch["kill_switch"] = kill_switch_engaged()
+        sketch["instant_audit"] = INSTANT_AUDIT
+        sketch["consultation"] = CONSULTATION
+        _print(sketch)
+        return 0
+
+    return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/action-gate/src/mcp_server_action_gate/cost.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+
+
+@dataclass(frozen=True)
+class CostAvoidanceEstimate:
+    unattended_high_tier_denied: int
+    write_simulated: int
+    fde_minutes_saved: float
+    hourly_operator_cost_usd: float
+    ungated_blast_cost_usd: float
+    estimated_operator_savings_usd: float
+    estimated_blast_avoidance_usd: float
+    estimated_total_avoidance_usd: float
+    notes: list[str]
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+
+def estimate_cost_avoidance(
+    *,
+    unattended_high_tier_denied: int,
+    write_simulated: int,
+    fde_minutes_saved: float,
+    hourly_operator_cost_usd: float = 200.0,
+    ungated_blast_cost_usd: float = 25000.0,
+) -> CostAvoidanceEstimate:
+    """
+    Illustrative Gate/Prove cost sketch for Instant Audit / consultation.
+
+    Defaults are mid-market FDE assumptions — not a quote.
+    """
+    operator = round((fde_minutes_saved / 60.0) * hourly_operator_cost_usd, 2)
+    blast = round(unattended_high_tier_denied * ungated_blast_cost_usd, 2)
+    return CostAvoidanceEstimate(
+        unattended_high_tier_denied=unattended_high_tier_denied,
+        write_simulated=write_simulated,
+        fde_minutes_saved=fde_minutes_saved,
+        hourly_operator_cost_usd=hourly_operator_cost_usd,
+        ungated_blast_cost_usd=ungated_blast_cost_usd,
+        estimated_operator_savings_usd=operator,
+        estimated_blast_avoidance_usd=blast,
+        estimated_total_avoidance_usd=round(operator + blast, 2),
+        notes=[
+            "illustrative_not_a_quote",
+            "unattended_high_tier_denied_avoids_agent_executing_destructive_tools",
+            "write_simulated_is_gate_prove_no_side_effects",
+            "compare_to_instant_audit_499_evaluation_instrument",
+        ],
+    )