From a6a1aa21dfd2d8d5b4c8120594e5e75c195e0bdd Mon Sep 17 00:00:00 2001 From: ProfRandom92 Date: Fri, 22 May 2026 10:49:29 +0200 Subject: [PATCH] feat: add MCP context local tool adapter --- docs/mcp_context_layer.md | 11 +++ scripts/mcp_context_tool.py | 141 ++++++++++++++++++++++++++++++++ tests/test_mcp_context_layer.py | 100 ++++++++++++++++++++++ 3 files changed, 252 insertions(+) create mode 100644 scripts/mcp_context_tool.py diff --git a/docs/mcp_context_layer.md b/docs/mcp_context_layer.md index 8a05a15..440a01d 100644 --- a/docs/mcp_context_layer.md +++ b/docs/mcp_context_layer.md @@ -100,6 +100,17 @@ python scripts/mcp_context_cli.py \ Add `--json` to emit deterministic JSON containing the compact replay payload, optional prompt context, and optional validation result. +## Local tool adapter + +`scripts/mcp_context_tool.py` accepts one JSON request from stdin or +`--request-file` and emits one deterministic JSON response. It is a local +adapter for later MCP wrapping, not a server or runtime tool executor. + +```bash +printf '%s\n' '{"tool":"build_replay_payload","params":{"fixture":"fixtures/mcp_trace_replay_v1/original","render_prompt":true,"validate":true}}' \ + | python scripts/mcp_context_tool.py +``` + ## Relationship to MCP This layer augments context integrity for MCP-compatible systems. It is not an diff --git a/scripts/mcp_context_tool.py b/scripts/mcp_context_tool.py new file mode 100644 index 0000000..5dd29a0 --- /dev/null +++ b/scripts/mcp_context_tool.py @@ -0,0 +1,141 @@ +"""One-shot JSON tool adapter for the deterministic MCP context layer.""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import Any + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from scripts.mcp_context_cli import load_fixture_context +from src.comptext_v7.mcp import build_replay_payload, render_prompt_context, validate_replay_payload + + +def _json_response(payload: dict[str, Any]) -> str: + return json.dumps(payload, indent=2, sort_keys=True) + "\n" + + +def _load_request(path: Path | None) -> dict[str, Any]: + try: + raw = path.read_text(encoding="utf-8") if path is not None else sys.stdin.read() + except FileNotFoundError as exc: + raise RuntimeError(f"missing request file: {path.as_posix() if path is not None else ''}") from exc + try: + request = json.loads(raw) + except json.JSONDecodeError as exc: + raise RuntimeError(f"invalid JSON request: {exc}") from exc + if not isinstance(request, dict): + raise RuntimeError("request must be a JSON object") + return request + + +def _params(request: dict[str, Any]) -> dict[str, Any]: + params = request.get("params", {}) + if not isinstance(params, dict): + raise RuntimeError("request.params must be a JSON object") + return params + + +def _tool_name(request: dict[str, Any]) -> str: + tool = request.get("tool") + if not isinstance(tool, str) or not tool: + raise RuntimeError("request.tool must be a non-empty string") + return tool + + +def _payload_from_params(params: dict[str, Any]) -> dict[str, Any]: + payload = params.get("payload") + if isinstance(payload, dict): + return payload + + fixture = params.get("fixture") + if not isinstance(fixture, str) or not fixture: + raise RuntimeError("request.params requires payload object or fixture path") + return build_replay_payload(load_fixture_context(Path(fixture))) + + +def _handle_build_replay_payload(params: dict[str, Any]) -> dict[str, Any]: + fixture = params.get("fixture") + if not isinstance(fixture, str) or not fixture: + raise RuntimeError("build_replay_payload requires params.fixture") + + payload = build_replay_payload(load_fixture_context(Path(fixture))) + result: dict[str, Any] = {"payload": payload} + + validation = validate_replay_payload(payload) if params.get("validate") is True else None + if validation is not None: + result["validation"] = validation + if params.get("render_prompt") is True: + prompt_payload = {**payload} + if validation is not None: + prompt_payload["validation"] = validation + result["prompt_context"] = render_prompt_context(prompt_payload) + return result + + +def _handle_render_prompt_context(params: dict[str, Any]) -> dict[str, Any]: + payload = _payload_from_params(params) + if params.get("validate") is True: + payload = {**payload, "validation": validate_replay_payload(payload)} + return {"prompt_context": render_prompt_context(payload)} + + +def _handle_validate_replay_payload(params: dict[str, Any]) -> dict[str, Any]: + return {"validation": validate_replay_payload(_payload_from_params(params))} + + +def handle_request(request: dict[str, Any]) -> dict[str, Any]: + tool = _tool_name(request) + params = _params(request) + + if tool == "build_replay_payload": + result = _handle_build_replay_payload(params) + elif tool == "render_prompt_context": + result = _handle_render_prompt_context(params) + elif tool == "validate_replay_payload": + result = _handle_validate_replay_payload(params) + else: + raise RuntimeError(f"unsupported tool: {tool}") + + return {"ok": True, "result": result, "tool": tool} + + +def _parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Run one deterministic MCP context-layer tool request.") + parser.add_argument("--request-file", type=Path, help="JSON request file. Reads stdin when omitted.") + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = _parse_args(sys.argv[1:] if argv is None else argv) + tool: str | None = None + try: + request = _load_request(args.request_file) + request_tool = request.get("tool") + tool = request_tool if isinstance(request_tool, str) else None + response = handle_request(request) + sys.stdout.write(_json_response(response)) + return 0 + except Exception as exc: + sys.stdout.write( + _json_response( + { + "error": { + "message": str(exc), + "type": exc.__class__.__name__, + }, + "ok": False, + "tool": tool, + } + ) + ) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_mcp_context_layer.py b/tests/test_mcp_context_layer.py index c400948..e44ce08 100644 --- a/tests/test_mcp_context_layer.py +++ b/tests/test_mcp_context_layer.py @@ -26,6 +26,7 @@ FIXTURE_ROOT = Path("fixtures/mcp_trace_replay_v1/original") ARTIFACT_PATH = Path("artifacts/mcp_context_layer_example.json") CLI_PATH = Path("scripts/mcp_context_cli.py") +TOOL_PATH = Path("scripts/mcp_context_tool.py") def _load_fixture_context() -> dict[str, object]: @@ -300,6 +301,16 @@ def _run_cli(*args: str) -> str: return completed.stdout +def _run_tool(request: dict[str, object], *args: str, check: bool = True) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [sys.executable, str(TOOL_PATH), *args], + input=json.dumps(request, sort_keys=True), + check=check, + capture_output=True, + text=True, + ) + + def test_mcp_context_cli_json_output_is_deterministic() -> None: args = ( "--fixture", @@ -394,3 +405,92 @@ def test_mcp_context_cli_load_json_requires_json_object(tmp_path: Path) -> None: _load_json(list_path) assert str(excinfo.value) == f"fixture file must contain a JSON object: {list_path.as_posix()}" + + +def test_mcp_context_tool_build_request_is_deterministic() -> None: + request = { + "tool": "build_replay_payload", + "params": { + "fixture": str(FIXTURE_ROOT), + "render_prompt": True, + "validate": True, + }, + } + + first = _run_tool(request).stdout + second = _run_tool(request).stdout + + assert first == second + assert first.endswith("\n") + response = json.loads(first) + assert set(response.keys()) == {"ok", "result", "tool"} + assert response["ok"] is True + assert response["tool"] == "build_replay_payload" + + result = response["result"] + assert result["payload"] == build_replay_payload(_load_fixture_context()) + assert result["validation"] == validate_replay_payload(result["payload"]) + assert result["prompt_context"] == render_prompt_context( + { + **result["payload"], + "validation": result["validation"], + } + ) + assert '"events"' not in first + assert '"dependency_graph"' not in first + assert '"permission_scopes"' not in first + + +def test_mcp_context_tool_request_file_validate_payload(tmp_path: Path) -> None: + request_path = tmp_path / "request.json" + payload = build_replay_payload(_load_fixture_context()) + request_path.write_text( + json.dumps( + { + "tool": "validate_replay_payload", + "params": {"payload": payload}, + }, + sort_keys=True, + ) + + "\n", + encoding="utf-8", + ) + + completed = subprocess.run( + [sys.executable, str(TOOL_PATH), "--request-file", str(request_path)], + check=True, + capture_output=True, + text=True, + ) + + assert json.loads(completed.stdout) == { + "ok": True, + "result": { + "validation": { + "admissible": True, + "failure_labels": [], + "issues": [], + } + }, + "tool": "validate_replay_payload", + } + + +def test_mcp_context_tool_invalid_request_returns_deterministic_error() -> None: + request = { + "tool": "build_replay_payload", + "params": {"fixture": "fixtures/mcp_trace_replay_v1/original/missing"}, + } + + completed = _run_tool(request, check=False) + + assert completed.returncode == 1 + assert completed.stderr == "" + assert json.loads(completed.stdout) == { + "error": { + "message": "missing required fixture file: fixtures/mcp_trace_replay_v1/original/missing/trace.json", + "type": "RuntimeError", + }, + "ok": False, + "tool": "build_replay_payload", + }