# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. from __future__ import annotations from typing import List, Union, Optional from typing_extensions import Literal import httpx from ... import _legacy_response from ..._types import Body, Omit, Query, Headers, NoneType, NotGiven, omit, not_given from ..._utils import maybe_transform, async_maybe_transform from ..._compat import cached_property from ..._resource import SyncAPIResource, AsyncAPIResource from ..._response import ( StreamedBinaryAPIResponse, AsyncStreamedBinaryAPIResponse, to_streamed_response_wrapper, async_to_streamed_response_wrapper, to_custom_streamed_response_wrapper, async_to_custom_streamed_response_wrapper, ) from ..._base_client import make_request_options from ...types.realtime import ( call_refer_params, call_accept_params, call_create_params, call_reject_params, ) from ...types.responses.response_prompt_param import ResponsePromptParam from ...types.realtime.realtime_truncation_param import RealtimeTruncationParam from ...types.realtime.realtime_audio_config_param import RealtimeAudioConfigParam from ...types.realtime.realtime_tools_config_param import RealtimeToolsConfigParam from ...types.realtime.realtime_tracing_config_param import RealtimeTracingConfigParam from ...types.realtime.realtime_tool_choice_config_param import RealtimeToolChoiceConfigParam from ...types.realtime.realtime_session_create_request_param import RealtimeSessionCreateRequestParam __all__ = ["Calls", "AsyncCalls"] class Calls(SyncAPIResource): @cached_property def with_raw_response(self) -> CallsWithRawResponse: """ This property can be used as a prefix for any HTTP method call to return the raw response object instead of the parsed content. For more information, see https://www.github.com/openai/openai-python#accessing-raw-response-data-eg-headers """ return CallsWithRawResponse(self) @cached_property def with_streaming_response(self) -> CallsWithStreamingResponse: """ An alternative to `.with_raw_response` that doesn't eagerly read the response body. For more information, see https://www.github.com/openai/openai-python#with_streaming_response """ return CallsWithStreamingResponse(self) def create( self, *, sdp: str, session: RealtimeSessionCreateRequestParam | Omit = omit, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = not_given, ) -> _legacy_response.HttpxBinaryResponseContent: """ Create a new Realtime API call over WebRTC and receive the SDP answer needed to complete the peer connection. Args: sdp: WebRTC Session Description Protocol (SDP) offer generated by the caller. session: Realtime session object configuration. extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ extra_headers = {"Accept": "application/sdp", **(extra_headers or {})} return self._post( "/realtime/calls", body=maybe_transform( { "sdp": sdp, "session": session, }, call_create_params.CallCreateParams, ), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=_legacy_response.HttpxBinaryResponseContent, ) def accept( self, call_id: str, *, type: Literal["realtime"], audio: RealtimeAudioConfigParam | Omit = omit, include: List[Literal["item.input_audio_transcription.logprobs"]] | Omit = omit, instructions: str | Omit = omit, max_output_tokens: Union[int, Literal["inf"]] | Omit = omit, model: Union[ str, Literal[ "gpt-realtime", "gpt-realtime-2025-08-28", "gpt-4o-realtime-preview", "gpt-4o-realtime-preview-2024-10-01", "gpt-4o-realtime-preview-2024-12-17", "gpt-4o-realtime-preview-2025-06-03", "gpt-4o-mini-realtime-preview", "gpt-4o-mini-realtime-preview-2024-12-17", "gpt-realtime-mini", "gpt-realtime-mini-2025-10-06", "gpt-audio-mini", "gpt-audio-mini-2025-10-06", ], ] | Omit = omit, output_modalities: List[Literal["text", "audio"]] | Omit = omit, prompt: Optional[ResponsePromptParam] | Omit = omit, tool_choice: RealtimeToolChoiceConfigParam | Omit = omit, tools: RealtimeToolsConfigParam | Omit = omit, tracing: Optional[RealtimeTracingConfigParam] | Omit = omit, truncation: RealtimeTruncationParam | Omit = omit, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = not_given, ) -> None: """ Accept an incoming SIP call and configure the realtime session that will handle it. Args: type: The type of session to create. Always `realtime` for the Realtime API. audio: Configuration for input and output audio. include: Additional fields to include in server outputs. `item.input_audio_transcription.logprobs`: Include logprobs for input audio transcription. instructions: The default system instructions (i.e. system message) prepended to model calls. This field allows the client to guide the model on desired responses. The model can be instructed on response content and format, (e.g. "be extremely succinct", "act friendly", "here are examples of good responses") and on audio behavior (e.g. "talk quickly", "inject emotion into your voice", "laugh frequently"). The instructions are not guaranteed to be followed by the model, but they provide guidance to the model on the desired behavior. Note that the server sets default instructions which will be used if this field is not set and are visible in the `session.created` event at the start of the session. max_output_tokens: Maximum number of output tokens for a single assistant response, inclusive of tool calls. Provide an integer between 1 and 4096 to limit output tokens, or `inf` for the maximum available tokens for a given model. Defaults to `inf`. model: The Realtime model used for this session. output_modalities: The set of modalities the model can respond with. It defaults to `["audio"]`, indicating that the model will respond with audio plus a transcript. `["text"]` can be used to make the model respond with text only. It is not possible to request both `text` and `audio` at the same time. prompt: Reference to a prompt template and its variables. [Learn more](https://platform.openai.com/docs/guides/text?api-mode=responses#reusable-prompts). tool_choice: How the model chooses tools. Provide one of the string modes or force a specific function/MCP tool. tools: Tools available to the model. tracing: Realtime API can write session traces to the [Traces Dashboard](/logs?api=traces). Set to null to disable tracing. Once tracing is enabled for a session, the configuration cannot be modified. `auto` will create a trace for the session with default values for the workflow name, group id, and metadata. truncation: Controls how the realtime conversation is truncated prior to model inference. The default is `auto`. extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ if not call_id: raise ValueError(f"Expected a non-empty value for `call_id` but received {call_id!r}") extra_headers = {"Accept": "*/*", **(extra_headers or {})} return self._post( f"/realtime/calls/{call_id}/accept", body=maybe_transform( { "type": type, "audio": audio, "include": include, "instructions": instructions, "max_output_tokens": max_output_tokens, "model": model, "output_modalities": output_modalities, "prompt": prompt, "tool_choice": tool_choice, "tools": tools, "tracing": tracing, "truncation": truncation, }, call_accept_params.CallAcceptParams, ), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=NoneType, ) def hangup( self, call_id: str, *, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = not_given, ) -> None: """ End an active Realtime API call, whether it was initiated over SIP or WebRTC. Args: extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ if not call_id: raise ValueError(f"Expected a non-empty value for `call_id` but received {call_id!r}") extra_headers = {"Accept": "*/*", **(extra_headers or {})} return self._post( f"/realtime/calls/{call_id}/hangup", options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=NoneType, ) def refer( self, call_id: str, *, target_uri: str, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = not_given, ) -> None: """ Transfer an active SIP call to a new destination using the SIP REFER verb. Args: target_uri: URI that should appear in the SIP Refer-To header. Supports values like `tel:+14155550123` or `sip:agent@example.com`. extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ if not call_id: raise ValueError(f"Expected a non-empty value for `call_id` but received {call_id!r}") extra_headers = {"Accept": "*/*", **(extra_headers or {})} return self._post( f"/realtime/calls/{call_id}/refer", body=maybe_transform({"target_uri": target_uri}, call_refer_params.CallReferParams), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=NoneType, ) def reject( self, call_id: str, *, status_code: int | Omit = omit, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = not_given, ) -> None: """ Decline an incoming SIP call by returning a SIP status code to the caller. Args: status_code: SIP response code to send back to the caller. Defaults to `603` (Decline) when omitted. extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ if not call_id: raise ValueError(f"Expected a non-empty value for `call_id` but received {call_id!r}") extra_headers = {"Accept": "*/*", **(extra_headers or {})} return self._post( f"/realtime/calls/{call_id}/reject", body=maybe_transform({"status_code": status_code}, call_reject_params.CallRejectParams), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=NoneType, ) class AsyncCalls(AsyncAPIResource): @cached_property def with_raw_response(self) -> AsyncCallsWithRawResponse: """ This property can be used as a prefix for any HTTP method call to return the raw response object instead of the parsed content. For more information, see https://www.github.com/openai/openai-python#accessing-raw-response-data-eg-headers """ return AsyncCallsWithRawResponse(self) @cached_property def with_streaming_response(self) -> AsyncCallsWithStreamingResponse: """ An alternative to `.with_raw_response` that doesn't eagerly read the response body. For more information, see https://www.github.com/openai/openai-python#with_streaming_response """ return AsyncCallsWithStreamingResponse(self) async def create( self, *, sdp: str, session: RealtimeSessionCreateRequestParam | Omit = omit, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = not_given, ) -> _legacy_response.HttpxBinaryResponseContent: """ Create a new Realtime API call over WebRTC and receive the SDP answer needed to complete the peer connection. Args: sdp: WebRTC Session Description Protocol (SDP) offer generated by the caller. session: Realtime session object configuration. extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ extra_headers = {"Accept": "application/sdp", **(extra_headers or {})} return await self._post( "/realtime/calls", body=await async_maybe_transform( { "sdp": sdp, "session": session, }, call_create_params.CallCreateParams, ), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=_legacy_response.HttpxBinaryResponseContent, ) async def accept( self, call_id: str, *, type: Literal["realtime"], audio: RealtimeAudioConfigParam | Omit = omit, include: List[Literal["item.input_audio_transcription.logprobs"]] | Omit = omit, instructions: str | Omit = omit, max_output_tokens: Union[int, Literal["inf"]] | Omit = omit, model: Union[ str, Literal[ "gpt-realtime", "gpt-realtime-2025-08-28", "gpt-4o-realtime-preview", "gpt-4o-realtime-preview-2024-10-01", "gpt-4o-realtime-preview-2024-12-17", "gpt-4o-realtime-preview-2025-06-03", "gpt-4o-mini-realtime-preview", "gpt-4o-mini-realtime-preview-2024-12-17", "gpt-realtime-mini", "gpt-realtime-mini-2025-10-06", "gpt-audio-mini", "gpt-audio-mini-2025-10-06", ], ] | Omit = omit, output_modalities: List[Literal["text", "audio"]] | Omit = omit, prompt: Optional[ResponsePromptParam] | Omit = omit, tool_choice: RealtimeToolChoiceConfigParam | Omit = omit, tools: RealtimeToolsConfigParam | Omit = omit, tracing: Optional[RealtimeTracingConfigParam] | Omit = omit, truncation: RealtimeTruncationParam | Omit = omit, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = not_given, ) -> None: """ Accept an incoming SIP call and configure the realtime session that will handle it. Args: type: The type of session to create. Always `realtime` for the Realtime API. audio: Configuration for input and output audio. include: Additional fields to include in server outputs. `item.input_audio_transcription.logprobs`: Include logprobs for input audio transcription. instructions: The default system instructions (i.e. system message) prepended to model calls. This field allows the client to guide the model on desired responses. The model can be instructed on response content and format, (e.g. "be extremely succinct", "act friendly", "here are examples of good responses") and on audio behavior (e.g. "talk quickly", "inject emotion into your voice", "laugh frequently"). The instructions are not guaranteed to be followed by the model, but they provide guidance to the model on the desired behavior. Note that the server sets default instructions which will be used if this field is not set and are visible in the `session.created` event at the start of the session. max_output_tokens: Maximum number of output tokens for a single assistant response, inclusive of tool calls. Provide an integer between 1 and 4096 to limit output tokens, or `inf` for the maximum available tokens for a given model. Defaults to `inf`. model: The Realtime model used for this session. output_modalities: The set of modalities the model can respond with. It defaults to `["audio"]`, indicating that the model will respond with audio plus a transcript. `["text"]` can be used to make the model respond with text only. It is not possible to request both `text` and `audio` at the same time. prompt: Reference to a prompt template and its variables. [Learn more](https://platform.openai.com/docs/guides/text?api-mode=responses#reusable-prompts). tool_choice: How the model chooses tools. Provide one of the string modes or force a specific function/MCP tool. tools: Tools available to the model. tracing: Realtime API can write session traces to the [Traces Dashboard](/logs?api=traces). Set to null to disable tracing. Once tracing is enabled for a session, the configuration cannot be modified. `auto` will create a trace for the session with default values for the workflow name, group id, and metadata. truncation: Controls how the realtime conversation is truncated prior to model inference. The default is `auto`. extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ if not call_id: raise ValueError(f"Expected a non-empty value for `call_id` but received {call_id!r}") extra_headers = {"Accept": "*/*", **(extra_headers or {})} return await self._post( f"/realtime/calls/{call_id}/accept", body=await async_maybe_transform( { "type": type, "audio": audio, "include": include, "instructions": instructions, "max_output_tokens": max_output_tokens, "model": model, "output_modalities": output_modalities, "prompt": prompt, "tool_choice": tool_choice, "tools": tools, "tracing": tracing, "truncation": truncation, }, call_accept_params.CallAcceptParams, ), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=NoneType, ) async def hangup( self, call_id: str, *, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = not_given, ) -> None: """ End an active Realtime API call, whether it was initiated over SIP or WebRTC. Args: extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ if not call_id: raise ValueError(f"Expected a non-empty value for `call_id` but received {call_id!r}") extra_headers = {"Accept": "*/*", **(extra_headers or {})} return await self._post( f"/realtime/calls/{call_id}/hangup", options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=NoneType, ) async def refer( self, call_id: str, *, target_uri: str, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = not_given, ) -> None: """ Transfer an active SIP call to a new destination using the SIP REFER verb. Args: target_uri: URI that should appear in the SIP Refer-To header. Supports values like `tel:+14155550123` or `sip:agent@example.com`. extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ if not call_id: raise ValueError(f"Expected a non-empty value for `call_id` but received {call_id!r}") extra_headers = {"Accept": "*/*", **(extra_headers or {})} return await self._post( f"/realtime/calls/{call_id}/refer", body=await async_maybe_transform({"target_uri": target_uri}, call_refer_params.CallReferParams), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=NoneType, ) async def reject( self, call_id: str, *, status_code: int | Omit = omit, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, extra_query: Query | None = None, extra_body: Body | None = None, timeout: float | httpx.Timeout | None | NotGiven = not_given, ) -> None: """ Decline an incoming SIP call by returning a SIP status code to the caller. Args: status_code: SIP response code to send back to the caller. Defaults to `603` (Decline) when omitted. extra_headers: Send extra headers extra_query: Add additional query parameters to the request extra_body: Add additional JSON properties to the request timeout: Override the client-level default timeout for this request, in seconds """ if not call_id: raise ValueError(f"Expected a non-empty value for `call_id` but received {call_id!r}") extra_headers = {"Accept": "*/*", **(extra_headers or {})} return await self._post( f"/realtime/calls/{call_id}/reject", body=await async_maybe_transform({"status_code": status_code}, call_reject_params.CallRejectParams), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=NoneType, ) class CallsWithRawResponse: def __init__(self, calls: Calls) -> None: self._calls = calls self.create = _legacy_response.to_raw_response_wrapper( calls.create, ) self.accept = _legacy_response.to_raw_response_wrapper( calls.accept, ) self.hangup = _legacy_response.to_raw_response_wrapper( calls.hangup, ) self.refer = _legacy_response.to_raw_response_wrapper( calls.refer, ) self.reject = _legacy_response.to_raw_response_wrapper( calls.reject, ) class AsyncCallsWithRawResponse: def __init__(self, calls: AsyncCalls) -> None: self._calls = calls self.create = _legacy_response.async_to_raw_response_wrapper( calls.create, ) self.accept = _legacy_response.async_to_raw_response_wrapper( calls.accept, ) self.hangup = _legacy_response.async_to_raw_response_wrapper( calls.hangup, ) self.refer = _legacy_response.async_to_raw_response_wrapper( calls.refer, ) self.reject = _legacy_response.async_to_raw_response_wrapper( calls.reject, ) class CallsWithStreamingResponse: def __init__(self, calls: Calls) -> None: self._calls = calls self.create = to_custom_streamed_response_wrapper( calls.create, StreamedBinaryAPIResponse, ) self.accept = to_streamed_response_wrapper( calls.accept, ) self.hangup = to_streamed_response_wrapper( calls.hangup, ) self.refer = to_streamed_response_wrapper( calls.refer, ) self.reject = to_streamed_response_wrapper( calls.reject, ) class AsyncCallsWithStreamingResponse: def __init__(self, calls: AsyncCalls) -> None: self._calls = calls self.create = async_to_custom_streamed_response_wrapper( calls.create, AsyncStreamedBinaryAPIResponse, ) self.accept = async_to_streamed_response_wrapper( calls.accept, ) self.hangup = async_to_streamed_response_wrapper( calls.hangup, ) self.refer = async_to_streamed_response_wrapper( calls.refer, ) self.reject = async_to_streamed_response_wrapper( calls.reject, )