67 lines
1.8 KiB
Python
67 lines
1.8 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, 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:
|
|
loop = DeviceLoop(cfg, profile, auth)
|
|
await loop.run(ws)
|
|
|
|
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()
|