Class WebSocketHandler

evi/evi-python-function-calling/main.py:15–168

Handler for containing the EVI WebSocket and associated socket handling behavior.

from utils import print_prompt, extract_top_n_emotions, print_emotion_scores

class WebSocketHandler:
    """Handler for containing the EVI WebSocket and associated socket handling behavior."""

    def __init__(self):
        """Construct the WebSocketHandler, initially assigning the socket to None and the byte stream to a new Stream object."""
        self.socket = None
        self.byte_strs = Stream.new()

    def set_socket(self, socket: AsyncChatSocketClient):
        """Set the socket.
        
        This method assigns the provided asynchronous WebSocket connection
        to the instance variable `self.socket`. It is invoked after successfully
        establishing a connection using the client's connect method.

        Args:
            socket (AsyncChatSocketClient): EVI asynchronous WebSocket returned by the client's connect method.
        """
        self.socket = socket

    async def handle_tool_call(self, message: ToolCallMessage) -> Union[ToolCallMessage, ToolErrorMessage]:
        """Functionality which executes when a tool call is invoked.
        
        Args:
            message (ToolCallMessage): The message sent when a tool call is invoked. See it in the API Reference [here](https://dev.hume.ai/reference/empathic-voice-interface-evi/chat/chat#receive.Tool%20Call%20Message.name).
        
        Returns:
            Union[ToolResponseMessage, ToolErrorMessage]: Returns a [ToolResponseMessage](https://dev.hume.ai/reference/empathic-voice-interface-evi/chat/chat#send.Tool%20Response%20Message.type) if the tool call is succesful or a [ToolErrorMessage](https://dev.hume.ai/reference/empathic-voice-interface-evi/chat/chat#send.Tool%20Error%20Message.type) if the tool call fails. 
        """

        # Obtain the name, ID, and parameters of the tool call
        tool_name = message.name
        tool_call_id = message.tool_call_id

        # Parse the stringified JSON parameters into a dictionary
        try:
            tool_parameters = json.loads(message.parameters)
        except json.JSONDecodeError:
            resp = ToolErrorMessage(
                tool_call_id=tool_call_id,
                content="Invalid parameters format.",
                error="JSONDecodeError"
            )
            await self.socket.send_tool_error(resp)
            print(f"(Sent ToolErrorMessage for tool_call_id {tool_call_id} due to JSON decode error.)\n")
            return

        if tool_name == "get_current_weather":
            obtained_location = tool_parameters.get('location')
            obtained_format = tool_parameters.get('format', 'text')

            if not obtained_location:
                resp = ToolErrorMessage(
                    tool_call_id=tool_call_id,
                    content="Missing 'location' parameter.",
                    error="MissingParameter"
                )
                await self.socket.send_tool_error(resp)