feat: add MCP stdio proxy gated by IntentLedger

CHANGELOG.md

@@ -5,6 +5,16 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.2.0] - 2026-09-06
+
+### Added
+
+- `actgate mcp --upstream <cmd...>`: stdio MCP proxy that forwards `tools/list`,
+  proposes ledger intents on `tools/call`, and calls upstream only after approve.
+- Pending first call returns `ACTGATE_PENDING intent_id=...` (no upstream side effect).
+- Identical approved `tools/call` executes once; deny never hits upstream.
+- Stdlib JSON-RPC Content-Length framing (no MCP SDK dependency).
+
 ## [0.1.0] - 2026-09-06
 
 ### Added

README.md

@@ -4,21 +4,21 @@
 [![Python](https://img.shields.io/badge/python-3.10%2B-blue)](pyproject.toml)
 [![License: MIT](https://img.shields.io/badge/license-MIT-green)](LICENSE)
 
-Local IntentLedger: propose a tool action, record approve or deny in an
-append-only hash-chained ledger, verify the chain before you trust it.
+Local IntentLedger and MCP stdio proxy: propose a tool action, approve or deny
+in an append-only hash-chained ledger, then let an identical tools/call reach
+one upstream MCP server.
 
-This is not an MCP proxy yet. This is not a SaaS. Everything runs offline
-against files on disk.
-
-dry-run and approve record decisions only; they do not execute tools.
+This is not a SaaS. The ledger path stays on disk. dry-run and approve record
+decisions only; they do not execute tools. The MCP proxy is what executes, and
+only after approve.
 
 ## Install
 
 ```
 pip install -e .[dev]
 ```
 
-## Quickstart
+## CLI quickstart
 
 ```
 actgate init
@@ -29,13 +29,29 @@ actgate verify
 actgate list
 ```
 
-Deny path:
+## MCP proxy
+
+Point your MCP client at ActGate instead of the upstream server:
 
 ```
-actgate deny <intent_id> --reason "too broad"
-# exits 1
+actgate init
+actgate mcp --upstream python -m some_mcp_server
 ```
 
+Flow:
+
+1. Client `tools/list` is forwarded to upstream. Only `tools/call` is gated;
+   other methods are forwarded.
+2. First `tools/call` for a tool+args writes a propose event and returns
+   `ACTGATE_PENDING intent_id=...` (upstream is not called).
+3. Human: `actgate approve <intent_id>` (or `actgate deny <intent_id>`).
+4. Identical subsequent `tools/call` (same tool and args) runs upstream once
+   and appends an `execute` event. A third call returns already-executed.
+5. Denied intents never hit upstream.
+
+Bare `verify` checks hash-chain integrity only. Set `ACTGATE_SEAL_KEY` for
+optional HMAC seals, or pass `verify --require-seal`.
+
 ## Exit codes
 
 | Code | Meaning |
@@ -57,27 +73,11 @@ actgate deny <intent_id> --reason "too broad"
 }
 ```
 
-Provide either `args` or `args_hash` (sha256 of canonical JSON args). Optional
-`blast_tags` and `requested_mode`.
-
-## Ledger
-
-`.actgate/ledger.jsonl` is append-only. Each line has `prev_hash` / `entry_hash`
-(sha256). Bare `verify` checks chain integrity only: a rewritten but
-internally consistent chain still passes. It is not a signature check unless
-you opt in.
-
-Optional authenticity: set `ACTGATE_SEAL_KEY` when writing so entries get an
-HMAC seal. Then `verify` (with the key set) requires matching seals, or pass
-`verify --require-seal` to fail when seals are missing.
-
-Path escapes outside the ledger root are rejected (exit 2).
-
 ## What this is not
 
-- Not an MCP proxy (yet)
 - Not a hosted approval product
-- No network calls in the core path
+- Not a policy DSL
+- No network calls in the ledger core path (the MCP proxy talks to a local upstream process)
 
 ## Development
 

actgate/__init__.py

@@ -1,3 +1,3 @@
-"""ActGate: local IntentLedger for tool-action propose / approve / deny."""
+"""ActGate: local IntentLedger and MCP tool-call proxy."""
 
-__version__ = "0.1.0"
+__version__ = "0.2.0"

actgate/cli.py

@@ -12,6 +12,8 @@
 from actgate.core.intent import Intent, IntentError, build_intent
 from actgate.core.ledger import Ledger, LedgerError, resolve_ledger_path
 from actgate.core.verify import verify_ledger
+from actgate.core.mcp_proxy import run_proxy
+from actgate.core.mcp_rpc import RpcError
 
 
 def _eprint(msg: str) -> None:
@@ -203,6 +205,21 @@ def cmd_list(args: argparse.Namespace) -> int:
     return 0
 
 
+
+
+def cmd_mcp(args: argparse.Namespace) -> int:
+    """Stdio MCP proxy: gate tools/call via IntentLedger before upstream."""
+    if not args.upstream:
+        _eprint("mcp requires --upstream <cmd...>")
+        return 2
+    root = Path(args.root).resolve() if getattr(args, "root", None) else Path.cwd()
+    path = getattr(args, "ledger", None)
+    try:
+        return run_proxy(root=root, upstream_cmd=list(args.upstream), ledger_path=path)
+    except (RpcError, OSError, LedgerError) as exc:
+        _eprint(str(exc))
+        return 2
+
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(
         prog="actgate",
@@ -255,6 +272,19 @@ def build_parser() -> argparse.ArgumentParser:
     p_list = sub.add_parser("list", help="list intents and status")
     p_list.set_defaults(func=cmd_list)
 
+    p_mcp = sub.add_parser(
+        "mcp",
+        help="stdio MCP proxy: gate tools/call until ledger approve",
+    )
+    p_mcp.add_argument(
+        "--upstream",
+        nargs="+",
+        metavar="CMD",
+        required=True,
+        help="upstream MCP server command and args",
+    )
+    p_mcp.set_defaults(func=cmd_mcp)
+
     return parser
 
 

actgate/core/mcp_proxy.py

@@ -0,0 +1,219 @@
+"""MCP stdio proxy gated by IntentLedger."""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+from actgate.core.intent import Intent, build_intent, hash_args
+from actgate.core.ledger import Ledger
+from actgate.core.mcp_rpc import RpcError, read_message, write_message
+
+
+def _args_match(intent: dict[str, Any], tool: str, arguments: dict[str, Any] | None) -> bool:
+    if intent.get("tool") != tool:
+        return False
+    args = intent.get("args")
+    if args is not None:
+        return args == (arguments or {})
+    args_hash = intent.get("args_hash")
+    if args_hash and arguments is not None:
+        return args_hash == hash_args(arguments)
+    return arguments in (None, {})
+
+
+def _find_proposal(ledger: Ledger, tool: str, arguments: dict[str, Any] | None) -> dict[str, Any] | None:
+    for entry in ledger.read_entries():
+        if entry.get("action") != "propose":
+            continue
+        intent = entry.get("intent") or {}
+        if _args_match(intent, tool, arguments):
+            return entry
+    return None
+
+
+def _was_executed(ledger: Ledger, intent_id: str) -> bool:
+    for entry in ledger.read_entries():
+        if entry.get("intent_id") == intent_id and entry.get("action") == "execute":
+            return True
+    return False
+
+
+def _pending_result(intent_id: str) -> dict[str, Any]:
+    return {
+        "content": [
+            {
+                "type": "text",
+                "text": f"ACTGATE_PENDING intent_id={intent_id}",
+            }
+        ],
+        "isError": True,
+        "_actgate": {"status": "pending", "intent_id": intent_id},
+    }
+
+
+def _denied_result(intent_id: str, reason: str | None) -> dict[str, Any]:
+    why = reason or "denied"
+    return {
+        "content": [{"type": "text", "text": f"ACTGATE_DENIED intent_id={intent_id}: {why}"}],
+        "isError": True,
+        "_actgate": {"status": "denied", "intent_id": intent_id},
+    }
+
+
+class McpProxy:
+    def __init__(self, ledger: Ledger, upstream_cmd: list[str]) -> None:
+        self.ledger = ledger
+        self.upstream_cmd = upstream_cmd
+        self._proc: subprocess.Popen[bytes] | None = None
+        self._client_in = sys.stdin.buffer
+        self._client_out = sys.stdout.buffer
+
+    def start_upstream(self) -> None:
+        self._proc = subprocess.Popen(
+            self.upstream_cmd,
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=sys.stderr,
+            bufsize=0,
+        )
+
+    def close(self) -> None:
+        if self._proc is None:
+            return
+        if self._proc.stdin:
+            self._proc.stdin.close()
+        self._proc.terminate()
+        try:
+            self._proc.wait(timeout=2)
+        except subprocess.TimeoutExpired:
+            self._proc.kill()
+        self._proc = None
+
+    def _upstream_request(self, method: str, params: dict[str, Any] | None, req_id: Any) -> dict[str, Any]:
+        assert self._proc and self._proc.stdin and self._proc.stdout
+        msg: dict[str, Any] = {"jsonrpc": "2.0", "method": method, "id": req_id}
+        if params is not None:
+            msg["params"] = params
+        write_message(self._proc.stdin, msg)
+        reply = read_message(self._proc.stdout)
+        if reply is None:
+            raise RpcError("upstream closed")
+        return reply
+
+    def _upstream_notify(self, method: str, params: dict[str, Any] | None = None) -> None:
+        assert self._proc and self._proc.stdin
+        msg: dict[str, Any] = {"jsonrpc": "2.0", "method": method}
+        if params is not None:
+            msg["params"] = params
+        write_message(self._proc.stdin, msg)
+
+    def _handle_tools_call(self, req_id: Any, params: dict[str, Any]) -> dict[str, Any]:
+        tool = params.get("name") or ""
+        arguments = params.get("arguments")
+        if arguments is None:
+            arguments = {}
+        if not isinstance(arguments, dict):
+            return {
+                "jsonrpc": "2.0",
+                "id": req_id,
+                "error": {"code": -32602, "message": "arguments must be an object"},
+            }
+
+        self.ledger.ensure()
+        proposal = _find_proposal(self.ledger, tool, arguments)
+        if proposal is None:
+            intent = build_intent(tool=tool, args=arguments, requested_mode="execute")
+            self.ledger.append("propose", intent=intent)
+            return {"jsonrpc": "2.0", "id": req_id, "result": _pending_result(intent.id)}
+
+        intent_id = proposal["intent_id"]
+        decision = self.ledger.latest_decision(intent_id)
+        if decision is None:
+            return {"jsonrpc": "2.0", "id": req_id, "result": _pending_result(intent_id)}
+        if decision.get("action") == "deny":
+            return {
+                "jsonrpc": "2.0",
+                "id": req_id,
+                "result": _denied_result(intent_id, decision.get("reason")),
+            }
+        if _was_executed(self.ledger, intent_id):
+            return {
+                "jsonrpc": "2.0",
+                "id": req_id,
+                "result": {
+                    "content": [
+                        {
+                            "type": "text",
+                            "text": f"ACTGATE_ALREADY_EXECUTED intent_id={intent_id}",
+                        }
+                    ],
+                    "isError": True,
+                    "_actgate": {"status": "already_executed", "intent_id": intent_id},
+                },
+            }
+
+        # Record execute before upstream so a crash after success cannot double-call.
+        intent_obj = Intent.from_dict(proposal["intent"])
+        self.ledger.append("execute", intent=intent_obj, outcome="started")
+        upstream = self._upstream_request("tools/call", params, req_id)
+        return upstream
+
+    def handle(self, message: dict[str, Any]) -> dict[str, Any] | None:
+        """Handle one client message. Returns response or None for notifications."""
+        if "method" not in message:
+            return {
+                "jsonrpc": "2.0",
+                "id": message.get("id"),
+                "error": {"code": -32600, "message": "invalid request"},
+            }
+        method = message["method"]
+        req_id = message.get("id")
+        params = message.get("params") or {}
+        if not isinstance(params, dict):
+            params = {}
+
+        # notifications (no id)
+        if req_id is None:
+            if method == "notifications/initialized":
+                self._upstream_notify(method, params or None)
+            return None
+
+        if method == "tools/call":
+            return self._handle_tools_call(req_id, params)
+
+        # forward initialize, tools/list, ping, etc.
+        return self._upstream_request(method, params if params else None, req_id)
+
+    def run(self) -> int:
+        self.ledger.ensure()
+        self.start_upstream()
+        try:
+            while True:
+                msg = read_message(self._client_in)
+                if msg is None:
+                    return 0
+                try:
+                    reply = self.handle(msg)
+                except RpcError as exc:
+                    if "id" in msg:
+                        write_message(
+                            self._client_out,
+                            {
+                                "jsonrpc": "2.0",
+                                "id": msg.get("id"),
+                                "error": {"code": -32000, "message": str(exc)},
+                            },
+                        )
+                    return 1
+                if reply is not None:
+                    write_message(self._client_out, reply)
+        finally:
+            self.close()
+
+
+def run_proxy(root: Path, upstream_cmd: list[str], ledger_path: Path | str | None = None) -> int:
+    ledger = Ledger.open(root=root, path=ledger_path)
+    return McpProxy(ledger, upstream_cmd).run()