# 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 # pylint: skip-file """Code generated by Speakeasy (https://speakeasy.com). DO NOT EDIT.""" from .basesdk import AsyncBaseSDK, BaseSDK from . import errors, models, types, utils from ._hooks import AfterParseErrorContext, HookContext, ResponseContext 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"""Creates a new interaction. :param body: The request body. :param api_version: API version for request routing. :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, background: bool = ..., cached_content: str = ..., environment: interactions.CreateModelInteractionEnvironmentParam = ..., generation_config: interactions.GenerationConfigParam = ..., input: interactions.InteractionsInputParam = ..., labels: Dict[str, str] = ..., model: interactions.Model, previous_interaction_id: str = ..., response_format: interactions.CreateModelInteractionResponseFormatParam = ..., response_mime_type: str = ..., response_modalities: List[interactions.ResponseModality] = ..., safety_settings: List[interactions.SafetySettingParam] = ..., service_tier: interactions.ServiceTier = ..., store: bool = ..., stream: Union[Literal[False], None] = None, system_instruction: str = ..., tools: List[interactions.ToolParam] = ..., webhook_config: interactions.WebhookConfigParam = ..., 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"""Creates a new interaction. :param api_version: API version for request routing. :param background: Input only. Whether to run the model interaction in the background. :param cached_content: The name of the cached content used as context to serve the prediction. Note: only used in explicit caching, where users can have control over caching (e.g. what content to cache) and enjoy guaranteed cost savings. Format: cachedContents/{cachedContent} :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 input: The input for the interaction. :param labels: The labels with user-defined metadata for the request. Label keys and values can be no longer than 63 characters (Unicode codepoints) and can only contain lowercase letters, numeric characters, underscores, and dashes. International characters are allowed. Label values are optional. Label keys must start with a letter. :param model: The model that will complete your prompt.\n\nSee [models](https://ai.google.dev/gemini-api/docs/models) for additional details. :param previous_interaction_id: The ID of the previous interaction, if any. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param response_mime_type: The mime type of the response. This is required if response_format is set. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param safety_settings: Safety settings for the interaction. :param service_tier: :param store: Input only. Whether to store the response and request for later retrieval. :param stream: Input only. Whether the interaction will be streamed. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param webhook_config: Message for configuring webhook events for a request. :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, background: bool = ..., cached_content: str = ..., environment: interactions.CreateModelInteractionEnvironmentParam = ..., generation_config: interactions.GenerationConfigParam = ..., input: interactions.InteractionsInputParam = ..., labels: Dict[str, str] = ..., model: interactions.Model, previous_interaction_id: str = ..., response_format: interactions.CreateModelInteractionResponseFormatParam = ..., response_mime_type: str = ..., response_modalities: List[interactions.ResponseModality] = ..., safety_settings: List[interactions.SafetySettingParam] = ..., service_tier: interactions.ServiceTier = ..., store: bool = ..., stream: Literal[True], system_instruction: str = ..., tools: List[interactions.ToolParam] = ..., webhook_config: interactions.WebhookConfigParam = ..., 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"""Creates a new interaction. :param api_version: API version for request routing. :param background: Input only. Whether to run the model interaction in the background. :param cached_content: The name of the cached content used as context to serve the prediction. Note: only used in explicit caching, where users can have control over caching (e.g. what content to cache) and enjoy guaranteed cost savings. Format: cachedContents/{cachedContent} :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 input: The input for the interaction. :param labels: The labels with user-defined metadata for the request. Label keys and values can be no longer than 63 characters (Unicode codepoints) and can only contain lowercase letters, numeric characters, underscores, and dashes. International characters are allowed. Label values are optional. Label keys must start with a letter. :param model: The model that will complete your prompt.\n\nSee [models](https://ai.google.dev/gemini-api/docs/models) for additional details. :param previous_interaction_id: The ID of the previous interaction, if any. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param response_mime_type: The mime type of the response. This is required if response_format is set. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param safety_settings: Safety settings for the interaction. :param service_tier: :param store: Input only. Whether to store the response and request for later retrieval. :param stream: Input only. Whether the interaction will be streamed. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param webhook_config: Message for configuring webhook events for a request. :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, agent_config: interactions.CreateAgentInteractionAgentConfigParam = ..., background: bool = ..., environment: interactions.CreateAgentInteractionEnvironmentParam = ..., input: interactions.InteractionsInputParam = ..., labels: Dict[str, str] = ..., previous_interaction_id: str = ..., response_format: interactions.CreateAgentInteractionResponseFormatParam = ..., response_mime_type: str = ..., response_modalities: List[interactions.ResponseModality] = ..., safety_settings: List[interactions.SafetySettingParam] = ..., service_tier: interactions.ServiceTier = ..., store: bool = ..., stream: Union[Literal[False], None] = None, system_instruction: str = ..., tools: List[interactions.ToolParam] = ..., webhook_config: interactions.WebhookConfigParam = ..., 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"""Creates a new interaction. :param api_version: API version for request routing. :param agent: The agent to interact with. :param agent_config: Configuration parameters for the agent interaction. :param background: Input only. Whether to run the model interaction in the background. :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 input: The input for the interaction. :param labels: The labels with user-defined metadata for the request. Label keys and values can be no longer than 63 characters (Unicode codepoints) and can only contain lowercase letters, numeric characters, underscores, and dashes. International characters are allowed. Label values are optional. Label keys must start with a letter. :param previous_interaction_id: The ID of the previous interaction, if any. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param response_mime_type: The mime type of the response. This is required if response_format is set. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param safety_settings: Safety settings for the interaction. :param service_tier: :param store: Input only. Whether to store the response and request for later retrieval. :param stream: Input only. Whether the interaction will be streamed. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param webhook_config: Message for configuring webhook events for a request. :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, agent_config: interactions.CreateAgentInteractionAgentConfigParam = ..., background: bool = ..., environment: interactions.CreateAgentInteractionEnvironmentParam = ..., input: interactions.InteractionsInputParam = ..., labels: Dict[str, str] = ..., previous_interaction_id: str = ..., response_format: interactions.CreateAgentInteractionResponseFormatParam = ..., response_mime_type: str = ..., response_modalities: List[interactions.ResponseModality] = ..., safety_settings: List[interactions.SafetySettingParam] = ..., service_tier: interactions.ServiceTier = ..., store: bool = ..., stream: Literal[True], system_instruction: str = ..., tools: List[interactions.ToolParam] = ..., webhook_config: interactions.WebhookConfigParam = ..., 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"""Creates a new interaction. :param api_version: API version for request routing. :param agent: The agent to interact with. :param agent_config: Configuration parameters for the agent interaction. :param background: Input only. Whether to run the model interaction in the background. :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 input: The input for the interaction. :param labels: The labels with user-defined metadata for the request. Label keys and values can be no longer than 63 characters (Unicode codepoints) and can only contain lowercase letters, numeric characters, underscores, and dashes. International characters are allowed. Label values are optional. Label keys must start with a letter. :param previous_interaction_id: The ID of the previous interaction, if any. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param response_mime_type: The mime type of the response. This is required if response_format is set. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param safety_settings: Safety settings for the interaction. :param service_tier: :param store: Input only. Whether to store the response and request for later retrieval. :param stream: Input only. Whether the interaction will be streamed. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param webhook_config: Message for configuring webhook events for a request. :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"""Creates a new interaction. :param api_version: API version for request routing. :param background: Input only. Whether to run the model interaction in the background. :param cached_content: The name of the cached content used as context to serve the prediction. Note: only used in explicit caching, where users can have control over caching (e.g. what content to cache) and enjoy guaranteed cost savings. Format: cachedContents/{cachedContent} :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 input: The input for the interaction. :param labels: The labels with user-defined metadata for the request. Label keys and values can be no longer than 63 characters (Unicode codepoints) and can only contain lowercase letters, numeric characters, underscores, and dashes. International characters are allowed. Label values are optional. Label keys must start with a letter. :param model: The model that will complete your prompt.\n\nSee [models](https://ai.google.dev/gemini-api/docs/models) for additional details. :param previous_interaction_id: The ID of the previous interaction, if any. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param response_mime_type: The mime type of the response. This is required if response_format is set. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param safety_settings: Safety settings for the interaction. :param service_tier: :param store: Input only. Whether to store the response and request for later retrieval. :param stream: Input only. Whether the interaction will be streamed. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param webhook_config: Message for configuring webhook events for a request. :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"""Creates a new interaction. :param api_version: API version for request routing. :param background: Input only. Whether to run the model interaction in the background. :param cached_content: The name of the cached content used as context to serve the prediction. Note: only used in explicit caching, where users can have control over caching (e.g. what content to cache) and enjoy guaranteed cost savings. Format: cachedContents/{cachedContent} :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 input: The input for the interaction. :param labels: The labels with user-defined metadata for the request. Label keys and values can be no longer than 63 characters (Unicode codepoints) and can only contain lowercase letters, numeric characters, underscores, and dashes. International characters are allowed. Label values are optional. Label keys must start with a letter. :param model: The model that will complete your prompt.\n\nSee [models](https://ai.google.dev/gemini-api/docs/models) for additional details. :param previous_interaction_id: The ID of the previous interaction, if any. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param response_mime_type: The mime type of the response. This is required if response_format is set. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param safety_settings: Safety settings for the interaction. :param service_tier: :param store: Input only. Whether to store the response and request for later retrieval. :param stream: Input only. Whether the interaction will be streamed. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param webhook_config: Message for configuring webhook events for a request. :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 "cached_content" in body_kwargs and "agent" in body_kwargs: raise ValueError("Cannot supply both 'cached_content' and 'agent'.") if "cached_content" in body_kwargs and "agent_config" in body_kwargs: raise ValueError("Cannot supply both 'cached_content' 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'." ) 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'.") 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=["interactions"], extensions=None, 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_, ) 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"""Deletes the interaction by id. :param id: Required. The name of the interaction to delete. :param api_version: API version for request routing. :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( api_version=api_version, id=id, ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request( method="DELETE", path="/{api_version}/interactions/{interactionsId}", 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=["interactions"], extensions=None, 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_, ) @overload def get( self, id: str, *, api_version: Optional[str] = None, include_input: Optional[bool] = None, last_event_id: Optional[str] = None, stream: Union[Literal[False], None] = 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"""Retrieves the full details of a single interaction based on its `Interaction.id`. :param id: Required. The name of the interaction to retrieve. :param api_version: API version for request routing. :param include_input: If true, includes the input in the response. :param last_event_id: If set, resumes the interaction stream from the chunk after the event marked by the event id. Can only be used if `stream` is true. :param stream: If true, streams the interaction events as Server-Sent Events. :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, *, api_version: Optional[str] = None, include_input: Optional[bool] = None, last_event_id: Optional[str] = None, stream: Literal[True], 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"""Retrieves the full details of a single interaction based on its `Interaction.id`. :param id: Required. The name of the interaction to retrieve. :param api_version: API version for request routing. :param include_input: If true, includes the input in the response. :param last_event_id: If set, resumes the interaction stream from the chunk after the event marked by the event id. Can only be used if `stream` is true. :param stream: If true, streams the interaction events as Server-Sent Events. :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, *, api_version: Optional[str] = None, include_input: Optional[bool] = None, last_event_id: Optional[str] = None, stream: bool, 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"""Retrieves the full details of a single interaction based on its `Interaction.id`. :param id: Required. The name of the interaction to retrieve. :param api_version: API version for request routing. :param include_input: If true, includes the input in the response. :param last_event_id: If set, resumes the interaction stream from the chunk after the event marked by the event id. Can only be used if `stream` is true. :param stream: If true, streams the interaction events as Server-Sent Events. :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, *, api_version: Optional[str] = None, include_input: Optional[bool] = None, last_event_id: Optional[str] = None, stream: Optional[bool] = 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"""Retrieves the full details of a single interaction based on its `Interaction.id`. :param id: Required. The name of the interaction to retrieve. :param api_version: API version for request routing. :param include_input: If true, includes the input in the response. :param last_event_id: If set, resumes the interaction stream from the chunk after the event marked by the event id. Can only be used if `stream` is true. :param stream: If true, streams the interaction events as Server-Sent Events. :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( api_version=api_version, include_input=include_input, id=id, last_event_id=last_event_id, stream=stream, ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request( method="GET", path="/{api_version}/interactions/{interactionsId}", 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=["interactions"], extensions=None, 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 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"""Cancels an interaction by id. This only applies to background interactions that are still running. :param id: Required. The name of the interaction to cancel. :param api_version: API version for request routing. :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( api_version=api_version, id=id, ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request( method="POST", path="/{api_version}/interactions/{interactionsId}/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=["interactions"], extensions=None, 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.delete = response_helpers.to_raw_response_wrapper( sdk.delete, "extra_headers" ) self.get = response_helpers.to_raw_response_wrapper(sdk.get, "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.delete = response_helpers.to_streamed_response_wrapper( sdk.delete, "extra_headers" ) self.get = response_helpers.to_streamed_response_wrapper( sdk.get, "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"""Creates a new interaction. :param body: The request body. :param api_version: API version for request routing. :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, background: bool = ..., cached_content: str = ..., environment: interactions.CreateModelInteractionEnvironmentParam = ..., generation_config: interactions.GenerationConfigParam = ..., input: interactions.InteractionsInputParam = ..., labels: Dict[str, str] = ..., model: interactions.Model, previous_interaction_id: str = ..., response_format: interactions.CreateModelInteractionResponseFormatParam = ..., response_mime_type: str = ..., response_modalities: List[interactions.ResponseModality] = ..., safety_settings: List[interactions.SafetySettingParam] = ..., service_tier: interactions.ServiceTier = ..., store: bool = ..., stream: Union[Literal[False], None] = None, system_instruction: str = ..., tools: List[interactions.ToolParam] = ..., webhook_config: interactions.WebhookConfigParam = ..., 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"""Creates a new interaction. :param api_version: API version for request routing. :param background: Input only. Whether to run the model interaction in the background. :param cached_content: The name of the cached content used as context to serve the prediction. Note: only used in explicit caching, where users can have control over caching (e.g. what content to cache) and enjoy guaranteed cost savings. Format: cachedContents/{cachedContent} :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 input: The input for the interaction. :param labels: The labels with user-defined metadata for the request. Label keys and values can be no longer than 63 characters (Unicode codepoints) and can only contain lowercase letters, numeric characters, underscores, and dashes. International characters are allowed. Label values are optional. Label keys must start with a letter. :param model: The model that will complete your prompt.\n\nSee [models](https://ai.google.dev/gemini-api/docs/models) for additional details. :param previous_interaction_id: The ID of the previous interaction, if any. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param response_mime_type: The mime type of the response. This is required if response_format is set. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param safety_settings: Safety settings for the interaction. :param service_tier: :param store: Input only. Whether to store the response and request for later retrieval. :param stream: Input only. Whether the interaction will be streamed. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param webhook_config: Message for configuring webhook events for a request. :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, background: bool = ..., cached_content: str = ..., environment: interactions.CreateModelInteractionEnvironmentParam = ..., generation_config: interactions.GenerationConfigParam = ..., input: interactions.InteractionsInputParam = ..., labels: Dict[str, str] = ..., model: interactions.Model, previous_interaction_id: str = ..., response_format: interactions.CreateModelInteractionResponseFormatParam = ..., response_mime_type: str = ..., response_modalities: List[interactions.ResponseModality] = ..., safety_settings: List[interactions.SafetySettingParam] = ..., service_tier: interactions.ServiceTier = ..., store: bool = ..., stream: Literal[True], system_instruction: str = ..., tools: List[interactions.ToolParam] = ..., webhook_config: interactions.WebhookConfigParam = ..., 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"""Creates a new interaction. :param api_version: API version for request routing. :param background: Input only. Whether to run the model interaction in the background. :param cached_content: The name of the cached content used as context to serve the prediction. Note: only used in explicit caching, where users can have control over caching (e.g. what content to cache) and enjoy guaranteed cost savings. Format: cachedContents/{cachedContent} :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 input: The input for the interaction. :param labels: The labels with user-defined metadata for the request. Label keys and values can be no longer than 63 characters (Unicode codepoints) and can only contain lowercase letters, numeric characters, underscores, and dashes. International characters are allowed. Label values are optional. Label keys must start with a letter. :param model: The model that will complete your prompt.\n\nSee [models](https://ai.google.dev/gemini-api/docs/models) for additional details. :param previous_interaction_id: The ID of the previous interaction, if any. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param response_mime_type: The mime type of the response. This is required if response_format is set. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param safety_settings: Safety settings for the interaction. :param service_tier: :param store: Input only. Whether to store the response and request for later retrieval. :param stream: Input only. Whether the interaction will be streamed. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param webhook_config: Message for configuring webhook events for a request. :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, agent_config: interactions.CreateAgentInteractionAgentConfigParam = ..., background: bool = ..., environment: interactions.CreateAgentInteractionEnvironmentParam = ..., input: interactions.InteractionsInputParam = ..., labels: Dict[str, str] = ..., previous_interaction_id: str = ..., response_format: interactions.CreateAgentInteractionResponseFormatParam = ..., response_mime_type: str = ..., response_modalities: List[interactions.ResponseModality] = ..., safety_settings: List[interactions.SafetySettingParam] = ..., service_tier: interactions.ServiceTier = ..., store: bool = ..., stream: Union[Literal[False], None] = None, system_instruction: str = ..., tools: List[interactions.ToolParam] = ..., webhook_config: interactions.WebhookConfigParam = ..., 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"""Creates a new interaction. :param api_version: API version for request routing. :param agent: The agent to interact with. :param agent_config: Configuration parameters for the agent interaction. :param background: Input only. Whether to run the model interaction in the background. :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 input: The input for the interaction. :param labels: The labels with user-defined metadata for the request. Label keys and values can be no longer than 63 characters (Unicode codepoints) and can only contain lowercase letters, numeric characters, underscores, and dashes. International characters are allowed. Label values are optional. Label keys must start with a letter. :param previous_interaction_id: The ID of the previous interaction, if any. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param response_mime_type: The mime type of the response. This is required if response_format is set. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param safety_settings: Safety settings for the interaction. :param service_tier: :param store: Input only. Whether to store the response and request for later retrieval. :param stream: Input only. Whether the interaction will be streamed. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param webhook_config: Message for configuring webhook events for a request. :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, agent_config: interactions.CreateAgentInteractionAgentConfigParam = ..., background: bool = ..., environment: interactions.CreateAgentInteractionEnvironmentParam = ..., input: interactions.InteractionsInputParam = ..., labels: Dict[str, str] = ..., previous_interaction_id: str = ..., response_format: interactions.CreateAgentInteractionResponseFormatParam = ..., response_mime_type: str = ..., response_modalities: List[interactions.ResponseModality] = ..., safety_settings: List[interactions.SafetySettingParam] = ..., service_tier: interactions.ServiceTier = ..., store: bool = ..., stream: Literal[True], system_instruction: str = ..., tools: List[interactions.ToolParam] = ..., webhook_config: interactions.WebhookConfigParam = ..., 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"""Creates a new interaction. :param api_version: API version for request routing. :param agent: The agent to interact with. :param agent_config: Configuration parameters for the agent interaction. :param background: Input only. Whether to run the model interaction in the background. :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 input: The input for the interaction. :param labels: The labels with user-defined metadata for the request. Label keys and values can be no longer than 63 characters (Unicode codepoints) and can only contain lowercase letters, numeric characters, underscores, and dashes. International characters are allowed. Label values are optional. Label keys must start with a letter. :param previous_interaction_id: The ID of the previous interaction, if any. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param response_mime_type: The mime type of the response. This is required if response_format is set. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param safety_settings: Safety settings for the interaction. :param service_tier: :param store: Input only. Whether to store the response and request for later retrieval. :param stream: Input only. Whether the interaction will be streamed. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param webhook_config: Message for configuring webhook events for a request. :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"""Creates a new interaction. :param api_version: API version for request routing. :param background: Input only. Whether to run the model interaction in the background. :param cached_content: The name of the cached content used as context to serve the prediction. Note: only used in explicit caching, where users can have control over caching (e.g. what content to cache) and enjoy guaranteed cost savings. Format: cachedContents/{cachedContent} :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 input: The input for the interaction. :param labels: The labels with user-defined metadata for the request. Label keys and values can be no longer than 63 characters (Unicode codepoints) and can only contain lowercase letters, numeric characters, underscores, and dashes. International characters are allowed. Label values are optional. Label keys must start with a letter. :param model: The model that will complete your prompt.\n\nSee [models](https://ai.google.dev/gemini-api/docs/models) for additional details. :param previous_interaction_id: The ID of the previous interaction, if any. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param response_mime_type: The mime type of the response. This is required if response_format is set. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param safety_settings: Safety settings for the interaction. :param service_tier: :param store: Input only. Whether to store the response and request for later retrieval. :param stream: Input only. Whether the interaction will be streamed. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param webhook_config: Message for configuring webhook events for a request. :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"""Creates a new interaction. :param api_version: API version for request routing. :param background: Input only. Whether to run the model interaction in the background. :param cached_content: The name of the cached content used as context to serve the prediction. Note: only used in explicit caching, where users can have control over caching (e.g. what content to cache) and enjoy guaranteed cost savings. Format: cachedContents/{cachedContent} :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 input: The input for the interaction. :param labels: The labels with user-defined metadata for the request. Label keys and values can be no longer than 63 characters (Unicode codepoints) and can only contain lowercase letters, numeric characters, underscores, and dashes. International characters are allowed. Label values are optional. Label keys must start with a letter. :param model: The model that will complete your prompt.\n\nSee [models](https://ai.google.dev/gemini-api/docs/models) for additional details. :param previous_interaction_id: The ID of the previous interaction, if any. :param response_format: Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field. :param response_mime_type: The mime type of the response. This is required if response_format is set. :param response_modalities: The requested modalities of the response (TEXT, IMAGE, AUDIO). :param safety_settings: Safety settings for the interaction. :param service_tier: :param store: Input only. Whether to store the response and request for later retrieval. :param stream: Input only. Whether the interaction will be streamed. :param system_instruction: System instruction for the interaction. :param tools: A list of tool declarations the model may call during interaction. :param webhook_config: Message for configuring webhook events for a request. :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 "cached_content" in body_kwargs and "agent" in body_kwargs: raise ValueError("Cannot supply both 'cached_content' and 'agent'.") if "cached_content" in body_kwargs and "agent_config" in body_kwargs: raise ValueError("Cannot supply both 'cached_content' 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'." ) 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'.") 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=["interactions"], extensions=None, 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_, ) 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"""Deletes the interaction by id. :param id: Required. The name of the interaction to delete. :param api_version: API version for request routing. :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( api_version=api_version, id=id, ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request_async( method="DELETE", path="/{api_version}/interactions/{interactionsId}", 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=["interactions"], extensions=None, 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_, ) @overload async def get( self, id: str, *, api_version: Optional[str] = None, include_input: Optional[bool] = None, last_event_id: Optional[str] = None, stream: Union[Literal[False], None] = 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"""Retrieves the full details of a single interaction based on its `Interaction.id`. :param id: Required. The name of the interaction to retrieve. :param api_version: API version for request routing. :param include_input: If true, includes the input in the response. :param last_event_id: If set, resumes the interaction stream from the chunk after the event marked by the event id. Can only be used if `stream` is true. :param stream: If true, streams the interaction events as Server-Sent Events. :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, *, api_version: Optional[str] = None, include_input: Optional[bool] = None, last_event_id: Optional[str] = None, stream: Literal[True], 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"""Retrieves the full details of a single interaction based on its `Interaction.id`. :param id: Required. The name of the interaction to retrieve. :param api_version: API version for request routing. :param include_input: If true, includes the input in the response. :param last_event_id: If set, resumes the interaction stream from the chunk after the event marked by the event id. Can only be used if `stream` is true. :param stream: If true, streams the interaction events as Server-Sent Events. :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, *, api_version: Optional[str] = None, include_input: Optional[bool] = None, last_event_id: Optional[str] = None, stream: bool, 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"""Retrieves the full details of a single interaction based on its `Interaction.id`. :param id: Required. The name of the interaction to retrieve. :param api_version: API version for request routing. :param include_input: If true, includes the input in the response. :param last_event_id: If set, resumes the interaction stream from the chunk after the event marked by the event id. Can only be used if `stream` is true. :param stream: If true, streams the interaction events as Server-Sent Events. :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, *, api_version: Optional[str] = None, include_input: Optional[bool] = None, last_event_id: Optional[str] = None, stream: Optional[bool] = 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"""Retrieves the full details of a single interaction based on its `Interaction.id`. :param id: Required. The name of the interaction to retrieve. :param api_version: API version for request routing. :param include_input: If true, includes the input in the response. :param last_event_id: If set, resumes the interaction stream from the chunk after the event marked by the event id. Can only be used if `stream` is true. :param stream: If true, streams the interaction events as Server-Sent Events. :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( api_version=api_version, include_input=include_input, id=id, last_event_id=last_event_id, stream=stream, ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request_async( method="GET", path="/{api_version}/interactions/{interactionsId}", 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=["interactions"], extensions=None, 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 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"""Cancels an interaction by id. This only applies to background interactions that are still running. :param id: Required. The name of the interaction to cancel. :param api_version: API version for request routing. :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( api_version=api_version, id=id, ) _speakeasy_response_mode, http_headers = response_helpers.consume_response_mode( http_headers ) req = self._build_request_async( method="POST", path="/{api_version}/interactions/{interactionsId}/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=["interactions"], extensions=None, 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.delete = response_helpers.async_to_raw_response_wrapper( sdk.delete, "extra_headers" ) self.get = response_helpers.async_to_raw_response_wrapper( sdk.get, "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.delete = response_helpers.async_to_streamed_response_wrapper( sdk.delete, "extra_headers" ) self.get = response_helpers.async_to_streamed_response_wrapper( sdk.get, "extra_headers" ) self.cancel = response_helpers.async_to_streamed_response_wrapper( sdk.cancel, "extra_headers" )