File size: 2,678 Bytes
7dfd4df 038ad63 7dfd4df 038ad63 7dfd4df ee3a0ed 7dfd4df ee3a0ed 7dfd4df | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 | """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}
|