From cfe5622bec0f578c8da04f59713853b99d329468 Mon Sep 17 00:00:00 2001 From: kartsan03 Date: Sun, 6 Sep 2026 01:34:04 +0200 Subject: [PATCH 1/5] feat: add MCP stdio proxy gated by IntentLedger Forward tools/list from upstream. Propose on tools/call; execute once after approve; deny never hits upstream. Stdlib Content-Length JSON-RPC framing. --- actgate/__init__.py | 4 +- actgate/cli.py | 29 +++++ actgate/core/mcp_proxy.py | 227 ++++++++++++++++++++++++++++++++++++++ actgate/core/mcp_rpc.py | 57 ++++++++++ pyproject.toml | 4 +- 5 files changed, 317 insertions(+), 4 deletions(-) create mode 100644 actgate/core/mcp_proxy.py create mode 100644 actgate/core/mcp_rpc.py 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..5216438 100644 --- a/actgate/cli.py +++ b/actgate/cli.py @@ -12,6 +12,7 @@ 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 def _eprint(msg: str) -> None: @@ -203,6 +204,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 Exception as exc: # noqa: BLE001 — surface proxy failures as setup errors + _eprint(str(exc)) + return 2 + def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="actgate", @@ -255,6 +271,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..fa715e4 --- /dev/null +++ b/actgate/core/mcp_proxy.py @@ -0,0 +1,227 @@ +"""MCP stdio proxy gated by IntentLedger.""" + +from __future__ import annotations + +import json +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}, + }, + } + + upstream = self._upstream_request("tools/call", params, req_id) + intent_obj = Intent.from_dict(proposal["intent"]) + if "error" in upstream: + self.ledger.append( + "execute", + intent=intent_obj, + outcome="error", + detail=upstream.get("error"), + ) + return upstream + self.ledger.append("execute", intent=intent_obj, outcome="ok") + 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 100644 index 0000000..12c0a9e --- /dev/null +++ b/actgate/core/mcp_rpc.py @@ -0,0 +1,57 @@ +"""Minimal MCP/JSON-RPC stdio framing (Content-Length).""" + +from __future__ import annotations + +import json +import sys +from typing import Any, BinaryIO, TextIO + + +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 = f"Content-Length: {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 + + +def write_text_line(stream: TextIO, text: str) -> None: + stream.write(text) + if not text.endswith("\n"): + stream.write("\n") + stream.flush() 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" From e6cd858774ab70fb0051088f7e73f891740548c8 Mon Sep 17 00:00:00 2001 From: kartsan03 Date: Sun, 6 Sep 2026 01:34:15 +0200 Subject: [PATCH 2/5] test: cover MCP pending approve deny and single execute Fake upstream echo tool; assert unapproved blocks and ledger verify stays green. --- tests/fake_mcp_upstream.py | 106 +++++++++++++++++++++++++++++++ tests/test_mcp_proxy.py | 124 +++++++++++++++++++++++++++++++++++++ 2 files changed, 230 insertions(+) create mode 100644 tests/fake_mcp_upstream.py create mode 100644 tests/test_mcp_proxy.py diff --git a/tests/fake_mcp_upstream.py b/tests/fake_mcp_upstream.py new file mode 100644 index 0000000..987cac5 --- /dev/null +++ b/tests/fake_mcp_upstream.py @@ -0,0 +1,106 @@ +"""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")) + + +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": + 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..2632c78 --- /dev/null +++ b/tests/test_mcp_proxy.py @@ -0,0 +1,124 @@ +"""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) -> 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: + 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 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) From aa0781c6c38585856be6b2500295fb4808a027dd Mon Sep 17 00:00:00 2001 From: kartsan03 Date: Sun, 6 Sep 2026 01:34:15 +0200 Subject: [PATCH 3/5] docs: document MCP proxy flow and bump changelog to 0.2.0 Remove not-an-MCP-proxy-yet wording; describe approve then execute. --- CHANGELOG.md | 10 ++++++++++ README.md | 55 ++++++++++++++++++++++++++-------------------------- 2 files changed, 37 insertions(+), 28 deletions(-) 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..7e0b5db 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,28 @@ 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. +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 +72,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 From 39622d741f3ef2d4e074f21e87ea951fe406d6fd Mon Sep 17 00:00:00 2001 From: kartsan03 Date: Sun, 6 Sep 2026 01:39:35 +0200 Subject: [PATCH 4/5] fix: drop unused mcp helpers and record execute before upstream Remove dead write_text_line/json imports. Append execute before tools/call to upstream so a crash after success cannot double-call. Narrow cmd_mcp excepts. --- actgate/cli.py | 3 ++- actgate/core/mcp_proxy.py | 14 +++----------- actgate/core/mcp_rpc.py | 12 ++---------- 3 files changed, 7 insertions(+), 22 deletions(-) mode change 100644 => 100755 actgate/core/mcp_rpc.py diff --git a/actgate/cli.py b/actgate/cli.py index 5216438..f4e0bc3 100644 --- a/actgate/cli.py +++ b/actgate/cli.py @@ -13,6 +13,7 @@ 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: @@ -215,7 +216,7 @@ def cmd_mcp(args: argparse.Namespace) -> int: path = getattr(args, "ledger", None) try: return run_proxy(root=root, upstream_cmd=list(args.upstream), ledger_path=path) - except Exception as exc: # noqa: BLE001 — surface proxy failures as setup errors + except (RpcError, OSError, LedgerError) as exc: _eprint(str(exc)) return 2 diff --git a/actgate/core/mcp_proxy.py b/actgate/core/mcp_proxy.py index fa715e4..d389e5c 100644 --- a/actgate/core/mcp_proxy.py +++ b/actgate/core/mcp_proxy.py @@ -2,7 +2,6 @@ from __future__ import annotations -import json import subprocess import sys from pathlib import Path @@ -156,17 +155,10 @@ def _handle_tools_call(self, req_id: Any, params: dict[str, Any]) -> dict[str, A }, } - upstream = self._upstream_request("tools/call", params, req_id) + # Record execute before upstream so a crash after success cannot double-call. intent_obj = Intent.from_dict(proposal["intent"]) - if "error" in upstream: - self.ledger.append( - "execute", - intent=intent_obj, - outcome="error", - detail=upstream.get("error"), - ) - return upstream - self.ledger.append("execute", intent=intent_obj, outcome="ok") + 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: diff --git a/actgate/core/mcp_rpc.py b/actgate/core/mcp_rpc.py old mode 100644 new mode 100755 index 12c0a9e..4ca579a --- a/actgate/core/mcp_rpc.py +++ b/actgate/core/mcp_rpc.py @@ -3,8 +3,7 @@ from __future__ import annotations import json -import sys -from typing import Any, BinaryIO, TextIO +from typing import Any, BinaryIO class RpcError(RuntimeError): @@ -13,7 +12,7 @@ class RpcError(RuntimeError): def write_message(stream: BinaryIO, message: dict[str, Any]) -> None: body = json.dumps(message, ensure_ascii=False, separators=(",", ":")).encode("utf-8") - header = f"Content-Length: {len(body)}\r\n\r\n".encode("ascii") + header = ("Content-Length: %d" % len(body) + "\r\n\r\n").encode("ascii") stream.write(header) stream.write(body) stream.flush() @@ -48,10 +47,3 @@ def read_message(stream: BinaryIO) -> dict[str, Any] | None: return json.loads(body.decode("utf-8")) except json.JSONDecodeError as exc: raise RpcError(f"invalid JSON body: {exc}") from exc - - -def write_text_line(stream: TextIO, text: str) -> None: - stream.write(text) - if not text.endswith("\n"): - stream.write("\n") - stream.flush() From 8c5c8dbb7fa9f40106c7d9255098244948590331 Mon Sep 17 00:00:00 2001 From: kartsan03 Date: Sun, 6 Sep 2026 01:39:35 +0200 Subject: [PATCH 5/5] test: assert approval is tool+args scoped; count upstream calls Different args/tools get new pending not the hi-approved execute. Doc that only tools/call is gated. --- README.md | 3 ++- tests/fake_mcp_upstream.py | 7 ++++++ tests/test_mcp_proxy.py | 46 +++++++++++++++++++++++++++++++++++++- 3 files changed, 54 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 7e0b5db..cb14375 100644 --- a/README.md +++ b/README.md @@ -40,7 +40,8 @@ actgate mcp --upstream python -m some_mcp_server Flow: -1. Client `tools/list` is forwarded to upstream. +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 `). diff --git a/tests/fake_mcp_upstream.py b/tests/fake_mcp_upstream.py index 987cac5..8dc6975 100644 --- a/tests/fake_mcp_upstream.py +++ b/tests/fake_mcp_upstream.py @@ -30,6 +30,9 @@ def read_message() -> dict[str, Any] | None: return json.loads(body.decode("utf-8")) +CALLS = 0 +CALLS_PATH = __import__("os").environ.get("ACTGATE_FAKE_CALLS") + TOOLS = [ { "name": "echo", @@ -68,6 +71,10 @@ def main() -> int: 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": diff --git a/tests/test_mcp_proxy.py b/tests/test_mcp_proxy.py index 2632c78..1815168 100644 --- a/tests/test_mcp_proxy.py +++ b/tests/test_mcp_proxy.py @@ -37,7 +37,9 @@ def root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: return tmp_path -def test_tools_call_pending_then_approve_executes_once(root: Path) -> None: +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], @@ -84,6 +86,7 @@ def test_tools_call_pending_then_approve_executes_once(root: Path) -> None: 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() @@ -122,3 +125,44 @@ def test_unapproved_never_hits_upstream(root: Path) -> None: 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)