Skip to content

symfonic.evals.http_target

http_target

The deployed-application target: the shipped JSON chat API over HTTP.

Split from :mod:symfonic.evals.targets when that module went over its size budget. The split is along the line the two halves already had: the targets there drive an Agent in this process, and this one drives an application across a network, which is why it is the only module in the package that imports httpx and the only one declared as an egress path.

HttpChatTarget

HttpChatTarget(base_url: str, *, headers_for_scope: HeaderFactory, restart: RestartHook | None = None, evidence_for_run: EvidenceLoader | None = None, client_factory: Callable[[], AsyncClient] | None = None)

Evaluate the shipped JSON chat API with explicit scope credentials.

headers_for_scope is the authorization boundary: a scenario names a harmless alias and the deployment decides which authenticated headers that alias receives. Raw credentials therefore never enter reports or fixtures.

It reports no capabilities and no turn_inputs, which is the honest answer for this protocol: the API carries a query and returns an answer, so nothing about a turn through it evidences a composed capability. Optional packs therefore resolve to not-applicable against it by name rather than failing on a step it would refuse to deliver.

Source code in src/symfonic/evals/http_target.py
def __init__(
    self,
    base_url: str,
    *,
    headers_for_scope: HeaderFactory,
    restart: RestartHook | None = None,
    evidence_for_run: EvidenceLoader | None = None,
    client_factory: Callable[[], httpx.AsyncClient] | None = None,
) -> None:
    if not base_url:
        raise ValueError("HttpChatTarget requires a base URL")
    if not callable(headers_for_scope):
        raise TypeError("headers_for_scope must be callable")
    self._base_url = base_url.rstrip("/")
    self._headers_for_scope = headers_for_scope
    self._restart_hook = restart
    self._evidence_for_run = evidence_for_run
    self._client_factory = client_factory or (
        lambda: httpx.AsyncClient(base_url=self._base_url)
    )
    self._client: httpx.AsyncClient | None = None
    self._sessions: dict[tuple[str, str], str] = {}