Something went wrong. Try again.
The agent that grows with you
Something went wrong. Try again.
4.2 kB · 124 lines
Python
at bb/theme
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125"""Wire protocol for gateway ↔ node RPC.
Everything is a JSON object with the same envelope shape:
Request: {"type": <str>, "id": <str>, "token": <str>, "payload": <dict>} Response: {"type": "<req-type>_res", "id": <req-id>, "payload": <dict>} Error: {"type": "error", "id": <req-id>, "error": <str>}
Requests must carry the shared bearer token (set up via``hermes meet node approve`` on the gateway and read off disk on theserver). Mismatched tokens are rejected before dispatch."""
from __future__ import annotations
import jsonimport uuidfrom typing import Any, Dict, Tuple
VALID_REQUEST_TYPES = frozenset({ "start_bot", "stop", "status", "transcript", "say", "ping",})
def make_request( type: str, token: str, payload: Dict[str, Any], req_id: str | None = None,) -> Dict[str, Any]: """Construct a request envelope.
``req_id`` is auto-generated (uuid4 hex) when not supplied so callers can correlate async responses. """ if not isinstance(type, str) or not type: raise ValueError("type must be a non-empty string") if type not in VALID_REQUEST_TYPES: raise ValueError(f"unknown request type: {type!r}") if not isinstance(token, str): raise ValueError("token must be a string") if not isinstance(payload, dict): raise ValueError("payload must be a dict") return { "type": type, "id": req_id or uuid.uuid4().hex, "token": token, "payload": payload, }
def make_response(req_id: str, payload: Dict[str, Any]) -> Dict[str, Any]: """Build a success response. The caller supplies the *request* type; we suffix it with ``_res`` so clients can assert they got the right reply.
For simplicity we don't require the type here — clients usually just key off ``id``. But we still emit a generic ``*_res`` envelope. """ if not isinstance(payload, dict): raise ValueError("payload must be a dict") return {"type": "response", "id": req_id, "payload": payload}
def make_error(req_id: str, error: str) -> Dict[str, Any]: return {"type": "error", "id": req_id, "error": str(error)}
def encode(msg: Dict[str, Any]) -> str: """Serialize a message envelope to a JSON string.""" return json.dumps(msg, separators=(",", ":"), ensure_ascii=False)
def decode(raw: str) -> Dict[str, Any]: """Parse a JSON envelope, raising ValueError on anything malformed.
Minimal type validation: must be an object, must contain ``type`` and ``id``. Heavier validation (token match, payload shape) happens in :func:`validate_request` on the server side. """ try: obj = json.loads(raw) except (TypeError, json.JSONDecodeError) as exc: raise ValueError(f"malformed JSON: {exc}") from exc if not isinstance(obj, dict): raise ValueError("envelope must be a JSON object") if "type" not in obj or not isinstance(obj["type"], str): raise ValueError("envelope missing string 'type'") if "id" not in obj or not isinstance(obj["id"], str): raise ValueError("envelope missing string 'id'") return obj
def validate_request(msg: Dict[str, Any], expected_token: str) -> Tuple[bool, str]: """Check a decoded request against the server's shared token.
Returns ``(True, "")`` when the envelope is acceptable or ``(False, <reason>)`` otherwise. Reason strings are safe to surface back to the client in an error envelope. """ if not isinstance(msg, dict): return False, "envelope must be a dict" t = msg.get("type") if not isinstance(t, str) or not t: return False, "missing or non-string 'type'" if t not in VALID_REQUEST_TYPES: return False, f"unknown request type: {t!r}" if not isinstance(msg.get("id"), str) or not msg.get("id"): return False, "missing or non-string 'id'" token = msg.get("token") if not isinstance(token, str) or not token: return False, "missing token" if token != expected_token: return False, "token mismatch" payload = msg.get("payload") if not isinstance(payload, dict): return False, "payload must be a dict" return True, ""