# Voice-only quickstart (/en/realtime-media/rtc/voice-quickstart/python)

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

This Python quickstart shows you how to create a basic Voice Calling app using the Agora RTC SDK.

## Understand the tech

To start a Voice Calling 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.

* **Send and receive audio**: All users can publish streams to the channel and subscribe to audio streams published by other users in the channel.

![Video calling workflow](https://assets-docs.agora.io/images/voice-sdk/get-started-sdk-voice.svg)

## Prerequisites

* A microphone

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

## Set up your project

This section shows you how to set up your Python project and install the Agora RTC 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 Voice Calling

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

### Import Agora classes

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

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

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

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

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

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

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

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

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

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/manage-connection-and-quality/cloud-proxy.mdx) to use Agora services normally.

### Next steps

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/authentication-workflow.mdx).

### API reference

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