from contextlib import asynccontextmanager import datetime import os from pathlib import Path import subprocess import httpx from fastapi import FastAPI, Request, Response from fastapi.responses import HTMLResponse children: dict[str, subprocess.Popen] = {} def resolve_data_dir() -> Path: data_dir = Path(os.environ.get("HERMES_DATA_DIR", "/data")) if data_dir.is_dir() and os.access(data_dir, os.W_OK): return data_dir fallback = Path.home() / "hermes_data" fallback.mkdir(parents=True, exist_ok=True) os.environ["HERMES_DATA_DIR"] = str(fallback) return fallback def write_initial_config(data_dir: Path) -> None: env_path = data_dir / ".env" config_path = data_dir / "config.yaml" if not env_path.exists(): env_path.write_text( "OPENAI_API_KEY={}\nANTHROPIC_API_KEY={}\n".format( os.environ.get("OPENAI_API_KEY", ""), os.environ.get("ANTHROPIC_API_KEY", ""), ) ) if not config_path.exists(): config_path.write_text( "model: claude-opus-4-5\ndata_dir: {}\n".format(data_dir) ) def start_child(name: str, command: list[str], env: dict[str, str]) -> None: proc = children.get(name) if proc and proc.poll() is None: return log_dir = Path("/app/logs") log_dir.mkdir(parents=True, exist_ok=True) out = (log_dir / f"{name}.log").open("ab", buffering=0) err = (log_dir / f"{name}_err.log").open("ab", buffering=0) children[name] = subprocess.Popen(command, stdout=out, stderr=err, env=env) @asynccontextmanager async def lifespan(app: FastAPI): data_dir = resolve_data_dir() write_initial_config(data_dir) env = os.environ.copy() env["HERMES_HOME"] = str(data_dir) env["HERMES_DATA_DIR"] = str(data_dir) env.setdefault("API_SERVER_ENABLED", "true") env.setdefault("API_SERVER_HOST", "127.0.0.1") env.setdefault("API_SERVER_PORT", "8642") start_child("hermes_gateway", ["hermes", "gateway", "run", "--replace"], env) yield for proc in children.values(): if proc.poll() is None: proc.terminate() app = FastAPI(lifespan=lifespan) @app.get("/", response_class=HTMLResponse) async def root(): data_dir = Path(os.environ.get("HERMES_DATA_DIR", "/data")) storage_ok = data_dir.is_dir() and os.access(data_dir, os.W_OK) gateway_ok = children.get("hermes_gateway") and children["hermes_gateway"].poll() is None ts = datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC") hermes_state = "Running" if gateway_ok else "Starting..." storage_state = f"{data_dir} writable" if storage_ok else "Local fallback" return f""" TestClaw
Hermes autonomous research container
Hermes Gateway
{hermes_state}
API proxy: /api/
Control Surface
Available
TestClaw status and routing are served directly by this Space.
Storage
{storage_state}
Attach persistent storage at /data for durable memory.
""" @app.get("/health") async def health(): return { "status": "ok", "hermes_gateway": bool(children.get("hermes_gateway") and children["hermes_gateway"].poll() is None), "ts": datetime.datetime.utcnow().isoformat(), } async def proxy_request(request: Request, target_base: str, path: str) -> Response: url = f"{target_base}/{path}" if request.url.query: url = f"{url}?{request.url.query}" headers = { key: value for key, value in request.headers.items() if key.lower() not in {"host", "content-length"} } body = await request.body() async with httpx.AsyncClient(timeout=60.0, follow_redirects=False) as client: upstream = await client.request( request.method, url, headers=headers, content=body, ) response_headers = { key: value for key, value in upstream.headers.items() if key.lower() not in {"content-encoding", "transfer-encoding", "connection"} } return Response(upstream.content, status_code=upstream.status_code, headers=response_headers) @app.api_route("/api/{path:path}", methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"]) async def api_proxy(request: Request, path: str = ""): return await proxy_request(request, "http://127.0.0.1:8642", path) @app.api_route("/hermes/{path:path}", methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"]) async def hermes_proxy(request: Request, path: str = ""): return Response( "Hermes web dashboard is disabled in the slim Space image. Use /api/ for gateway access.", status_code=503, media_type="text/plain", )