45 lines
1.4 KiB
Python
45 lines
1.4 KiB
Python
"""Backend API client for assistant config."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, Optional
|
|
|
|
import aiohttp
|
|
from loguru import logger
|
|
|
|
from app.config import settings
|
|
|
|
|
|
async def fetch_assistant_config(assistant_id: str) -> Optional[Dict[str, Any]]:
|
|
"""Fetch assistant config payload from backend API.
|
|
|
|
Expected response shape:
|
|
{
|
|
"assistant": {...},
|
|
"voice": {...} | null
|
|
}
|
|
"""
|
|
if not settings.backend_url:
|
|
logger.warning("BACKEND_URL not set; skipping assistant config fetch")
|
|
return None
|
|
|
|
url = f"{settings.backend_url.rstrip('/')}/api/assistants/{assistant_id}/config"
|
|
timeout = aiohttp.ClientTimeout(total=settings.backend_timeout_sec)
|
|
|
|
try:
|
|
async with aiohttp.ClientSession(timeout=timeout) as session:
|
|
async with session.get(url) as resp:
|
|
if resp.status == 404:
|
|
logger.warning(f"Assistant config not found: {assistant_id}")
|
|
return None
|
|
resp.raise_for_status()
|
|
payload = await resp.json()
|
|
if not isinstance(payload, dict):
|
|
logger.warning("Assistant config payload is not a dict; ignoring")
|
|
return None
|
|
return payload
|
|
except Exception as exc:
|
|
logger.warning(f"Failed to fetch assistant config ({assistant_id}): {exc}")
|
|
return None
|
|
|