INNER CODE UNIT · Python

_require_stream_response

genome-spy/genome-spy · packages/app-agent/server/app/main.py:313

def _require_stream_response(response: ProviderResponse | None) -> ProviderResponse:
    """Require a final provider stream event to include a response."""
    if response is None:
        raise ProviderError("Provider stream ended without a response.")

    return response


def _build_final_stream_payload(
    response: ProviderResponse, message: str, duration_ms: int
) -> dict[str, object]:
    """Build the final SSE payload for a completed provider turn."""
    payload: dict[str, object] = {
        "response": {
            "type": response.type,
            "message": response.message,
            **(
                {

View source record →