# This file was auto-generated by Fern from our API Definition. import typing from json.decoder import JSONDecodeError from ...core.api_error import ApiError from ...core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ...core.http_response import AsyncHttpResponse, HttpResponse from ...core.jsonable_encoder import jsonable_encoder from ...core.request_options import RequestOptions from ...core.unchecked_base_model import construct_type from ...errors.unprocessable_entity_error import UnprocessableEntityError from ...types.conversation_signed_url_response_model import ConversationSignedUrlResponseModel from ...types.evaluation_success_result import EvaluationSuccessResult from ...types.get_conversation_response_model import GetConversationResponseModel from ...types.get_conversations_page_response_model import GetConversationsPageResponseModel from ...types.http_validation_error import HttpValidationError from ...types.token_response_model import TokenResponseModel from .types.conversations_list_request_summary_mode import ConversationsListRequestSummaryMode class RawConversationsClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def get_signed_url( self, *, agent_id: str, include_conversation_id: typing.Optional[bool] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[ConversationSignedUrlResponseModel]: """ Get a signed url to start a conversation with an agent with an agent that requires authorization Parameters ---------- agent_id : str The id of the agent you're taking the action on. include_conversation_id : typing.Optional[bool] Whether to include a conversation_id with the response. If included, the conversation_signature cannot be used again. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ConversationSignedUrlResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( "v1/convai/conversation/get-signed-url", method="GET", params={ "agent_id": agent_id, "include_conversation_id": include_conversation_id, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ConversationSignedUrlResponseModel, construct_type( type_=ConversationSignedUrlResponseModel, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get_webrtc_token( self, *, agent_id: str, participant_name: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[TokenResponseModel]: """ Get a WebRTC session token for real-time communication. Parameters ---------- agent_id : str The id of the agent you're taking the action on. participant_name : typing.Optional[str] Optional custom participant name. If not provided, user ID will be used request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[TokenResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( "v1/convai/conversation/token", method="GET", params={ "agent_id": agent_id, "participant_name": participant_name, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( TokenResponseModel, construct_type( type_=TokenResponseModel, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def list( self, *, cursor: typing.Optional[str] = None, agent_id: typing.Optional[str] = None, call_successful: typing.Optional[EvaluationSuccessResult] = None, call_start_before_unix: typing.Optional[int] = None, call_start_after_unix: typing.Optional[int] = None, call_duration_min_secs: typing.Optional[int] = None, call_duration_max_secs: typing.Optional[int] = None, rating_max: typing.Optional[int] = None, rating_min: typing.Optional[int] = None, has_feedback_comment: typing.Optional[bool] = None, user_id: typing.Optional[str] = None, evaluation_params: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, data_collection_params: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, tool_names: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, page_size: typing.Optional[int] = None, summary_mode: typing.Optional[ConversationsListRequestSummaryMode] = None, search: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[GetConversationsPageResponseModel]: """ Get all conversations of agents that user owns. With option to restrict to a specific agent. Parameters ---------- cursor : typing.Optional[str] Used for fetching next page. Cursor is returned in the response. agent_id : typing.Optional[str] The id of the agent you're taking the action on. call_successful : typing.Optional[EvaluationSuccessResult] The result of the success evaluation call_start_before_unix : typing.Optional[int] Unix timestamp (in seconds) to filter conversations up to this start date. call_start_after_unix : typing.Optional[int] Unix timestamp (in seconds) to filter conversations after to this start date. call_duration_min_secs : typing.Optional[int] Minimum call duration in seconds. call_duration_max_secs : typing.Optional[int] Maximum call duration in seconds. rating_max : typing.Optional[int] Maximum overall rating (1-5). rating_min : typing.Optional[int] Minimum overall rating (1-5). has_feedback_comment : typing.Optional[bool] Filter conversations with user feedback comments. user_id : typing.Optional[str] Filter conversations by the user ID who initiated them. evaluation_params : typing.Optional[typing.Union[str, typing.Sequence[str]]] Evaluation filters. Repeat param. Format: criteria_id:result. Example: eval=value_framing:success data_collection_params : typing.Optional[typing.Union[str, typing.Sequence[str]]] Data collection filters. Repeat param. Format: id:op:value where op is one of eq|neq|gt|gte|lt|lte|in|exists|missing. For in, pipe-delimit values. tool_names : typing.Optional[typing.Union[str, typing.Sequence[str]]] Filter conversations by tool names used during the call. page_size : typing.Optional[int] How many conversations to return at maximum. Can not exceed 100, defaults to 30. summary_mode : typing.Optional[ConversationsListRequestSummaryMode] Whether to include transcript summaries in the response. search : typing.Optional[str] Full-text or fuzzy search over transcript messages request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[GetConversationsPageResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( "v1/convai/conversations", method="GET", params={ "cursor": cursor, "agent_id": agent_id, "call_successful": call_successful, "call_start_before_unix": call_start_before_unix, "call_start_after_unix": call_start_after_unix, "call_duration_min_secs": call_duration_min_secs, "call_duration_max_secs": call_duration_max_secs, "rating_max": rating_max, "rating_min": rating_min, "has_feedback_comment": has_feedback_comment, "user_id": user_id, "evaluation_params": evaluation_params, "data_collection_params": data_collection_params, "tool_names": tool_names, "page_size": page_size, "summary_mode": summary_mode, "search": search, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( GetConversationsPageResponseModel, construct_type( type_=GetConversationsPageResponseModel, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get( self, conversation_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[GetConversationResponseModel]: """ Get the details of a particular conversation Parameters ---------- conversation_id : str The id of the conversation you're taking the action on. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[GetConversationResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/conversations/{jsonable_encoder(conversation_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( GetConversationResponseModel, construct_type( type_=GetConversationResponseModel, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def delete( self, conversation_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[typing.Any]: """ Delete a particular conversation Parameters ---------- conversation_id : str The id of the conversation you're taking the action on. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[typing.Any] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/conversations/{jsonable_encoder(conversation_id)}", method="DELETE", request_options=request_options, ) try: if _response is None or not _response.text.strip(): return HttpResponse(response=_response, data=None) if 200 <= _response.status_code < 300: _data = typing.cast( typing.Any, construct_type( type_=typing.Any, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) class AsyncRawConversationsClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def get_signed_url( self, *, agent_id: str, include_conversation_id: typing.Optional[bool] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[ConversationSignedUrlResponseModel]: """ Get a signed url to start a conversation with an agent with an agent that requires authorization Parameters ---------- agent_id : str The id of the agent you're taking the action on. include_conversation_id : typing.Optional[bool] Whether to include a conversation_id with the response. If included, the conversation_signature cannot be used again. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ConversationSignedUrlResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( "v1/convai/conversation/get-signed-url", method="GET", params={ "agent_id": agent_id, "include_conversation_id": include_conversation_id, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ConversationSignedUrlResponseModel, construct_type( type_=ConversationSignedUrlResponseModel, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get_webrtc_token( self, *, agent_id: str, participant_name: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[TokenResponseModel]: """ Get a WebRTC session token for real-time communication. Parameters ---------- agent_id : str The id of the agent you're taking the action on. participant_name : typing.Optional[str] Optional custom participant name. If not provided, user ID will be used request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[TokenResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( "v1/convai/conversation/token", method="GET", params={ "agent_id": agent_id, "participant_name": participant_name, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( TokenResponseModel, construct_type( type_=TokenResponseModel, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def list( self, *, cursor: typing.Optional[str] = None, agent_id: typing.Optional[str] = None, call_successful: typing.Optional[EvaluationSuccessResult] = None, call_start_before_unix: typing.Optional[int] = None, call_start_after_unix: typing.Optional[int] = None, call_duration_min_secs: typing.Optional[int] = None, call_duration_max_secs: typing.Optional[int] = None, rating_max: typing.Optional[int] = None, rating_min: typing.Optional[int] = None, has_feedback_comment: typing.Optional[bool] = None, user_id: typing.Optional[str] = None, evaluation_params: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, data_collection_params: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, tool_names: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, page_size: typing.Optional[int] = None, summary_mode: typing.Optional[ConversationsListRequestSummaryMode] = None, search: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[GetConversationsPageResponseModel]: """ Get all conversations of agents that user owns. With option to restrict to a specific agent. Parameters ---------- cursor : typing.Optional[str] Used for fetching next page. Cursor is returned in the response. agent_id : typing.Optional[str] The id of the agent you're taking the action on. call_successful : typing.Optional[EvaluationSuccessResult] The result of the success evaluation call_start_before_unix : typing.Optional[int] Unix timestamp (in seconds) to filter conversations up to this start date. call_start_after_unix : typing.Optional[int] Unix timestamp (in seconds) to filter conversations after to this start date. call_duration_min_secs : typing.Optional[int] Minimum call duration in seconds. call_duration_max_secs : typing.Optional[int] Maximum call duration in seconds. rating_max : typing.Optional[int] Maximum overall rating (1-5). rating_min : typing.Optional[int] Minimum overall rating (1-5). has_feedback_comment : typing.Optional[bool] Filter conversations with user feedback comments. user_id : typing.Optional[str] Filter conversations by the user ID who initiated them. evaluation_params : typing.Optional[typing.Union[str, typing.Sequence[str]]] Evaluation filters. Repeat param. Format: criteria_id:result. Example: eval=value_framing:success data_collection_params : typing.Optional[typing.Union[str, typing.Sequence[str]]] Data collection filters. Repeat param. Format: id:op:value where op is one of eq|neq|gt|gte|lt|lte|in|exists|missing. For in, pipe-delimit values. tool_names : typing.Optional[typing.Union[str, typing.Sequence[str]]] Filter conversations by tool names used during the call. page_size : typing.Optional[int] How many conversations to return at maximum. Can not exceed 100, defaults to 30. summary_mode : typing.Optional[ConversationsListRequestSummaryMode] Whether to include transcript summaries in the response. search : typing.Optional[str] Full-text or fuzzy search over transcript messages request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[GetConversationsPageResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( "v1/convai/conversations", method="GET", params={ "cursor": cursor, "agent_id": agent_id, "call_successful": call_successful, "call_start_before_unix": call_start_before_unix, "call_start_after_unix": call_start_after_unix, "call_duration_min_secs": call_duration_min_secs, "call_duration_max_secs": call_duration_max_secs, "rating_max": rating_max, "rating_min": rating_min, "has_feedback_comment": has_feedback_comment, "user_id": user_id, "evaluation_params": evaluation_params, "data_collection_params": data_collection_params, "tool_names": tool_names, "page_size": page_size, "summary_mode": summary_mode, "search": search, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( GetConversationsPageResponseModel, construct_type( type_=GetConversationsPageResponseModel, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get( self, conversation_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[GetConversationResponseModel]: """ Get the details of a particular conversation Parameters ---------- conversation_id : str The id of the conversation you're taking the action on. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[GetConversationResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/convai/conversations/{jsonable_encoder(conversation_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( GetConversationResponseModel, construct_type( type_=GetConversationResponseModel, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def delete( self, conversation_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[typing.Any]: """ Delete a particular conversation Parameters ---------- conversation_id : str The id of the conversation you're taking the action on. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[typing.Any] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/convai/conversations/{jsonable_encoder(conversation_id)}", method="DELETE", request_options=request_options, ) try: if _response is None or not _response.text.strip(): return AsyncHttpResponse(response=_response, data=None) if 200 <= _response.status_code < 300: _data = typing.cast( typing.Any, construct_type( type_=typing.Any, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)