| """Authenticated agent endpoint; Cloudflare is the only intended caller."""
|
|
|
| import html |
| import re |
| import time
|
| import uuid
|
|
|
| from fastapi import APIRouter, HTTPException, Request |
|
|
| from src.agentcore.models import AgentRequest, AgentResponse
|
| from src.agentcore.orchestrator import _agent_query_inner, _log_agent_query
|
| from src.agentcore.utils import _get_runtime_config, _normalize_whitespace, _safe_metrics_increment, _sanitize_llm_response
|
| from src.services.html_sanitizer import sanitize_answer_html |
| from src.config import get_settings |
|
|
| router = APIRouter()
|
|
|
|
|
| @router.post("", response_model=AgentResponse)
|
| async def agent_query(request: Request, req: AgentRequest): |
| start_time = time.time() |
| settings = get_settings() |
| if (len(req.question) > settings.max_question_chars or |
| len(req.history) > settings.max_history_messages or |
| any(len(message.content) > settings.max_message_chars for message in req.history)): |
| raise HTTPException(status_code=422, detail="Question or history exceeds the configured text limit.") |
| supplied_id = request.headers.get("x-request-id", "") |
| request_id = supplied_id if re.fullmatch(r"[A-Za-z0-9_-]{8,80}", supplied_id) else str(uuid.uuid4()) |
| question = _normalize_whitespace(req.question) |
| if not question: |
| raise HTTPException(status_code=422, detail="Question must contain text.") |
| history = list(req.history)
|
| if history and history[-1].role == "user" and _normalize_whitespace(history[-1].content) == question:
|
| history.pop()
|
| is_canary = request.headers.get("x-libbee-canary") == "1"
|
| if not is_canary:
|
| _safe_metrics_increment("agent_requests")
|
| response = await _agent_query_inner(question, history, req.model, req.client_state, start_time)
|
| response.request_id = request_id
|
| response.answer = _sanitize_llm_response(response.answer)
|
| runtime_config = _get_runtime_config()
|
| announcement = str(runtime_config.get("announcement", "")).strip()
|
| if announcement and response.intent not in {"social_greeting", "social"}:
|
| response.answer += '<div class="libbee-announcement" role="note"><strong>Announcement:</strong> ' + html.escape(announcement) + "</div>"
|
| response.answer = sanitize_answer_html(response.answer)
|
| if not is_canary:
|
| _log_agent_query(question, response, start_time)
|
| return response
|
|
|
|
|
| @router.get("/test")
|
| async def test():
|
| return {"status": "ok"}
|
|
|
|
|
| @router.get("/rag-status")
|
| async def rag_status():
|
| try:
|
| from app import get_rag_service
|
| return get_rag_service().get_stats()
|
| except Exception:
|
| return {"ready": False}
|
|
|
