Janady13/ai-brain-core / adapter/adapter.py

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/ai-brain-core / adapter/adapter.py
Date	Not specified in source
Repository	Janady13/ai-brain-core
Revision	f7275582f13ada80bfb3b1008089800ca26f4a23
Branch	main
Path or archive member	adapter/adapter.py
Original SHA-256	f969bdc1b9d6e91b24b6748cea0254a16854d5d83bc59c0d9db5445e905f91c7
Git blob SHA	f7f84cef559a5a501bd683bfb7539eca1e6be2c9
Reading-copy SHA-256	f969bdc1b9d6e91b24b6748cea0254a16854d5d83bc59c0d9db5445e905f91c7

Original source ↗

Every recorded occurrence

These 3 records contain identical original source bytes. Their paths and revisions remain part of the archive.

Original record	Branch or archive	Revision / identifier	Original location
Janady13/ai-brain-core / adapter/adapter.py	main	f7275582f13ada80bfb3b1008089800ca26f4a23	Open original
Janady13/ai-brain-core / adapter/adapter.py	claude/agi-architecture-diagram-01GJ7CohfKvLb77yiKpHqd6d	365850e27004318edd582881225024026b118c2d	Open original
Janady13/ai-brain-core / adapter/adapter.py	healer-core	0fe4cbbaa91912149f1a8e32205371ff67017268	Open original

Recovered source transcript

Download complete source text ↓54 lines · 2.0 KB

#!/usr/bin/env python3
"""
Adapter service: exposes a minimal API over a UNIX domain socket path (or TCP localhost fallback)
and executes whitelisted action scripts found in adapter/actions/.
Auth: token via X-Adapter-Token header.
"""
import os, json, subprocess, shlex, logging
from pathlib import Path
from fastapi import FastAPI, Request, HTTPException
import uvicorn

BASE = Path(__file__).resolve().parent
ACTIONS_DIR = BASE / 'actions'
TOKEN_FILE = BASE / 'token.txt'
SOCKET_PATH = BASE / 'adapter.sock'

app = FastAPI()
logging.basicConfig(level=logging.INFO)

def read_token():
    if TOKEN_FILE.exists():
        return TOKEN_FILE.read_text().strip()
    return None

def is_whitelisted(action):
    # action script must exist in actions dir and be executable
    script = ACTIONS_DIR / f"{action}.sh"
    return script.exists() and os.access(script, os.X_OK)

@app.post("/{action}")
async def run_action(action: str, request: Request):
    token = request.headers.get('x-adapter-token')
    expected = read_token()
    if expected and token != expected:
        raise HTTPException(status_code=401, detail="invalid token")
    if not is_whitelisted(action):
        raise HTTPException(status_code=403, detail="action not allowed")
    payload = await request.json()
    # create unique log id
    jid = payload.get('job_id', 'manual')
    script = ACTIONS_DIR / f"{action}.sh"
    try:
        # run script with payload as JSON env var
        env = os.environ.copy()
        env['HYP_PAYLOAD'] = json.dumps(payload)
        proc = subprocess.run([str(script)], env=env, timeout=60, capture_output=True, text=True)
        return {"returncode": proc.returncode, "stdout": proc.stdout, "stderr": proc.stderr}
    except subprocess.TimeoutExpired:
        raise HTTPException(status_code=504, detail="action timeout")

if __name__ == "__main__":
    # start on TCP localhost for simplicity if UDS not available; recommend using systemd socket activation on Linux
    uvicorn.run(app, host="127.0.0.1", port=8800)