From 89afc9d7c4e48a9751bcf26fd4d04db2b23bba35 Mon Sep 17 00:00:00 2001 From: Alexander Whitestone Date: Wed, 15 Apr 2026 15:32:52 +0000 Subject: [PATCH] feat: crisis bridge HTTP API for web-agent integration (#99) --- crisis/bridge.py | 168 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 168 insertions(+) create mode 100644 crisis/bridge.py diff --git a/crisis/bridge.py b/crisis/bridge.py new file mode 100644 index 0000000..152d2a2 --- /dev/null +++ b/crisis/bridge.py @@ -0,0 +1,168 @@ +#!/usr/bin/env python3 +""" +Crisis Bridge — HTTP integration between the-door web and hermes-agent. + +Provides: +- GET /api/crisis/escalations — list recent escalation events +- GET /api/crisis/stats — aggregate statistics +- POST /api/crisis/log — log a new escalation (from hermes-agent) +- POST /api/crisis/resolve/:id — mark escalation as resolved + +Can be mounted as an ASGI/FastAPI sub-app or used standalone. +Falls back to a simple HTTP server if no framework is available. +""" + +import json +import os +from typing import Optional + +from .tracker import ( + log_escalation, + get_escalations, + mark_resolved, + get_stats, +) + + +def handle_crisis_api(method: str, path: str, body: Optional[str] = None, + hermes_home: Optional[str] = None) -> dict: + """ + Handle a crisis API request. Returns dict with status, headers, body. + + Args: + method: HTTP method (GET, POST) + path: Request path (e.g., "/api/crisis/escalations") + body: JSON request body (for POST) + hermes_home: Override HERMES_HOME path + + Returns: + {"status": int, "headers": dict, "body": str} + """ + # Normalize path + path = path.rstrip("/") + + # GET /api/crisis/escalations + if method == "GET" and path == "/api/crisis/escalations": + params = _parse_query(path) + events = get_escalations( + limit=int(params.get("limit", 50)), + source=params.get("source"), + level=params.get("level"), + session_id=params.get("session_id"), + since=params.get("since"), + hermes_home=hermes_home, + ) + return _json_response(200, {"events": events, "count": len(events)}) + + # GET /api/crisis/stats + if method == "GET" and path == "/api/crisis/stats": + stats = get_stats(hermes_home=hermes_home) + return _json_response(200, stats) + + # POST /api/crisis/log + if method == "POST" and path == "/api/crisis/log": + if not body: + return _json_response(400, {"error": "Missing request body"}) + try: + data = json.loads(body) + except json.JSONDecodeError: + return _json_response(400, {"error": "Invalid JSON"}) + + required = ["source", "session_id", "level", "indicators"] + missing = [f for f in required if f not in data] + if missing: + return _json_response(400, {"error": f"Missing fields: {missing}"}) + + event = log_escalation( + source=data["source"], + session_id=data["session_id"], + level=data["level"], + indicators=data.get("indicators", []), + score=data.get("score", 0.0), + action_taken=data.get("action_taken", ""), + hermes_home=hermes_home, + ) + return _json_response(201, event) + + # POST /api/crisis/resolve/:id + if method == "POST" and path.startswith("/api/crisis/resolve/"): + event_id = path.split("/")[-1] + if mark_resolved(event_id, hermes_home=hermes_home): + return _json_response(200, {"resolved": True, "id": event_id}) + return _json_response(404, {"error": "Event not found"}) + + return _json_response(404, {"error": "Not found"}) + + +def _parse_query(path: str) -> dict: + """Extract query parameters from path.""" + params = {} + if "?" in path: + query = path.split("?", 1)[1] + for pair in query.split("&"): + if "=" in pair: + k, v = pair.split("=", 1) + params[k] = v + return params + + +def _json_response(status: int, body: dict) -> dict: + """Format a JSON response.""" + return { + "status": status, + "headers": {"Content-Type": "application/json"}, + "body": json.dumps(body, indent=2), + } + + +# --------------------------------------------------------------------------- +# Standalone server (for development / testing) +# --------------------------------------------------------------------------- + +def run_standalone(host: str = "127.0.0.1", port: int = 8650, + hermes_home: Optional[str] = None): + """Run a minimal HTTP server for the crisis API.""" + try: + from http.server import HTTPServer, BaseHTTPRequestHandler + except ImportError: + print("http.server not available") + return + + class Handler(BaseHTTPRequestHandler): + def do_GET(self): + result = handle_crisis_api("GET", self.path, hermes_home=hermes_home) + self._send(result) + + def do_POST(self): + length = int(self.headers.get("Content-Length", 0)) + body = self.rfile.read(length).decode() if length > 0 else None + result = handle_crisis_api("POST", self.path, body, hermes_home=hermes_home) + self._send(result) + + def _send(self, result): + self.send_response(result["status"]) + for k, v in result["headers"].items(): + self.send_header(k, v) + self.end_headers() + self.wfile.write(result["body"].encode()) + + def log_message(self, format, *args): + pass # Suppress default logging + + server = HTTPServer((host, port), Handler) + print(f"Crisis bridge running at http://{host}:{port}") + print(f" GET /api/crisis/escalations") + print(f" GET /api/crisis/stats") + print(f" POST /api/crisis/log") + print(f" POST /api/crisis/resolve/:id") + server.serve_forever() + + +if __name__ == "__main__": + import argparse + parser = argparse.ArgumentParser(description="Crisis Bridge API Server") + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=8650) + parser.add_argument("--hermes-home", default=None) + args = parser.parse_args() + run_standalone(args.host, args.port, args.hermes_home)