client.py

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

import typing

from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper
from ..core.request_options import RequestOptions
from .raw_client import AsyncRawAgentManagementClient, RawAgentManagementClient
from .types.agent_think_agent_management_request_on_listening_action import (
    AgentThinkAgentManagementRequestOnListeningAction,
)
from .types.agent_think_agent_management_request_on_speaking_action import (
    AgentThinkAgentManagementRequestOnSpeakingAction,
)
from .types.agent_think_agent_management_request_on_thinking_action import (
    AgentThinkAgentManagementRequestOnThinkingAction,
)
from .types.agent_think_agent_management_response import AgentThinkAgentManagementResponse

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


class AgentManagementClient:
    def __init__(self, *, client_wrapper: SyncClientWrapper):
        self._raw_client = RawAgentManagementClient(client_wrapper=client_wrapper)

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

        Returns
        -------
        RawAgentManagementClient
        """
        return self._raw_client

    def agent_think(
        self,
        appid: str,
        agent_id: str,
        *,
        text: str,
        on_listening_action: typing.Optional[AgentThinkAgentManagementRequestOnListeningAction] = OMIT,
        on_thinking_action: typing.Optional[AgentThinkAgentManagementRequestOnThinkingAction] = OMIT,
        on_speaking_action: typing.Optional[AgentThinkAgentManagementRequestOnSpeakingAction] = OMIT,
        interruptable: typing.Optional[bool] = OMIT,
        metadata: typing.Optional[typing.Dict[str, str]] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> AgentThinkAgentManagementResponse:
        """
        Send a custom text instruction to the specified conversational AI agent instance.

        The instruction is injected into the current conversation pipeline as user input, and the agent processes and responds to it following the standard user input logic.

        Use this endpoint for the following scenarios:
        - **Implicit instruction injection**: Inject hidden context or directives into the conversation.
        - **Client-side event triggering**: Notify the agent of client-side events, such as a user clicking a button.
        - **Voice and text collaboration**: Combine text instructions with voice input for richer interaction.

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

        agent_id : str
            The agent instance ID you obtained after successfully calling `join` to start a conversational AI agent.

        text : str
            The custom instruction text to inject into the current conversation pipeline. The system processes this as user input.

        on_listening_action : typing.Optional[AgentThinkAgentManagementRequestOnListeningAction]
            The action to take when the agent is in a listening state:
            - `inject`: Inject the custom text instruction into the current turn without interrupting it.
            - `interrupt`: Immediately interrupt the current flow and initiate a new round of dialogue.
            - `ignore`: Ignore the request.

        on_thinking_action : typing.Optional[AgentThinkAgentManagementRequestOnThinkingAction]
            The action to take when the agent is in a thinking state:
            - `interrupt`: Interrupt the current state and start a new conversation turn.
            - `ignore`: Ignore the request.

        on_speaking_action : typing.Optional[AgentThinkAgentManagementRequestOnSpeakingAction]
            The action to take when the agent is in a speaking state:
            - `interrupt`: Interrupt the current state and start a new conversation turn.
            - `ignore`: Ignore the request.

        interruptable : typing.Optional[bool]
            Whether user speech can interrupt the injected instruction:
            - `true`: User speech can interrupt the instruction.
            - `false`: User speech cannot interrupt the instruction.

        metadata : typing.Optional[typing.Dict[str, str]]
            Custom metadata in key-value pair format. Use this field to pass additional business information such as identifiers or model references.

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

        Returns
        -------
        AgentThinkAgentManagementResponse
            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.agent_management.agent_think(
            appid="appid",
            agent_id="agentId",
            text="The user just clicked the purchase button.",
            on_listening_action="inject",
            on_thinking_action="interrupt",
            on_speaking_action="ignore",
            interruptable=True,
            metadata={"publisher": "user123", "model": "deepseek-r1"},
        )
        """
        _response = self._raw_client.agent_think(
            appid,
            agent_id,
            text=text,
            on_listening_action=on_listening_action,
            on_thinking_action=on_thinking_action,
            on_speaking_action=on_speaking_action,
            interruptable=interruptable,
            metadata=metadata,
            request_options=request_options,
        )
        return _response.data


class AsyncAgentManagementClient:
    def __init__(self, *, client_wrapper: AsyncClientWrapper):
        self._raw_client = AsyncRawAgentManagementClient(client_wrapper=client_wrapper)

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

        Returns
        -------
        AsyncRawAgentManagementClient
        """
        return self._raw_client

    async def agent_think(
        self,
        appid: str,
        agent_id: str,
        *,
        text: str,
        on_listening_action: typing.Optional[AgentThinkAgentManagementRequestOnListeningAction] = OMIT,
        on_thinking_action: typing.Optional[AgentThinkAgentManagementRequestOnThinkingAction] = OMIT,
        on_speaking_action: typing.Optional[AgentThinkAgentManagementRequestOnSpeakingAction] = OMIT,
        interruptable: typing.Optional[bool] = OMIT,
        metadata: typing.Optional[typing.Dict[str, str]] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> AgentThinkAgentManagementResponse:
        """
        Send a custom text instruction to the specified conversational AI agent instance.

        The instruction is injected into the current conversation pipeline as user input, and the agent processes and responds to it following the standard user input logic.

        Use this endpoint for the following scenarios:
        - **Implicit instruction injection**: Inject hidden context or directives into the conversation.
        - **Client-side event triggering**: Notify the agent of client-side events, such as a user clicking a button.
        - **Voice and text collaboration**: Combine text instructions with voice input for richer interaction.

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

        agent_id : str
            The agent instance ID you obtained after successfully calling `join` to start a conversational AI agent.

        text : str
            The custom instruction text to inject into the current conversation pipeline. The system processes this as user input.

        on_listening_action : typing.Optional[AgentThinkAgentManagementRequestOnListeningAction]
            The action to take when the agent is in a listening state:
            - `inject`: Inject the custom text instruction into the current turn without interrupting it.
            - `interrupt`: Immediately interrupt the current flow and initiate a new round of dialogue.
            - `ignore`: Ignore the request.

        on_thinking_action : typing.Optional[AgentThinkAgentManagementRequestOnThinkingAction]
            The action to take when the agent is in a thinking state:
            - `interrupt`: Interrupt the current state and start a new conversation turn.
            - `ignore`: Ignore the request.

        on_speaking_action : typing.Optional[AgentThinkAgentManagementRequestOnSpeakingAction]
            The action to take when the agent is in a speaking state:
            - `interrupt`: Interrupt the current state and start a new conversation turn.
            - `ignore`: Ignore the request.

        interruptable : typing.Optional[bool]
            Whether user speech can interrupt the injected instruction:
            - `true`: User speech can interrupt the instruction.
            - `false`: User speech cannot interrupt the instruction.

        metadata : typing.Optional[typing.Dict[str, str]]
            Custom metadata in key-value pair format. Use this field to pass additional business information such as identifiers or model references.

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

        Returns
        -------
        AgentThinkAgentManagementResponse
            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.agent_management.agent_think(
                appid="appid",
                agent_id="agentId",
                text="The user just clicked the purchase button.",
                on_listening_action="inject",
                on_thinking_action="interrupt",
                on_speaking_action="ignore",
                interruptable=True,
                metadata={"publisher": "user123", "model": "deepseek-r1"},
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.agent_think(
            appid,
            agent_id,
            text=text,
            on_listening_action=on_listening_action,
            on_thinking_action=on_thinking_action,
            on_speaking_action=on_speaking_action,
            interruptable=interruptable,
            metadata=metadata,
            request_options=request_options,
        )
        return _response.data