diff --git a/test/test_github_dispatch_mcp.py b/test/test_github_dispatch_mcp.py new file mode 100644 index 0000000..056d9b8 --- /dev/null +++ b/test/test_github_dispatch_mcp.py @@ -0,0 +1,166 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import json +import os +import subprocess +import unittest +from pathlib import Path + + +REPO_ROOT = Path(__file__).resolve().parents[1] +SCRIPT = REPO_ROOT / "tools" / "run_github_dispatch_mcp.sh" + + +def write_message(proc: subprocess.Popen[str], payload: dict) -> None: + body = json.dumps(payload) + proc.stdin.write(f"Content-Length: {len(body.encode('utf-8'))}\r\n\r\n{body}") + proc.stdin.flush() + + +def read_message(proc: subprocess.Popen[str]) -> dict: + headers = {} + while True: + line = proc.stdout.readline() + if not line: + raise AssertionError("unexpected EOF while reading MCP headers") + if line in ("\r\n", "\n"): + break + key, _, value = line.partition(":") + headers[key.strip().lower()] = value.strip() + + length = int(headers["content-length"]) + body = proc.stdout.read(length) + return json.loads(body) + + +class GitHubDispatchMcpTests(unittest.TestCase): + def test_server_supports_initialize_and_tools_list(self): + env = os.environ.copy() + env.pop("KILL_LIFE_GITHUB_TOKEN", None) + env.pop("GITHUB_TOKEN", None) + proc = subprocess.Popen( + ["bash", str(SCRIPT)], + cwd=REPO_ROOT, + env=env, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + + try: + write_message( + proc, + {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}, + ) + initialize = read_message(proc) + self.assertEqual(initialize["result"]["serverInfo"]["name"], "github-dispatch") + self.assertEqual(initialize["result"]["protocolVersion"], "2025-03-26") + + write_message( + proc, + {"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}, + ) + tools_list = read_message(proc) + tool_names = {tool["name"] for tool in tools_list["result"]["tools"]} + self.assertEqual( + tool_names, + {"list_allowlisted_workflows", "dispatch_workflow", "get_dispatch_status"}, + ) + finally: + proc.terminate() + proc.wait(timeout=5) + if proc.stdin: + proc.stdin.close() + if proc.stdout: + proc.stdout.close() + if proc.stderr: + proc.stderr.close() + + def test_list_allowlisted_workflows_succeeds_without_token(self): + env = os.environ.copy() + env.pop("KILL_LIFE_GITHUB_TOKEN", None) + env.pop("GITHUB_TOKEN", None) + proc = subprocess.Popen( + ["bash", str(SCRIPT)], + cwd=REPO_ROOT, + env=env, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + + try: + write_message( + proc, + { + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": {"name": "list_allowlisted_workflows", "arguments": {}}, + }, + ) + response = read_message(proc) + self.assertFalse(response["result"]["isError"]) + self.assertGreater( + len(response["result"]["structuredContent"]["workflows"]), + 0, + ) + finally: + proc.terminate() + proc.wait(timeout=5) + if proc.stdin: + proc.stdin.close() + if proc.stdout: + proc.stdout.close() + if proc.stderr: + proc.stderr.close() + + def test_dispatch_workflow_returns_structured_missing_secret_error(self): + env = os.environ.copy() + env.pop("KILL_LIFE_GITHUB_TOKEN", None) + env.pop("GITHUB_TOKEN", None) + proc = subprocess.Popen( + ["bash", str(SCRIPT)], + cwd=REPO_ROOT, + env=env, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + + try: + write_message( + proc, + { + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": { + "name": "dispatch_workflow", + "arguments": {"workflow_file": "badges.yml"}, + }, + }, + ) + response = read_message(proc) + self.assertTrue(response["result"]["isError"]) + self.assertEqual( + response["result"]["structuredContent"]["error"]["code"], + "missing_secret", + ) + finally: + proc.terminate() + proc.wait(timeout=5) + if proc.stdin: + proc.stdin.close() + if proc.stdout: + proc.stdout.close() + if proc.stderr: + proc.stderr.close() + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/github_dispatch_mcp.py b/tools/github_dispatch_mcp.py new file mode 100755 index 0000000..6ed2821 --- /dev/null +++ b/tools/github_dispatch_mcp.py @@ -0,0 +1,219 @@ +#!/usr/bin/env python3 +"""Local MCP server for GitHub workflow_dispatch operations.""" + +from __future__ import annotations + +import asyncio +import os +import sys +from pathlib import Path +from typing import Any + +from mcp_stdio import ( # type: ignore + PROTOCOL_VERSION, + error_tool_result, + make_error, + make_response, + ok_tool_result, + read_message, + write_message, +) + +ROOT = Path(__file__).resolve().parents[1] +MASCARADE_DIR = Path(os.environ.get("MASCARADE_DIR", ROOT.parent / "mascarade")).resolve() +MASCARADE_CORE_DIR = MASCARADE_DIR / "core" + +if str(MASCARADE_CORE_DIR) not in sys.path: + sys.path.insert(0, str(MASCARADE_CORE_DIR)) + +from mascarade.integrations.github_dispatch import ( # noqa: E402 + DEFAULT_GITHUB_REPO, + GitHubDispatchAuthError, + GitHubDispatchClient, + GitHubDispatchError, + list_allowlisted_workflows, +) + + +TOOLS = [ + { + "name": "list_allowlisted_workflows", + "description": "List the GitHub Actions workflows allowed for dispatch in this workspace.", + "inputSchema": {"type": "object", "properties": {}}, + }, + { + "name": "dispatch_workflow", + "description": "Dispatch an allowlisted GitHub Actions workflow.", + "inputSchema": { + "type": "object", + "properties": { + "workflow_file": { + "type": "string", + "description": "Allowlisted workflow file to dispatch", + }, + "ref": {"type": "string", "description": "Git ref to dispatch against"}, + "inputs": { + "type": "object", + "description": "Workflow dispatch inputs", + }, + }, + "required": ["workflow_file"], + }, + }, + { + "name": "get_dispatch_status", + "description": "Resolve the structured status of a previously dispatched workflow.", + "inputSchema": { + "type": "object", + "properties": { + "dispatch_id": { + "type": "string", + "description": "Dispatch identifier returned by dispatch_workflow", + } + }, + "required": ["dispatch_id"], + }, + }, +] + + +async def tool_list_allowlisted_workflows(_: dict[str, Any]) -> dict[str, Any]: + workflows = list_allowlisted_workflows() + payload = {"ok": True, "repo": DEFAULT_GITHUB_REPO, "workflows": workflows} + return ok_tool_result( + f"{len(workflows)} allowlisted GitHub workflow(s)", + payload, + ) + + +async def tool_dispatch_workflow(arguments: dict[str, Any]) -> dict[str, Any]: + workflow_file = str(arguments.get("workflow_file", "")).strip() + ref = str(arguments.get("ref", "")).strip() or None + raw_inputs = arguments.get("inputs") + inputs = raw_inputs if isinstance(raw_inputs, dict) else None + if not workflow_file: + return error_tool_result( + "Missing required argument 'workflow_file'", + { + "ok": False, + "error": { + "code": "invalid_arguments", + "message": "workflow_file is required", + }, + }, + ) + + client = GitHubDispatchClient() + try: + payload = await client.dispatch_workflow(workflow_file, ref=ref, inputs=inputs) + return ok_tool_result( + f"Accepted GitHub dispatch for {payload['workflow_file']}", + {"ok": True, **payload}, + ) + except GitHubDispatchAuthError as exc: + return error_tool_result( + str(exc), + {"ok": False, "error": {"code": "missing_secret", "message": str(exc)}}, + ) + except GitHubDispatchError as exc: + return error_tool_result( + str(exc), + {"ok": False, "error": {"code": "dispatch_failed", "message": str(exc)}}, + ) + finally: + await client.close() + + +async def tool_get_dispatch_status(arguments: dict[str, Any]) -> dict[str, Any]: + dispatch_id = str(arguments.get("dispatch_id", "")).strip() + if not dispatch_id: + return error_tool_result( + "Missing required argument 'dispatch_id'", + { + "ok": False, + "error": { + "code": "invalid_arguments", + "message": "dispatch_id is required", + }, + }, + ) + + client = GitHubDispatchClient() + try: + payload = await client.get_dispatch_status(dispatch_id) + return ok_tool_result( + f"Dispatch {dispatch_id} status is {payload['status']}", + {"ok": True, **payload}, + ) + except GitHubDispatchError as exc: + return error_tool_result( + str(exc), + {"ok": False, "error": {"code": "status_failed", "message": str(exc)}}, + ) + finally: + await client.close() + + +def serve_mcp() -> int: + while True: + request = read_message() + if request is None: + return 0 + + method = request.get("method") + request_id = request.get("id") + params = request.get("params") or {} + + if method == "initialize": + write_message( + make_response( + request_id, + { + "protocolVersion": PROTOCOL_VERSION, + "capabilities": {"tools": {"listChanged": False}}, + "serverInfo": {"name": "github-dispatch", "version": "1.0.0"}, + }, + ) + ) + continue + + if method == "notifications/initialized": + continue + + if method == "ping": + write_message(make_response(request_id, {})) + continue + + if method == "tools/list": + write_message(make_response(request_id, {"tools": TOOLS})) + continue + + if method == "tools/call": + tool_name = params.get("name") + arguments = params.get("arguments") or {} + + if tool_name == "list_allowlisted_workflows": + write_message( + make_response(request_id, asyncio.run(tool_list_allowlisted_workflows(arguments))) + ) + continue + if tool_name == "dispatch_workflow": + write_message( + make_response(request_id, asyncio.run(tool_dispatch_workflow(arguments))) + ) + continue + if tool_name == "get_dispatch_status": + write_message( + make_response(request_id, asyncio.run(tool_get_dispatch_status(arguments))) + ) + continue + + write_message(make_error(request_id, -32602, f"Unknown tool: {tool_name}")) + continue + + if request_id is not None: + write_message(make_error(request_id, -32601, f"Method not found: {method}")) + + +if __name__ == "__main__": + raise SystemExit(serve_mcp()) diff --git a/tools/github_dispatch_mcp_smoke.py b/tools/github_dispatch_mcp_smoke.py new file mode 100755 index 0000000..4a5cb1a --- /dev/null +++ b/tools/github_dispatch_mcp_smoke.py @@ -0,0 +1,168 @@ +#!/usr/bin/env python3 +"""Smoke checks for the local GitHub dispatch MCP server.""" + +from __future__ import annotations + +import argparse +import os +from pathlib import Path + +from mcp_smoke_common import ( + PROTOCOL_VERSION, + SmokeError, + call_tool, + emit_payload, + initialize, + list_tools, + spawn_server, + terminate_server, +) + +ROOT = Path(__file__).resolve().parents[1] +SERVER = ROOT / "tools" / "run_github_dispatch_mcp.sh" + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--timeout", type=float, default=15.0) + parser.add_argument("--json", action="store_true") + parser.add_argument("--quick", action="store_true") + parser.add_argument( + "--live", + action="store_true", + help="Dispatch an allowlisted workflow when a GitHub token is configured.", + ) + parser.add_argument( + "--workflow-file", + default="repo_state.yml", + help="Allowlisted workflow file to use when --live is enabled.", + ) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + token_configured = bool( + os.getenv("KILL_LIFE_GITHUB_TOKEN") or os.getenv("GITHUB_TOKEN") + ) + proc = spawn_server(["bash", str(SERVER)], ROOT) + payload = { + "status": "failed", + "protocol_version": None, + "server_name": "github-dispatch", + "tool_count": 0, + "checks": [], + "token_configured": token_configured, + "live_requested": args.live, + "live_validation": "skipped" if not args.live else "pending", + "error": None, + } + + try: + init = initialize(proc, args.timeout, "kill-life-github-dispatch-mcp-smoke") + tools = list_tools(proc, args.timeout) + tool_names = {tool.get("name") for tool in tools} + expected = {"list_allowlisted_workflows", "dispatch_workflow", "get_dispatch_status"} + if expected - tool_names: + raise SmokeError(f"github-dispatch tools missing: {sorted(expected - tool_names)}") + + payload["protocol_version"] = init.get("protocolVersion", PROTOCOL_VERSION) + payload["server_name"] = (init.get("serverInfo") or {}).get("name", "github-dispatch") + payload["tool_count"] = len(tools) + payload["checks"] = ["initialize", "tools/list"] + + if args.quick: + if token_configured: + payload["status"] = "ready" + payload["live_validation"] = "skipped" + else: + payload["status"] = "degraded" + payload["live_validation"] = "missing_secret" + payload["error"] = "KILL_LIFE_GITHUB_TOKEN or GITHUB_TOKEN missing" + return emit_payload(payload, json_output=args.json) + + list_result = call_tool( + proc, + args.timeout, + 3, + "list_allowlisted_workflows", + {}, + ) + if list_result.get("isError"): + raise SmokeError("list_allowlisted_workflows returned isError=true") + structured = list_result.get("structuredContent") or {} + workflows = (structured.get("workflows") or []) if isinstance(structured, dict) else [] + if not workflows: + raise SmokeError("list_allowlisted_workflows returned no workflows") + payload["checks"].append("list_allowlisted_workflows") + + if not token_configured: + dispatch_result = call_tool( + proc, + args.timeout, + 4, + "dispatch_workflow", + {"workflow_file": args.workflow_file}, + ) + if not dispatch_result.get("isError"): + raise SmokeError("dispatch_workflow unexpectedly succeeded without token") + payload["checks"].append("dispatch_workflow_missing_secret") + payload["status"] = "degraded" + payload["live_validation"] = "missing_secret" + payload["error"] = "KILL_LIFE_GITHUB_TOKEN or GITHUB_TOKEN missing" + return emit_payload(payload, json_output=args.json) + + if not args.live: + payload["status"] = "ready" + payload["live_validation"] = "skipped" + return emit_payload(payload, json_output=args.json) + + dispatch_result = call_tool( + proc, + args.timeout, + 5, + "dispatch_workflow", + {"workflow_file": args.workflow_file}, + ) + if dispatch_result.get("isError"): + structured = dispatch_result.get("structuredContent") or {} + raise SmokeError( + ((structured.get("error") or {}).get("message")) + if isinstance(structured, dict) + else "dispatch_workflow failed" + ) + payload["checks"].append("dispatch_workflow") + + dispatch_structured = dispatch_result.get("structuredContent") or {} + dispatch_id = dispatch_structured.get("dispatch_id") if isinstance(dispatch_structured, dict) else None + if not isinstance(dispatch_id, str) or not dispatch_id: + raise SmokeError("dispatch_workflow returned no dispatch_id") + + status_result = call_tool( + proc, + args.timeout, + 6, + "get_dispatch_status", + {"dispatch_id": dispatch_id}, + ) + if status_result.get("isError"): + structured = status_result.get("structuredContent") or {} + raise SmokeError( + ((structured.get("error") or {}).get("message")) + if isinstance(structured, dict) + else "get_dispatch_status failed" + ) + payload["checks"].append("get_dispatch_status") + payload["status"] = "ready" + payload["live_validation"] = "passed" + return emit_payload(payload, json_output=args.json) + except Exception as exc: + payload["status"] = "failed" + payload["error"] = str(exc) + return emit_payload(payload, json_output=args.json) + finally: + terminate_server(proc) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/run_github_dispatch_mcp.sh b/tools/run_github_dispatch_mcp.sh new file mode 100755 index 0000000..52cc002 --- /dev/null +++ b/tools/run_github_dispatch_mcp.sh @@ -0,0 +1,24 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +MASCARADE_DIR="${MASCARADE_DIR:-$ROOT_DIR/../mascarade}" +CORE_PYTHON="${MASCARADE_CORE_PYTHON:-$MASCARADE_DIR/core/.venv/bin/python}" +SERVER_SCRIPT="$ROOT_DIR/tools/github_dispatch_mcp.py" + +if [[ "${1:-}" == "--doctor" ]]; then + cat <&2 + exit 1 +} + +exec "$CORE_PYTHON" "$SERVER_SCRIPT" "$@"