diff --git a/docs/quickstart.md b/docs/quickstart.md index f12f38fa9..add897197 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -159,6 +159,35 @@ print(response.choices[0].message.content) Reasoning also works with streaming — each chunk may include `chunk.choices[0].delta.reasoning`. +## Gemini Interactions + +Gemini maps text-only `responses()` calls to Google's [Interactions API](https://ai.google.dev/gemini-api/docs/interactions). The existing `completion()` method continues to use `generateContent`. + +```python +from any_llm import AnyLLM + +llm = AnyLLM.create("gemini") +response = llm.responses( + model="gemini-3.8-flash", + input_data="Explain why the sky is blue.", +) +print(response.output_text) +``` + +Pass `stream=True` to `responses()` for a synchronous event iterator. With the async API, await `aresponses()` and consume the returned async iterator: + +```python +stream = await llm.aresponses( + model="gemini-3.8-flash", + input_data="Explain why the sky is blue.", + stream=True, +) +async for event in stream: + print(event.type) +``` + +Gemini text Interactions accept string input, instructions, maximum output tokens, per-request timeouts, and streaming. This first slice does not expose `store`, so Google's default storage behavior applies. Other Responses parameters, including media, tools, reasoning controls, structured output, chaining, metadata, and background execution, raise `UnsupportedParameterError`. + ## Embeddings `embedding` and `aembedding` allow you to create vector embeddings from text using the same unified interface across providers. diff --git a/pyproject.toml b/pyproject.toml index 985350fb5..7a8d0d59d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,7 +39,8 @@ mistral = [ anthropic = [] gemini = [ - "google-genai>=1.51.0", + # 2.17.0 first models the stable Interactions API's diagnostic errors field. + "google-genai>=2.17.0", "google-cloud-storage", ] diff --git a/src/any_llm/providers/gemini/gemini.py b/src/any_llm/providers/gemini/gemini.py index a44bbb110..4a2c1eee8 100644 --- a/src/any_llm/providers/gemini/gemini.py +++ b/src/any_llm/providers/gemini/gemini.py @@ -1,13 +1,17 @@ import os +from collections.abc import AsyncIterator from typing import Any from google import genai from google.genai import types from typing_extensions import override -from any_llm.exceptions import MissingApiKeyError +from any_llm.exceptions import MissingApiKeyError, UnsupportedParameterError +from any_llm.types.responses import Response, ResponsesParams, ResponseStreamEvent from .base import GoogleProvider +from .interactions import convert_interaction_to_response, convert_responses_params +from .interactions_stream import convert_interaction_stream class GeminiProvider(GoogleProvider): @@ -17,6 +21,9 @@ class GeminiProvider(GoogleProvider): PROVIDER_DOCUMENTATION_URL = "https://ai.google.dev/gemini-api/docs" ENV_API_KEY_NAME = "GEMINI_API_KEY/GOOGLE_API_KEY" ENV_API_BASE_NAME = "GOOGLE_GEMINI_BASE_URL" + SUPPORTS_RESPONSES = True + + _interactions_api_version: str @override def _verify_and_set_api_key(self, api_key: str | None = None) -> str | None: @@ -29,6 +36,18 @@ def _verify_and_set_api_key(self, api_key: str | None = None) -> str | None: @override def _init_client(self, api_key: str | None = None, api_base: str | None = None, **kwargs: Any) -> None: + http_options = kwargs.get("http_options") + if isinstance(http_options, dict): + configured_api_version = http_options.get("api_version") + elif isinstance(http_options, types.HttpOptions): + configured_api_version = http_options.api_version + else: + configured_api_version = None + # Interactions is GA in v1, while the SDK defaults the shared Gemini + # Developer API client to v1beta for generateContent preview features. + # https://ai.google.dev/gemini-api/docs/api-versions + self._interactions_api_version = configured_api_version or "v1" + if api_base: http_options = kwargs.pop("http_options", None) if http_options is None: @@ -44,3 +63,23 @@ def _init_client(self, api_key: str | None = None, api_base: str | None = None, GoogleProvider._merge_timeout_into_http_options(timeout, kwargs) self.client = genai.Client(api_key=api_key, **kwargs) + + @override + async def _aresponses( + self, params: ResponsesParams, **kwargs: Any + ) -> Response | AsyncIterator[ResponseStreamEvent]: + if kwargs.pop("extra_body", None) is not None: + parameter_name = "extra_body" + raise UnsupportedParameterError(parameter_name, self.PROVIDER_NAME) + create_kwargs = convert_responses_params( + params, + self.PROVIDER_NAME, + api_version=self._interactions_api_version, + ) + if (timeout := kwargs.get("timeout")) is not None: + create_kwargs["timeout"] = timeout + if params.stream: + stream = await self.client.aio.interactions.create(**create_kwargs) + return convert_interaction_stream(stream, model=params.model) + interaction = await self.client.aio.interactions.create(**create_kwargs) + return convert_interaction_to_response(interaction) diff --git a/src/any_llm/providers/gemini/interactions.py b/src/any_llm/providers/gemini/interactions.py new file mode 100644 index 000000000..f5f01b613 --- /dev/null +++ b/src/any_llm/providers/gemini/interactions.py @@ -0,0 +1,197 @@ +from __future__ import annotations + +from datetime import datetime +from typing import TYPE_CHECKING, Any, Literal + +from google.genai.interactions import ( + InteractionSseEventInteraction, + ModelOutputStep, + TextContent, + Usage, +) +from openai.types.responses import ( + Response as OpenAIResponse, +) +from openai.types.responses import ( + ResponseOutputMessage, + ResponseOutputText, + ResponseStatus, + ResponseUsage, +) +from openai.types.responses.response_usage import InputTokensDetails, OutputTokensDetails + +from any_llm.exceptions import UnsupportedParameterError + +if TYPE_CHECKING: + from collections.abc import Sequence + + # google-genai's public re-export is correct at runtime, but mypy resolves + # Interaction to its generated request union. Keep the private import type-only. + # https://github.com/googleapis/python-genai/blob/v2.22.0/google/genai/interactions.py + from google.genai._gaos.types.interactions import Interaction + + from any_llm.types.responses import ResponsesParams + + +# The stable v1 schema has six statuses. google-genai 2.17.0 also accepts +# queued and budget_exceeded, so normalize those SDK values without exposing +# them through the narrower OpenAI Response status type. +# https://ai.google.dev/api/interactions-api-v1#Interaction +_STATUS_MAP: dict[str, ResponseStatus] = { + "completed": "completed", + "failed": "failed", + "in_progress": "in_progress", + "cancelled": "cancelled", + "queued": "queued", + "incomplete": "incomplete", + "requires_action": "incomplete", + "budget_exceeded": "incomplete", +} + + +def _iso_to_epoch(value: str | None) -> float: + if not value: + return 0.0 + try: + return datetime.fromisoformat(value).timestamp() + except ValueError: + return 0.0 + + +def _map_status(status: object) -> ResponseStatus: + return _STATUS_MAP.get(str(status), "in_progress") + + +def _convert_usage(usage: Usage | None) -> ResponseUsage | None: + if usage is None: + return None + input_tokens = usage.total_input_tokens or 0 + output_tokens = usage.total_output_tokens or 0 + return ResponseUsage( + input_tokens=input_tokens, + input_tokens_details=InputTokensDetails( + cached_tokens=usage.total_cached_tokens or 0, + cache_write_tokens=0, + ), + output_tokens=output_tokens, + output_tokens_details=OutputTokensDetails(reasoning_tokens=usage.total_thought_tokens or 0), + total_tokens=usage.total_tokens if usage.total_tokens is not None else input_tokens + output_tokens, + ) + + +def _message_status(status: ResponseStatus) -> Literal["completed", "in_progress", "incomplete"]: + if status == "completed": + return "completed" + if status in {"in_progress", "queued"}: + return "in_progress" + return "incomplete" + + +def _messages_from_steps( + steps: Sequence[object] | None, + status: ResponseStatus, +) -> list[ResponseOutputMessage]: + messages: list[ResponseOutputMessage] = [] + for step in steps or []: + if not isinstance(step, ModelOutputStep): + continue + text_parts = [part.text for part in step.content or [] if isinstance(part, TextContent)] + if not text_parts: + continue + text = "".join(text_parts) + output_index = len(messages) + messages.append( + ResponseOutputMessage( + id=f"msg-{output_index}", + type="message", + role="assistant", + status=_message_status(status), + content=[ResponseOutputText(type="output_text", text=text, annotations=[])], + ) + ) + return messages + + +def _response_from_interaction( + interaction: Interaction | InteractionSseEventInteraction, + *, + fallback_model: str = "", +) -> OpenAIResponse: + status = _map_status(interaction.status) + previous_response_id = None + instructions = None + metadata = None + response_error = None + if not isinstance(interaction, InteractionSseEventInteraction): + previous_response_id = interaction.previous_interaction_id + instructions = interaction.system_instruction + metadata = interaction.labels + if interaction.errors: + first_error = interaction.errors[0] + response_error = { + "code": "server_error", + "message": first_error.message or first_error.code or "Gemini interaction failed", + } + + return OpenAIResponse.model_validate( + { + "id": interaction.id or "", + "created_at": _iso_to_epoch(interaction.created), + "error": response_error, + "instructions": instructions, + "metadata": metadata, + "model": str(interaction.model or fallback_model), + "object": "response", + "output": _messages_from_steps(interaction.steps, status), + "parallel_tool_calls": False, + "status": status, + "tool_choice": "auto", + "tools": [], + "previous_response_id": previous_response_id, + "usage": _convert_usage(interaction.usage), + } + ) + + +def convert_interaction_to_response(interaction: Interaction) -> OpenAIResponse: + """Normalize the text subset of a Gemini Interaction resource.""" + return _response_from_interaction(interaction) + + +def convert_responses_params( + params: ResponsesParams, + provider_name: str, + *, + api_version: str, +) -> dict[str, Any]: + """Translate the supported Responses subset into Interactions arguments.""" + if not isinstance(params.input, str): + parameter_name = "input" + raise UnsupportedParameterError(parameter_name, provider_name) + + supported = { + "model", + "input", + "instructions", + "max_output_tokens", + "stream", + } + unsupported = params.model_dump(exclude_none=True).keys() - supported + if unsupported: + raise UnsupportedParameterError(min(unsupported), provider_name) + + # The SDK has separate streaming and non-streaming overloads, while optional + # fields must be absent rather than None. Any stays at this generated SDK + # boundary because object values cannot satisfy either unpacked overload. + create_kwargs: dict[str, Any] = { + "api_version": api_version, + "model": params.model, + "input": params.input, + } + if params.instructions is not None: + create_kwargs["system_instruction"] = params.instructions + if params.max_output_tokens is not None: + create_kwargs["generation_config"] = {"max_output_tokens": params.max_output_tokens} + if params.stream: + create_kwargs["stream"] = True + return create_kwargs diff --git a/src/any_llm/providers/gemini/interactions_stream.py b/src/any_llm/providers/gemini/interactions_stream.py new file mode 100644 index 000000000..743334395 --- /dev/null +++ b/src/any_llm/providers/gemini/interactions_stream.py @@ -0,0 +1,326 @@ +from __future__ import annotations + +from inspect import isawaitable +from typing import TYPE_CHECKING, NoReturn, Protocol, assert_never, runtime_checkable + +from google.genai.interactions import ( + ErrorEvent, + InteractionCompletedEvent, + InteractionCreatedEvent, + InteractionSSEEvent, + InteractionStatusUpdate, + ModelOutputStep, + StepDelta, + StepStart, + StepStop, + TextContent, + TextDelta, + UnknownInteractionSSEEvent, + UnknownStepDeltaData, +) +from openai.types.responses import ( + ResponseCompletedEvent, + ResponseContentPartAddedEvent, + ResponseContentPartDoneEvent, + ResponseCreatedEvent, + ResponseFailedEvent, + ResponseIncompleteEvent, + ResponseInProgressEvent, + ResponseOutputItemAddedEvent, + ResponseOutputItemDoneEvent, + ResponseOutputMessage, + ResponseOutputText, + ResponseTextDeltaEvent, + ResponseTextDoneEvent, +) + +from any_llm.exceptions import ProviderError +from any_llm.logging import logger + +from .interactions import _response_from_interaction + +if TYPE_CHECKING: + from collections.abc import AsyncGenerator, AsyncIterator, Awaitable + + from openai.types.responses import Response as OpenAIResponse + + from any_llm.types.responses import ResponseStreamEvent + + +@runtime_checkable +class _Closeable(Protocol): + def close(self) -> Awaitable[None] | None: ... + + +def _terminal_event(response: OpenAIResponse, sequence_number: int) -> ResponseStreamEvent: + if response.status == "completed": + return ResponseCompletedEvent( + type="response.completed", + sequence_number=sequence_number, + response=response, + ) + if response.status == "failed": + return ResponseFailedEvent( + type="response.failed", + sequence_number=sequence_number, + response=response, + ) + return ResponseIncompleteEvent( + type="response.incomplete", + sequence_number=sequence_number, + response=response, + ) + + +def _raise_stream_error(message: str, *, code: str | None = None) -> NoReturn: + raise ProviderError(message, provider_name="gemini", code=code) + + +class _TextStreamState: + """Track Gemini steps while emitting the OpenAI text event lifecycle. + + Gemini streams step.start, step.delta, and step.stop. OpenAI consumers + expect item and content-part boundaries around text deltas, so this state + stays in the provider adapter instead of leaking Gemini events to callers. + https://ai.google.dev/gemini-api/docs/interactions/streaming + https://platform.openai.com/docs/api-reference/responses-streaming + """ + + def __init__(self, model: str) -> None: + self.model = model + self.sequence = 0 + self.started = False + self.open_steps: set[int] = set() + self.text_by_step: dict[int, str] = {} + self.output_index_by_step: dict[int, int] = {} + + def convert(self, event: InteractionSSEEvent) -> tuple[list[ResponseStreamEvent], bool]: + terminal = False + if isinstance(event, InteractionCreatedEvent): + converted = self._created(event) + elif isinstance(event, StepStart): + converted = self._step_started(event) + elif isinstance(event, StepDelta): + converted = self._step_delta(event) + elif isinstance(event, StepStop): + converted = self._step_stopped(event) + elif isinstance(event, InteractionCompletedEvent): + converted = [self._completed(event)] + terminal = True + elif isinstance(event, ErrorEvent): + self._error(event) + elif isinstance(event, UnknownInteractionSSEEvent): + logger.warning("Skipping unknown Gemini Interactions event: %s", event.event_type) + converted = [] + elif isinstance(event, InteractionStatusUpdate): + if not self.started: + _raise_stream_error("Gemini interaction stream emitted a status update before interaction.created") + converted = [] + else: + assert_never(event) + return converted, terminal + + def incomplete(self) -> NoReturn: + _raise_stream_error("Gemini interaction stream ended before interaction.completed") + + def _next_sequence(self) -> int: + sequence = self.sequence + self.sequence += 1 + return sequence + + def _created(self, event: InteractionCreatedEvent) -> list[ResponseStreamEvent]: + if self.started: + _raise_stream_error("Gemini interaction stream emitted interaction.created more than once") + self.started = True + response = _response_from_interaction(event.interaction, fallback_model=self.model) + return [ + ResponseCreatedEvent( + type="response.created", + sequence_number=self._next_sequence(), + response=response, + ), + ResponseInProgressEvent( + type="response.in_progress", + sequence_number=self._next_sequence(), + response=response.model_copy(update={"status": "in_progress"}), + ), + ] + + def _step_started(self, event: StepStart) -> list[ResponseStreamEvent]: + if not self.started: + _raise_stream_error("Gemini interaction stream emitted step.start before interaction.created") + if event.index in self.open_steps or event.index in self.text_by_step: + _raise_stream_error(f"Gemini interaction stream started step {event.index} more than once") + self.open_steps.add(event.index) + if not isinstance(event.step, ModelOutputStep): + return [] + + prefix = "".join(part.text for part in event.step.content or [] if isinstance(part, TextContent)) + self.text_by_step[event.index] = prefix + output_index = len(self.output_index_by_step) + self.output_index_by_step[event.index] = output_index + item_id = f"msg-{output_index}" + events: list[ResponseStreamEvent] = [ + ResponseOutputItemAddedEvent( + type="response.output_item.added", + sequence_number=self._next_sequence(), + output_index=output_index, + item=ResponseOutputMessage( + id=item_id, + type="message", + role="assistant", + status="in_progress", + content=[], + ), + ), + ResponseContentPartAddedEvent( + type="response.content_part.added", + sequence_number=self._next_sequence(), + item_id=item_id, + output_index=output_index, + content_index=0, + part=ResponseOutputText(type="output_text", text="", annotations=[]), + ), + ] + if prefix: + events.append(self._text_delta(event.index, prefix)) + return events + + def _step_delta(self, event: StepDelta) -> list[ResponseStreamEvent]: + if not self.started: + _raise_stream_error("Gemini interaction stream emitted step.delta before interaction.created") + if event.index not in self.open_steps: + _raise_stream_error(f"Gemini interaction stream emitted a delta before step.start for step {event.index}") + if isinstance(event.delta, UnknownStepDeltaData): + logger.warning("Skipping unknown Gemini Interactions step delta: %s", event.delta.raw) + return [] + if not isinstance(event.delta, TextDelta): + return [] + if event.index not in self.text_by_step: + _raise_stream_error(f"Gemini interaction stream emitted text for non-model step {event.index}") + self.text_by_step[event.index] += event.delta.text + return [self._text_delta(event.index, event.delta.text)] + + def _text_delta(self, step_index: int, text: str) -> ResponseTextDeltaEvent: + output_index = self.output_index_by_step[step_index] + return ResponseTextDeltaEvent( + type="response.output_text.delta", + sequence_number=self._next_sequence(), + item_id=f"msg-{output_index}", + output_index=output_index, + content_index=0, + delta=text, + logprobs=[], + ) + + def _step_stopped(self, event: StepStop) -> list[ResponseStreamEvent]: + if not self.started: + _raise_stream_error("Gemini interaction stream emitted step.stop before interaction.created") + if event.index not in self.open_steps: + _raise_stream_error(f"Gemini interaction stream stopped unknown step {event.index}") + self.open_steps.remove(event.index) + if event.index not in self.text_by_step: + return [] + + text = self.text_by_step[event.index] + output_index = self.output_index_by_step[event.index] + item_id = f"msg-{output_index}" + completed_part = ResponseOutputText(type="output_text", text=text, annotations=[]) + completed_item = ResponseOutputMessage( + id=item_id, + type="message", + role="assistant", + status="completed", + content=[completed_part], + ) + return [ + ResponseTextDoneEvent( + type="response.output_text.done", + sequence_number=self._next_sequence(), + item_id=item_id, + output_index=output_index, + content_index=0, + text=text, + logprobs=[], + ), + ResponseContentPartDoneEvent( + type="response.content_part.done", + sequence_number=self._next_sequence(), + item_id=item_id, + output_index=output_index, + content_index=0, + part=completed_part, + ), + ResponseOutputItemDoneEvent( + type="response.output_item.done", + sequence_number=self._next_sequence(), + output_index=output_index, + item=completed_item, + ), + ] + + def _completed(self, event: InteractionCompletedEvent) -> ResponseStreamEvent: + if not self.started: + _raise_stream_error("Gemini interaction stream completed before interaction.created") + if self.open_steps: + _raise_stream_error(f"Gemini interaction stream completed before step.stop for step {min(self.open_steps)}") + response = _response_from_interaction(event.interaction, fallback_model=self.model) + if not response.output: + response = response.model_copy(update={"output": self._stream_messages()}) + return _terminal_event(response, self._next_sequence()) + + def _stream_messages(self) -> list[ResponseOutputMessage]: + return [ + ResponseOutputMessage( + id=f"msg-{self.output_index_by_step[step_index]}", + type="message", + role="assistant", + status="completed", + content=[ResponseOutputText(type="output_text", text=text, annotations=[])], + ) + for step_index, text in sorted( + self.text_by_step.items(), + key=lambda item: self.output_index_by_step[item[0]], + ) + ] + + @staticmethod + def _error(event: ErrorEvent) -> NoReturn: + message = ( + event.error.message if event.error is not None and event.error.message else "Gemini interaction failed" + ) + code = event.error.code if event.error is not None else None + _raise_stream_error(message, code=code) + + +async def convert_interaction_stream( + stream: AsyncIterator[InteractionSSEEvent], + *, + model: str, +) -> AsyncGenerator[ResponseStreamEvent]: + """Normalize a create stream into the OpenAI text event lifecycle.""" + state = _TextStreamState(model) + primary_error: BaseException | None = None + try: + async for event in stream: + events, terminal = state.convert(event) + for converted_event in events: + yield converted_event + if terminal: + return + state.incomplete() + except BaseException as error: + if not isinstance(error, GeneratorExit): + primary_error = error + raise + finally: + try: + if isinstance(stream, _Closeable) and isawaitable(close_result := stream.close()): + await close_result + except BaseException as close_error: + if primary_error is None: + raise + logger.warning( + "Failed to close Gemini Interactions stream while handling another error", + exc_info=close_error, + ) diff --git a/tests/unit/providers/test_gemini_interactions.py b/tests/unit/providers/test_gemini_interactions.py new file mode 100644 index 000000000..7ae76b5a0 --- /dev/null +++ b/tests/unit/providers/test_gemini_interactions.py @@ -0,0 +1,831 @@ +import asyncio +import json +import logging +from collections.abc import AsyncIterator +from unittest.mock import AsyncMock, patch + +import httpx +import pytest +from google.genai import types +from google.genai._gaos.types.interactions import Interaction +from google.genai.interactions import ( + ArgumentsDelta, + Error, + ErrorEvent, + InteractionCompletedEvent, + InteractionCreatedEvent, + InteractionSSEEvent, + InteractionSseEventInteraction, + InteractionSseEventInteractionStatus, + InteractionStatusUpdate, + ModelOutputStep, + Step, + StepDelta, + StepStart, + StepStop, + TextContent, + TextDelta, + UnknownInteractionSSEEvent, + UnknownStepDeltaData, + Usage, + UserInputStep, +) +from openai.types.responses import ( + ResponseCompletedEvent, + ResponseContentPartAddedEvent, + ResponseContentPartDoneEvent, + ResponseCreatedEvent, + ResponseFailedEvent, + ResponseIncompleteEvent, + ResponseInProgressEvent, + ResponseOutputItemAddedEvent, + ResponseOutputItemDoneEvent, + ResponseOutputMessage, + ResponseOutputText, + ResponseTextDeltaEvent, + ResponseTextDoneEvent, +) + +from any_llm.exceptions import InvalidRequestError, ProviderError, UnsupportedParameterError +from any_llm.providers.gemini import GeminiProvider +from any_llm.providers.gemini.base import GoogleProvider +from any_llm.providers.gemini.interactions import ( + convert_interaction_to_response, + convert_responses_params, +) +from any_llm.providers.gemini.interactions_stream import convert_interaction_stream +from any_llm.providers.vertexai import VertexaiProvider +from any_llm.types.responses import Response, ResponsesParams, ResponseStreamEvent + + +def _interaction( + *, + status: str = "completed", + created: str = "2026-01-02T03:04:05Z", + steps: list[object] | None = None, + usage: Usage | None = None, +) -> Interaction: + if steps is None: + steps = [ModelOutputStep(content=[TextContent(text="Hello")])] + if usage is None: + usage = Usage( + total_input_tokens=4, + total_output_tokens=2, + total_tokens=6, + total_cached_tokens=1, + total_thought_tokens=3, + ) + return Interaction.model_validate( + { + "id": "int-123", + "status": status, + "model": "gemini-3.8-flash", + "created": created, + "previous_interaction_id": "int-previous", + "system_instruction": "Be concise", + "labels": {"team": "sdk"}, + "steps": steps, + "usage": usage, + } + ) + + +async def _events(*events: InteractionSSEEvent) -> AsyncIterator[InteractionSSEEvent]: + for event in events: + yield event + + +def _created(*, model: str | None = None) -> InteractionCreatedEvent: + return InteractionCreatedEvent( + interaction=InteractionSseEventInteraction( + id="int-123", + status="in_progress", + model=model, + ) + ) + + +def _completed( + status: InteractionSseEventInteractionStatus = "completed", + *, + model: str | None = None, + steps: list[Step] | None = None, +) -> InteractionCompletedEvent: + return InteractionCompletedEvent( + interaction=InteractionSseEventInteraction( + id="int-123", + status=status, + model=model, + steps=steps, + ) + ) + + +async def _converted_events(*events: InteractionSSEEvent, model: str = "requested") -> list[ResponseStreamEvent]: + return [event async for event in convert_interaction_stream(_events(*events), model=model)] + + +def test_gemini_enables_responses_without_changing_shared_google_provider() -> None: + assert GeminiProvider.SUPPORTS_RESPONSES is True + assert GoogleProvider.SUPPORTS_RESPONSES is False + assert VertexaiProvider.SUPPORTS_RESPONSES is False + + +def test_convert_interaction_maps_text_status_metadata_and_usage() -> None: + response = convert_interaction_to_response(_interaction()) + + assert isinstance(response, Response) + assert response.id == "int-123" + assert response.status == "completed" + assert response.model == "gemini-3.8-flash" + assert response.created_at == 1767323045.0 + assert response.previous_response_id == "int-previous" + assert response.instructions == "Be concise" + assert response.metadata == {"team": "sdk"} + assert response.output_text == "Hello" + message = response.output[0] + assert isinstance(message, ResponseOutputMessage) + assert message.id == "msg-0" + assert response.usage is not None + assert response.usage.input_tokens == 4 + assert response.usage.output_tokens == 2 + assert response.usage.total_tokens == 6 + assert response.usage.input_tokens_details.cached_tokens == 1 + assert response.usage.output_tokens_details.reasoning_tokens == 3 + + +def test_convert_interaction_preserves_explicit_zero_total_usage() -> None: + usage = Usage(total_input_tokens=2, total_output_tokens=3, total_tokens=0) + response = convert_interaction_to_response(_interaction(usage=usage)) + + assert response.usage is not None + assert response.usage.total_tokens == 0 + + +def test_convert_interaction_preserves_absent_usage() -> None: + interaction = _interaction() + interaction.usage = None + + assert convert_interaction_to_response(interaction).usage is None + + +def test_convert_interaction_preserves_empty_text_output() -> None: + response = convert_interaction_to_response(_interaction(steps=[ModelOutputStep(content=[TextContent(text="")])])) + + assert len(response.output) == 1 + assert response.output_text == "" + + +def test_convert_interaction_skips_unsupported_steps_and_content() -> None: + interaction = _interaction( + steps=[ + {"type": "future_step", "future": True}, + ModelOutputStep.model_validate({"content": [{"type": "future_content", "future": True}]}), + ModelOutputStep.model_validate( + { + "content": [ + {"type": "future_content", "future": True}, + {"type": "text", "text": "kept"}, + ] + } + ), + ] + ) + + response = convert_interaction_to_response(interaction) + + assert response.output_text == "kept" + assert len(response.output) == 1 + assert response.output[0].id == "msg-0" + + +def test_convert_interaction_maps_provider_error_without_raw_side_channel() -> None: + interaction = _interaction(status="failed", steps=[]) + interaction.errors = [Error(code="gateway_timeout", message="deadline expired")] + + response = convert_interaction_to_response(interaction) + + assert response.error is not None + assert response.error.code == "server_error" + assert response.error.message == "deadline expired" + assert not any(name.startswith("gemini_") for name in response.model_dump()) + + +def test_convert_interaction_handles_unknown_status_and_invalid_timestamp() -> None: + response = convert_interaction_to_response(_interaction(status="future_status", created="invalid")) + + assert response.status == "in_progress" + assert response.created_at == 0.0 + + +@pytest.mark.parametrize( + ("gemini_status", "expected_status"), + [ + ("queued", "queued"), + ("requires_action", "incomplete"), + ("budget_exceeded", "incomplete"), + ], +) +def test_convert_interaction_normalizes_extended_sdk_statuses( + gemini_status: str, + expected_status: str, +) -> None: + assert convert_interaction_to_response(_interaction(status=gemini_status)).status == expected_status + + +def test_convert_responses_params_maps_only_reviewed_text_subset() -> None: + params = ResponsesParams( + model="gemini-3.8-flash", + input="Hello", + instructions="", + max_output_tokens=0, + stream=True, + ) + + assert convert_responses_params(params, "gemini", api_version="v1") == { + "api_version": "v1", + "model": "gemini-3.8-flash", + "input": "Hello", + "system_instruction": "", + "generation_config": {"max_output_tokens": 0}, + "stream": True, + } + + +def test_convert_responses_params_rejects_non_string_input() -> None: + params = ResponsesParams(model="gemini-3.8-flash", input=[{"type": "input_text", "text": "Hello"}]) + + with pytest.raises(UnsupportedParameterError, match="input"): + convert_responses_params(params, "gemini", api_version="v1") + + +@pytest.mark.parametrize( + ("parameter", "value"), + [ + ("tools", [{"type": "function", "name": "lookup"}]), + ("reasoning", {"effort": "low"}), + ("response_format", {"type": "json_object"}), + ("background", True), + ("temperature", 0.2), + ("store", False), + ("metadata", {}), + ("previous_response_id", "int-previous"), + ], +) +def test_convert_responses_params_rejects_unimplemented_surface(parameter: str, value: object) -> None: + params = ResponsesParams.model_validate({"model": "gemini-3.8-flash", "input": "Hello", parameter: value}) + + with pytest.raises(UnsupportedParameterError, match=parameter): + convert_responses_params(params, "gemini", api_version="v1") + + +@pytest.mark.asyncio +async def test_convert_interaction_stream_maps_text_and_terminal_snapshot() -> None: + started = StepStart(index=0, step=ModelOutputStep(content=[TextContent(text="Hello")])) + delta = StepDelta(index=0, delta=TextDelta(text=" world")) + status = InteractionStatusUpdate(interaction_id="int-123", status="in_progress") + stopped = StepStop(index=0) + + result = await _converted_events( + _created(model="gemini-3.8-flash"), + status, + started, + delta, + stopped, + _completed(model="gemini-3.8-flash"), + model="gemini-3.8-flash", + ) + + assert [event.type for event in result] == [ + "response.created", + "response.in_progress", + "response.output_item.added", + "response.content_part.added", + "response.output_text.delta", + "response.output_text.delta", + "response.output_text.done", + "response.content_part.done", + "response.output_item.done", + "response.completed", + ] + assert [event.sequence_number for event in result] == list(range(10)) + assert isinstance(result[0], ResponseCreatedEvent) + assert result[0].response.model == "gemini-3.8-flash" + assert isinstance(result[1], ResponseInProgressEvent) + assert isinstance(result[2], ResponseOutputItemAddedEvent) + assert isinstance(result[2].item, ResponseOutputMessage) + assert result[2].item.content == [] + assert isinstance(result[3], ResponseContentPartAddedEvent) + assert isinstance(result[3].part, ResponseOutputText) + assert result[3].part.text == "" + assert isinstance(result[4], ResponseTextDeltaEvent) + assert result[4].delta == "Hello" + assert isinstance(result[5], ResponseTextDeltaEvent) + assert result[5].delta == " world" + assert isinstance(result[6], ResponseTextDoneEvent) + assert result[6].text == "Hello world" + assert isinstance(result[7], ResponseContentPartDoneEvent) + assert isinstance(result[7].part, ResponseOutputText) + assert result[7].part.text == "Hello world" + assert isinstance(result[8], ResponseOutputItemDoneEvent) + assert isinstance(result[8].item, ResponseOutputMessage) + assert isinstance(result[8].item.content[0], ResponseOutputText) + assert result[8].item.content[0].text == "Hello world" + terminal = result[9] + assert isinstance(terminal, ResponseCompletedEvent) + assert terminal.response.output_text == "Hello world" + + +@pytest.mark.asyncio +async def test_convert_interaction_stream_keeps_output_indices_contiguous() -> None: + user_started = StepStart(index=0, step=UserInputStep()) + model_started = StepStart(index=1, step=ModelOutputStep()) + + result = await _converted_events( + _created(), + user_started, + StepStop(index=0), + model_started, + StepDelta(index=1, delta=TextDelta(text="Hello")), + StepStop(index=1), + _completed(), + ) + + added = next(event for event in result if isinstance(event, ResponseOutputItemAddedEvent)) + assert added.output_index == 0 + assert added.item.id == "msg-0" + terminal = result[-1] + assert isinstance(terminal, ResponseCompletedEvent) + assert terminal.response.output[0].id == "msg-0" + + +@pytest.mark.asyncio +async def test_convert_interaction_stream_orders_terminal_messages_by_output_index() -> None: + result = await _converted_events( + _created(), + StepStart(index=7, step=ModelOutputStep()), + StepStart(index=2, step=ModelOutputStep()), + StepDelta(index=2, delta=TextDelta(text="second")), + StepStop(index=2), + StepDelta(index=7, delta=TextDelta(text="first")), + StepStop(index=7), + _completed(), + ) + + terminal = result[-1] + assert isinstance(terminal, ResponseCompletedEvent) + assert [message.id for message in terminal.response.output] == ["msg-0", "msg-1"] + assert terminal.response.output_text == "firstsecond" + + +@pytest.mark.asyncio +async def test_convert_interaction_stream_uses_terminal_steps_when_present() -> None: + result = await _converted_events( + _created(), + _completed(steps=[ModelOutputStep(content=[TextContent(text="terminal")])]), + ) + + terminal = result[-1] + assert isinstance(terminal, ResponseCompletedEvent) + assert terminal.response.model == "requested" + assert terminal.response.output_text == "terminal" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("status", "event_type", "message_status"), + [ + ("failed", "response.failed", "incomplete"), + ("incomplete", "response.incomplete", "incomplete"), + ], +) +async def test_convert_interaction_stream_maps_non_success_terminal_status( + status: InteractionSseEventInteractionStatus, + event_type: str, + message_status: str, +) -> None: + result = await _converted_events( + _created(), + _completed(status, steps=[ModelOutputStep(content=[TextContent(text="partial")])]), + ) + + terminal = result[-1] + assert isinstance(terminal, ResponseFailedEvent | ResponseIncompleteEvent) + assert terminal.type == event_type + assert isinstance(terminal.response.output[0], ResponseOutputMessage) + assert terminal.response.output[0].status == message_status + + +@pytest.mark.asyncio +async def test_convert_interaction_stream_logs_and_skips_unknown_event(caplog: pytest.LogCaptureFixture) -> None: + unknown = UnknownInteractionSSEEvent(raw={"event_type": "future.event", "value": 1}) + + with caplog.at_level(logging.WARNING, logger="any_llm"): + result = await _converted_events(_created(), unknown, _completed()) + + assert [event.type for event in result] == [ + "response.created", + "response.in_progress", + "response.completed", + ] + assert "Skipping unknown Gemini Interactions event" in caplog.text + + +@pytest.mark.asyncio +async def test_convert_interaction_stream_logs_and_skips_unknown_delta(caplog: pytest.LogCaptureFixture) -> None: + started = StepStart(index=0, step=ModelOutputStep()) + non_text = StepDelta(index=0, delta=ArgumentsDelta(arguments="{}")) + unknown = StepDelta(index=0, delta=UnknownStepDeltaData(raw={"type": "future_delta", "value": 1})) + stopped = StepStop(index=0) + + with caplog.at_level(logging.WARNING, logger="any_llm"): + result = await _converted_events(_created(), started, non_text, unknown, stopped, _completed()) + + assert [event.type for event in result] == [ + "response.created", + "response.in_progress", + "response.output_item.added", + "response.content_part.added", + "response.output_text.done", + "response.content_part.done", + "response.output_item.done", + "response.completed", + ] + assert "Skipping unknown Gemini Interactions step delta" in caplog.text + + +@pytest.mark.asyncio +async def test_convert_interaction_stream_raises_error_event() -> None: + error = ErrorEvent.model_validate( + {"event_type": "error", "error": {"code": "gateway_timeout", "message": "deadline expired"}} + ) + + with pytest.raises(ProviderError, match="deadline expired") as raised: + _ = [event async for event in convert_interaction_stream(_events(error), model="requested")] + + assert raised.value.code == "gateway_timeout" + + +@pytest.mark.asyncio +async def test_convert_interaction_stream_rejects_missing_terminal_event() -> None: + with pytest.raises(ProviderError, match=r"before interaction\.completed"): + await _converted_events(_created()) + + +@pytest.mark.asyncio +async def test_convert_interaction_stream_rejects_terminal_before_created() -> None: + with pytest.raises(ProviderError, match=r"before interaction\.created"): + await _converted_events(_completed("failed")) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("events", "message"), + [ + ( + [StepStart(index=0, step=ModelOutputStep())], + "step.start before interaction.created", + ), + ( + [InteractionStatusUpdate(interaction_id="int-123", status="in_progress")], + "status update before interaction.created", + ), + ( + [StepDelta(index=0, delta=TextDelta(text="unexpected"))], + "step.delta before interaction.created", + ), + ( + [StepStop(index=0)], + "step.stop before interaction.created", + ), + ( + [_created(), _created()], + "interaction.created more than once", + ), + ( + [ + _created(), + StepStart(index=0, step=ModelOutputStep()), + StepStart(index=0, step=ModelOutputStep()), + ], + "started step 0 more than once", + ), + ( + [ + _created(), + StepDelta(index=0, delta=TextDelta(text="unexpected")), + ], + "delta before step.start", + ), + ( + [ + _created(), + StepStart(index=0, step=UserInputStep()), + StepDelta(index=0, delta=TextDelta(text="unexpected")), + ], + "text for non-model step", + ), + ( + [_created(), StepStop(index=0)], + "stopped unknown step", + ), + ( + [ + _created(), + StepStart(index=0, step=ModelOutputStep()), + _completed(), + ], + "before step.stop", + ), + ], +) +async def test_convert_interaction_stream_rejects_malformed_order( + events: list[InteractionSSEEvent], + message: str, +) -> None: + with pytest.raises(ProviderError, match=message): + await _converted_events(*events) + + +@pytest.mark.asyncio +async def test_convert_interaction_stream_closes_source_when_consumer_stops() -> None: + stream = AsyncMock() + stream.__aiter__.return_value = [_created()] + + converted = convert_interaction_stream(stream, model="requested") + await anext(converted) + await converted.aclose() + + stream.close.assert_awaited_once_with() + + +@pytest.mark.asyncio +async def test_convert_interaction_stream_propagates_close_error_after_success() -> None: + stream = AsyncMock() + stream.__aiter__.return_value = [_created(), _completed()] + stream.close.side_effect = RuntimeError("close failed") + + with pytest.raises(RuntimeError, match="close failed"): + _ = [event async for event in convert_interaction_stream(stream, model="requested")] + + +@pytest.mark.asyncio +async def test_convert_interaction_stream_preserves_primary_error_when_close_fails( + caplog: pytest.LogCaptureFixture, +) -> None: + stream = AsyncMock() + stream.__aiter__.return_value = [ + ErrorEvent.model_validate( + {"event_type": "error", "error": {"code": "gateway_timeout", "message": "request failed"}} + ) + ] + stream.close.side_effect = RuntimeError("close failed") + + with ( + caplog.at_level(logging.WARNING, logger="any_llm"), + pytest.raises(ProviderError, match="request failed"), + ): + _ = [event async for event in convert_interaction_stream(stream, model="requested")] + + assert "Failed to close Gemini Interactions stream" in caplog.text + + +@pytest.mark.asyncio +async def test_convert_interaction_stream_propagates_cancellation_and_closes_source() -> None: + stream = AsyncMock() + + async def blocked_events() -> AsyncIterator[InteractionSSEEvent]: + yield _created() + await asyncio.Event().wait() + + stream.__aiter__.side_effect = blocked_events + converted = convert_interaction_stream(stream, model="requested") + await anext(converted) + await anext(converted) + pending = asyncio.create_task(anext(converted)) + await asyncio.sleep(0) + pending.cancel() + + with pytest.raises(asyncio.CancelledError): + await pending + + stream.close.assert_awaited_once_with() + + +@pytest.mark.asyncio +async def test_aresponses_defaults_only_interactions_requests_to_v1() -> None: + with patch("any_llm.providers.gemini.gemini.genai.Client") as client_class: + client = client_class.return_value + client.aio.interactions.create = AsyncMock(return_value=_interaction()) + provider = GeminiProvider(api_key="test-key") + result = await provider.aresponses( + "gemini-3.8-flash", + "Hello", + instructions="Be concise", + timeout=1.5, + ) + + assert isinstance(result, Response) + assert result.output_text == "Hello" + client_class.assert_called_once_with(api_key="test-key") + client.aio.interactions.create.assert_awaited_once_with( + api_version="v1", + model="gemini-3.8-flash", + input="Hello", + system_instruction="Be concise", + timeout=1.5, + ) + + +@pytest.mark.asyncio +async def test_aresponses_preserves_explicit_v1beta_client_configuration() -> None: + with patch("any_llm.providers.gemini.gemini.genai.Client") as client_class: + client = client_class.return_value + client.aio.interactions.create = AsyncMock(return_value=_interaction()) + provider = GeminiProvider(api_key="test-key", http_options={"api_version": "v1beta"}) + await provider.aresponses("gemini-3.8-flash", "Hello") + + assert client_class.call_args.kwargs["http_options"] == {"api_version": "v1beta"} + assert client.aio.interactions.create.await_args.kwargs["api_version"] == "v1beta" + + +@pytest.mark.asyncio +async def test_aresponses_rejects_openai_extra_body() -> None: + with patch("any_llm.providers.gemini.gemini.genai.Client"): + provider = GeminiProvider(api_key="test-key") + with pytest.raises(UnsupportedParameterError, match="extra_body"): + await provider.aresponses( + "gemini-3.8-flash", + "Hello", + extra_body={"future": True}, + ) + + +@pytest.mark.asyncio +async def test_real_sdk_serializes_stable_interactions_path_and_body() -> None: + requests: list[httpx.Request] = [] + + async def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response( + 200, + json={ + "id": "int-123", + "status": "completed", + "model": "gemini-3.8-flash", + "steps": [{"type": "model_output", "content": [{"type": "text", "text": "Hello"}]}], + }, + ) + + http_client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + try: + provider = GeminiProvider( + api_key="test-key", + api_base="https://example.test", + http_options=types.HttpOptions(httpx_async_client=http_client), + ) + response = await provider.aresponses( + "gemini-3.8-flash", + "Hello", + instructions="", + max_output_tokens=0, + ) + finally: + await http_client.aclose() + + assert isinstance(response, Response) + assert response.output_text == "Hello" + assert len(requests) == 1 + assert str(requests[0].url) == "https://example.test/v1/interactions" + assert requests[0].headers["x-goog-api-key"] == "test-key" + assert json.loads(requests[0].content) == { + "input": "Hello", + "model": "gemini-3.8-flash", + "generation_config": {"max_output_tokens": 0}, + "system_instruction": "", + } + + +@pytest.mark.asyncio +async def test_real_sdk_parses_stream_events_and_done_sentinel() -> None: + requests: list[httpx.Request] = [] + event_payloads = [ + { + "event_type": "interaction.created", + "interaction": {"id": "int-123", "status": "in_progress"}, + }, + { + "event_type": "step.start", + "index": 0, + "step": {"type": "model_output", "content": [{"type": "text", "text": ""}]}, + }, + {"event_type": "step.delta", "index": 0, "delta": {"type": "text", "text": "Hello"}}, + {"event_type": "step.stop", "index": 0}, + { + "event_type": "interaction.completed", + "interaction": {"id": "int-123", "status": "completed"}, + }, + ] + body = "".join(f"data: {json.dumps(payload)}\n\n" for payload in event_payloads) + "data: [DONE]\n\n" + + async def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=body) + + http_client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + try: + provider = GeminiProvider( + api_key="test-key", + api_base="https://example.test", + http_options=types.HttpOptions(httpx_async_client=http_client), + ) + response = await provider.aresponses("gemini-3.8-flash", "Hello", stream=True) + assert isinstance(response, AsyncIterator) + events = [event async for event in response] + finally: + await http_client.aclose() + + assert [event.type for event in events] == [ + "response.created", + "response.in_progress", + "response.output_item.added", + "response.content_part.added", + "response.output_text.delta", + "response.output_text.done", + "response.content_part.done", + "response.output_item.done", + "response.completed", + ] + terminal = events[-1] + assert isinstance(terminal, ResponseCompletedEvent) + assert terminal.response.output_text == "Hello" + assert str(requests[0].url) == "https://example.test/v1/interactions" + assert json.loads(requests[0].content)["stream"] is True + + +@pytest.mark.asyncio +async def test_real_sdk_http_error_uses_unified_error_mapping( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("ANY_LLM_UNIFIED_EXCEPTIONS", "1") + + async def handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response( + 400, + headers={"content-type": "application/json"}, + json={"error": {"code": "INVALID_ARGUMENT", "message": "invalid input"}}, + ) + + http_client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + try: + provider = GeminiProvider( + api_key="test-key", + api_base="https://example.test", + http_options=types.HttpOptions(httpx_async_client=http_client), + ) + with pytest.raises(InvalidRequestError) as raised: + await provider.aresponses("gemini-3.8-flash", "Hello") + finally: + await http_client.aclose() + + assert raised.value.status_code == 400 + assert raised.value.code == "INVALID_ARGUMENT" + + +@pytest.mark.asyncio +async def test_real_sdk_timeout_uses_unified_error_mapping( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("ANY_LLM_UNIFIED_EXCEPTIONS", "1") + + async def handler(request: httpx.Request) -> httpx.Response: + message = "read timed out" + raise httpx.ReadTimeout(message, request=request) + + http_client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + try: + provider = GeminiProvider( + api_key="test-key", + api_base="https://example.test", + http_options=types.HttpOptions(httpx_async_client=http_client), + ) + with pytest.raises(ProviderError) as raised: + await provider.aresponses("gemini-3.8-flash", "Hello", timeout=0.1) + finally: + await http_client.aclose() + + assert raised.value.status_code is None + original = raised.value.original_exception + assert original is not None + assert type(original).__name__ == "APITimeoutError" + assert isinstance(original.__cause__, httpx.ReadTimeout) + + +def test_responses_calls_interactions_synchronously() -> None: + with patch("any_llm.providers.gemini.gemini.genai.Client") as client_class: + client_class.return_value.aio.interactions.create = AsyncMock(return_value=_interaction()) + provider = GeminiProvider(api_key="test-key") + response = provider.responses("gemini-3.8-flash", "Hello") + + assert isinstance(response, Response) + assert response.output_text == "Hello"