# Quickstart (/en/realtime-media/interactive-live-streaming/quickstart/python)

> For AI agents: see the complete documentation index at [llms.txt](/llms.txt).

This page provides a step-by-step guide on how to create a basic Interactive Live Streaming app using the Agora Video SDK.

    ## Understand the tech [#understand-the-tech-12]

    To start a Interactive Live Streaming session, implement the following steps in your app:

    * **Initialize the Agora Engine**: Before calling other APIs, create and initialize an Agora Engine instance.

    * **Join a channel**: Call methods to create and join a channel.

      * **Join as a host**: A live streaming event has one or more hosts. A host publishes audio and video to the channel. Hosts can also subscribe to streams from other hosts.

      * **Join as audience**: Audience members can only subscribe to streams published by hosts.

    * **Send and receive audio and video**: Hosts publish streams to the channel. Audience members subscribe to audio and video streams published by hosts.

    ![Streaming workflow](https://assets-docs.agora.io/images/video-sdk/get-started-ils-bs.svg)

    ## Prerequisites [#prerequisites-12]

    * A camera and a microphone

    * A valid Agora account and project. Please refer to [Agora account management](manage-agora-account.md) for details.

    ## Set up your project [#set-up-your-project-12]

    This section shows you how to set up your Python project and install the Agora Video SDK.

    1. Install the build tools for compiling the SDK.

       ```text
       sudo apt install build-essential python3-dev
       ```

    2. To implement structured, asynchronous event handling in Python, install the `pyee` library.

       ```bash
       pip3 install pyee
       ```

    3. Install the Agora server side Python SDK.

       ```bash
       pip3 install agora-python-server-sdk
       ```

    <CalloutContainer type="info">
      <CalloutDescription>
        The Python SDK is a server side SDK.
      </CalloutDescription>
    </CalloutContainer>

    ## Implement Interactive Live Streaming [#implement-interactive-live-streaming-12]

    This section guides you through the implementation of basic real-time audio and video interaction in your app.

    ### Import Agora classes [#import-agora-classes-2]

    Import the relevant Agora SDK classes and interfaces:

    ```python
    from agora.rtc.agora_base import (
      AudioScenarioType,
      ChannelProfileType,
      ClientRoleType,
    )
    from agora.rtc.agora_service import (
      AgoraService,
      AgoraServiceConfig,
      RTCConnConfig,
    )
    from agora.rtc.audio_frame_observer import AudioFrame, IAudioFrameObserver
    from agora.rtc.audio_pcm_data_sender import PcmAudioFrame
    from agora.rtc.local_user import LocalUser
    from agora.rtc.local_user_observer import IRTCLocalUserObserver
    from agora.rtc.rtc_connection import RTCConnection, RTCConnInfo
    from agora.rtc.rtc_connection_observer import IRTCConnectionObserver
    ```

    ### Initialize the engine [#initialize-the-engine-9]

    The following code defines the `RtcEngine` class, which initializes and configures the `AgoraService`. The class constructor takes an `appid` as input, configures the Agora service, and initializes it. You use the `RtcEngine` class to interact with the Agora SDK in this demo.

    ```python
    class RtcEngine:
      def __init__(self, appid: str, appcert: str):
        self.appid = appid
        self.appcert = appcert

        if not appid:
          raise Exception("App ID is required)")

        config = AgoraServiceConfig()
        config.audio_scenario = AudioScenarioType.AUDIO_SCENARIO_CHORUS
        config.appid = appid
        config.log_path = os.path.join(
          os.path.dirname(
            os.path.dirname(
              os.path.dirname(os.path.join(os.path.abspath(__file__)))
            )
          ),
          "agorasdk.log",
        )
        self.agora_service = AgoraService()
        self.agora_service.initialize(config)
    ```

    ### Join a channel [#join-a-channel-11]

    To asynchronously join a channel, implement a `Channel` class. When you create an instance of the class, the initializer sets up the necessary components for joining a channel. It takes an instance of `RtcEngine`, a `channelId`, and a `uid` as parameters. During initialization, the code creates an event emitter, configures the connection for broadcasting, and registers an event observer for channel events. It also sets up the local user’s audio configuration to enable audio streaming.

    <CalloutContainer type="info">
      <CalloutDescription>
        UIDs in the Python SDK are set using a string value. Agora recommends using only numerical values for UID strings to ensure compatibility with all Agora products and extensions.
      </CalloutDescription>
    </CalloutContainer>

    ```python
    class Channel:
      def __init__(self, rtc: "RtcEngine", options: RtcOptions) -> None:
        self.loop = asyncio.get_event_loop()

        # Create the event emitter
        self.emitter = AsyncIOEventEmitter(self.loop)

        self.connection_state = 0
        self.options = options
        self.remote_users = dict[int, Any]()
        self.rtc = rtc
        self.chat = Chat(self)
        self.channelId = options.channel_name
        self.uid = options.uid
        self.enable_pcm_dump = options.enable_pcm_dump
        self.token = options.build_token(rtc.appid, rtc.appcert) if rtc.appcert else ""
        conn_config = RTCConnConfig(
          client_role_type=ClientRoleType.CLIENT_ROLE_BROADCASTER,
          channel_profile=ChannelProfileType.CHANNEL_PROFILE_LIVE_BROADCASTING,
        )
        self.connection = self.rtc.agora_service.create_rtc_connection(conn_config)

        self.channel_event_observer = ChannelEventObserver(
          self.emitter,
          options=options,
        )
        self.connection.register_observer(self.channel_event_observer)

        self.local_user = self.connection.get_local_user()
        self.local_user.set_playback_audio_frame_before_mixing_parameters(
          options.channels, options.sample_rate
        )
        self.local_user.register_local_user_observer(self.channel_event_observer)
        self.local_user.register_audio_frame_observer(self.channel_event_observer)
        # self.local_user.subscribe_all_audio()

        self.media_node_factory = self.rtc.agora_service.create_media_node_factory()
        self.audio_pcm_data_sender = (
          self.media_node_factory.create_audio_pcm_data_sender()
        )
        self.audio_track = self.rtc.agora_service.create_custom_audio_track_pcm(
          self.audio_pcm_data_sender
        )
        self.audio_track.set_enabled(1)
        self.local_user.publish_audio(self.audio_track)

        self.stream_id = self.connection.create_data_stream(False, False)
        self.received_chunks = {}
        self.waiting_message = None
        self.msg_id = ""
        self.msg_index = ""

        self.on(
          "user_joined",
          lambda agora_rtc_conn, user_id: self.remote_users.update({user_id: True}),
        )
        self.on(
          "user_left",
          lambda agora_rtc_conn, user_id, reason: self.remote_users.pop(
            user_id, None
          ),
        )
    ```

    The following code uses the `Channel` class to join a channel. It sets up a `future` to handle the connection state and returns a `Channel` object when the connection is successfully established.

    ```python
    async def connect(self) -> None:
      """
      Connects to a channel.

      Parameters:
        channelId: The channel ID.
        uid: The user ID.

      Returns:
        Channel: The connected channel.
      """
      if self.connection_state == 3:
        return

      future = asyncio.Future()

      def callback(agora_rtc_conn: RTCConnection, conn_info: RTCConnInfo, reason):
        logger.info(f"Connection state changed: {conn_info.state}")
        if conn_info.state == 3: # Connection successful
          future.set_result(None)
        elif conn_info.state == 5: # Connection failed
          future.set_exception(
            Exception(f"Connection failed with state: {conn_info.state}")
          )

      self.on("connection_state_changed", callback)
      logger.info(f"Connecting to channel {self.channelId} with token {self.token}")
      self.connection.connect(self.token, self.channelId, f"{self.uid}")

      if self.enable_pcm_dump:
        agora_parameter = self.connection.get_agora_parameter()
        agora_parameter.set_parameters("{\"che.audio.frame_dump\":{\"location\":\"all\",\"action\":\"start\",\"max_size_bytes\":\"120000000\",\"uuid\":\"123456789\",\"duration\":\"1200000\"}}")

      try:
        await future
      except Exception as e:
        raise Exception(
          f"Failed to connect to channel {self.channelId}: {str(e)}"
        ) from e
      finally:
        self.off("connection_state_changed", callback)
    ```

    ### Handle connection and channel events [#handle-connection-and-channel-events]

    To listen for channel and connection events, such as users joining or leaving the channel, and connection state changes, implement the `ChannelEventObserver` class. This class enables you to respond to SDK events.

    ```python
    class ChannelEventObserver(
      IRTCConnectionObserver, IRTCLocalUserObserver, IAudioFrameObserver
    ):
      def __init__(self, event_emitter: AsyncIOEventEmitter, options: RtcOptions) -> None:
        self.loop = asyncio.get_event_loop()
        self.emitter = event_emitter
        self.audio_streams = dict[int, AudioStream]()
        self.options = options

      def emit_event(self, event_name: str, *args):
        """Helper function to emit events."""
        self.loop.call_soon_threadsafe(self.emitter.emit, event_name, *args)

      def on_connected(
        self, agora_rtc_conn: RTCConnection, conn_info: RTCConnInfo, reason
      ):
        logger.info(f"Connected to RTC: {agora_rtc_conn} {conn_info} {reason}")
        self.emit_event("connection_state_changed", agora_rtc_conn, conn_info, reason)

      def on_disconnected(
        self, agora_rtc_conn: RTCConnection, conn_info: RTCConnInfo, reason
      ):
        logger.info(f"Disconnected from RTC: {agora_rtc_conn} {conn_info} {reason}")
        self.emit_event("connection_state_changed", agora_rtc_conn, conn_info, reason)

      def on_connecting(
        self, agora_rtc_conn: RTCConnection, conn_info: RTCConnInfo, reason
      ):
        logger.info(f"Connecting to RTC: {agora_rtc_conn} {conn_info} {reason}")
        self.emit_event("connection_state_changed", agora_rtc_conn, conn_info, reason)

      def on_connection_failure(self, agora_rtc_conn, conn_info, reason):
        logger.error(f"Connection failure: {agora_rtc_conn} {conn_info} {reason}")
        self.emit_event("connection_state_changed", agora_rtc_conn, conn_info, reason)

      def on_user_joined(self, agora_rtc_conn: RTCConnection, user_id):
        logger.info(f"User joined: {agora_rtc_conn} {user_id}")
        self.emit_event("user_joined", agora_rtc_conn, user_id)

      def on_user_left(self, agora_rtc_conn: RTCConnection, user_id, reason):
        logger.info(f"User left: {agora_rtc_conn} {user_id} {reason}")
        self.emit_event("user_left", agora_rtc_conn, user_id, reason)

      def handle_received_chunk(self, json_chunk):
        chunk = json.loads(json_chunk)
        msg_id = chunk["msg_id"]
        part_idx = chunk["part_idx"]
        total_parts = chunk["total_parts"]
        if msg_id not in self.received_chunks:
          self.received_chunks[msg_id] = {"parts": {}, "total_parts": total_parts}
        if (
          part_idx not in self.received_chunks[msg_id]["parts"]
          and 0 <= part_idx < total_parts
        ):
          self.received_chunks[msg_id]["parts"][part_idx] = chunk
          if len(self.received_chunks[msg_id]["parts"]) == total_parts:
            # all parts received, now recomposing original message and get rid it from dict
            sorted_parts = sorted(
              self.received_chunks[msg_id]["parts"].values(),
              key=lambda c: c["part_idx"],
            )
            full_message = "".join(part["content"] for part in sorted_parts)
            del self.received_chunks[msg_id]
            return full_message, msg_id
        return (None, None)

      def on_stream_message(
        self, agora_local_user: LocalUser, user_id, stream_id, data, length
      ):
        # logger.info(f"Stream message", agora_local_user, user_id, stream_id, length)
        (reassembled_message, msg_id) = self.handle_received_chunk(data)
        if reassembled_message is not None:
          logger.info(f"Reassembled message: {msg_id} {reassembled_message}")

      def on_audio_subscribe_state_changed(
        self,
        agora_local_user,
        channel,
        user_id,
        old_state,
        new_state,
        elapse_since_last_state,
      ):
        logger.info(
          f"Audio subscribe state changed: {user_id} {new_state} {elapse_since_last_state}"
        )
        self.emit_event(
          "audio_subscribe_state_changed",
          agora_local_user,
          channel,
          user_id,
          old_state,
          new_state,
          elapse_since_last_state,
        )

      def on_playback_audio_frame_before_mixing(
        self, agora_local_user: LocalUser, channelId, uid, frame: AudioFrame
      ):
        audio_frame = PcmAudioFrame()
        audio_frame.samples_per_channel = frame.samples_per_channel
        audio_frame.bytes_per_sample = frame.bytes_per_sample
        audio_frame.number_of_channels = frame.channels
        audio_frame.sample_rate = self.options.sample_rate
        audio_frame.data = frame.buffer

        self.loop.call_soon_threadsafe(
          self.audio_streams[uid].queue.put_nowait, audio_frame
        )
        return 0
    ```

    ### Subscribe to an audio stream [#subscribe-to-an-audio-stream]

    To asynchronously subscribe to audio streams for a specific user identified by their `uid`, refer to the following code. The method sets up a callback to monitor changes in the audio subscription state and handles the result based on whether the subscription is successfully established.

    ```python
    async def subscribe_audio(self, uid: int) -> None:
      """
      Subscribes to the audio of a user.

      Parameters:
        uid: The user ID to subscribe to.
      """
      future = asyncio.Future()

      def callback(
        agora_local_user,
        channel,
        user_id,
        old_state,
        new_state,
        elapse_since_last_state,
      ):
        if new_state == 3: # Successfully subscribed
          future.set_result(None)

      self.on("audio_subscribe_state_changed", callback)
      self.local_user.subscribe_audio(uid)

      try:
        await future
      except Exception as e:
        raise Exception(
          f"Audio subscription failed for user {uid}: {str(e)}"
        ) from e
      finally:
        self.off("audio_subscribe_state_changed", callback)
    ```

    ### Unsubscribe from an audio stream [#unsubscribe-from-an-audio-stream]

    To unsubscribe from an audio stream, implement an asynchronous method similar to `subscribe_audio` and use the following code to unsubscribe:

    ```python
    self.local_user.unsubscribe_audio(uid)
    ```

    ### Disconnect from the service [#disconnect-from-the-service]

    To leave a channel, disconnect from Agora SDRTN® and release resources, refer to he following code.

    ```python
    async def disconnect(self) -> None:
      """
      Disconnects the channel.
      """
      if self.connection_state == 1:
        return

      disconnected_future = asyncio.Future[None]()

      def callback(agora_rtc_conn: RTCConnection, conn_info: RTCConnInfo, reason):
        self.off("connection_state_changed", callback)
        if conn_info.state == 1:
          disconnected_future.set_result(None)

      self.on("connection_state_changed", callback)
      self.connection.disconnect()
      await disconnected_future
    ```

    ### Complete code [#complete-code]

    The `rtc.py` script integrates the code components presented in this section into reusable Python classes that you can extend for your own applications.

    **Complete code for `rtc.py`**

    ```python
    import asyncio
    import json
    import logging
    import os
    from typing import Any, AsyncIterator

    from agora.rtc.agora_base import (
      AudioScenarioType,
      ChannelProfileType,
      ClientRoleType,
    )
    from agora.rtc.agora_service import (
      AgoraService,
      AgoraServiceConfig,
      RTCConnConfig,
    )
    from agora.rtc.audio_frame_observer import AudioFrame, IAudioFrameObserver
    from agora.rtc.audio_pcm_data_sender import PcmAudioFrame
    from agora.rtc.local_user import LocalUser
    from agora.rtc.local_user_observer import IRTCLocalUserObserver
    from agora.rtc.rtc_connection import RTCConnection, RTCConnInfo
    from agora.rtc.rtc_connection_observer import IRTCConnectionObserver
    from pyee.asyncio import AsyncIOEventEmitter

    from .logger import setup_logger
    from .token_builder.realtimekit_token_builder import RealtimekitTokenBuilder

    # Set up the logger with color and timestamp support
    logger = setup_logger(name=__name__, log_level=logging.INFO)

    class RtcOptions:
      def __init__(
        self,
        *,
        channel_name: str = None,
        uid: int = 0,
        sample_rate: int = 24000,
        channels: int = 1,
        enable_pcm_dump: bool = False,
      ):
        self.channel_name = channel_name
        self.uid = uid
        self.sample_rate = sample_rate
        self.channels = channels
        self.enable_pcm_dump = enable_pcm_dump

      def build_token(self, appid: str, appcert: str) -> str:
        return RealtimekitTokenBuilder.build_token(
          appid, appcert, self.channel_name, self.uid
        )

    class AudioStream:
      def __init__(self) -> None:
        self.queue: asyncio.Queue = asyncio.Queue()

      def __aiter__(self) -> AsyncIterator[PcmAudioFrame]:
        return self

      async def __anext__(self) -> PcmAudioFrame:
        item = await self.queue.get()
        if item is None:
          raise StopAsyncIteration

        return item

    class ChannelEventObserver(
      IRTCConnectionObserver, IRTCLocalUserObserver, IAudioFrameObserver
    ):
      def __init__(self, event_emitter: AsyncIOEventEmitter, options: RtcOptions) -> None:
        self.loop = asyncio.get_event_loop()
        self.emitter = event_emitter
        self.audio_streams = dict[int, AudioStream]()
        self.options = options

      def emit_event(self, event_name: str, *args):
        """Helper function to emit events."""
        self.loop.call_soon_threadsafe(self.emitter.emit, event_name, *args)

      def on_connected(
        self, agora_rtc_conn: RTCConnection, conn_info: RTCConnInfo, reason
      ):
        logger.info(f"Connected to RTC: {agora_rtc_conn} {conn_info} {reason}")
        self.emit_event("connection_state_changed", agora_rtc_conn, conn_info, reason)

      def on_disconnected(
        self, agora_rtc_conn: RTCConnection, conn_info: RTCConnInfo, reason
      ):
        logger.info(f"Disconnected from RTC: {agora_rtc_conn} {conn_info} {reason}")
        self.emit_event("connection_state_changed", agora_rtc_conn, conn_info, reason)

      def on_connecting(
        self, agora_rtc_conn: RTCConnection, conn_info: RTCConnInfo, reason
      ):
        logger.info(f"Connecting to RTC: {agora_rtc_conn} {conn_info} {reason}")
        self.emit_event("connection_state_changed", agora_rtc_conn, conn_info, reason)

      def on_connection_failure(self, agora_rtc_conn, conn_info, reason):
        logger.error(f"Connection failure: {agora_rtc_conn} {conn_info} {reason}")
        self.emit_event("connection_state_changed", agora_rtc_conn, conn_info, reason)

      def on_user_joined(self, agora_rtc_conn: RTCConnection, user_id):
        logger.info(f"User joined: {agora_rtc_conn} {user_id}")
        self.emit_event("user_joined", agora_rtc_conn, user_id)

      def on_user_left(self, agora_rtc_conn: RTCConnection, user_id, reason):
        logger.info(f"User left: {agora_rtc_conn} {user_id} {reason}")
        self.emit_event("user_left", agora_rtc_conn, user_id, reason)

      def handle_received_chunk(self, json_chunk):
        chunk = json.loads(json_chunk)
        msg_id = chunk["msg_id"]
        part_idx = chunk["part_idx"]
        total_parts = chunk["total_parts"]
        if msg_id not in self.received_chunks:
          self.received_chunks[msg_id] = {"parts": {}, "total_parts": total_parts}
        if (
          part_idx not in self.received_chunks[msg_id]["parts"]
          and 0 <= part_idx < total_parts
        ):
          self.received_chunks[msg_id]["parts"][part_idx] = chunk
          if len(self.received_chunks[msg_id]["parts"]) == total_parts:
            # all parts received, now recomposing original message and get rid it from dict
            sorted_parts = sorted(
              self.received_chunks[msg_id]["parts"].values(),
              key=lambda c: c["part_idx"],
            )
            full_message = "".join(part["content"] for part in sorted_parts)
            del self.received_chunks[msg_id]
            return full_message, msg_id
        return (None, None)

      def on_stream_message(
        self, agora_local_user: LocalUser, user_id, stream_id, data, length
      ):
        # logger.info(f"Stream message", agora_local_user, user_id, stream_id, length)
        (reassembled_message, msg_id) = self.handle_received_chunk(data)
        if reassembled_message is not None:
          logger.info(f"Reassembled message: {msg_id} {reassembled_message}")

      def on_audio_subscribe_state_changed(
        self,
        agora_local_user,
        channel,
        user_id,
        old_state,
        new_state,
        elapse_since_last_state,
      ):
        logger.info(
          f"Audio subscribe state changed: {user_id} {new_state} {elapse_since_last_state}"
        )
        self.emit_event(
          "audio_subscribe_state_changed",
          agora_local_user,
          channel,
          user_id,
          old_state,
          new_state,
          elapse_since_last_state,
        )

      def on_playback_audio_frame_before_mixing(
        self, agora_local_user: LocalUser, channelId, uid, frame: AudioFrame
      ):
        audio_frame = PcmAudioFrame()
        audio_frame.samples_per_channel = frame.samples_per_channel
        audio_frame.bytes_per_sample = frame.bytes_per_sample
        audio_frame.number_of_channels = frame.channels
        audio_frame.sample_rate = self.options.sample_rate
        audio_frame.data = frame.buffer

        # print(
        #   "on_playback_audio_frame_before_mixing",
        #   audio_frame.samples_per_channel,
        #   audio_frame.bytes_per_sample,
        #   audio_frame.number_of_channels,
        #   audio_frame.sample_rate,
        #   len(audio_frame.data),
        # )
        self.loop.call_soon_threadsafe(
          self.audio_streams[uid].queue.put_nowait, audio_frame
        )
        return 0

    class Channel:
      def __init__(self, rtc: "RtcEngine", options: RtcOptions) -> None:
        self.loop = asyncio.get_event_loop()

        # Create the event emitter
        self.emitter = AsyncIOEventEmitter(self.loop)

        self.connection_state = 0
        self.options = options
        self.remote_users = dict[int, Any]()
        self.rtc = rtc
        self.chat = Chat(self)
        self.channelId = options.channel_name
        self.uid = options.uid
        self.enable_pcm_dump = options.enable_pcm_dump
        self.token = options.build_token(rtc.appid, rtc.appcert) if rtc.appcert else ""
        conn_config = RTCConnConfig(
          client_role_type=ClientRoleType.CLIENT_ROLE_BROADCASTER,
          channel_profile=ChannelProfileType.CHANNEL_PROFILE_LIVE_BROADCASTING,
        )
        self.connection = self.rtc.agora_service.create_rtc_connection(conn_config)

        self.channel_event_observer = ChannelEventObserver(
          self.emitter,
          options=options,
        )
        self.connection.register_observer(self.channel_event_observer)

        self.local_user = self.connection.get_local_user()
        self.local_user.set_playback_audio_frame_before_mixing_parameters(
          options.channels, options.sample_rate
        )
        self.local_user.register_local_user_observer(self.channel_event_observer)
        self.local_user.register_audio_frame_observer(self.channel_event_observer)
        # self.local_user.subscribe_all_audio()

        self.media_node_factory = self.rtc.agora_service.create_media_node_factory()
        self.audio_pcm_data_sender = (
          self.media_node_factory.create_audio_pcm_data_sender()
        )
        self.audio_track = self.rtc.agora_service.create_custom_audio_track_pcm(
          self.audio_pcm_data_sender
        )
        self.audio_track.set_enabled(1)
        self.local_user.publish_audio(self.audio_track)

        self.stream_id = self.connection.create_data_stream(False, False)
        self.received_chunks = {}
        self.waiting_message = None
        self.msg_id = ""
        self.msg_index = ""

        self.on(
          "user_joined",
          lambda agora_rtc_conn, user_id: self.remote_users.update({user_id: True}),
        )
        self.on(
          "user_left",
          lambda agora_rtc_conn, user_id, reason: self.remote_users.pop(
            user_id, None
          ),
        )

        def handle_audio_subscribe_state_changed(
          agora_local_user,
          channel,
          user_id,
          old_state,
          new_state,
          elapse_since_last_state,
        ):
          if new_state == 3: # Successfully subscribed
            self.channel_event_observer.audio_streams.update(
              {user_id: AudioStream()}
            )
          elif new_state == 0:
            self.channel_event_observer.audio_streams.pop(user_id, None)

        self.on("audio_subscribe_state_changed", handle_audio_subscribe_state_changed)
        self.on(
          "connection_state_changed",
          lambda agora_rtc_conn, conn_info, reason: setattr(
            self, "connection_state", conn_info.state
          ),
        )

      async def connect(self) -> None:
        """
        Connects to a channel.

        Parameters:
          channelId: The channel ID.
          uid: The user ID.

        Returns:
          Channel: The connected channel.
        """
        if self.connection_state == 3:
          return

        future = asyncio.Future()

        def callback(agora_rtc_conn: RTCConnection, conn_info: RTCConnInfo, reason):
          logger.info(f"Connection state changed: {conn_info.state}")
          if conn_info.state == 3: # Connection successful
            future.set_result(None)
          elif conn_info.state == 5: # Connection failed
            future.set_exception(
              Exception(f"Connection failed with state: {conn_info.state}")
            )

        self.on("connection_state_changed", callback)
        logger.info(f"Connecting to channel {self.channelId} with token {self.token}")
        self.connection.connect(self.token, self.channelId, f"{self.uid}")

        if self.enable_pcm_dump:
          agora_parameter = self.connection.get_agora_parameter()
          agora_parameter.set_parameters("{"che.audio.frame_dump":{"location":"all","action":"start","max_size_bytes":"120000000","uuid":"123456789","duration":"1200000"}}")

        try:
          await future
        except Exception as e:
          raise Exception(
            f"Failed to connect to channel {self.channelId}: {str(e)}"
          ) from e
        finally:
          self.off("connection_state_changed", callback)

      async def disconnect(self) -> None:
        """
        Disconnects the channel.
        """
        if self.connection_state == 1:
          return

        disconnected_future = asyncio.Future[None]()

        def callback(agora_rtc_conn: RTCConnection, conn_info: RTCConnInfo, reason):
          self.off("connection_state_changed", callback)
          if conn_info.state == 1:
            disconnected_future.set_result(None)

        self.on("connection_state_changed", callback)
        self.connection.disconnect()
        await disconnected_future

      def get_audio_frames(self, uid: int) -> AudioStream:
        """
        Returns the audio frames from the channel.

        Returns:
          AudioStream: The audio stream.
        """
        return self.channel_event_observer.audio_streams[uid]

      async def push_audio_frame(self, frame: bytes) -> None:
        """
        Pushes an audio frame to the channel.

        Parameters:
          frame: The audio frame to push.
        """
        audio_frame = PcmAudioFrame()
        audio_frame.data = bytearray(frame)
        audio_frame.timestamp = 0
        audio_frame.bytes_per_sample = 2
        audio_frame.number_of_channels = self.options.channels
        audio_frame.sample_rate = self.options.sample_rate
        audio_frame.samples_per_channel = int(
          len(frame) / audio_frame.bytes_per_sample / audio_frame.number_of_channels
        )

        ret = self.audio_pcm_data_sender.send_audio_pcm_data(audio_frame)
        logger.info(f"Pushed audio frame: {ret}, audio frame length: {len(frame)}")
        if ret < 0:
          raise Exception(f"Failed to send audio frame: {ret}, audio frame length: {len(frame)}")

      async def clear_sender_audio_buffer(self) -> None:
        """
        Clears the audio buffer which is used to send.
        """
        self.audio_track.clear_sender_buffer()

      async def subscribe_audio(self, uid: int) -> None:
        """
        Subscribes to the audio of a user.

        Parameters:
          uid: The user ID to subscribe to.
        """
        future = asyncio.Future()

        def callback(
          agora_local_user,
          channel,
          user_id,
          old_state,
          new_state,
          elapse_since_last_state,
        ):
          if new_state == 3: # Successfully subscribed
            future.set_result(None)
          # elif new_state == 1: # Subscription failed
          #   future.set_exception(
          #     Exception(
          #       f"Failed to subscribe {user_id} audio: state changed from {old_state} to {new_state}"
          #     )
          #   )

        self.on("audio_subscribe_state_changed", callback)
        self.local_user.subscribe_audio(uid)

        try:
          await future
        except Exception as e:
          raise Exception(
            f"Audio subscription failed for user {uid}: {str(e)}"
          ) from e
        finally:
          self.off("audio_subscribe_state_changed", callback)

      async def unsubscribe_audio(self, uid: int) -> None:
        """
        Unsubscribes from the audio of a user.

        Parameters:
          uid: The user ID to unsubscribe from.
        """
        future = asyncio.Future()

        def callback(
          agora_local_user,
          channel,
          user_id,
          old_state,
          new_state,
          elapse_since_last_state,
        ):
          if new_state == 3: # Successfully unsubscribed
            future.set_result(None)
          else: # Failed to unsubscribe
            future.set_exception(
              Exception(
                f"Failed to unsubscribe {user_id} audio: state changed from {old_state} to {new_state}"
              )
            )

        self.on("audio_subscribe_state_changed", callback)
        self.local_user.unsubscribe_audio(uid)

        try:
          await future
        except Exception as e:
          raise Exception(
            f"Audio unsubscription failed for user {uid}: {str(e)}"
          ) from e
        finally:
          self.off("audio_subscribe_state_changed", callback)

      def _split_string_into_chunks(
        self, long_string, msg_id, chunk_size=300
      ) -> list[dict[str:Any]]:
        """
        Splits a long string into chunks of a given size.

        Parameters:
          long_string: The string to split.
          msg_id: The message ID.
          chunk_size: The size of each chunk.

        Returns:
          list[dict[str: Any]]: The list of chunks.

        """
        total_parts = (len(long_string) + chunk_size - 1) // chunk_size
        json_chunks = []
        for idx in range(total_parts):
          start = idx * chunk_size
          end = min(start + chunk_size, len(long_string))
          chunk = {
            "msg_id": msg_id,
            "part_idx": idx,
            "total_parts": total_parts,
            "content": long_string[start:end],
          }
          json_chunk = json.dumps(chunk, ensure_ascii=False)
          json_chunks.append(json_chunk)
        return json_chunks

      async def send_stream_message(self, data: str, msg_id: str) -> None:
        """
        Sends a stream message to the channel.

        Parameters:
          data: The data to send.
          msg_id: The message ID.
        """

        chunks = self._split_string_into_chunks(data, msg_id)
        for chunk in chunks:
          self.connection.send_stream_message(self.stream_id, chunk)

      def on(self, event_name: str, callback):
        """
        Allows external components to subscribe to events.

        Parameters:
          event_name: The name of the event to subscribe to.
          callback: The callback to call when the event is emitted.

        """
        self.emitter.on(event_name, callback)

      def once(self, event_name: str, callback):
        """
        Allows external components to subscribe to events once.

        Parameters:
          event_name: The name of the event to subscribe to.
          callback: The callback to call when the event is emitted.
        """
        self.emitter.once(event_name, callback)

      def off(self, event_name: str, callback):
        """
        Allows external components to unsubscribe from events.

        Parameters:
          event_name: The name of the event to unsubscribe from.
          callback: The callback to remove from the event.
        """
        self.emitter.remove_listener(event_name, callback)

    class ChatMessage:
      def __init__(self, message: str, msg_id: str) -> None:
        self.message = message
        self.msg_id = msg_id

    class Chat:
      def __init__(self, channel: Channel) -> None:
        self.channel = channel
        self.loop = self.channel.loop
        self.queue = asyncio.Queue()

        def log_exception(t: asyncio.Task[Any]) -> None:
          if not t.cancelled() and t.exception():
            logger.error(
              "unhandled exception",
              exc_info=t.exception(),
            )

        asyncio.create_task(self._process_message()).add_done_callback(log_exception)

      async def send_message(self, item: ChatMessage) -> None:
        """
        Sends a message to the channel.

        Parameters:
          item: The message to send.
        """
        await self.queue.put(item)
        # await self.queue.put_nowait(item)

      async def _process_message(self) -> None:
        """
        Processes messages in the queue.
        """

        while True:
          item: ChatMessage = await self.queue.get()
          await self.channel.send_stream_message(item.message, item.msg_id)
          self.queue.task_done()
          # await asyncio.sleep(0)

    class RtcEngine:
      def __init__(self, appid: str, appcert: str):
        self.appid = appid
        self.appcert = appcert

        if not appid:
          raise Exception("App ID is required)")

        config = AgoraServiceConfig()
        config.audio_scenario = AudioScenarioType.AUDIO_SCENARIO_CHORUS
        config.appid = appid
        config.log_path = os.path.join(
          os.path.dirname(
            os.path.dirname(
              os.path.dirname(os.path.join(os.path.abspath(__file__)))
            )
          ),
          "agorasdk.log",
        )
        self.agora_service = AgoraService()
        self.agora_service.initialize(config)

      def create_channel(self, options: RtcOptions) -> Channel:
        """
        Creates a channel.

        Parameters:
          channelId: The channel ID.
          uid: The user ID.

        Returns:
          Channel: The created channel.
        """
        return Channel(self, options)

      def destroy(self) -> None:
        """
        Destroys the RTC engine.
        """
        self.agora_service.release()
    ```

    ## Test the sample code [#test-the-sample-code-12]

    Follow these steps to test the demo code:

    1. Create a file named `rtc.py` and paste the [complete source code](#complete-code) into this file.

    2. Create a file named `main.py` in the same folder as `rtc.py` and copy the following code to the file:

       ```python
       import asyncio
       from rtc import RtcEngine # Import the RtcEngine class from rtc.py

       async def main():
         appid = "<Your app Id>" # Replace with your Agora App ID
         channelId = "demo" # Replace with your desired channel ID
         uid = "123" # Replace with your unique user ID

         rtc_engine = RtcEngine(appid)
         channel = await rtc_engine.connect(channelId, uid)

         # Keep the script running to listen for events
         await asyncio.Event().wait()

       if __name__ == "__main__":
         asyncio.run(main())
       ```

    3. To specify the audio parameters, create a folder named `realtimeapi` and add a file `util.py` containing the following code:

       ```python
       # Number of audio channels (1 for mono, 2 for stereo)
       CHANNELS = 2

       # Sample rate for audio processing (in Hz)
       SAMPLE_RATE = 44100 # Common sample rates include 8000, 16000, 44100, 48000
       ```

    4. To run the app, execute the following command in your terminal:

       ```bash
       python3 main.py
       ```

    You see output similar to the following:

    ```text
    Initialization result: 0
    LocalUserCB _on_audio_track_publish_start: 123456607304576 123456864576600
    ```

    Congratulations! You have successfully connected to the Agora SDRTN® and joined a channel.

    ## Reference [#reference-12]

    This section contains content that completes the information on this page, or points you to documentation that explains other aspects to this product.

    * If a firewall is deployed in your network environment, refer to [Connect with Cloud Proxy](build/optimize-quality-and-connection/cloud-proxy.mdx) to use Agora services normally.

    ### Next steps [#next-steps-12]

    After implementing the quickstart sample, read the following documents to learn more:

    * To ensure communication security in a test or production environment, best practice is to obtain and use a token from an authentication server. For details, see [Secure authentication with tokens](build/authenticate-users/use-tokens.mdx).

    ### API reference [#api-reference-11]

    * [`rtc.py`](https://api-ref.agora.io/en/voice-sdk/python/rtc-py-api.html)