# 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.serialization import convert_and_respect_annotation_metadata from ...core.unchecked_base_model import construct_type from ...errors.unprocessable_entity_error import UnprocessableEntityError from ...types.create_phone_number_response_model import CreatePhoneNumberResponseModel from ...types.http_validation_error import HttpValidationError from ...types.inbound_sip_trunk_config_request_model import InboundSipTrunkConfigRequestModel from ...types.livekit_stack_type import LivekitStackType from ...types.outbound_sip_trunk_config_request_model import OutboundSipTrunkConfigRequestModel from .types.phone_numbers_create_request_body import PhoneNumbersCreateRequestBody from .types.phone_numbers_get_response import PhoneNumbersGetResponse from .types.phone_numbers_list_response_item import PhoneNumbersListResponseItem from .types.phone_numbers_update_response import PhoneNumbersUpdateResponse # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawPhoneNumbersClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def list( self, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[typing.List[PhoneNumbersListResponseItem]]: """ Retrieve all Phone Numbers Parameters ---------- request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[typing.List[PhoneNumbersListResponseItem]] Successful Response """ _response = self._client_wrapper.httpx_client.request( "v1/convai/phone-numbers", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( typing.List[PhoneNumbersListResponseItem], construct_type( type_=typing.List[PhoneNumbersListResponseItem], # 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 create( self, *, request: PhoneNumbersCreateRequestBody, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[CreatePhoneNumberResponseModel]: """ Import Phone Number from provider configuration (Twilio or SIP trunk) Parameters ---------- request : PhoneNumbersCreateRequestBody request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[CreatePhoneNumberResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( "v1/convai/phone-numbers", method="POST", json=convert_and_respect_annotation_metadata( object_=request, annotation=PhoneNumbersCreateRequestBody, direction="write" ), headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( CreatePhoneNumberResponseModel, construct_type( type_=CreatePhoneNumberResponseModel, # 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, phone_number_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[PhoneNumbersGetResponse]: """ Retrieve Phone Number details by ID Parameters ---------- phone_number_id : str The id of an agent. This is returned on agent creation. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[PhoneNumbersGetResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/phone-numbers/{jsonable_encoder(phone_number_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( PhoneNumbersGetResponse, construct_type( type_=PhoneNumbersGetResponse, # 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, phone_number_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[typing.Any]: """ Delete Phone Number by ID Parameters ---------- phone_number_id : str The id of an agent. This is returned on agent creation. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[typing.Any] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/phone-numbers/{jsonable_encoder(phone_number_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) def update( self, phone_number_id: str, *, agent_id: typing.Optional[str] = OMIT, inbound_trunk_config: typing.Optional[InboundSipTrunkConfigRequestModel] = OMIT, outbound_trunk_config: typing.Optional[OutboundSipTrunkConfigRequestModel] = OMIT, livekit_stack: typing.Optional[LivekitStackType] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[PhoneNumbersUpdateResponse]: """ Update assigned agent of a phone number Parameters ---------- phone_number_id : str The id of an agent. This is returned on agent creation. agent_id : typing.Optional[str] inbound_trunk_config : typing.Optional[InboundSipTrunkConfigRequestModel] outbound_trunk_config : typing.Optional[OutboundSipTrunkConfigRequestModel] livekit_stack : typing.Optional[LivekitStackType] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[PhoneNumbersUpdateResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/convai/phone-numbers/{jsonable_encoder(phone_number_id)}", method="PATCH", json={ "agent_id": agent_id, "inbound_trunk_config": convert_and_respect_annotation_metadata( object_=inbound_trunk_config, annotation=InboundSipTrunkConfigRequestModel, direction="write" ), "outbound_trunk_config": convert_and_respect_annotation_metadata( object_=outbound_trunk_config, annotation=OutboundSipTrunkConfigRequestModel, direction="write" ), "livekit_stack": livekit_stack, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( PhoneNumbersUpdateResponse, construct_type( type_=PhoneNumbersUpdateResponse, # 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 AsyncRawPhoneNumbersClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def list( self, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[typing.List[PhoneNumbersListResponseItem]]: """ Retrieve all Phone Numbers Parameters ---------- request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[typing.List[PhoneNumbersListResponseItem]] Successful Response """ _response = await self._client_wrapper.httpx_client.request( "v1/convai/phone-numbers", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( typing.List[PhoneNumbersListResponseItem], construct_type( type_=typing.List[PhoneNumbersListResponseItem], # 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 create( self, *, request: PhoneNumbersCreateRequestBody, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[CreatePhoneNumberResponseModel]: """ Import Phone Number from provider configuration (Twilio or SIP trunk) Parameters ---------- request : PhoneNumbersCreateRequestBody request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[CreatePhoneNumberResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( "v1/convai/phone-numbers", method="POST", json=convert_and_respect_annotation_metadata( object_=request, annotation=PhoneNumbersCreateRequestBody, direction="write" ), headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( CreatePhoneNumberResponseModel, construct_type( type_=CreatePhoneNumberResponseModel, # 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, phone_number_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[PhoneNumbersGetResponse]: """ Retrieve Phone Number details by ID Parameters ---------- phone_number_id : str The id of an agent. This is returned on agent creation. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[PhoneNumbersGetResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/convai/phone-numbers/{jsonable_encoder(phone_number_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( PhoneNumbersGetResponse, construct_type( type_=PhoneNumbersGetResponse, # 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, phone_number_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[typing.Any]: """ Delete Phone Number by ID Parameters ---------- phone_number_id : str The id of an agent. This is returned on agent creation. 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/phone-numbers/{jsonable_encoder(phone_number_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) async def update( self, phone_number_id: str, *, agent_id: typing.Optional[str] = OMIT, inbound_trunk_config: typing.Optional[InboundSipTrunkConfigRequestModel] = OMIT, outbound_trunk_config: typing.Optional[OutboundSipTrunkConfigRequestModel] = OMIT, livekit_stack: typing.Optional[LivekitStackType] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[PhoneNumbersUpdateResponse]: """ Update assigned agent of a phone number Parameters ---------- phone_number_id : str The id of an agent. This is returned on agent creation. agent_id : typing.Optional[str] inbound_trunk_config : typing.Optional[InboundSipTrunkConfigRequestModel] outbound_trunk_config : typing.Optional[OutboundSipTrunkConfigRequestModel] livekit_stack : typing.Optional[LivekitStackType] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[PhoneNumbersUpdateResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/convai/phone-numbers/{jsonable_encoder(phone_number_id)}", method="PATCH", json={ "agent_id": agent_id, "inbound_trunk_config": convert_and_respect_annotation_metadata( object_=inbound_trunk_config, annotation=InboundSipTrunkConfigRequestModel, direction="write" ), "outbound_trunk_config": convert_and_respect_annotation_metadata( object_=outbound_trunk_config, annotation=OutboundSipTrunkConfigRequestModel, direction="write" ), "livekit_stack": livekit_stack, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( PhoneNumbersUpdateResponse, construct_type( type_=PhoneNumbersUpdateResponse, # 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)