src/gaia_agent/client.py

from pathlib import Path
from urllib.parse import quote

import httpx

from gaia_agent.models import Answer, Question, ScoreResponse

DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
GAIA_VALIDATION_FILES_URL = (
    "https://huggingface.co/datasets/evatan/GAIA-modified/resolve/main/2023/validation"
)
REQUEST_TIMEOUT_SECONDS = 60.0


class ScoringClient:
    def __init__(self, base_url: str = DEFAULT_API_URL) -> None:
        self._client = httpx.Client(
            base_url=base_url.rstrip("/"), timeout=REQUEST_TIMEOUT_SECONDS, follow_redirects=True
        )

    def close(self) -> None:
        self._client.close()

    def __enter__(self) -> "ScoringClient":
        return self

    def __exit__(self, *_: object) -> None:
        self.close()

    def questions(self) -> list[Question]:
        response = self._client.get("/questions")
        response.raise_for_status()
        return [Question.model_validate(item) for item in response.json()]

    def download_attachment(self, question: Question, directory: Path) -> Path | None:
        if not question.file_name:
            return None
        directory.mkdir(parents=True, exist_ok=True)
        destination = directory / Path(question.file_name).name
        response = self._client.get(f"/files/{quote(question.task_id, safe='')}")
        if response.status_code == httpx.codes.NOT_FOUND:
            response = httpx.get(
                f"{GAIA_VALIDATION_FILES_URL}/{quote(destination.name, safe='')}",
                timeout=REQUEST_TIMEOUT_SECONDS,
                follow_redirects=True,
            )
        response.raise_for_status()
        destination.write_bytes(response.content)
        return destination

    def submit(self, username: str, agent_code: str, answers: list[Answer]) -> ScoreResponse:
        payload = {
            "username": username,
            "agent_code": agent_code,
            "answers": [answer.model_dump() for answer in answers],
        }
        response = self._client.post("/submit", json=payload)
        response.raise_for_status()
        return ScoreResponse.model_validate(response.json())