diff --git a/CHANGELOG.md b/CHANGELOG.md index 600dcba..c930b9b 100644 --- a/CHANGELOG.md +++ b/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 `: 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 diff --git a/README.md b/README.md index d2958e8..cb14375 100644 --- a/README.md +++ b/README.md @@ -4,13 +4,13 @@ [![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 @@ -18,7 +18,7 @@ dry-run and approve record decisions only; they do not execute tools. 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 --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 ` (or `actgate deny `). +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 --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 diff --git a/actgate/__init__.py b/actgate/__init__.py index cf5c1af..d6a331d 100644 --- a/actgate/__init__.py +++ b/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" diff --git a/actgate/cli.py b/actgate/cli.py index 46a22d7..f4e0bc3 100644 --- a/actgate/cli.py +++ b/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 ") + 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 diff --git a/actgate/core/mcp_proxy.py b/actgate/core/mcp_proxy.py new file mode 100644 index 0000000..d389e5c --- /dev/null +++ b/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() diff --git a/actgate/core/mcp_rpc.py b/actgate/core/mcp_rpc.py new file mode 100755 index 0000000..4ca579a --- /dev/null +++ b/actgate/core/mcp_rpc.py @@ -0,0 +1,49 @@ +"""Minimal MCP/JSON-RPC stdio framing (Content-Length).""" + +from __future__ import annotations + +import json +from typing import Any, BinaryIO + + +class RpcError(RuntimeError): + """Transport or protocol failure.""" + + +def write_message(stream: BinaryIO, message: dict[str, Any]) -> None: + body = json.dumps(message, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + header = ("Content-Length: %d" % len(body) + "\r\n\r\n").encode("ascii") + stream.write(header) + stream.write(body) + stream.flush() + + +def read_message(stream: BinaryIO) -> dict[str, Any] | None: + """Read one framed message. Returns None on clean EOF before a header.""" + headers: dict[str, str] = {} + while True: + line = stream.readline() + if not line: + if not headers: + return None + raise RpcError("unexpected EOF in headers") + if line in (b"\r\n", b"\n"): + break + try: + text = line.decode("ascii").rstrip("\r\n") + except UnicodeDecodeError as exc: + raise RpcError(f"invalid header encoding: {exc}") from exc + if ":" not in text: + raise RpcError(f"malformed header: {text!r}") + key, value = text.split(":", 1) + headers[key.strip().lower()] = value.strip() + if "content-length" not in headers: + raise RpcError("missing Content-Length") + length = int(headers["content-length"]) + body = stream.read(length) + if len(body) != length: + raise RpcError("unexpected EOF in body") + try: + return json.loads(body.decode("utf-8")) + except json.JSONDecodeError as exc: + raise RpcError(f"invalid JSON body: {exc}") from exc diff --git a/pyproject.toml b/pyproject.toml index 6037331..ab5c76e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,7 @@ [project] name = "actgate" -version = "0.1.0" -description = "Local intent ledger and approval gate for tool actions" +version = "0.2.0" +description = "Local intent ledger and MCP tool-call proxy with approval gate" readme = "README.md" license = "MIT" requires-python = ">=3.10" diff --git a/tests/fake_mcp_upstream.py b/tests/fake_mcp_upstream.py new file mode 100644 index 0000000..8dc6975 --- /dev/null +++ b/tests/fake_mcp_upstream.py @@ -0,0 +1,113 @@ +"""Minimal MCP upstream for tests: one echo tool.""" + +from __future__ import annotations + +import json +import sys +from typing import Any + +# Inline framing so the script is standalone when run as subprocess. +def write_message(message: dict[str, Any]) -> None: + body = json.dumps(message, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + sys.stdout.buffer.write(f"Content-Length: {len(body)}\r\n\r\n".encode("ascii")) + sys.stdout.buffer.write(body) + sys.stdout.buffer.flush() + + +def read_message() -> dict[str, Any] | None: + headers: dict[str, str] = {} + while True: + line = sys.stdin.buffer.readline() + if not line: + return None if not headers else (_ for _ in ()).throw(RuntimeError("EOF")) + if line in (b"\r\n", b"\n"): + break + text = line.decode("ascii").rstrip("\r\n") + key, value = text.split(":", 1) + headers[key.strip().lower()] = value.strip() + length = int(headers["content-length"]) + body = sys.stdin.buffer.read(length) + return json.loads(body.decode("utf-8")) + + +CALLS = 0 +CALLS_PATH = __import__("os").environ.get("ACTGATE_FAKE_CALLS") + +TOOLS = [ + { + "name": "echo", + "description": "Echo arguments", + "inputSchema": { + "type": "object", + "properties": {"text": {"type": "string"}}, + "required": ["text"], + }, + } +] + + +def main() -> int: + while True: + msg = read_message() + if msg is None: + return 0 + method = msg.get("method") + req_id = msg.get("id") + params = msg.get("params") or {} + if req_id is None: + continue + if method == "initialize": + write_message( + { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "protocolVersion": "2024-11-05", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "fake-upstream", "version": "0.0.1"}, + }, + } + ) + elif method == "tools/list": + write_message({"jsonrpc": "2.0", "id": req_id, "result": {"tools": TOOLS}}) + elif method == "tools/call": + global CALLS + CALLS += 1 + if CALLS_PATH: + open(CALLS_PATH, "w", encoding="utf-8").write(str(CALLS)) + name = params.get("name") + arguments = params.get("arguments") or {} + if name != "echo": + write_message( + { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": -32601, "message": f"unknown tool {name}"}, + } + ) + else: + text = arguments.get("text", "") + write_message( + { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": f"echo:{text}"}], + "isError": False, + }, + } + ) + elif method == "ping": + write_message({"jsonrpc": "2.0", "id": req_id, "result": {}}) + else: + write_message( + { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": -32601, "message": f"method not found: {method}"}, + } + ) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_mcp_proxy.py b/tests/test_mcp_proxy.py new file mode 100644 index 0000000..1815168 --- /dev/null +++ b/tests/test_mcp_proxy.py @@ -0,0 +1,168 @@ +"""MCP proxy: pending until approve, then execute once.""" + +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +from actgate.cli import main +from actgate.core.ledger import Ledger +from actgate.core.mcp_rpc import read_message, write_message +from actgate.core.verify import verify_ledger + + +FAKE = Path(__file__).resolve().parent / "fake_mcp_upstream.py" + + +def _rpc(proc: subprocess.Popen[bytes], method: str, params: dict | None, req_id: int) -> dict: + assert proc.stdin and proc.stdout + msg: dict = {"jsonrpc": "2.0", "method": method, "id": req_id} + if params is not None: + msg["params"] = params + write_message(proc.stdin, msg) + reply = read_message(proc.stdout) + assert reply is not None + return reply + + +@pytest.fixture +def root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + monkeypatch.chdir(tmp_path) + monkeypatch.delenv("ACTGATE_SEAL_KEY", raising=False) + assert main(["init"]) == 0 + return tmp_path + + +def test_tools_call_pending_then_approve_executes_once(root: Path, monkeypatch: pytest.MonkeyPatch) -> None: + calls = root / "fake_calls.txt" + monkeypatch.setenv("ACTGATE_FAKE_CALLS", str(calls)) + upstream = [sys.executable, str(FAKE)] + proc = subprocess.Popen( + [sys.executable, "-m", "actgate", "mcp", "--upstream", *upstream], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + cwd=root, + bufsize=0, + ) + try: + init = _rpc( + proc, + "initialize", + { + "protocolVersion": "2024-11-05", + "capabilities": {}, + "clientInfo": {"name": "test", "version": "0"}, + }, + 1, + ) + assert "result" in init + listed = _rpc(proc, "tools/list", {}, 2) + assert listed["result"]["tools"][0]["name"] == "echo" + + pending = _rpc(proc, "tools/call", {"name": "echo", "arguments": {"text": "hi"}}, 3) + body = pending["result"] + assert body.get("isError") is True + assert "ACTGATE_PENDING" in body["content"][0]["text"] + intent_id = body["_actgate"]["intent_id"] + + # Upstream must not have been called yet: ledger has propose only + entries = Ledger.open(root=root).read_entries() + assert [e["action"] for e in entries] == ["propose"] + + assert main(["approve", intent_id]) == 0 + + done = _rpc(proc, "tools/call", {"name": "echo", "arguments": {"text": "hi"}}, 4) + assert done["result"]["content"][0]["text"] == "echo:hi" + assert done["result"].get("isError") is False + + again = _rpc(proc, "tools/call", {"name": "echo", "arguments": {"text": "hi"}}, 5) + assert again["result"].get("isError") is True + assert "ACTGATE_ALREADY_EXECUTED" in again["result"]["content"][0]["text"] + + actions = [e["action"] for e in Ledger.open(root=root).read_entries()] + assert actions.count("execute") == 1 + assert calls.read_text(encoding="utf-8") == "1" + assert verify_ledger(Ledger.open(root=root)).ok + finally: + proc.terminate() + proc.wait(timeout=3) + + +def test_unapproved_never_hits_upstream(root: Path) -> None: + upstream = [sys.executable, str(FAKE)] + proc = subprocess.Popen( + [sys.executable, "-m", "actgate", "mcp", "--upstream", *upstream], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + cwd=root, + bufsize=0, + ) + try: + _rpc( + proc, + "initialize", + { + "protocolVersion": "2024-11-05", + "capabilities": {}, + "clientInfo": {"name": "test", "version": "0"}, + }, + 1, + ) + pending = _rpc(proc, "tools/call", {"name": "echo", "arguments": {"text": "x"}}, 2) + iid = pending["result"]["_actgate"]["intent_id"] + assert main(["deny", iid, "--reason", "nope"]) == 1 + denied = _rpc(proc, "tools/call", {"name": "echo", "arguments": {"text": "x"}}, 3) + assert denied["result"].get("isError") is True + assert "ACTGATE_DENIED" in denied["result"]["content"][0]["text"] + actions = [e["action"] for e in Ledger.open(root=root).read_entries()] + assert "execute" not in actions + finally: + proc.terminate() + proc.wait(timeout=3) + + +def test_approval_does_not_cover_different_args(root: Path) -> None: + upstream = [sys.executable, str(FAKE)] + proc = subprocess.Popen( + [sys.executable, "-m", "actgate", "mcp", "--upstream", *upstream], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + cwd=root, + bufsize=0, + ) + try: + _rpc( + proc, + "initialize", + { + "protocolVersion": "2024-11-05", + "capabilities": {}, + "clientInfo": {"name": "test", "version": "0"}, + }, + 1, + ) + pending = _rpc(proc, "tools/call", {"name": "echo", "arguments": {"text": "hi"}}, 2) + iid = pending["result"]["_actgate"]["intent_id"] + assert main(["approve", iid]) == 0 + + other = _rpc(proc, "tools/call", {"name": "echo", "arguments": {"text": "bye"}}, 3) + assert other["result"].get("isError") is True + assert "ACTGATE_PENDING" in other["result"]["content"][0]["text"] + assert other["result"]["_actgate"]["intent_id"] != iid + + other_tool = _rpc(proc, "tools/call", {"name": "other", "arguments": {"text": "hi"}}, 4) + assert other_tool["result"].get("isError") is True + assert "ACTGATE_PENDING" in other_tool["result"]["content"][0]["text"] + + actions = [e["action"] for e in Ledger.open(root=root).read_entries()] + assert "execute" not in actions + finally: + proc.terminate() + proc.wait(timeout=3)