GeminiRealtimeClient

async def add_event(self, event: Optional[RealtimeEvent]):
    await self._eventQueue.put(event)

async def get_event(self) -> Optional[RealtimeEvent]:
    return await self._eventQueue.get()

async def queue_input_audio_buffer_delta(self, audio: str) -> None:
    """queue InputAudioBufferDelta.

    Args:
        audio (str): The audio.
    """
    await self.add_event(InputAudioBufferDelta(delta=audio, item_id=None, raw_message=dict()))

async def send_function_result(self, call_id: str, result: str) -> None:
    """Send the result of a function call to the Gemini Realtime API.

    Args:
        call_id (str): The ID of the function call.
        result (str): The result of the function call.
    """
    msg = {
        "tool_response": {"function_responses": [{"id": call_id, "response": {"result": {"string_value": result}}}]}
    }
    if self._is_reading_events:
        await self.connection.send(json.dumps(msg))

async def send_text(self, *, role: Role, text: str, turn_complete: bool = True) -> None:
    """Send a text message to the Gemini Realtime API.

    Args:
        role: The role of the message.
        text: The text of the message.
        turn_complete: A flag indicating if the turn is complete.
    """
    msg = {
        "client_content": {
            "turn_complete": turn_complete,
            "turns": [{"role": role, "parts": [{"text": text}]}],
        }
    }
    if self._is_reading_events:
        await self.connection.send(json.dumps(msg))

async def send_audio(self, audio: str) -> None:
    """Send audio to the Gemini Realtime API.

    Args:
        audio (str): The audio to send.
    """
    msg = {
        "realtime_input": {
            "media_chunks": [
                {
                    "data": audio,
                    "mime_type": "audio/pcm",
                }
            ]
        }
    }
    await self.queue_input_audio_buffer_delta(audio)
    if self._is_reading_events:
        await self.connection.send(json.dumps(msg))

async def truncate_audio(self, audio_end_ms: int, content_index: int, item_id: str) -> None:
    self.logger.info("This is not natively supported by Gemini Realtime API.")
    pass

async def session_update(self, session_options: dict[str, Any]) -> None:
    """Record session updates to be applied when the connection is established.

    Args:
        session_options (dict[str, Any]): The session options to update.
    """
    if self._is_reading_events:
        self.logger.warning("Is reading events. Session update will be ignored.")
    else:
        self._pending_session_updates.update(session_options)

@asynccontextmanager
async def connect(self) -> AsyncGenerator[None, None]:
    """Connect to the Gemini Realtime API."""
    try:
        async with connect(
            self._base_url, additional_headers={"Content-Type": "application/json"}
        ) as self._connection:
            yield
    finally:
        self._connection = None

async def read_events(self) -> AsyncGenerator[RealtimeEvent, None]:
    """Read Events from the Gemini Realtime Client"""
    if self._connection is None:
        raise RuntimeError("Client is not connected, call connect() first.")
    await self._initialize_session()

    self._is_reading_events = True

    async for event in self._read_events():
        yield event

@classmethod
def get_factory(
    cls, llm_config: dict[str, Any], logger: Logger, **kwargs: Any
) -> Optional[Callable[[], "RealtimeClientProtocol"]]:
    """Create a Realtime API client.

    Args:
        llm_config: The LLM config for the client.
        logger: The logger for the client.
        **kwargs: Additional arguments.

    Returns:
        RealtimeClientProtocol: The Realtime API client is returned if the model matches the pattern
    """
    if llm_config["config_list"][0].get("api_type") == "google" and list(kwargs.keys()) == []:
        return lambda: GeminiRealtimeClient(llm_config=llm_config, logger=logger, **kwargs)
    return None