client.py

# This file was auto-generated by Fern from our API Definition.

import typing

from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper
from ..core.pagination import AsyncPager, SyncPager
from ..core.request_options import RequestOptions
from .raw_client import AsyncRawTelephonyClient, RawTelephonyClient
from .types.call_telephony_request_properties import CallTelephonyRequestProperties
from .types.call_telephony_request_sip import CallTelephonyRequestSip
from .types.call_telephony_response import CallTelephonyResponse
from .types.get_telephony_response import GetTelephonyResponse
from .types.hangup_telephony_response import HangupTelephonyResponse
from .types.list_telephony_request_type import ListTelephonyRequestType
from .types.list_telephony_response import ListTelephonyResponse
from .types.list_telephony_response_data_list_item import ListTelephonyResponseDataListItem

# this is used as the default value for optional parameters
OMIT = typing.cast(typing.Any, ...)


class TelephonyClient:
    def __init__(self, *, client_wrapper: SyncClientWrapper):
        self._raw_client = RawTelephonyClient(client_wrapper=client_wrapper)

    @property
    def with_raw_response(self) -> RawTelephonyClient:
        """
        Retrieves a raw implementation of this client that returns raw responses.

        Returns
        -------
        RawTelephonyClient
        """
        return self._raw_client

    def list(
        self,
        appid: str,
        *,
        number: typing.Optional[str] = None,
        from_time: typing.Optional[int] = None,
        to_time: typing.Optional[int] = None,
        type: typing.Optional[ListTelephonyRequestType] = None,
        limit: typing.Optional[int] = None,
        cursor: typing.Optional[str] = None,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> SyncPager[ListTelephonyResponseDataListItem, ListTelephonyResponse]:
        """
        Query historical call records for a specified appid based on the filter criteria.

        Parameters
        ----------
        appid : str
            The App ID of the project.

        number : typing.Optional[str]
            Filter by phone number. Can be either the calling number or the called number.

        from_time : typing.Optional[int]
            Query list start timestamp (in seconds). Default is 60 days ago.

        to_time : typing.Optional[int]
            Query list end timestamp (in seconds). Default is current time.

        type : typing.Optional[ListTelephonyRequestType]
            Call type filter:
            - `inbound`: Inbound call.
            - `outbound`: Outbound call.

            If not specified, all call types are returned.

        limit : typing.Optional[int]
            Maximum number of items returned in a single page.

        cursor : typing.Optional[str]
            Pagination cursor. Use the `agent_id` from the previous page as the cursor for the next page.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        SyncPager[ListTelephonyResponseDataListItem, ListTelephonyResponse]
            Request was successful. The response body contains the result of the request.

        Examples
        --------
        from agora_agent import Agora

        client = Agora(
            authorization="YOUR_AUTHORIZATION",
            username="YOUR_USERNAME",
            password="YOUR_PASSWORD",
        )
        response = client.telephony.list(
            appid="appid",
        )
        for item in response:
            yield item
        # alternatively, you can paginate page-by-page
        for page in response.iter_pages():
            yield page
        """
        return self._raw_client.list(
            appid,
            number=number,
            from_time=from_time,
            to_time=to_time,
            type=type,
            limit=limit,
            cursor=cursor,
            request_options=request_options,
        )

    def call(
        self,
        appid: str,
        *,
        name: str,
        sip: CallTelephonyRequestSip,
        properties: CallTelephonyRequestProperties,
        pipeline_id: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> CallTelephonyResponse:
        """
        Initiate an outbound call to a specified number and create an agent to join the specified RTC channel.

        Use this endpoint to initiate an outbound call to the specified number and create an agent that joins the target RTC channel. The agent waits for the callee to answer.

        Parameters
        ----------
        appid : str
            The App ID of the project.

        name : str
            The name identifier of the call session.

        sip : CallTelephonyRequestSip
            SIP (Session Initiation Protocol) call configuration object.

        properties : CallTelephonyRequestProperties
            Call attribute configuration. The content of this field varies depending on the invocation method:
            - **Using pipeline ID**: Simply pass in `channel`, `token`, `agent_rtc_uid`, and `remote_rtc_uids`.
            - **Using complete configuration**: Pass in the complete parameters of the [Start a conversational AI agent](https://docs.agora.io/en/conversational-ai/rest-api/agent/join) `properties`, including all required fields such as `channel`, `token`, `agent_rtc_uid`, `remote_rtc_uids`, `tts`, and `llm`.

        pipeline_id : typing.Optional[str]
            The unique ID of a published project in AI Studio.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        CallTelephonyResponse
            Request was successful. The response body contains the result of the request.

        Examples
        --------
        from agora_agent import Agora
        from agora_agent.telephony import (
            CallTelephonyRequestProperties,
            CallTelephonyRequestSip,
        )

        client = Agora(
            authorization="YOUR_AUTHORIZATION",
            username="YOUR_USERNAME",
            password="YOUR_PASSWORD",
        )
        client.telephony.call(
            appid="appid",
            name="customer_service",
            sip=CallTelephonyRequestSip(
                to_number="+19876543210",
                from_number="+11234567890",
                rtc_uid="100",
                rtc_token="<agora_sip_rtc_token>",
            ),
            pipeline_id="fzufjlweixxxxnlp",
            properties=CallTelephonyRequestProperties(
                channel="<agora_channel>",
                token="<agora_channel_token>",
                agent_rtc_uid="111",
                remote_rtc_uids=["100"],
            ),
        )
        """
        _response = self._raw_client.call(
            appid, name=name, sip=sip, properties=properties, pipeline_id=pipeline_id, request_options=request_options
        )
        return _response.data

    def get(
        self, appid: str, agent_id: str, *, request_options: typing.Optional[RequestOptions] = None
    ) -> GetTelephonyResponse:
        """
        Retrieve the call status and related information of a specified agent.

        Parameters
        ----------
        appid : str
            The App ID of the project.

        agent_id : str
            The agent ID you obtained after successfully calling the API to initiate an outbound call.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        GetTelephonyResponse
            Request was successful. The response body contains the result of the request.

        Examples
        --------
        from agora_agent import Agora

        client = Agora(
            authorization="YOUR_AUTHORIZATION",
            username="YOUR_USERNAME",
            password="YOUR_PASSWORD",
        )
        client.telephony.get(
            appid="appid",
            agent_id="agent_id",
        )
        """
        _response = self._raw_client.get(appid, agent_id, request_options=request_options)
        return _response.data

    def hangup(
        self, appid: str, agent_id: str, *, request_options: typing.Optional[RequestOptions] = None
    ) -> HangupTelephonyResponse:
        """
        Instruct the agent to proactively hang up the ongoing call and leave the RTC channel.

        Parameters
        ----------
        appid : str
            The App ID of the project.

        agent_id : str
            The agent ID you obtained after successfully calling the API to initiate an outbound call.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        HangupTelephonyResponse
            Request was successful. The response body is empty.

        Examples
        --------
        from agora_agent import Agora

        client = Agora(
            authorization="YOUR_AUTHORIZATION",
            username="YOUR_USERNAME",
            password="YOUR_PASSWORD",
        )
        client.telephony.hangup(
            appid="appid",
            agent_id="agent_id",
        )
        """
        _response = self._raw_client.hangup(appid, agent_id, request_options=request_options)
        return _response.data


class AsyncTelephonyClient:
    def __init__(self, *, client_wrapper: AsyncClientWrapper):
        self._raw_client = AsyncRawTelephonyClient(client_wrapper=client_wrapper)

    @property
    def with_raw_response(self) -> AsyncRawTelephonyClient:
        """
        Retrieves a raw implementation of this client that returns raw responses.

        Returns
        -------
        AsyncRawTelephonyClient
        """
        return self._raw_client

    async def list(
        self,
        appid: str,
        *,
        number: typing.Optional[str] = None,
        from_time: typing.Optional[int] = None,
        to_time: typing.Optional[int] = None,
        type: typing.Optional[ListTelephonyRequestType] = None,
        limit: typing.Optional[int] = None,
        cursor: typing.Optional[str] = None,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> AsyncPager[ListTelephonyResponseDataListItem, ListTelephonyResponse]:
        """
        Query historical call records for a specified appid based on the filter criteria.

        Parameters
        ----------
        appid : str
            The App ID of the project.

        number : typing.Optional[str]
            Filter by phone number. Can be either the calling number or the called number.

        from_time : typing.Optional[int]
            Query list start timestamp (in seconds). Default is 60 days ago.

        to_time : typing.Optional[int]
            Query list end timestamp (in seconds). Default is current time.

        type : typing.Optional[ListTelephonyRequestType]
            Call type filter:
            - `inbound`: Inbound call.
            - `outbound`: Outbound call.

            If not specified, all call types are returned.

        limit : typing.Optional[int]
            Maximum number of items returned in a single page.

        cursor : typing.Optional[str]
            Pagination cursor. Use the `agent_id` from the previous page as the cursor for the next page.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        AsyncPager[ListTelephonyResponseDataListItem, ListTelephonyResponse]
            Request was successful. The response body contains the result of the request.

        Examples
        --------
        import asyncio

        from agora_agent import AsyncAgora

        client = AsyncAgora(
            authorization="YOUR_AUTHORIZATION",
            username="YOUR_USERNAME",
            password="YOUR_PASSWORD",
        )


        async def main() -> None:
            response = await client.telephony.list(
                appid="appid",
            )
            async for item in response:
                yield item

            # alternatively, you can paginate page-by-page
            async for page in response.iter_pages():
                yield page


        asyncio.run(main())
        """
        return await self._raw_client.list(
            appid,
            number=number,
            from_time=from_time,
            to_time=to_time,
            type=type,
            limit=limit,
            cursor=cursor,
            request_options=request_options,
        )

    async def call(
        self,
        appid: str,
        *,
        name: str,
        sip: CallTelephonyRequestSip,
        properties: CallTelephonyRequestProperties,
        pipeline_id: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> CallTelephonyResponse:
        """
        Initiate an outbound call to a specified number and create an agent to join the specified RTC channel.

        Use this endpoint to initiate an outbound call to the specified number and create an agent that joins the target RTC channel. The agent waits for the callee to answer.

        Parameters
        ----------
        appid : str
            The App ID of the project.

        name : str
            The name identifier of the call session.

        sip : CallTelephonyRequestSip
            SIP (Session Initiation Protocol) call configuration object.

        properties : CallTelephonyRequestProperties
            Call attribute configuration. The content of this field varies depending on the invocation method:
            - **Using pipeline ID**: Simply pass in `channel`, `token`, `agent_rtc_uid`, and `remote_rtc_uids`.
            - **Using complete configuration**: Pass in the complete parameters of the [Start a conversational AI agent](https://docs.agora.io/en/conversational-ai/rest-api/agent/join) `properties`, including all required fields such as `channel`, `token`, `agent_rtc_uid`, `remote_rtc_uids`, `tts`, and `llm`.

        pipeline_id : typing.Optional[str]
            The unique ID of a published project in AI Studio.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        CallTelephonyResponse
            Request was successful. The response body contains the result of the request.

        Examples
        --------
        import asyncio

        from agora_agent import AsyncAgora
        from agora_agent.telephony import (
            CallTelephonyRequestProperties,
            CallTelephonyRequestSip,
        )

        client = AsyncAgora(
            authorization="YOUR_AUTHORIZATION",
            username="YOUR_USERNAME",
            password="YOUR_PASSWORD",
        )


        async def main() -> None:
            await client.telephony.call(
                appid="appid",
                name="customer_service",
                sip=CallTelephonyRequestSip(
                    to_number="+19876543210",
                    from_number="+11234567890",
                    rtc_uid="100",
                    rtc_token="<agora_sip_rtc_token>",
                ),
                pipeline_id="fzufjlweixxxxnlp",
                properties=CallTelephonyRequestProperties(
                    channel="<agora_channel>",
                    token="<agora_channel_token>",
                    agent_rtc_uid="111",
                    remote_rtc_uids=["100"],
                ),
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.call(
            appid, name=name, sip=sip, properties=properties, pipeline_id=pipeline_id, request_options=request_options
        )
        return _response.data

    async def get(
        self, appid: str, agent_id: str, *, request_options: typing.Optional[RequestOptions] = None
    ) -> GetTelephonyResponse:
        """
        Retrieve the call status and related information of a specified agent.

        Parameters
        ----------
        appid : str
            The App ID of the project.

        agent_id : str
            The agent ID you obtained after successfully calling the API to initiate an outbound call.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        GetTelephonyResponse
            Request was successful. The response body contains the result of the request.

        Examples
        --------
        import asyncio

        from agora_agent import AsyncAgora

        client = AsyncAgora(
            authorization="YOUR_AUTHORIZATION",
            username="YOUR_USERNAME",
            password="YOUR_PASSWORD",
        )


        async def main() -> None:
            await client.telephony.get(
                appid="appid",
                agent_id="agent_id",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.get(appid, agent_id, request_options=request_options)
        return _response.data

    async def hangup(
        self, appid: str, agent_id: str, *, request_options: typing.Optional[RequestOptions] = None
    ) -> HangupTelephonyResponse:
        """
        Instruct the agent to proactively hang up the ongoing call and leave the RTC channel.

        Parameters
        ----------
        appid : str
            The App ID of the project.

        agent_id : str
            The agent ID you obtained after successfully calling the API to initiate an outbound call.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        HangupTelephonyResponse
            Request was successful. The response body is empty.

        Examples
        --------
        import asyncio

        from agora_agent import AsyncAgora

        client = AsyncAgora(
            authorization="YOUR_AUTHORIZATION",
            username="YOUR_USERNAME",
            password="YOUR_PASSWORD",
        )


        async def main() -> None:
            await client.telephony.hangup(
                appid="appid",
                agent_id="agent_id",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.hangup(appid, agent_id, request_options=request_options)
        return _response.data