feat(gemini): add text Interactions responses

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.

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",
 ]
 

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)

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