kunthawat
a582c1ff36
- auth from Authorization/Device-Id headers (firmware v2.5.0), frame fallback kept
- server hello carries transport=websocket + session_id (firmware ParseServerHello)
- /ota GET+POST returns {websocket:{url,token,version}, server_time:{timestamp,timezone_offset}, firmware:{version}}
- config.yaml: device registered by MAC, ota block added
- 21 tests pass (added OTA + header-auth coverage)
105 lines
3.2 KiB
Python
105 lines
3.2 KiB
Python
"""Hermes-Xiaozhi Bridge — FastAPI app (REQ-001, REQ-012).
|
|
|
|
Binds LAN/localhost only (REQ-012); the Cloudflare tunnel is external. The
|
|
single device-facing endpoint delegates the whole session to
|
|
:func:`xiaozhi.websocket.DeviceLoop`.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from fastapi import FastAPI, Request, WebSocket
|
|
|
|
from .config import Config
|
|
from .hermes import VoiceProfile
|
|
from .security import DeviceAuth
|
|
from .xiaozhi.websocket import DeviceLoop
|
|
|
|
|
|
def make_app(cfg: Config | None = None) -> FastAPI:
|
|
cfg = cfg or Config()
|
|
app = FastAPI(title="hermes-xiaozhi-bridge")
|
|
app.state.cfg = cfg
|
|
|
|
# AC-004: assert the voice contract at startup, not mid-conversation.
|
|
profile = VoiceProfile(cfg.hermes.reasoning, cfg.hermes.tools)
|
|
auth = DeviceAuth(
|
|
{d.device_id: d.token_hash for d in cfg.security.devices},
|
|
enabled=cfg.security.enabled,
|
|
)
|
|
|
|
@app.get("/health")
|
|
async def health() -> dict:
|
|
return {"status": "ok", "service": "hermes-xiaozhi-bridge"}
|
|
|
|
@app.websocket(cfg.ws_path)
|
|
async def xiaozhi_endpoint(ws: WebSocket) -> None:
|
|
# FIRMWARE handshake (xiaozhi v2.5.0): auth travels in HTTP headers,
|
|
# not the hello frame — Authorization: Bearer <token>, Device-Id: <MAC>.
|
|
headers = {k.lower(): v for k, v in ws.headers.items()}
|
|
device_id = headers.get("device-id", "")
|
|
authorization = headers.get("authorization", "")
|
|
loop = DeviceLoop(cfg, profile, auth, headers={
|
|
"device_id": device_id,
|
|
"authorization": authorization,
|
|
"client_id": headers.get("client-id", ""),
|
|
"protocol_version": headers.get("protocol-version", "1"),
|
|
})
|
|
await loop.run(ws)
|
|
|
|
if cfg.ota is not None:
|
|
|
|
def _ota_payload() -> dict:
|
|
import time
|
|
|
|
ota = cfg.ota
|
|
now_ms = int(time.time() * 1000)
|
|
return {
|
|
"websocket": {
|
|
"url": ota.server_ws_url,
|
|
"token": ota.server_token,
|
|
"version": ota.protocol_version,
|
|
},
|
|
"server_time": {
|
|
"timestamp": now_ms,
|
|
"timezone_offset": ota.timezone_minutes,
|
|
},
|
|
"firmware": {"version": ota.firmware_version},
|
|
}
|
|
|
|
@app.get("/ota")
|
|
async def ota_get() -> dict:
|
|
return _ota_payload()
|
|
|
|
@app.post("/ota")
|
|
async def ota_post(request: Request) -> dict:
|
|
return _ota_payload()
|
|
|
|
return app
|
|
|
|
|
|
def _load_config() -> Config:
|
|
"""Load config.yaml from the repo root; fall back to all-mock defaults."""
|
|
import os
|
|
|
|
path = os.environ.get(
|
|
"BRIDGE_CONFIG",
|
|
os.path.join(os.path.dirname(os.path.dirname(__file__)), "config.yaml"),
|
|
)
|
|
if os.path.exists(path):
|
|
return Config.from_file(path)
|
|
return Config()
|
|
|
|
|
|
# Module-level app so `uvicorn app.main:app` loads config.yaml (REQ: real
|
|
# token in production). `main()` is the equivalent entrypoint.
|
|
app = make_app(_load_config())
|
|
|
|
|
|
def main() -> None:
|
|
import uvicorn
|
|
|
|
uvicorn.run(app, host=app.state.cfg.host, port=app.state.cfg.port)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|