# Copyright 2026 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # # pyformat: disable """Code generated by Speakeasy (https://speakeasy.com). DO NOT EDIT.""" from . import errors, models, types, utils from ._hooks import AfterParseErrorContext, HookContext, ResponseContext from .basesdk import AsyncBaseSDK, BaseSDK from .types import BaseModel, OptionalNullable, UNSET, interactions from .utils import get_security_from_env, response_helpers from .utils.eventstreaming import AsyncStream, Stream from .utils.unmarshal_json_response import unmarshal_json_response import httpx from typing import Any, Dict, List, Literal, Mapping, Optional, Union, cast, overload class Interactions(BaseSDK): @property def with_raw_response(self): return InteractionsWithRawResponse(self) @property def with_streaming_response(self): return InteractionsWithStreamingResponse(self) @overload def create( self, *, request: Union[ models.CreateInteractionRequest, models.CreateInteractionRequestParam ], extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, extra_body: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> Union[interactions.Interaction, Stream[interactions.InteractionSSEEvent]]: r"""Creating an interaction Creates a new interaction. :param body: The request body. :param api_version: Which version of the API to use. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param extra_body: Additional JSON object fields to merge into request bodies. :param timeout: Override the default request timeout configuration for this method in seconds """ @overload def create( self, *, api_version: Optional[str] = None, model: interactions.Model, stream: Union[Literal[False], None] = None, store: bool = ..., background: bool = ..., system_instruction: str = ..., tools: List[interactions.ToolParam] = ..., response_modalities: List[interactions.ResponseModality] = ..., response_mime_type: str = ..., previous_interaction_id: str = ..., service_tier: interactions.ServiceTier = ..., webhook_config: interactions.WebhookConfigParam = ..., response_format: interactions.CreateModelInteractionResponseFormatParam = ..., environment: interactions.CreateModelInteractionEnvironmentParam = ..., generation_config: interactions.GenerationConfigParam = ..., safety_settings: List[interactions.SafetySettingParam] = ..., labels: Dict[str, str] = ..., input: interactions.InteractionsInputParam, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, extra_body: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> interactions.Interaction: r"""Creating an interaction Creates a new interaction. :param api_version: Which version of the API to use. :param model: The model that will complete your prompt.\n\nSee [models](https://ai.google.dev/gemini-api/docs/models) for additional details. :param stream: Input only. Whether the interaction will be streamed. :param store: Input only. Whether to store the response and request for later retrieval. :param background: Input only. Whether to run the model interaction in the background. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param response_mime_type: The mime type of the response. This is required if response_format is set. :param previous_interaction_id: The ID of the previous interaction, if any. :param service_tier: :param webhook_config: Message for configuring webhook events for a request. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param environment: The environment configuration for the interaction. Can be an object specifying remote environment sources or a string referencing an existing environment ID. :param generation_config: Configuration parameters for model interactions. :param safety_settings: Safety settings for the interaction. :param labels: The labels with user-defined metadata for the request. :param input: The input for the interaction. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param extra_body: Additional JSON object fields to merge into request bodies. :param timeout: Override the default request timeout configuration for this method in seconds """ @overload def create( self, *, api_version: Optional[str] = None, model: interactions.Model, stream: Literal[True], store: bool = ..., background: bool = ..., system_instruction: str = ..., tools: List[interactions.ToolParam] = ..., response_modalities: List[interactions.ResponseModality] = ..., response_mime_type: str = ..., previous_interaction_id: str = ..., service_tier: interactions.ServiceTier = ..., webhook_config: interactions.WebhookConfigParam = ..., response_format: interactions.CreateModelInteractionResponseFormatParam = ..., environment: interactions.CreateModelInteractionEnvironmentParam = ..., generation_config: interactions.GenerationConfigParam = ..., safety_settings: List[interactions.SafetySettingParam] = ..., labels: Dict[str, str] = ..., input: interactions.InteractionsInputParam, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, extra_body: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> Stream[interactions.InteractionSSEEvent]: r"""Creating an interaction Creates a new interaction. :param api_version: Which version of the API to use. :param model: The model that will complete your prompt.\n\nSee [models](https://ai.google.dev/gemini-api/docs/models) for additional details. :param stream: Input only. Whether the interaction will be streamed. :param store: Input only. Whether to store the response and request for later retrieval. :param background: Input only. Whether to run the model interaction in the background. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param response_mime_type: The mime type of the response. This is required if response_format is set. :param previous_interaction_id: The ID of the previous interaction, if any. :param service_tier: :param webhook_config: Message for configuring webhook events for a request. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param environment: The environment configuration for the interaction. Can be an object specifying remote environment sources or a string referencing an existing environment ID. :param generation_config: Configuration parameters for model interactions. :param safety_settings: Safety settings for the interaction. :param labels: The labels with user-defined metadata for the request. :param input: The input for the interaction. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param extra_body: Additional JSON object fields to merge into request bodies. :param timeout: Override the default request timeout configuration for this method in seconds """ @overload def create( self, *, api_version: Optional[str] = None, agent: interactions.AgentOption, stream: Union[Literal[False], None] = None, store: bool = ..., background: bool = ..., system_instruction: str = ..., tools: List[interactions.ToolParam] = ..., response_modalities: List[interactions.ResponseModality] = ..., response_mime_type: str = ..., previous_interaction_id: str = ..., service_tier: interactions.ServiceTier = ..., webhook_config: interactions.WebhookConfigParam = ..., response_format: interactions.CreateAgentInteractionResponseFormatParam = ..., environment: interactions.CreateAgentInteractionEnvironmentParam = ..., agent_config: interactions.CreateAgentInteractionAgentConfigParam = ..., safety_settings: List[interactions.SafetySettingParam] = ..., labels: Dict[str, str] = ..., input: interactions.InteractionsInputParam, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, extra_body: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> interactions.Interaction: r"""Creating an interaction Creates a new interaction. :param api_version: Which version of the API to use. :param agent: The agent to interact with. :param stream: Input only. Whether the interaction will be streamed. :param store: Input only. Whether to store the response and request for later retrieval. :param background: Input only. Whether to run the model interaction in the background. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param response_mime_type: The mime type of the response. This is required if response_format is set. :param previous_interaction_id: The ID of the previous interaction, if any. :param service_tier: :param webhook_config: Message for configuring webhook events for a request. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param environment: The environment configuration for the interaction. Can be an object specifying remote environment sources or a string referencing an existing environment ID. :param agent_config: Configuration parameters for the agent interaction. :param safety_settings: Safety settings for the interaction. :param labels: The labels with user-defined metadata for the request. :param input: The input for the interaction. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param extra_body: Additional JSON object fields to merge into request bodies. :param timeout: Override the default request timeout configuration for this method in seconds """ @overload def create( self, *, api_version: Optional[str] = None, agent: interactions.AgentOption, stream: Literal[True], store: bool = ..., background: bool = ..., system_instruction: str = ..., tools: List[interactions.ToolParam] = ..., response_modalities: List[interactions.ResponseModality] = ..., response_mime_type: str = ..., previous_interaction_id: str = ..., service_tier: interactions.ServiceTier = ..., webhook_config: interactions.WebhookConfigParam = ..., response_format: interactions.CreateAgentInteractionResponseFormatParam = ..., environment: interactions.CreateAgentInteractionEnvironmentParam = ..., agent_config: interactions.CreateAgentInteractionAgentConfigParam = ..., safety_settings: List[interactions.SafetySettingParam] = ..., labels: Dict[str, str] = ..., input: interactions.InteractionsInputParam, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, extra_body: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> Stream[interactions.InteractionSSEEvent]: r"""Creating an interaction Creates a new interaction. :param api_version: Which version of the API to use. :param agent: The agent to interact with. :param stream: Input only. Whether the interaction will be streamed. :param store: Input only. Whether to store the response and request for later retrieval. :param background: Input only. Whether to run the model interaction in the background. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param response_mime_type: The mime type of the response. This is required if response_format is set. :param previous_interaction_id: The ID of the previous interaction, if any. :param service_tier: :param webhook_config: Message for configuring webhook events for a request. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param environment: The environment configuration for the interaction. Can be an object specifying remote environment sources or a string referencing an existing environment ID. :param agent_config: Configuration parameters for the agent interaction. :param safety_settings: Safety settings for the interaction. :param labels: The labels with user-defined metadata for the request. :param input: The input for the interaction. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param extra_body: Additional JSON object fields to merge into request bodies. :param timeout: Override the default request timeout configuration for this method in seconds """ @overload def create( self, *, request: OptionalNullable[ Union[models.CreateInteractionRequest, models.CreateInteractionRequestParam] ] = UNSET, api_version: OptionalNullable[str] = UNSET, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, extra_body: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, **body_kwargs: Any, ) -> Union[interactions.Interaction, Stream[interactions.InteractionSSEEvent]]: r"""Creating an interaction Creates a new interaction. :param api_version: Which version of the API to use. :param model: The model that will complete your prompt.\n\nSee [models](https://ai.google.dev/gemini-api/docs/models) for additional details. :param stream: Input only. Whether the interaction will be streamed. :param store: Input only. Whether to store the response and request for later retrieval. :param background: Input only. Whether to run the model interaction in the background. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param response_mime_type: The mime type of the response. This is required if response_format is set. :param previous_interaction_id: The ID of the previous interaction, if any. :param service_tier: :param webhook_config: Message for configuring webhook events for a request. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param environment: The environment configuration for the interaction. Can be an object specifying remote environment sources or a string referencing an existing environment ID. :param generation_config: Configuration parameters for model interactions. :param safety_settings: Safety settings for the interaction. :param labels: The labels with user-defined metadata for the request. :param input: The input for the interaction. :param agent: The agent to interact with. :param agent_config: Configuration parameters for the agent interaction. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param extra_body: Additional JSON object fields to merge into request bodies. :param timeout: Override the default request timeout configuration for this method in seconds """ def create( self, *, request: OptionalNullable[ Union[models.CreateInteractionRequest, models.CreateInteractionRequestParam] ] = UNSET, api_version: OptionalNullable[str] = UNSET, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, extra_body: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, **body_kwargs: Any, ) -> Union[interactions.Interaction, Stream[interactions.InteractionSSEEvent]]: r"""Creating an interaction Creates a new interaction. :param api_version: Which version of the API to use. :param model: The model that will complete your prompt.\n\nSee [models](https://ai.google.dev/gemini-api/docs/models) for additional details. :param stream: Input only. Whether the interaction will be streamed. :param store: Input only. Whether to store the response and request for later retrieval. :param background: Input only. Whether to run the model interaction in the background. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param response_mime_type: The mime type of the response. This is required if response_format is set. :param previous_interaction_id: The ID of the previous interaction, if any. :param service_tier: :param webhook_config: Message for configuring webhook events for a request. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param environment: The environment configuration for the interaction. Can be an object specifying remote environment sources or a string referencing an existing environment ID. :param generation_config: Configuration parameters for model interactions. :param safety_settings: Safety settings for the interaction. :param labels: The labels with user-defined metadata for the request. :param input: The input for the interaction. :param agent: The agent to interact with. :param agent_config: Configuration parameters for the agent interaction. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param extra_body: Additional JSON object fields to merge into request bodies. :param timeout: Override the default request timeout configuration for this method in seconds """ if "model" in body_kwargs and "agent" in body_kwargs: raise ValueError("Cannot supply both 'model' and 'agent'.") if "model" in body_kwargs and "agent_config" in body_kwargs: raise ValueError("Cannot supply both 'model' and 'agent_config'.") if "generation_config" in body_kwargs and "agent" in body_kwargs: raise ValueError("Cannot supply both 'generation_config' and 'agent'.") if "generation_config" in body_kwargs and "agent_config" in body_kwargs: raise ValueError( "Cannot supply both 'generation_config' and 'agent_config'." ) base_url = None url_variables = None retries: OptionalNullable[utils.RetryConfig] = UNSET server_url = None http_headers = extra_headers timeout_ms = self._coerce_timeout_ms(timeout) if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) if request is not UNSET: request = cast( models.CreateInteractionRequest, request if isinstance(request, BaseModel) else utils.unmarshal( cast(Any, request), models.CreateInteractionRequest ), ) else: _body_kwargs = dict(body_kwargs) _body_kwargs = {k: v for k, v in _body_kwargs.items() if v is not UNSET} _request_kwargs: dict[str, Any] = {"api_version": api_version} _request_kwargs = { k: v for k, v in _request_kwargs.items() if v is not UNSET } _request_kwargs["body"] = _body_kwargs request = cast( models.CreateInteractionRequest, utils.unmarshal(_request_kwargs, models.CreateInteractionRequest), ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request( method="POST", path="/{api_version}/interactions", base_url=base_url, url_variables=url_variables, request=request, request_body_required=True, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="text/event-stream" if getattr(getattr(request, "body", None), "stream", False) is True else "application/json", http_headers=http_headers, extra_query_params=extra_query, _globals=models.CreateInteractionGlobals( api_version=self.sdk_configuration.globals.api_version, ), security=self.sdk_configuration.security, get_serialized_body=lambda: utils.serialize_request_body( request.body, False, False, "json", models.CreateInteractionRequestBody, extra_body=extra_body, ), allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config else: retries = utils.RetryConfig( "attempt-count-backoff", utils.BackoffStrategy(500, 8000, 2, 30000), True, max_retries=4, ) retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["408", "409", "429", "5XX"]) def _speakeasy_parse_response(http_res): response_data: Any = None if utils.match_response(http_res, "200", "application/json"): http_res_text = utils.stream_to_text(http_res) return unmarshal_json_response( interactions.Interaction, http_res, http_res_text, validate=False ) if utils.match_response(http_res, "200", "text/event-stream"): return Stream( http_res, lambda raw: unmarshal_json_response( interactions.InteractionSSEStreamEvent, http_res, raw, validate=False, ).data, sentinel="[DONE]", client_ref=self, ) if utils.match_response(http_res, "4XX", "application/json"): http_res_text = utils.stream_to_text(http_res) try: response_data = unmarshal_json_response( errors.CreateInteractionClientErrorData, http_res, http_res_text ) raise errors.CreateInteractionClientError( response_data, http_res, http_res_text ) except errors.ResponseValidationError as e: raise errors.GenAiDefaultError( "Error response body did not match expected schema", http_res, http_res_text, ) from e if utils.match_response(http_res, "5XX", "application/json"): http_res_text = utils.stream_to_text(http_res) try: response_data = unmarshal_json_response( errors.CreateInteractionServerErrorData, http_res, http_res_text ) raise errors.CreateInteractionServerError( response_data, http_res, http_res_text ) except errors.ResponseValidationError as e: raise errors.GenAiDefaultError( "Error response body did not match expected schema", http_res, http_res_text, ) from e http_res_text = utils.stream_to_text(http_res) raise errors.GenAiDefaultError( "Unexpected response received", http_res, http_res_text ) _speakeasy_hook_ctx = HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="CreateInteraction", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, types.Security ), tags=None, extensions={ "x-codeSamples": [ { "label": "simple", "lang": "sh", "source": 'curl -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "model": "gemini-3.6-flash",\n "input": "Hello, how are you?"\n }\'\n', }, { "label": "simple", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\ninteraction = client.interactions.create(\n model="gemini-3.6-flash",\n input="Hello, how are you?",\n)\nprint(interaction.output_text)\n', }, { "label": "simple", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\nconst interaction = await ai.interactions.create({\n model: 'gemini-3.6-flash',\n input: 'Hello, how are you?',\n});\nconsole.log(interaction.output_text);\n", }, { "label": "multi_turn", "lang": "sh", "source": 'curl -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "model": "gemini-3.6-flash",\n "input": [\n { "type": "user_input", "content": [{ "type": "text", "text": "Hello!" }] },\n { "type": "model_output", "content": [{ "type": "text", "text": "Hi there! How can I help you today?" }] },\n { "type": "user_input", "content": [{ "type": "text", "text": "What is the capital of France?" }] }\n ]\n }\'\n', }, { "label": "multi_turn", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\nresponse = client.interactions.create(\n model="gemini-3.6-flash",\n input=[\n { "type": "user_input", "content": [{ "type": "text", "text": "Hello!" }] },\n { "type": "model_output", "content": [{ "type": "text", "text": "Hi there! How can I help you today?" }] },\n { "type": "user_input", "content": [{ "type": "text", "text": "What is the capital of France?" }] }\n ]\n)\nprint(response.output_text)\n', }, { "label": "multi_turn", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\nconst interaction = await ai.interactions.create({\n model: 'gemini-3.6-flash',\n input: [\n { type: 'user_input', content: [{ type: 'text', text: 'Hello' }] },\n { type: 'model_output', content: [{ type: 'text', text: 'Hi there! How can I help you today?' }] },\n { type: 'user_input', content: [{ type: 'text', text: 'What is the capital of France?' }] }\n ]\n});\nconsole.log(interaction.output_text);\n", }, { "label": "multimodal_image", "lang": "sh", "source": 'curl -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "model": "gemini-3.6-flash",\n "input": [\n {\n "type": "text",\n "text": "What is in this picture?"\n },\n {\n "type": "image",\n "data": "BASE64_ENCODED_IMAGE",\n "mime_type": "image/png"\n }\n ]\n }\'\n', }, { "label": "multimodal_image", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\nresponse = client.interactions.create(\n model="gemini-3.6-flash",\n input=[\n { "type": "text", "text": "What is in this picture?" },\n { "type": "image", "data": "BASE64_ENCODED_IMAGE", "mime_type": "image/png" }\n ]\n)\nprint(response.output_text)\n', }, { "label": "multimodal_image", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\nconst interaction = await ai.interactions.create({\n model: 'gemini-3.6-flash',\n input: [\n { type: 'text', text: 'What is in this picture?' },\n { type: 'image', data: 'BASE64_ENCODED_IMAGE', mime_type: 'image/png' }\n ]\n});\nconsole.log(interaction.output_text);\n", }, { "label": "function_calling", "lang": "sh", "source": 'curl -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "model": "gemini-3.6-flash",\n "tools": [\n {\n "type": "function",\n "name": "get_weather",\n "description": "Get the current weather in a given location",\n "parameters": {\n "type": "object",\n "properties": {\n "location": {\n "type": "string",\n "description": "The city and state, e.g. San Francisco, CA"\n }\n },\n "required": [\n "location"\n ]\n }\n }\n ],\n "input": "What is the weather like in Boston, MA?"\n }\'\n', }, { "label": "function_calling", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\nresponse = client.interactions.create(\n model="gemini-3.6-flash",\n tools=[{\n "type": "function",\n "name": "get_weather",\n "description": "Get the current weather in a given location",\n "parameters": {\n "type": "object",\n "properties": {\n "location": {\n "type": "string",\n "description": "The city and state, e.g. San Francisco, CA"\n }\n },\n "required": ["location"]\n }\n }],\n input="What is the weather like in Boston, MA?"\n)\nprint(response.steps[-1])\n', }, { "label": "function_calling", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\nconst interaction = await ai.interactions.create({\n model: 'gemini-3.6-flash',\n tools: [{\n type: 'function',\n name: 'get_weather',\n description: 'Get the current weather in a given location',\n parameters: {\n type: 'object',\n properties: {\n location: {\n type: 'string',\n description: 'The city and state, e.g. San Francisco, CA'\n }\n },\n required: ['location']\n }\n }],\n input: 'What is the weather like in Boston, MA?'\n});\nconsole.log(interaction.steps.at(-1));\n", }, { "label": "deep_research", "lang": "sh", "source": 'curl -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "agent": "deep-research-pro-preview-12-2025",\n "input": "Find a cure to cancer",\n "background": true\n }\'\n', }, { "label": "deep_research", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\ninteraction = client.interactions.create(\n agent="deep-research-pro-preview-12-2025",\n input="find a cure to cancer",\n background=True,\n)\nprint(interaction.status)\n', }, { "label": "deep_research", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\nconst interaction = await ai.interactions.create({\n agent: 'deep-research-pro-preview-12-2025',\n input: 'find a cure to cancer',\n background: true,\n});\nconsole.log(interaction.status);\n", }, { "label": "antigravity", "lang": "sh", "source": 'curl -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "agent": "antigravity-preview-05-2026",\n "input": "Read Hacker News, summarize the top 5 stories, and save results as a markdown file.",\n "environment": "remote"\n }\'\n', }, { "label": "antigravity", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\ninteraction = client.interactions.create(\n agent="antigravity-preview-05-2026",\n input="Read Hacker News, summarize the top 5 stories, and save results as a markdown file.",\n environment="remote",\n)\nprint(interaction.output_text)\n', }, { "label": "antigravity", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\nconst interaction = await ai.interactions.create({\n agent: 'antigravity-preview-05-2026',\n input: 'Read Hacker News, summarize the top 5 stories, and save results as a markdown file.',\n environment: 'remote',\n});\nconsole.log(interaction.output_text);\n", }, { "label": "reuse_env", "lang": "sh", "source": '# Step 1: Create an interaction with a fresh remote environment.\nRESPONSE=$(curl -s -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -H "Api-Revision: 2026-05-20" \\\n -d \'{\n "agent": "antigravity-preview-05-2026",\n "input": "Write a hello world script at /workspace/hello.py.",\n "environment": "remote"\n }\')\nINTERACTION_ID=$(echo $RESPONSE | python3 -c "import sys,json; print(json.load(sys.stdin)[\'id\'])")\nENV_ID=$(echo $RESPONSE | python3 -c "import sys,json; print(json.load(sys.stdin)[\'environment_id\'])")\n\n# Step 2: Reuse the same environment in a follow-up interaction.\ncurl -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d "{\n \\"agent\\": \\"antigravity-preview-05-2026\\",\n \\"input\\": \\"Modify the script to accept a name argument and greet the user.\\",\n \\"environment\\": \\"$ENV_ID\\",\n \\"previous_interaction_id\\": \\"$INTERACTION_ID\\"\n }"\n', }, { "label": "reuse_env", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\n\n# Step 1: Create an interaction with a fresh remote environment.\ninteraction = client.interactions.create(\n agent="antigravity-preview-05-2026",\n input="Write a hello world script at /workspace/hello.py.",\n environment="remote",\n)\nprint(f"Environment ID: {interaction.environment_id}")\n\n# Step 2: Reuse the same environment in a follow-up interaction.\ninteraction_2 = client.interactions.create(\n agent="antigravity-preview-05-2026",\n input="Modify the script to accept a name argument and greet the user.",\n environment=interaction.environment_id,\n previous_interaction_id=interaction.id,\n)\nprint(interaction_2.output_text)\n', }, { "label": "reuse_env", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\n\n// Step 1: Create an interaction with a fresh remote environment.\nconst interaction = await ai.interactions.create({\n agent: 'antigravity-preview-05-2026',\n input: 'Write a hello world script at /workspace/hello.py.',\n environment: 'remote',\n});\nconsole.log(`Environment ID: ${interaction.environment_id}`);\n\n// Step 2: Reuse the same environment in a follow-up interaction.\nconst interaction2 = await ai.interactions.create({\n agent: 'antigravity-preview-05-2026',\n input: 'Modify the script to accept a name argument and greet the user.',\n environment: interaction.environment_id,\n previous_interaction_id: interaction.id,\n});\nconsole.log(interaction2.output_text);\n", }, { "label": "with_sources", "lang": "sh", "source": 'curl -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "agent": "antigravity-preview-05-2026",\n "input": "List all files under /workspace and summarize what you find.",\n "environment": {\n "type": "remote",\n "sources": [\n {\n "type": "repository",\n "source": "https://github.com/octocat/Spoon-Knife",\n "target": "/workspace/repo"\n },\n {\n "type": "inline",\n "content": "Focus on Python files only.",\n "target": "/workspace/notes.txt"\n }\n ]\n }\n }\'\n', }, { "label": "with_sources", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\ninteraction = client.interactions.create(\n agent="antigravity-preview-05-2026",\n input="List all files under /workspace and summarize what you find.",\n environment={\n "type": "remote",\n "sources": [\n {\n "type": "repository",\n "source": "https://github.com/octocat/Spoon-Knife",\n "target": "/workspace/repo",\n },\n {\n "type": "inline",\n "content": "Focus on Python files only.",\n "target": "/workspace/notes.txt",\n },\n ],\n },\n)\nprint(interaction.output_text)\n', }, { "label": "with_sources", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\nconst interaction = await ai.interactions.create({\n agent: 'antigravity-preview-05-2026',\n input: 'List all files under /workspace and summarize what you find.',\n environment: {\n type: 'remote',\n sources: [\n {\n type: 'repository',\n source: 'https://github.com/octocat/Spoon-Knife',\n target: '/workspace/repo',\n },\n {\n type: 'inline',\n content: 'Focus on Python files only.',\n target: '/workspace/notes.txt',\n },\n ],\n },\n});\nconsole.log(interaction.output_text);\n", }, { "label": "custom_agent", "lang": "sh", "source": '# Step 1: Create a custom agent.\ncurl -X POST https://generativelanguage.googleapis.com/v1beta/agents \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "id": "code-reviewer",\n "base_agent": "antigravity-preview-05-2026",\n "system_instruction": "You are a senior code reviewer. Check every file for bugs, style issues, and security vulnerabilities.",\n "base_environment": {\n "type": "remote",\n "sources": [{\n "type": "repository",\n "source": "https://github.com/octocat/Spoon-Knife",\n "target": "/workspace/repo"\n }]\n }\n }\'\n\n# Step 2: Use the custom agent.\ncurl -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "agent": "code-reviewer",\n "input": "Review the latest changes in /workspace/repo/src and file a summary.",\n "environment": "remote"\n }\'\n', }, { "label": "custom_agent", "lang": "python", "source": 'import uuid\nfrom google import genai\n\nclient = genai.Client()\n\n# Step 1: Create a custom agent.\nagent_id = f"code-reviewer-{uuid.uuid4().hex[:8]}"\nclient.agents.create(\n id=agent_id,\n base_agent="antigravity-preview-05-2026",\n system_instruction="You are a senior code reviewer. Check every file for bugs, style issues, and security vulnerabilities.",\n base_environment={\n "type": "remote",\n "sources": [{\n "type": "repository",\n "source": "https://github.com/octocat/Spoon-Knife",\n "target": "/workspace/repo",\n }],\n },\n)\n\n# Step 2: Use the custom agent.\nresult = client.interactions.create(\n agent=agent_id,\n input="Review the latest changes in /workspace/repo/src and file a summary.",\n environment="remote",\n)\nprint(result.output_text)\n\n# [cleanup]\nclient.agents.delete(agent_id)\n# [/cleanup]\n', }, { "label": "custom_agent", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\n\n// Step 1: Create a custom agent.\nconst agentId = `code-reviewer-${crypto.randomUUID().slice(0, 8)}`;\nawait ai.agents.create({\n id: agentId,\n base_agent: 'antigravity-preview-05-2026',\n system_instruction: 'You are a senior code reviewer. Check every file for bugs, style issues, and security vulnerabilities.',\n base_environment: {\n type: 'remote',\n sources: [{\n type: 'repository',\n source: 'https://github.com/octocat/Spoon-Knife',\n target: '/workspace/repo',\n }],\n },\n});\n\n// Step 2: Use the custom agent.\nconst result = await ai.interactions.create({\n agent: agentId,\n input: 'Review the latest changes in /workspace/repo/src and file a summary.',\n environment: 'remote',\n});\nconsole.log(result.output_text);\n\n// [cleanup]\nawait ai.agents.delete(agentId);\n// [/cleanup]\n", }, ] }, response=ResponseContext(mode=_speakeasy_response_mode, execution="sync"), ) http_res = self.do_request( hook_ctx=_speakeasy_hook_ctx, request=req, is_error_status_code=lambda c: utils.match_status_codes(["4XX", "5XX"], c), stream=getattr(getattr(request, "body", None), "stream", False) is True or _speakeasy_response_mode == "streaming", retry_config=retry_config, ) if _speakeasy_response_mode != "parsed": if utils.match_status_codes(["4XX", "5XX"], http_res.status_code): http_res.read() try: _speakeasy_parse_response(http_res) except Exception as parse_exc_: response_helpers.raise_parse_error( self.sdk_configuration.__dict__["_hooks"], AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) _speakeasy_response_cls = ( response_helpers.StreamedAPIResponse if _speakeasy_response_mode == "streaming" else response_helpers.APIResponse ) return cast( Any, _speakeasy_response_cls( raw=http_res, parser=_speakeasy_parse_response, mode=( "event_stream" if getattr(getattr(request, "body", None), "stream", False) is True else "buffered" ), client_ref=self, hook_ctx=AfterParseErrorContext(_speakeasy_hook_ctx), hooks=self.sdk_configuration.__dict__.get("_hooks"), ), ) try: return _speakeasy_parse_response(http_res) except Exception as parse_exc_: response_helpers.raise_parse_error( self.sdk_configuration.__dict__["_hooks"], AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) @overload def get( self, id: str, *, stream: Union[Literal[False], None] = None, last_event_id: Optional[str] = None, include_input: Optional[bool] = False, api_version: Optional[str] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> interactions.Interaction: r"""Retrieving an interaction Retrieves the full details of a single interaction based on its `Interaction.id`. :param id: The unique identifier of the interaction to retrieve. :param stream: If set to true, the generated content will be streamed incrementally. :param last_event_id: Optional. If set, resumes the interaction stream from the next chunk after the event marked by the event id. Can only be used if `stream` is true. :param include_input: If set to true, includes the input in the response. :param api_version: Which version of the API to use. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param timeout: Override the default request timeout configuration for this method in seconds """ @overload def get( self, id: str, *, stream: Literal[True], last_event_id: Optional[str] = None, include_input: Optional[bool] = False, api_version: Optional[str] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> Stream[interactions.InteractionSSEEvent]: r"""Retrieving an interaction Retrieves the full details of a single interaction based on its `Interaction.id`. :param id: The unique identifier of the interaction to retrieve. :param stream: If set to true, the generated content will be streamed incrementally. :param last_event_id: Optional. If set, resumes the interaction stream from the next chunk after the event marked by the event id. Can only be used if `stream` is true. :param include_input: If set to true, includes the input in the response. :param api_version: Which version of the API to use. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param timeout: Override the default request timeout configuration for this method in seconds """ @overload def get( self, id: str, *, stream: bool, last_event_id: Optional[str] = None, include_input: Optional[bool] = False, api_version: Optional[str] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> Union[interactions.Interaction, Stream[interactions.InteractionSSEEvent]]: r"""Retrieving an interaction Retrieves the full details of a single interaction based on its `Interaction.id`. :param id: The unique identifier of the interaction to retrieve. :param stream: If set to true, the generated content will be streamed incrementally. :param last_event_id: Optional. If set, resumes the interaction stream from the next chunk after the event marked by the event id. Can only be used if `stream` is true. :param include_input: If set to true, includes the input in the response. :param api_version: Which version of the API to use. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param timeout: Override the default request timeout configuration for this method in seconds """ def get( self, id: str, *, stream: Optional[bool] = False, last_event_id: Optional[str] = None, include_input: Optional[bool] = False, api_version: Optional[str] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> Union[interactions.Interaction, Stream[interactions.InteractionSSEEvent]]: r"""Retrieving an interaction Retrieves the full details of a single interaction based on its `Interaction.id`. :param id: The unique identifier of the interaction to retrieve. :param stream: If set to true, the generated content will be streamed incrementally. :param last_event_id: Optional. If set, resumes the interaction stream from the next chunk after the event marked by the event id. Can only be used if `stream` is true. :param include_input: If set to true, includes the input in the response. :param api_version: Which version of the API to use. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param timeout: Override the default request timeout configuration for this method in seconds """ base_url = None url_variables = None retries: OptionalNullable[utils.RetryConfig] = UNSET server_url = None http_headers = extra_headers timeout_ms = self._coerce_timeout_ms(timeout) if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.GetInteractionByIDRequest( id=id, stream=stream, last_event_id=last_event_id, include_input=include_input, api_version=api_version, ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request( method="GET", path="/{api_version}/interactions/{id}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="text/event-stream" if stream is True else "application/json", http_headers=http_headers, extra_query_params=extra_query, _globals=models.GetInteractionByIDGlobals( api_version=self.sdk_configuration.globals.api_version, ), security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config else: retries = utils.RetryConfig( "attempt-count-backoff", utils.BackoffStrategy(500, 8000, 2, 30000), True, max_retries=4, ) retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["408", "409", "429", "5XX"]) def _speakeasy_parse_response(http_res): response_data: Any = None if utils.match_response(http_res, "200", "application/json"): http_res_text = utils.stream_to_text(http_res) return unmarshal_json_response( interactions.Interaction, http_res, http_res_text, validate=False ) if utils.match_response(http_res, "200", "text/event-stream"): return Stream( http_res, lambda raw: unmarshal_json_response( interactions.InteractionSSEStreamEvent, http_res, raw, validate=False, ).data, sentinel="[DONE]", client_ref=self, ) if utils.match_response(http_res, "4XX", "application/json"): http_res_text = utils.stream_to_text(http_res) try: response_data = unmarshal_json_response( errors.GetInteractionByIDClientErrorData, http_res, http_res_text, ) raise errors.GetInteractionByIDClientError( response_data, http_res, http_res_text ) except errors.ResponseValidationError as e: raise errors.GenAiDefaultError( "Error response body did not match expected schema", http_res, http_res_text, ) from e if utils.match_response(http_res, "5XX", "application/json"): http_res_text = utils.stream_to_text(http_res) try: response_data = unmarshal_json_response( errors.GetInteractionByIDServerErrorData, http_res, http_res_text, ) raise errors.GetInteractionByIDServerError( response_data, http_res, http_res_text ) except errors.ResponseValidationError as e: raise errors.GenAiDefaultError( "Error response body did not match expected schema", http_res, http_res_text, ) from e http_res_text = utils.stream_to_text(http_res) raise errors.GenAiDefaultError( "Unexpected response received", http_res, http_res_text ) _speakeasy_hook_ctx = HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="getInteractionById", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, types.Security ), tags=None, extensions={ "x-codeSamples": [ { "label": "get", "lang": "sh", "source": '# [setup]\nINTERACTION_ID=$(curl -s -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -H "Api-Revision: 2026-05-20" \\\n -d \'{"model": "gemini-3.6-flash", "input": "Say hello."}\' \\\n | python3 -c "import sys,json; print(json.load(sys.stdin)[\'id\'])")\n# [/setup]\n\ncurl -X GET "https://generativelanguage.googleapis.com/v1beta/interactions/$INTERACTION_ID" \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Api-Revision: 2026-05-20"\n', }, { "label": "get", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\n\n# [setup]\ncreated = client.interactions.create(\n model="gemini-3.6-flash",\n input="Say hello."\n)\n# [/setup]\n\ninteraction = client.interactions.get(id=created.id)\nprint(interaction.status)\n', }, { "label": "get", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\n\n// [setup]\nconst created = await ai.interactions.create({\n model: 'gemini-3.6-flash',\n input: 'Say hello.'\n});\n// [/setup]\n\nconst interaction = await ai.interactions.get(created.id);\nconsole.log(interaction.status);\n", }, ] }, response=ResponseContext(mode=_speakeasy_response_mode, execution="sync"), ) http_res = self.do_request( hook_ctx=_speakeasy_hook_ctx, request=req, is_error_status_code=lambda c: utils.match_status_codes(["4XX", "5XX"], c), stream=stream is True or _speakeasy_response_mode == "streaming", retry_config=retry_config, ) if _speakeasy_response_mode != "parsed": if utils.match_status_codes(["4XX", "5XX"], http_res.status_code): http_res.read() try: _speakeasy_parse_response(http_res) except Exception as parse_exc_: response_helpers.raise_parse_error( self.sdk_configuration.__dict__["_hooks"], AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) _speakeasy_response_cls = ( response_helpers.StreamedAPIResponse if _speakeasy_response_mode == "streaming" else response_helpers.APIResponse ) return cast( Any, _speakeasy_response_cls( raw=http_res, parser=_speakeasy_parse_response, mode=("event_stream" if stream is True else "buffered"), client_ref=self, hook_ctx=AfterParseErrorContext(_speakeasy_hook_ctx), hooks=self.sdk_configuration.__dict__.get("_hooks"), ), ) try: return _speakeasy_parse_response(http_res) except Exception as parse_exc_: response_helpers.raise_parse_error( self.sdk_configuration.__dict__["_hooks"], AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) def delete( self, id: str, *, api_version: Optional[str] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ): r"""Deleting an interaction Deletes the interaction by id. :param id: The unique identifier of the interaction to delete. :param api_version: Which version of the API to use. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param timeout: Override the default request timeout configuration for this method in seconds """ base_url = None url_variables = None retries: OptionalNullable[utils.RetryConfig] = UNSET server_url = None http_headers = extra_headers timeout_ms = self._coerce_timeout_ms(timeout) if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.DeleteInteractionRequest( id=id, api_version=api_version, ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request( method="DELETE", path="/{api_version}/interactions/{id}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, extra_query_params=extra_query, _globals=models.DeleteInteractionGlobals( api_version=self.sdk_configuration.globals.api_version, ), security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config else: retries = utils.RetryConfig( "attempt-count-backoff", utils.BackoffStrategy(500, 8000, 2, 30000), True, max_retries=4, ) retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["408", "409", "429", "5XX"]) def _speakeasy_parse_response(http_res): response_data: Any = None if utils.match_response(http_res, "200", "*"): return if utils.match_response(http_res, "4XX", "application/json"): try: response_data = unmarshal_json_response( errors.DeleteInteractionClientErrorData, http_res ) raise errors.DeleteInteractionClientError(response_data, http_res) except errors.ResponseValidationError as e: raise errors.GenAiDefaultError( "Error response body did not match expected schema", http_res, http_res.text, ) from e if utils.match_response(http_res, "5XX", "application/json"): try: response_data = unmarshal_json_response( errors.DeleteInteractionServerErrorData, http_res ) raise errors.DeleteInteractionServerError(response_data, http_res) except errors.ResponseValidationError as e: raise errors.GenAiDefaultError( "Error response body did not match expected schema", http_res, http_res.text, ) from e raise errors.GenAiDefaultError("Unexpected response received", http_res) _speakeasy_hook_ctx = HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="deleteInteraction", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, types.Security ), tags=None, extensions={ "x-codeSamples": [ { "label": "delete", "lang": "sh", "source": '# [setup]\nINTERACTION_ID=$(curl -s -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\ -d \'{"model": "gemini-3.6-flash", "input": "Hello"}\' \\\n | python3 -c "import sys,json; print(json.load(sys.stdin)[\'id\'])")\n# [/setup]\n\ncurl -X DELETE "https://generativelanguage.googleapis.com/v1beta/interactions/$INTERACTION_ID" \\\n -H "x-goog-api-key: $GEMINI_API_KEY"\n', }, { "label": "delete", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\n\n# [setup]\ncreated = client.interactions.create(\n model="gemini-3.6-flash",\n input="Hello",\n)\n# [/setup]\n\nclient.interactions.delete(id=created.id)\nprint("Interaction deleted successfully.")\n', }, { "label": "delete", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\n\n// [setup]\nconst created = await ai.interactions.create({\n model: 'gemini-3.6-flash',\n input: 'Hello',\n});\n// [/setup]\n\nawait ai.interactions.delete(created.id);\nconsole.log('Interaction deleted successfully.');\n", }, ] }, response=ResponseContext(mode=_speakeasy_response_mode, execution="sync"), ) http_res = self.do_request( hook_ctx=_speakeasy_hook_ctx, request=req, is_error_status_code=lambda c: utils.match_status_codes(["4XX", "5XX"], c), stream=_speakeasy_response_mode == "streaming", retry_config=retry_config, ) if _speakeasy_response_mode != "parsed": if utils.match_status_codes(["4XX", "5XX"], http_res.status_code): http_res.read() try: _speakeasy_parse_response(http_res) except Exception as parse_exc_: response_helpers.raise_parse_error( self.sdk_configuration.__dict__["_hooks"], AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) _speakeasy_response_cls = ( response_helpers.StreamedAPIResponse if _speakeasy_response_mode == "streaming" else response_helpers.APIResponse ) return cast( Any, _speakeasy_response_cls( raw=http_res, parser=_speakeasy_parse_response, mode="buffered", client_ref=self, hook_ctx=AfterParseErrorContext(_speakeasy_hook_ctx), hooks=self.sdk_configuration.__dict__.get("_hooks"), ), ) try: return _speakeasy_parse_response(http_res) except Exception as parse_exc_: response_helpers.raise_parse_error( self.sdk_configuration.__dict__["_hooks"], AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) def cancel( self, id: str, *, api_version: Optional[str] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> interactions.Interaction: r"""Canceling an interaction Cancels an interaction by id. This only applies to background interactions that are still running. :param id: The unique identifier of the interaction to cancel. :param api_version: Which version of the API to use. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param timeout: Override the default request timeout configuration for this method in seconds """ base_url = None url_variables = None retries: OptionalNullable[utils.RetryConfig] = UNSET server_url = None http_headers = extra_headers timeout_ms = self._coerce_timeout_ms(timeout) if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.CancelInteractionByIDRequest( id=id, api_version=api_version, ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request( method="POST", path="/{api_version}/interactions/{id}/cancel", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, extra_query_params=extra_query, _globals=models.CancelInteractionByIDGlobals( api_version=self.sdk_configuration.globals.api_version, ), security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config else: retries = utils.RetryConfig( "attempt-count-backoff", utils.BackoffStrategy(500, 8000, 2, 30000), True, max_retries=4, ) retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["408", "409", "429", "5XX"]) def _speakeasy_parse_response(http_res): response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response( interactions.Interaction, http_res, validate=False ) if utils.match_response(http_res, "4XX", "application/json"): try: response_data = unmarshal_json_response( errors.CancelInteractionByIDClientErrorData, http_res ) raise errors.CancelInteractionByIDClientError( response_data, http_res ) except errors.ResponseValidationError as e: raise errors.GenAiDefaultError( "Error response body did not match expected schema", http_res, http_res.text, ) from e if utils.match_response(http_res, "5XX", "application/json"): try: response_data = unmarshal_json_response( errors.CancelInteractionByIDServerErrorData, http_res ) raise errors.CancelInteractionByIDServerError( response_data, http_res ) except errors.ResponseValidationError as e: raise errors.GenAiDefaultError( "Error response body did not match expected schema", http_res, http_res.text, ) from e raise errors.GenAiDefaultError("Unexpected response received", http_res) _speakeasy_hook_ctx = HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="cancelInteractionById", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, types.Security ), tags=None, extensions={ "x-codeSamples": [ { "label": "cancel", "lang": "sh", "source": '# [setup]\nINTERACTION_ID=$(curl -s -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\ -d \'{"model": "gemini-3.6-flash", "input": "Write a long essay about the history of computing.", "background": true}\' \\\n | python3 -c "import sys,json; print(json.load(sys.stdin)[\'id\'])")\n# [/setup]\n\ncurl -X POST "https://generativelanguage.googleapis.com/v1beta/interactions/$INTERACTION_ID/cancel" \\\n -H "x-goog-api-key: $GEMINI_API_KEY"\n', }, { "label": "cancel", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\n\n# Start a background interaction so it stays in-progress.\ncreated = client.interactions.create(\n model="gemini-3.6-flash",\n input="Write a long essay about the history of computing.",\n tools=[{"type": "computer_use"}],\n background=True,\n)\n\n# Cancel the in-progress interaction.\ninteraction = client.interactions.cancel(id=created.id)\nprint(interaction.status)\n', }, { "label": "cancel", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\n\n// Start a background interaction so it stays in-progress.\nconst created = await ai.interactions.create({\n model: 'gemini-3.6-flash',\n input: 'Write a long essay about the history of computing.',\n tools: [{ type: 'computer_use' }],\n background: true,\n});\n\n// Cancel the in-progress interaction.\nconst interaction = await ai.interactions.cancel(created.id);\nconsole.log(interaction.status);\n", }, ] }, response=ResponseContext(mode=_speakeasy_response_mode, execution="sync"), ) http_res = self.do_request( hook_ctx=_speakeasy_hook_ctx, request=req, is_error_status_code=lambda c: utils.match_status_codes(["4XX", "5XX"], c), stream=_speakeasy_response_mode == "streaming", retry_config=retry_config, ) if _speakeasy_response_mode != "parsed": if utils.match_status_codes(["4XX", "5XX"], http_res.status_code): http_res.read() try: _speakeasy_parse_response(http_res) except Exception as parse_exc_: response_helpers.raise_parse_error( self.sdk_configuration.__dict__["_hooks"], AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) _speakeasy_response_cls = ( response_helpers.StreamedAPIResponse if _speakeasy_response_mode == "streaming" else response_helpers.APIResponse ) return cast( Any, _speakeasy_response_cls( raw=http_res, parser=_speakeasy_parse_response, mode="buffered", client_ref=self, hook_ctx=AfterParseErrorContext(_speakeasy_hook_ctx), hooks=self.sdk_configuration.__dict__.get("_hooks"), ), ) try: return _speakeasy_parse_response(http_res) except Exception as parse_exc_: response_helpers.raise_parse_error( self.sdk_configuration.__dict__["_hooks"], AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) class InteractionsWithRawResponse: def __init__(self, sdk: Interactions) -> None: self._sdk = sdk self.create = response_helpers.to_raw_response_wrapper( sdk.create, "extra_headers" ) self.get = response_helpers.to_raw_response_wrapper(sdk.get, "extra_headers") self.delete = response_helpers.to_raw_response_wrapper( sdk.delete, "extra_headers" ) self.cancel = response_helpers.to_raw_response_wrapper( sdk.cancel, "extra_headers" ) class InteractionsWithStreamingResponse: def __init__(self, sdk: Interactions) -> None: self._sdk = sdk self.create = response_helpers.to_streamed_response_wrapper( sdk.create, "extra_headers" ) self.get = response_helpers.to_streamed_response_wrapper( sdk.get, "extra_headers" ) self.delete = response_helpers.to_streamed_response_wrapper( sdk.delete, "extra_headers" ) self.cancel = response_helpers.to_streamed_response_wrapper( sdk.cancel, "extra_headers" ) class AsyncInteractions(AsyncBaseSDK): @property def with_raw_response(self): return AsyncInteractionsWithRawResponse(self) @property def with_streaming_response(self): return AsyncInteractionsWithStreamingResponse(self) @overload async def create( self, *, request: Union[ models.CreateInteractionRequest, models.CreateInteractionRequestParam ], extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, extra_body: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> Union[interactions.Interaction, AsyncStream[interactions.InteractionSSEEvent]]: r"""Creating an interaction Creates a new interaction. :param body: The request body. :param api_version: Which version of the API to use. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param extra_body: Additional JSON object fields to merge into request bodies. :param timeout: Override the default request timeout configuration for this method in seconds """ @overload async def create( self, *, api_version: Optional[str] = None, model: interactions.Model, stream: Union[Literal[False], None] = None, store: bool = ..., background: bool = ..., system_instruction: str = ..., tools: List[interactions.ToolParam] = ..., response_modalities: List[interactions.ResponseModality] = ..., response_mime_type: str = ..., previous_interaction_id: str = ..., service_tier: interactions.ServiceTier = ..., webhook_config: interactions.WebhookConfigParam = ..., response_format: interactions.CreateModelInteractionResponseFormatParam = ..., environment: interactions.CreateModelInteractionEnvironmentParam = ..., generation_config: interactions.GenerationConfigParam = ..., safety_settings: List[interactions.SafetySettingParam] = ..., labels: Dict[str, str] = ..., input: interactions.InteractionsInputParam, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, extra_body: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> interactions.Interaction: r"""Creating an interaction Creates a new interaction. :param api_version: Which version of the API to use. :param model: The model that will complete your prompt.\n\nSee [models](https://ai.google.dev/gemini-api/docs/models) for additional details. :param stream: Input only. Whether the interaction will be streamed. :param store: Input only. Whether to store the response and request for later retrieval. :param background: Input only. Whether to run the model interaction in the background. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param response_mime_type: The mime type of the response. This is required if response_format is set. :param previous_interaction_id: The ID of the previous interaction, if any. :param service_tier: :param webhook_config: Message for configuring webhook events for a request. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param environment: The environment configuration for the interaction. Can be an object specifying remote environment sources or a string referencing an existing environment ID. :param generation_config: Configuration parameters for model interactions. :param safety_settings: Safety settings for the interaction. :param labels: The labels with user-defined metadata for the request. :param input: The input for the interaction. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param extra_body: Additional JSON object fields to merge into request bodies. :param timeout: Override the default request timeout configuration for this method in seconds """ @overload async def create( self, *, api_version: Optional[str] = None, model: interactions.Model, stream: Literal[True], store: bool = ..., background: bool = ..., system_instruction: str = ..., tools: List[interactions.ToolParam] = ..., response_modalities: List[interactions.ResponseModality] = ..., response_mime_type: str = ..., previous_interaction_id: str = ..., service_tier: interactions.ServiceTier = ..., webhook_config: interactions.WebhookConfigParam = ..., response_format: interactions.CreateModelInteractionResponseFormatParam = ..., environment: interactions.CreateModelInteractionEnvironmentParam = ..., generation_config: interactions.GenerationConfigParam = ..., safety_settings: List[interactions.SafetySettingParam] = ..., labels: Dict[str, str] = ..., input: interactions.InteractionsInputParam, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, extra_body: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> AsyncStream[interactions.InteractionSSEEvent]: r"""Creating an interaction Creates a new interaction. :param api_version: Which version of the API to use. :param model: The model that will complete your prompt.\n\nSee [models](https://ai.google.dev/gemini-api/docs/models) for additional details. :param stream: Input only. Whether the interaction will be streamed. :param store: Input only. Whether to store the response and request for later retrieval. :param background: Input only. Whether to run the model interaction in the background. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param response_mime_type: The mime type of the response. This is required if response_format is set. :param previous_interaction_id: The ID of the previous interaction, if any. :param service_tier: :param webhook_config: Message for configuring webhook events for a request. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param environment: The environment configuration for the interaction. Can be an object specifying remote environment sources or a string referencing an existing environment ID. :param generation_config: Configuration parameters for model interactions. :param safety_settings: Safety settings for the interaction. :param labels: The labels with user-defined metadata for the request. :param input: The input for the interaction. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param extra_body: Additional JSON object fields to merge into request bodies. :param timeout: Override the default request timeout configuration for this method in seconds """ @overload async def create( self, *, api_version: Optional[str] = None, agent: interactions.AgentOption, stream: Union[Literal[False], None] = None, store: bool = ..., background: bool = ..., system_instruction: str = ..., tools: List[interactions.ToolParam] = ..., response_modalities: List[interactions.ResponseModality] = ..., response_mime_type: str = ..., previous_interaction_id: str = ..., service_tier: interactions.ServiceTier = ..., webhook_config: interactions.WebhookConfigParam = ..., response_format: interactions.CreateAgentInteractionResponseFormatParam = ..., environment: interactions.CreateAgentInteractionEnvironmentParam = ..., agent_config: interactions.CreateAgentInteractionAgentConfigParam = ..., safety_settings: List[interactions.SafetySettingParam] = ..., labels: Dict[str, str] = ..., input: interactions.InteractionsInputParam, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, extra_body: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> interactions.Interaction: r"""Creating an interaction Creates a new interaction. :param api_version: Which version of the API to use. :param agent: The agent to interact with. :param stream: Input only. Whether the interaction will be streamed. :param store: Input only. Whether to store the response and request for later retrieval. :param background: Input only. Whether to run the model interaction in the background. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param response_mime_type: The mime type of the response. This is required if response_format is set. :param previous_interaction_id: The ID of the previous interaction, if any. :param service_tier: :param webhook_config: Message for configuring webhook events for a request. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param environment: The environment configuration for the interaction. Can be an object specifying remote environment sources or a string referencing an existing environment ID. :param agent_config: Configuration parameters for the agent interaction. :param safety_settings: Safety settings for the interaction. :param labels: The labels with user-defined metadata for the request. :param input: The input for the interaction. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param extra_body: Additional JSON object fields to merge into request bodies. :param timeout: Override the default request timeout configuration for this method in seconds """ @overload async def create( self, *, api_version: Optional[str] = None, agent: interactions.AgentOption, stream: Literal[True], store: bool = ..., background: bool = ..., system_instruction: str = ..., tools: List[interactions.ToolParam] = ..., response_modalities: List[interactions.ResponseModality] = ..., response_mime_type: str = ..., previous_interaction_id: str = ..., service_tier: interactions.ServiceTier = ..., webhook_config: interactions.WebhookConfigParam = ..., response_format: interactions.CreateAgentInteractionResponseFormatParam = ..., environment: interactions.CreateAgentInteractionEnvironmentParam = ..., agent_config: interactions.CreateAgentInteractionAgentConfigParam = ..., safety_settings: List[interactions.SafetySettingParam] = ..., labels: Dict[str, str] = ..., input: interactions.InteractionsInputParam, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, extra_body: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> AsyncStream[interactions.InteractionSSEEvent]: r"""Creating an interaction Creates a new interaction. :param api_version: Which version of the API to use. :param agent: The agent to interact with. :param stream: Input only. Whether the interaction will be streamed. :param store: Input only. Whether to store the response and request for later retrieval. :param background: Input only. Whether to run the model interaction in the background. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param response_mime_type: The mime type of the response. This is required if response_format is set. :param previous_interaction_id: The ID of the previous interaction, if any. :param service_tier: :param webhook_config: Message for configuring webhook events for a request. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param environment: The environment configuration for the interaction. Can be an object specifying remote environment sources or a string referencing an existing environment ID. :param agent_config: Configuration parameters for the agent interaction. :param safety_settings: Safety settings for the interaction. :param labels: The labels with user-defined metadata for the request. :param input: The input for the interaction. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param extra_body: Additional JSON object fields to merge into request bodies. :param timeout: Override the default request timeout configuration for this method in seconds """ @overload async def create( self, *, request: OptionalNullable[ Union[models.CreateInteractionRequest, models.CreateInteractionRequestParam] ] = UNSET, api_version: OptionalNullable[str] = UNSET, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, extra_body: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, **body_kwargs: Any, ) -> Union[interactions.Interaction, AsyncStream[interactions.InteractionSSEEvent]]: r"""Creating an interaction Creates a new interaction. :param api_version: Which version of the API to use. :param model: The model that will complete your prompt.\n\nSee [models](https://ai.google.dev/gemini-api/docs/models) for additional details. :param stream: Input only. Whether the interaction will be streamed. :param store: Input only. Whether to store the response and request for later retrieval. :param background: Input only. Whether to run the model interaction in the background. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param response_mime_type: The mime type of the response. This is required if response_format is set. :param previous_interaction_id: The ID of the previous interaction, if any. :param service_tier: :param webhook_config: Message for configuring webhook events for a request. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param environment: The environment configuration for the interaction. Can be an object specifying remote environment sources or a string referencing an existing environment ID. :param generation_config: Configuration parameters for model interactions. :param safety_settings: Safety settings for the interaction. :param labels: The labels with user-defined metadata for the request. :param input: The input for the interaction. :param agent: The agent to interact with. :param agent_config: Configuration parameters for the agent interaction. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param extra_body: Additional JSON object fields to merge into request bodies. :param timeout: Override the default request timeout configuration for this method in seconds """ async def create( self, *, request: OptionalNullable[ Union[models.CreateInteractionRequest, models.CreateInteractionRequestParam] ] = UNSET, api_version: OptionalNullable[str] = UNSET, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, extra_body: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, **body_kwargs: Any, ) -> Union[interactions.Interaction, AsyncStream[interactions.InteractionSSEEvent]]: r"""Creating an interaction Creates a new interaction. :param api_version: Which version of the API to use. :param model: The model that will complete your prompt.\n\nSee [models](https://ai.google.dev/gemini-api/docs/models) for additional details. :param stream: Input only. Whether the interaction will be streamed. :param store: Input only. Whether to store the response and request for later retrieval. :param background: Input only. Whether to run the model interaction in the background. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param response_mime_type: The mime type of the response. This is required if response_format is set. :param previous_interaction_id: The ID of the previous interaction, if any. :param service_tier: :param webhook_config: Message for configuring webhook events for a request. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param environment: The environment configuration for the interaction. Can be an object specifying remote environment sources or a string referencing an existing environment ID. :param generation_config: Configuration parameters for model interactions. :param safety_settings: Safety settings for the interaction. :param labels: The labels with user-defined metadata for the request. :param input: The input for the interaction. :param agent: The agent to interact with. :param agent_config: Configuration parameters for the agent interaction. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param extra_body: Additional JSON object fields to merge into request bodies. :param timeout: Override the default request timeout configuration for this method in seconds """ if "model" in body_kwargs and "agent" in body_kwargs: raise ValueError("Cannot supply both 'model' and 'agent'.") if "model" in body_kwargs and "agent_config" in body_kwargs: raise ValueError("Cannot supply both 'model' and 'agent_config'.") if "generation_config" in body_kwargs and "agent" in body_kwargs: raise ValueError("Cannot supply both 'generation_config' and 'agent'.") if "generation_config" in body_kwargs and "agent_config" in body_kwargs: raise ValueError( "Cannot supply both 'generation_config' and 'agent_config'." ) base_url = None url_variables = None retries: OptionalNullable[utils.RetryConfig] = UNSET server_url = None http_headers = extra_headers timeout_ms = self._coerce_timeout_ms(timeout) if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) if request is not UNSET: request = cast( models.CreateInteractionRequest, request if isinstance(request, BaseModel) else utils.unmarshal( cast(Any, request), models.CreateInteractionRequest ), ) else: _body_kwargs = dict(body_kwargs) _body_kwargs = {k: v for k, v in _body_kwargs.items() if v is not UNSET} _request_kwargs: dict[str, Any] = {"api_version": api_version} _request_kwargs = { k: v for k, v in _request_kwargs.items() if v is not UNSET } _request_kwargs["body"] = _body_kwargs request = cast( models.CreateInteractionRequest, utils.unmarshal(_request_kwargs, models.CreateInteractionRequest), ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request_async( method="POST", path="/{api_version}/interactions", base_url=base_url, url_variables=url_variables, request=request, request_body_required=True, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="text/event-stream" if getattr(getattr(request, "body", None), "stream", False) is True else "application/json", http_headers=http_headers, extra_query_params=extra_query, _globals=models.CreateInteractionGlobals( api_version=self.sdk_configuration.globals.api_version, ), security=self.sdk_configuration.security, get_serialized_body=lambda: utils.serialize_request_body( request.body, False, False, "json", models.CreateInteractionRequestBody, extra_body=extra_body, ), allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config else: retries = utils.RetryConfig( "attempt-count-backoff", utils.BackoffStrategy(500, 8000, 2, 30000), True, max_retries=4, ) retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["408", "409", "429", "5XX"]) async def _speakeasy_parse_response(http_res): response_data: Any = None if utils.match_response(http_res, "200", "application/json"): http_res_text = await utils.stream_to_text_async(http_res) return unmarshal_json_response( interactions.Interaction, http_res, http_res_text, validate=False ) if utils.match_response(http_res, "200", "text/event-stream"): return AsyncStream( http_res, lambda raw: unmarshal_json_response( interactions.InteractionSSEStreamEvent, http_res, raw, validate=False, ).data, sentinel="[DONE]", client_ref=self, ) if utils.match_response(http_res, "4XX", "application/json"): http_res_text = await utils.stream_to_text_async(http_res) try: response_data = unmarshal_json_response( errors.CreateInteractionClientErrorData, http_res, http_res_text ) raise errors.CreateInteractionClientError( response_data, http_res, http_res_text ) except errors.ResponseValidationError as e: raise errors.GenAiDefaultError( "Error response body did not match expected schema", http_res, http_res_text, ) from e if utils.match_response(http_res, "5XX", "application/json"): http_res_text = await utils.stream_to_text_async(http_res) try: response_data = unmarshal_json_response( errors.CreateInteractionServerErrorData, http_res, http_res_text ) raise errors.CreateInteractionServerError( response_data, http_res, http_res_text ) except errors.ResponseValidationError as e: raise errors.GenAiDefaultError( "Error response body did not match expected schema", http_res, http_res_text, ) from e http_res_text = await utils.stream_to_text_async(http_res) raise errors.GenAiDefaultError( "Unexpected response received", http_res, http_res_text ) _speakeasy_hook_ctx = HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="CreateInteraction", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, types.Security ), tags=None, extensions={ "x-codeSamples": [ { "label": "simple", "lang": "sh", "source": 'curl -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "model": "gemini-3.6-flash",\n "input": "Hello, how are you?"\n }\'\n', }, { "label": "simple", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\ninteraction = client.interactions.create(\n model="gemini-3.6-flash",\n input="Hello, how are you?",\n)\nprint(interaction.output_text)\n', }, { "label": "simple", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\nconst interaction = await ai.interactions.create({\n model: 'gemini-3.6-flash',\n input: 'Hello, how are you?',\n});\nconsole.log(interaction.output_text);\n", }, { "label": "multi_turn", "lang": "sh", "source": 'curl -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "model": "gemini-3.6-flash",\n "input": [\n { "type": "user_input", "content": [{ "type": "text", "text": "Hello!" }] },\n { "type": "model_output", "content": [{ "type": "text", "text": "Hi there! How can I help you today?" }] },\n { "type": "user_input", "content": [{ "type": "text", "text": "What is the capital of France?" }] }\n ]\n }\'\n', }, { "label": "multi_turn", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\nresponse = client.interactions.create(\n model="gemini-3.6-flash",\n input=[\n { "type": "user_input", "content": [{ "type": "text", "text": "Hello!" }] },\n { "type": "model_output", "content": [{ "type": "text", "text": "Hi there! How can I help you today?" }] },\n { "type": "user_input", "content": [{ "type": "text", "text": "What is the capital of France?" }] }\n ]\n)\nprint(response.output_text)\n', }, { "label": "multi_turn", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\nconst interaction = await ai.interactions.create({\n model: 'gemini-3.6-flash',\n input: [\n { type: 'user_input', content: [{ type: 'text', text: 'Hello' }] },\n { type: 'model_output', content: [{ type: 'text', text: 'Hi there! How can I help you today?' }] },\n { type: 'user_input', content: [{ type: 'text', text: 'What is the capital of France?' }] }\n ]\n});\nconsole.log(interaction.output_text);\n", }, { "label": "multimodal_image", "lang": "sh", "source": 'curl -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "model": "gemini-3.6-flash",\n "input": [\n {\n "type": "text",\n "text": "What is in this picture?"\n },\n {\n "type": "image",\n "data": "BASE64_ENCODED_IMAGE",\n "mime_type": "image/png"\n }\n ]\n }\'\n', }, { "label": "multimodal_image", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\nresponse = client.interactions.create(\n model="gemini-3.6-flash",\n input=[\n { "type": "text", "text": "What is in this picture?" },\n { "type": "image", "data": "BASE64_ENCODED_IMAGE", "mime_type": "image/png" }\n ]\n)\nprint(response.output_text)\n', }, { "label": "multimodal_image", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\nconst interaction = await ai.interactions.create({\n model: 'gemini-3.6-flash',\n input: [\n { type: 'text', text: 'What is in this picture?' },\n { type: 'image', data: 'BASE64_ENCODED_IMAGE', mime_type: 'image/png' }\n ]\n});\nconsole.log(interaction.output_text);\n", }, { "label": "function_calling", "lang": "sh", "source": 'curl -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "model": "gemini-3.6-flash",\n "tools": [\n {\n "type": "function",\n "name": "get_weather",\n "description": "Get the current weather in a given location",\n "parameters": {\n "type": "object",\n "properties": {\n "location": {\n "type": "string",\n "description": "The city and state, e.g. San Francisco, CA"\n }\n },\n "required": [\n "location"\n ]\n }\n }\n ],\n "input": "What is the weather like in Boston, MA?"\n }\'\n', }, { "label": "function_calling", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\nresponse = client.interactions.create(\n model="gemini-3.6-flash",\n tools=[{\n "type": "function",\n "name": "get_weather",\n "description": "Get the current weather in a given location",\n "parameters": {\n "type": "object",\n "properties": {\n "location": {\n "type": "string",\n "description": "The city and state, e.g. San Francisco, CA"\n }\n },\n "required": ["location"]\n }\n }],\n input="What is the weather like in Boston, MA?"\n)\nprint(response.steps[-1])\n', }, { "label": "function_calling", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\nconst interaction = await ai.interactions.create({\n model: 'gemini-3.6-flash',\n tools: [{\n type: 'function',\n name: 'get_weather',\n description: 'Get the current weather in a given location',\n parameters: {\n type: 'object',\n properties: {\n location: {\n type: 'string',\n description: 'The city and state, e.g. San Francisco, CA'\n }\n },\n required: ['location']\n }\n }],\n input: 'What is the weather like in Boston, MA?'\n});\nconsole.log(interaction.steps.at(-1));\n", }, { "label": "deep_research", "lang": "sh", "source": 'curl -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "agent": "deep-research-pro-preview-12-2025",\n "input": "Find a cure to cancer",\n "background": true\n }\'\n', }, { "label": "deep_research", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\ninteraction = client.interactions.create(\n agent="deep-research-pro-preview-12-2025",\n input="find a cure to cancer",\n background=True,\n)\nprint(interaction.status)\n', }, { "label": "deep_research", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\nconst interaction = await ai.interactions.create({\n agent: 'deep-research-pro-preview-12-2025',\n input: 'find a cure to cancer',\n background: true,\n});\nconsole.log(interaction.status);\n", }, { "label": "antigravity", "lang": "sh", "source": 'curl -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "agent": "antigravity-preview-05-2026",\n "input": "Read Hacker News, summarize the top 5 stories, and save results as a markdown file.",\n "environment": "remote"\n }\'\n', }, { "label": "antigravity", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\ninteraction = client.interactions.create(\n agent="antigravity-preview-05-2026",\n input="Read Hacker News, summarize the top 5 stories, and save results as a markdown file.",\n environment="remote",\n)\nprint(interaction.output_text)\n', }, { "label": "antigravity", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\nconst interaction = await ai.interactions.create({\n agent: 'antigravity-preview-05-2026',\n input: 'Read Hacker News, summarize the top 5 stories, and save results as a markdown file.',\n environment: 'remote',\n});\nconsole.log(interaction.output_text);\n", }, { "label": "reuse_env", "lang": "sh", "source": '# Step 1: Create an interaction with a fresh remote environment.\nRESPONSE=$(curl -s -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -H "Api-Revision: 2026-05-20" \\\n -d \'{\n "agent": "antigravity-preview-05-2026",\n "input": "Write a hello world script at /workspace/hello.py.",\n "environment": "remote"\n }\')\nINTERACTION_ID=$(echo $RESPONSE | python3 -c "import sys,json; print(json.load(sys.stdin)[\'id\'])")\nENV_ID=$(echo $RESPONSE | python3 -c "import sys,json; print(json.load(sys.stdin)[\'environment_id\'])")\n\n# Step 2: Reuse the same environment in a follow-up interaction.\ncurl -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d "{\n \\"agent\\": \\"antigravity-preview-05-2026\\",\n \\"input\\": \\"Modify the script to accept a name argument and greet the user.\\",\n \\"environment\\": \\"$ENV_ID\\",\n \\"previous_interaction_id\\": \\"$INTERACTION_ID\\"\n }"\n', }, { "label": "reuse_env", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\n\n# Step 1: Create an interaction with a fresh remote environment.\ninteraction = client.interactions.create(\n agent="antigravity-preview-05-2026",\n input="Write a hello world script at /workspace/hello.py.",\n environment="remote",\n)\nprint(f"Environment ID: {interaction.environment_id}")\n\n# Step 2: Reuse the same environment in a follow-up interaction.\ninteraction_2 = client.interactions.create(\n agent="antigravity-preview-05-2026",\n input="Modify the script to accept a name argument and greet the user.",\n environment=interaction.environment_id,\n previous_interaction_id=interaction.id,\n)\nprint(interaction_2.output_text)\n', }, { "label": "reuse_env", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\n\n// Step 1: Create an interaction with a fresh remote environment.\nconst interaction = await ai.interactions.create({\n agent: 'antigravity-preview-05-2026',\n input: 'Write a hello world script at /workspace/hello.py.',\n environment: 'remote',\n});\nconsole.log(`Environment ID: ${interaction.environment_id}`);\n\n// Step 2: Reuse the same environment in a follow-up interaction.\nconst interaction2 = await ai.interactions.create({\n agent: 'antigravity-preview-05-2026',\n input: 'Modify the script to accept a name argument and greet the user.',\n environment: interaction.environment_id,\n previous_interaction_id: interaction.id,\n});\nconsole.log(interaction2.output_text);\n", }, { "label": "with_sources", "lang": "sh", "source": 'curl -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "agent": "antigravity-preview-05-2026",\n "input": "List all files under /workspace and summarize what you find.",\n "environment": {\n "type": "remote",\n "sources": [\n {\n "type": "repository",\n "source": "https://github.com/octocat/Spoon-Knife",\n "target": "/workspace/repo"\n },\n {\n "type": "inline",\n "content": "Focus on Python files only.",\n "target": "/workspace/notes.txt"\n }\n ]\n }\n }\'\n', }, { "label": "with_sources", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\ninteraction = client.interactions.create(\n agent="antigravity-preview-05-2026",\n input="List all files under /workspace and summarize what you find.",\n environment={\n "type": "remote",\n "sources": [\n {\n "type": "repository",\n "source": "https://github.com/octocat/Spoon-Knife",\n "target": "/workspace/repo",\n },\n {\n "type": "inline",\n "content": "Focus on Python files only.",\n "target": "/workspace/notes.txt",\n },\n ],\n },\n)\nprint(interaction.output_text)\n', }, { "label": "with_sources", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\nconst interaction = await ai.interactions.create({\n agent: 'antigravity-preview-05-2026',\n input: 'List all files under /workspace and summarize what you find.',\n environment: {\n type: 'remote',\n sources: [\n {\n type: 'repository',\n source: 'https://github.com/octocat/Spoon-Knife',\n target: '/workspace/repo',\n },\n {\n type: 'inline',\n content: 'Focus on Python files only.',\n target: '/workspace/notes.txt',\n },\n ],\n },\n});\nconsole.log(interaction.output_text);\n", }, { "label": "custom_agent", "lang": "sh", "source": '# Step 1: Create a custom agent.\ncurl -X POST https://generativelanguage.googleapis.com/v1beta/agents \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "id": "code-reviewer",\n "base_agent": "antigravity-preview-05-2026",\n "system_instruction": "You are a senior code reviewer. Check every file for bugs, style issues, and security vulnerabilities.",\n "base_environment": {\n "type": "remote",\n "sources": [{\n "type": "repository",\n "source": "https://github.com/octocat/Spoon-Knife",\n "target": "/workspace/repo"\n }]\n }\n }\'\n\n# Step 2: Use the custom agent.\ncurl -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -d \'{\n "agent": "code-reviewer",\n "input": "Review the latest changes in /workspace/repo/src and file a summary.",\n "environment": "remote"\n }\'\n', }, { "label": "custom_agent", "lang": "python", "source": 'import uuid\nfrom google import genai\n\nclient = genai.Client()\n\n# Step 1: Create a custom agent.\nagent_id = f"code-reviewer-{uuid.uuid4().hex[:8]}"\nclient.agents.create(\n id=agent_id,\n base_agent="antigravity-preview-05-2026",\n system_instruction="You are a senior code reviewer. Check every file for bugs, style issues, and security vulnerabilities.",\n base_environment={\n "type": "remote",\n "sources": [{\n "type": "repository",\n "source": "https://github.com/octocat/Spoon-Knife",\n "target": "/workspace/repo",\n }],\n },\n)\n\n# Step 2: Use the custom agent.\nresult = client.interactions.create(\n agent=agent_id,\n input="Review the latest changes in /workspace/repo/src and file a summary.",\n environment="remote",\n)\nprint(result.output_text)\n\n# [cleanup]\nclient.agents.delete(agent_id)\n# [/cleanup]\n', }, { "label": "custom_agent", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\n\n// Step 1: Create a custom agent.\nconst agentId = `code-reviewer-${crypto.randomUUID().slice(0, 8)}`;\nawait ai.agents.create({\n id: agentId,\n base_agent: 'antigravity-preview-05-2026',\n system_instruction: 'You are a senior code reviewer. Check every file for bugs, style issues, and security vulnerabilities.',\n base_environment: {\n type: 'remote',\n sources: [{\n type: 'repository',\n source: 'https://github.com/octocat/Spoon-Knife',\n target: '/workspace/repo',\n }],\n },\n});\n\n// Step 2: Use the custom agent.\nconst result = await ai.interactions.create({\n agent: agentId,\n input: 'Review the latest changes in /workspace/repo/src and file a summary.',\n environment: 'remote',\n});\nconsole.log(result.output_text);\n\n// [cleanup]\nawait ai.agents.delete(agentId);\n// [/cleanup]\n", }, ] }, response=ResponseContext(mode=_speakeasy_response_mode, execution="async"), ) http_res = await self.do_request_async( hook_ctx=_speakeasy_hook_ctx, request=req, is_error_status_code=lambda c: utils.match_status_codes(["4XX", "5XX"], c), stream=getattr(getattr(request, "body", None), "stream", False) is True or _speakeasy_response_mode == "streaming", retry_config=retry_config, ) if _speakeasy_response_mode != "parsed": if utils.match_status_codes(["4XX", "5XX"], http_res.status_code): await http_res.aread() try: await _speakeasy_parse_response(http_res) except Exception as parse_exc_: await response_helpers.raise_parse_error_async( self.sdk_configuration.__dict__.get("_async_hooks"), self.sdk_configuration.__dict__.get("_hooks"), AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) _speakeasy_response_cls = ( response_helpers.AsyncStreamedAPIResponse if _speakeasy_response_mode == "streaming" else response_helpers.AsyncAPIResponse ) return cast( Any, _speakeasy_response_cls( raw=http_res, parser=_speakeasy_parse_response, mode=( "event_stream" if getattr(getattr(request, "body", None), "stream", False) is True else "buffered" ), client_ref=self, hook_ctx=AfterParseErrorContext(_speakeasy_hook_ctx), hooks=self.sdk_configuration.__dict__.get("_hooks"), async_hooks=self.sdk_configuration.__dict__.get("_async_hooks"), ), ) try: return await _speakeasy_parse_response(http_res) except Exception as parse_exc_: await response_helpers.raise_parse_error_async( self.sdk_configuration.__dict__.get("_async_hooks"), self.sdk_configuration.__dict__.get("_hooks"), AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) @overload async def get( self, id: str, *, stream: Union[Literal[False], None] = None, last_event_id: Optional[str] = None, include_input: Optional[bool] = False, api_version: Optional[str] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> interactions.Interaction: r"""Retrieving an interaction Retrieves the full details of a single interaction based on its `Interaction.id`. :param id: The unique identifier of the interaction to retrieve. :param stream: If set to true, the generated content will be streamed incrementally. :param last_event_id: Optional. If set, resumes the interaction stream from the next chunk after the event marked by the event id. Can only be used if `stream` is true. :param include_input: If set to true, includes the input in the response. :param api_version: Which version of the API to use. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param timeout: Override the default request timeout configuration for this method in seconds """ @overload async def get( self, id: str, *, stream: Literal[True], last_event_id: Optional[str] = None, include_input: Optional[bool] = False, api_version: Optional[str] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> AsyncStream[interactions.InteractionSSEEvent]: r"""Retrieving an interaction Retrieves the full details of a single interaction based on its `Interaction.id`. :param id: The unique identifier of the interaction to retrieve. :param stream: If set to true, the generated content will be streamed incrementally. :param last_event_id: Optional. If set, resumes the interaction stream from the next chunk after the event marked by the event id. Can only be used if `stream` is true. :param include_input: If set to true, includes the input in the response. :param api_version: Which version of the API to use. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param timeout: Override the default request timeout configuration for this method in seconds """ @overload async def get( self, id: str, *, stream: bool, last_event_id: Optional[str] = None, include_input: Optional[bool] = False, api_version: Optional[str] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> Union[interactions.Interaction, AsyncStream[interactions.InteractionSSEEvent]]: r"""Retrieving an interaction Retrieves the full details of a single interaction based on its `Interaction.id`. :param id: The unique identifier of the interaction to retrieve. :param stream: If set to true, the generated content will be streamed incrementally. :param last_event_id: Optional. If set, resumes the interaction stream from the next chunk after the event marked by the event id. Can only be used if `stream` is true. :param include_input: If set to true, includes the input in the response. :param api_version: Which version of the API to use. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param timeout: Override the default request timeout configuration for this method in seconds """ async def get( self, id: str, *, stream: Optional[bool] = False, last_event_id: Optional[str] = None, include_input: Optional[bool] = False, api_version: Optional[str] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> Union[interactions.Interaction, AsyncStream[interactions.InteractionSSEEvent]]: r"""Retrieving an interaction Retrieves the full details of a single interaction based on its `Interaction.id`. :param id: The unique identifier of the interaction to retrieve. :param stream: If set to true, the generated content will be streamed incrementally. :param last_event_id: Optional. If set, resumes the interaction stream from the next chunk after the event marked by the event id. Can only be used if `stream` is true. :param include_input: If set to true, includes the input in the response. :param api_version: Which version of the API to use. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param timeout: Override the default request timeout configuration for this method in seconds """ base_url = None url_variables = None retries: OptionalNullable[utils.RetryConfig] = UNSET server_url = None http_headers = extra_headers timeout_ms = self._coerce_timeout_ms(timeout) if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.GetInteractionByIDRequest( id=id, stream=stream, last_event_id=last_event_id, include_input=include_input, api_version=api_version, ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request_async( method="GET", path="/{api_version}/interactions/{id}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="text/event-stream" if stream is True else "application/json", http_headers=http_headers, extra_query_params=extra_query, _globals=models.GetInteractionByIDGlobals( api_version=self.sdk_configuration.globals.api_version, ), security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config else: retries = utils.RetryConfig( "attempt-count-backoff", utils.BackoffStrategy(500, 8000, 2, 30000), True, max_retries=4, ) retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["408", "409", "429", "5XX"]) async def _speakeasy_parse_response(http_res): response_data: Any = None if utils.match_response(http_res, "200", "application/json"): http_res_text = await utils.stream_to_text_async(http_res) return unmarshal_json_response( interactions.Interaction, http_res, http_res_text, validate=False ) if utils.match_response(http_res, "200", "text/event-stream"): return AsyncStream( http_res, lambda raw: unmarshal_json_response( interactions.InteractionSSEStreamEvent, http_res, raw, validate=False, ).data, sentinel="[DONE]", client_ref=self, ) if utils.match_response(http_res, "4XX", "application/json"): http_res_text = await utils.stream_to_text_async(http_res) try: response_data = unmarshal_json_response( errors.GetInteractionByIDClientErrorData, http_res, http_res_text, ) raise errors.GetInteractionByIDClientError( response_data, http_res, http_res_text ) except errors.ResponseValidationError as e: raise errors.GenAiDefaultError( "Error response body did not match expected schema", http_res, http_res_text, ) from e if utils.match_response(http_res, "5XX", "application/json"): http_res_text = await utils.stream_to_text_async(http_res) try: response_data = unmarshal_json_response( errors.GetInteractionByIDServerErrorData, http_res, http_res_text, ) raise errors.GetInteractionByIDServerError( response_data, http_res, http_res_text ) except errors.ResponseValidationError as e: raise errors.GenAiDefaultError( "Error response body did not match expected schema", http_res, http_res_text, ) from e http_res_text = await utils.stream_to_text_async(http_res) raise errors.GenAiDefaultError( "Unexpected response received", http_res, http_res_text ) _speakeasy_hook_ctx = HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="getInteractionById", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, types.Security ), tags=None, extensions={ "x-codeSamples": [ { "label": "get", "lang": "sh", "source": '# [setup]\nINTERACTION_ID=$(curl -s -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\\n -H "Api-Revision: 2026-05-20" \\\n -d \'{"model": "gemini-3.6-flash", "input": "Say hello."}\' \\\n | python3 -c "import sys,json; print(json.load(sys.stdin)[\'id\'])")\n# [/setup]\n\ncurl -X GET "https://generativelanguage.googleapis.com/v1beta/interactions/$INTERACTION_ID" \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Api-Revision: 2026-05-20"\n', }, { "label": "get", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\n\n# [setup]\ncreated = client.interactions.create(\n model="gemini-3.6-flash",\n input="Say hello."\n)\n# [/setup]\n\ninteraction = client.interactions.get(id=created.id)\nprint(interaction.status)\n', }, { "label": "get", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\n\n// [setup]\nconst created = await ai.interactions.create({\n model: 'gemini-3.6-flash',\n input: 'Say hello.'\n});\n// [/setup]\n\nconst interaction = await ai.interactions.get(created.id);\nconsole.log(interaction.status);\n", }, ] }, response=ResponseContext(mode=_speakeasy_response_mode, execution="async"), ) http_res = await self.do_request_async( hook_ctx=_speakeasy_hook_ctx, request=req, is_error_status_code=lambda c: utils.match_status_codes(["4XX", "5XX"], c), stream=stream is True or _speakeasy_response_mode == "streaming", retry_config=retry_config, ) if _speakeasy_response_mode != "parsed": if utils.match_status_codes(["4XX", "5XX"], http_res.status_code): await http_res.aread() try: await _speakeasy_parse_response(http_res) except Exception as parse_exc_: await response_helpers.raise_parse_error_async( self.sdk_configuration.__dict__.get("_async_hooks"), self.sdk_configuration.__dict__.get("_hooks"), AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) _speakeasy_response_cls = ( response_helpers.AsyncStreamedAPIResponse if _speakeasy_response_mode == "streaming" else response_helpers.AsyncAPIResponse ) return cast( Any, _speakeasy_response_cls( raw=http_res, parser=_speakeasy_parse_response, mode=("event_stream" if stream is True else "buffered"), client_ref=self, hook_ctx=AfterParseErrorContext(_speakeasy_hook_ctx), hooks=self.sdk_configuration.__dict__.get("_hooks"), async_hooks=self.sdk_configuration.__dict__.get("_async_hooks"), ), ) try: return await _speakeasy_parse_response(http_res) except Exception as parse_exc_: await response_helpers.raise_parse_error_async( self.sdk_configuration.__dict__.get("_async_hooks"), self.sdk_configuration.__dict__.get("_hooks"), AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) async def delete( self, id: str, *, api_version: Optional[str] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ): r"""Deleting an interaction Deletes the interaction by id. :param id: The unique identifier of the interaction to delete. :param api_version: Which version of the API to use. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param timeout: Override the default request timeout configuration for this method in seconds """ base_url = None url_variables = None retries: OptionalNullable[utils.RetryConfig] = UNSET server_url = None http_headers = extra_headers timeout_ms = self._coerce_timeout_ms(timeout) if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.DeleteInteractionRequest( id=id, api_version=api_version, ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request_async( method="DELETE", path="/{api_version}/interactions/{id}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, extra_query_params=extra_query, _globals=models.DeleteInteractionGlobals( api_version=self.sdk_configuration.globals.api_version, ), security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config else: retries = utils.RetryConfig( "attempt-count-backoff", utils.BackoffStrategy(500, 8000, 2, 30000), True, max_retries=4, ) retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["408", "409", "429", "5XX"]) async def _speakeasy_parse_response(http_res): response_data: Any = None if utils.match_response(http_res, "200", "*"): return if utils.match_response(http_res, "4XX", "application/json"): try: response_data = unmarshal_json_response( errors.DeleteInteractionClientErrorData, http_res ) raise errors.DeleteInteractionClientError(response_data, http_res) except errors.ResponseValidationError as e: raise errors.GenAiDefaultError( "Error response body did not match expected schema", http_res, http_res.text, ) from e if utils.match_response(http_res, "5XX", "application/json"): try: response_data = unmarshal_json_response( errors.DeleteInteractionServerErrorData, http_res ) raise errors.DeleteInteractionServerError(response_data, http_res) except errors.ResponseValidationError as e: raise errors.GenAiDefaultError( "Error response body did not match expected schema", http_res, http_res.text, ) from e raise errors.GenAiDefaultError("Unexpected response received", http_res) _speakeasy_hook_ctx = HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="deleteInteraction", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, types.Security ), tags=None, extensions={ "x-codeSamples": [ { "label": "delete", "lang": "sh", "source": '# [setup]\nINTERACTION_ID=$(curl -s -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\ -d \'{"model": "gemini-3.6-flash", "input": "Hello"}\' \\\n | python3 -c "import sys,json; print(json.load(sys.stdin)[\'id\'])")\n# [/setup]\n\ncurl -X DELETE "https://generativelanguage.googleapis.com/v1beta/interactions/$INTERACTION_ID" \\\n -H "x-goog-api-key: $GEMINI_API_KEY"\n', }, { "label": "delete", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\n\n# [setup]\ncreated = client.interactions.create(\n model="gemini-3.6-flash",\n input="Hello",\n)\n# [/setup]\n\nclient.interactions.delete(id=created.id)\nprint("Interaction deleted successfully.")\n', }, { "label": "delete", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\n\n// [setup]\nconst created = await ai.interactions.create({\n model: 'gemini-3.6-flash',\n input: 'Hello',\n});\n// [/setup]\n\nawait ai.interactions.delete(created.id);\nconsole.log('Interaction deleted successfully.');\n", }, ] }, response=ResponseContext(mode=_speakeasy_response_mode, execution="async"), ) http_res = await self.do_request_async( hook_ctx=_speakeasy_hook_ctx, request=req, is_error_status_code=lambda c: utils.match_status_codes(["4XX", "5XX"], c), stream=_speakeasy_response_mode == "streaming", retry_config=retry_config, ) if _speakeasy_response_mode != "parsed": if utils.match_status_codes(["4XX", "5XX"], http_res.status_code): await http_res.aread() try: await _speakeasy_parse_response(http_res) except Exception as parse_exc_: await response_helpers.raise_parse_error_async( self.sdk_configuration.__dict__.get("_async_hooks"), self.sdk_configuration.__dict__.get("_hooks"), AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) _speakeasy_response_cls = ( response_helpers.AsyncStreamedAPIResponse if _speakeasy_response_mode == "streaming" else response_helpers.AsyncAPIResponse ) return cast( Any, _speakeasy_response_cls( raw=http_res, parser=_speakeasy_parse_response, mode="buffered", client_ref=self, hook_ctx=AfterParseErrorContext(_speakeasy_hook_ctx), hooks=self.sdk_configuration.__dict__.get("_hooks"), async_hooks=self.sdk_configuration.__dict__.get("_async_hooks"), ), ) try: return await _speakeasy_parse_response(http_res) except Exception as parse_exc_: await response_helpers.raise_parse_error_async( self.sdk_configuration.__dict__.get("_async_hooks"), self.sdk_configuration.__dict__.get("_hooks"), AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) async def cancel( self, id: str, *, api_version: Optional[str] = None, extra_headers: Optional[Mapping[str, str]] = None, extra_query: Optional[Mapping[str, Any]] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, ) -> interactions.Interaction: r"""Canceling an interaction Cancels an interaction by id. This only applies to background interactions that are still running. :param id: The unique identifier of the interaction to cancel. :param api_version: Which version of the API to use. :param extra_headers: Additional headers to set or replace on requests. :param extra_query: Additional query parameters to append to requests. :param timeout: Override the default request timeout configuration for this method in seconds """ base_url = None url_variables = None retries: OptionalNullable[utils.RetryConfig] = UNSET server_url = None http_headers = extra_headers timeout_ms = self._coerce_timeout_ms(timeout) if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.CancelInteractionByIDRequest( id=id, api_version=api_version, ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request_async( method="POST", path="/{api_version}/interactions/{id}/cancel", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, extra_query_params=extra_query, _globals=models.CancelInteractionByIDGlobals( api_version=self.sdk_configuration.globals.api_version, ), security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config else: retries = utils.RetryConfig( "attempt-count-backoff", utils.BackoffStrategy(500, 8000, 2, 30000), True, max_retries=4, ) retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["408", "409", "429", "5XX"]) async def _speakeasy_parse_response(http_res): response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response( interactions.Interaction, http_res, validate=False ) if utils.match_response(http_res, "4XX", "application/json"): try: response_data = unmarshal_json_response( errors.CancelInteractionByIDClientErrorData, http_res ) raise errors.CancelInteractionByIDClientError( response_data, http_res ) except errors.ResponseValidationError as e: raise errors.GenAiDefaultError( "Error response body did not match expected schema", http_res, http_res.text, ) from e if utils.match_response(http_res, "5XX", "application/json"): try: response_data = unmarshal_json_response( errors.CancelInteractionByIDServerErrorData, http_res ) raise errors.CancelInteractionByIDServerError( response_data, http_res ) except errors.ResponseValidationError as e: raise errors.GenAiDefaultError( "Error response body did not match expected schema", http_res, http_res.text, ) from e raise errors.GenAiDefaultError("Unexpected response received", http_res) _speakeasy_hook_ctx = HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="cancelInteractionById", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, types.Security ), tags=None, extensions={ "x-codeSamples": [ { "label": "cancel", "lang": "sh", "source": '# [setup]\nINTERACTION_ID=$(curl -s -X POST https://generativelanguage.googleapis.com/v1beta/interactions \\\n -H "x-goog-api-key: $GEMINI_API_KEY" \\\n -H "Content-Type: application/json" \\ -d \'{"model": "gemini-3.6-flash", "input": "Write a long essay about the history of computing.", "background": true}\' \\\n | python3 -c "import sys,json; print(json.load(sys.stdin)[\'id\'])")\n# [/setup]\n\ncurl -X POST "https://generativelanguage.googleapis.com/v1beta/interactions/$INTERACTION_ID/cancel" \\\n -H "x-goog-api-key: $GEMINI_API_KEY"\n', }, { "label": "cancel", "lang": "python", "source": 'from google import genai\n\nclient = genai.Client()\n\n# Start a background interaction so it stays in-progress.\ncreated = client.interactions.create(\n model="gemini-3.6-flash",\n input="Write a long essay about the history of computing.",\n tools=[{"type": "computer_use"}],\n background=True,\n)\n\n# Cancel the in-progress interaction.\ninteraction = client.interactions.cancel(id=created.id)\nprint(interaction.status)\n', }, { "label": "cancel", "lang": "javascript", "source": "import {GoogleGenAI} from '@google/genai';\n\nconst ai = new GoogleGenAI({});\n\n// Start a background interaction so it stays in-progress.\nconst created = await ai.interactions.create({\n model: 'gemini-3.6-flash',\n input: 'Write a long essay about the history of computing.',\n tools: [{ type: 'computer_use' }],\n background: true,\n});\n\n// Cancel the in-progress interaction.\nconst interaction = await ai.interactions.cancel(created.id);\nconsole.log(interaction.status);\n", }, ] }, response=ResponseContext(mode=_speakeasy_response_mode, execution="async"), ) http_res = await self.do_request_async( hook_ctx=_speakeasy_hook_ctx, request=req, is_error_status_code=lambda c: utils.match_status_codes(["4XX", "5XX"], c), stream=_speakeasy_response_mode == "streaming", retry_config=retry_config, ) if _speakeasy_response_mode != "parsed": if utils.match_status_codes(["4XX", "5XX"], http_res.status_code): await http_res.aread() try: await _speakeasy_parse_response(http_res) except Exception as parse_exc_: await response_helpers.raise_parse_error_async( self.sdk_configuration.__dict__.get("_async_hooks"), self.sdk_configuration.__dict__.get("_hooks"), AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) _speakeasy_response_cls = ( response_helpers.AsyncStreamedAPIResponse if _speakeasy_response_mode == "streaming" else response_helpers.AsyncAPIResponse ) return cast( Any, _speakeasy_response_cls( raw=http_res, parser=_speakeasy_parse_response, mode="buffered", client_ref=self, hook_ctx=AfterParseErrorContext(_speakeasy_hook_ctx), hooks=self.sdk_configuration.__dict__.get("_hooks"), async_hooks=self.sdk_configuration.__dict__.get("_async_hooks"), ), ) try: return await _speakeasy_parse_response(http_res) except Exception as parse_exc_: await response_helpers.raise_parse_error_async( self.sdk_configuration.__dict__.get("_async_hooks"), self.sdk_configuration.__dict__.get("_hooks"), AfterParseErrorContext(_speakeasy_hook_ctx), http_res, parse_exc_, ) class AsyncInteractionsWithRawResponse: def __init__(self, sdk: AsyncInteractions) -> None: self._sdk = sdk self.create = response_helpers.async_to_raw_response_wrapper( sdk.create, "extra_headers" ) self.get = response_helpers.async_to_raw_response_wrapper( sdk.get, "extra_headers" ) self.delete = response_helpers.async_to_raw_response_wrapper( sdk.delete, "extra_headers" ) self.cancel = response_helpers.async_to_raw_response_wrapper( sdk.cancel, "extra_headers" ) class AsyncInteractionsWithStreamingResponse: def __init__(self, sdk: AsyncInteractions) -> None: self._sdk = sdk self.create = response_helpers.async_to_streamed_response_wrapper( sdk.create, "extra_headers" ) self.get = response_helpers.async_to_streamed_response_wrapper( sdk.get, "extra_headers" ) self.delete = response_helpers.async_to_streamed_response_wrapper( sdk.delete, "extra_headers" ) self.cancel = response_helpers.async_to_streamed_response_wrapper( sdk.cancel, "extra_headers" )