from __future__ import annotations import argparse import json import os import sqlite3 import sys import urllib.error import urllib.request from pathlib import Path from typing import Any, Protocol from .models import Role class OperationsError(RuntimeError): pass class ToolCaller(Protocol): def call(self, name: str, arguments: dict[str, Any]) -> dict[str, Any]: ... class HTTPMCPClient: def __init__(self, endpoint: str, token: str, timeout: float = 10.0): self.endpoint = endpoint self.token = token self.timeout = timeout self._request_id = 0 def call(self, name: str, arguments: dict[str, Any]) -> dict[str, Any]: self._request_id += 1 payload = { "jsonrpc": "2.0", "id": self._request_id, "method": "tools/call", "params": {"name": name, "arguments": arguments}, } request = urllib.request.Request( self.endpoint, data=json.dumps(payload, separators=(",", ":")).encode(), headers={ "Authorization": f"Bearer {self.token}", "Content-Type": "application/json", }, method="POST", ) try: with urllib.request.urlopen(request, timeout=self.timeout) as response: result = json.load(response) except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as error: raise OperationsError(f"MCP request failed: {error}") from error if "error" in result: raise OperationsError(f"JSON-RPC error: {result['error']}") tool_result = result.get("result", {}) if tool_result.get("isError"): content = tool_result.get("content", []) message = content[0].get("text", "unknown tool error") if content else "unknown tool error" raise OperationsError(message) content = tool_result.get("content", []) if not content or content[0].get("type") != "text": raise OperationsError("MCP tool returned no JSON text content") try: return json.loads(content[0]["text"]) except json.JSONDecodeError as error: raise OperationsError("MCP tool returned invalid JSON text") from error def load_manifest(path: Path) -> dict[str, Any]: try: manifest = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: raise OperationsError(f"Cannot read manifest: {error}") from error project = manifest.get("project") actors = manifest.get("actors") if not isinstance(project, dict) or not isinstance(actors, list) or not actors: raise OperationsError("Manifest requires a project object and non-empty actors array") for field in ("project_id", "name"): if not isinstance(project.get(field), str) or not project[field]: raise OperationsError(f"Manifest project.{field} must be a non-empty string") seen_ids: set[str] = set() seen_subjects: set[str] = set() for index, actor in enumerate(actors): if not isinstance(actor, dict): raise OperationsError(f"Manifest actor {index} must be an object") for field in ("actor_id", "auth_subject", "display_name", "role"): if not isinstance(actor.get(field), str) or not actor[field]: raise OperationsError(f"Manifest actor {index}.{field} must be a non-empty string") try: role = Role(actor["role"]) except ValueError as error: raise OperationsError(f"Manifest actor {index} has invalid role") from error if role is Role.SYSTEM: raise OperationsError("SYSTEM is a transport bootstrap identity, not a project actor") if actor["actor_id"] in seen_ids or actor["auth_subject"] in seen_subjects: raise OperationsError("Actor IDs and auth subjects must be unique") seen_ids.add(actor["actor_id"]) seen_subjects.add(actor["auth_subject"]) return manifest def bootstrap_project(client: ToolCaller, manifest: dict[str, Any]) -> dict[str, Any]: project = manifest["project"] project_id = project["project_id"] created = client.call( "create_project", { "project_id": project_id, "name": project["name"], "description": project.get("description", ""), "idempotency_key": f"bootstrap:{project_id}:project:v1", }, ) registered = [] for actor in manifest["actors"]: registered.append( client.call( "register_actor", { "project_id": project_id, **actor, "idempotency_key": f"bootstrap:{project_id}:actor:{actor['actor_id']}:v1", }, ) ) return {"project": created, "actors": registered} def backup_database(source: Path, destination: Path) -> dict[str, Any]: if source.resolve() == destination.resolve(): raise OperationsError("Backup destination must differ from source") destination.parent.mkdir(parents=True, exist_ok=True) try: with sqlite3.connect(source) as source_db, sqlite3.connect(destination) as backup_db: source_db.backup(backup_db) integrity = backup_db.execute("PRAGMA integrity_check").fetchone()[0] cursor_row = backup_db.execute( "SELECT COALESCE(MAX(cursor), 0) FROM event_log" ).fetchone() except sqlite3.Error as error: raise OperationsError(f"Database backup failed: {error}") from error if integrity != "ok": raise OperationsError(f"Backup integrity check failed: {integrity}") return {"destination": str(destination), "integrity": integrity, "event_cursor": cursor_row[0]} def _bootstrap_command(arguments: argparse.Namespace) -> int: token = os.environ.get(arguments.token_env) if not token: raise OperationsError(f"Required secret environment variable is unset: {arguments.token_env}") manifest = load_manifest(arguments.manifest) result = bootstrap_project(HTTPMCPClient(arguments.endpoint, token), manifest) print(json.dumps(result, indent=2)) return 0 def _backup_command(arguments: argparse.Namespace) -> int: print(json.dumps(backup_database(arguments.source, arguments.destination), indent=2)) return 0 def parser() -> argparse.ArgumentParser: result = argparse.ArgumentParser(description="Project Bus operational tooling") subparsers = result.add_subparsers(dest="command", required=True) bootstrap = subparsers.add_parser("bootstrap", help="Provision a project and actor roster") bootstrap.add_argument("--endpoint", required=True) bootstrap.add_argument("--manifest", type=Path, required=True) bootstrap.add_argument("--token-env", default="PROJECT_BUS_BOOTSTRAP_TOKEN") bootstrap.set_defaults(handler=_bootstrap_command) backup = subparsers.add_parser("backup", help="Create and integrity-check a live SQLite backup") backup.add_argument("--source", type=Path, required=True) backup.add_argument("--destination", type=Path, required=True) backup.set_defaults(handler=_backup_command) return result def main() -> None: arguments = parser().parse_args() try: raise SystemExit(arguments.handler(arguments)) except OperationsError as error: print(f"error: {error}", file=sys.stderr) raise SystemExit(2) from error if __name__ == "__main__": main()