Vonage Video Connector Transport

src/pipecat/runner/run.py

@@ -108,8 +108,10 @@ from pipecat.runner.types import (
     DailyRunnerArguments,
     RunnerArguments,
     SmallWebRTCRunnerArguments,
+    VonageRunnerArguments,
     WebSocketRunnerArguments,
 )
+from pipecat.runner.vonage import configure as configure_vonage
 
 try:
     import uvicorn
@@ -983,6 +985,25 @@ async def _run_daily_direct(args: argparse.Namespace):
         await bot_module.bot(runner_args)
 
 
+async def _run_vonage():
+    """Run Vonage bot (no FastAPI server)."""
+    logger.info("Running Vonage transport...")
+
+    application_id, session_id, token = await configure_vonage()
+    runner_args = VonageRunnerArguments(
+        application_id=application_id, session_id=session_id, token=token
+    )
+    runner_args.handle_sigint = True
+
+    # Get the bot module and run it directly
+    bot_module = _get_bot_module()
+
+    print(f"Joining Vonage session: {runner_args.session_id}")
+    print()
+
+    await bot_module.bot(runner_args)
+
+
 def _validate_and_clean_proxy(proxy: str) -> str:
     """Validate and clean proxy hostname, removing protocol if present."""
     if not proxy:
@@ -1062,7 +1083,7 @@ def main(parser: argparse.ArgumentParser | None = None):
         "-t",
         "--transport",
         type=str,
-        choices=["daily", "webrtc", *TELEPHONY_TRANSPORTS],
+        choices=["daily", "vonage", "webrtc", *TELEPHONY_TRANSPORTS],
         default=None,
         help=(
             "Restrict the server to a single transport and set it as the default for /start. "
@@ -1169,6 +1190,12 @@ def main(parser: argparse.ArgumentParser | None = None):
         if args.proxy:
             print(f"   → XML webhook: http://{args.host}:{args.port}/")
         print(f"   → WebSocket:   ws://{args.host}:{args.port}/ws")
+    elif args.transport == "vonage":
+        print()
+        print(f"🚀 Bot ready!")
+        asyncio.run(_run_vonage())
+        print()
+        return
     print()
 
     RUNNER_DOWNLOADS_FOLDER = args.folder

src/pipecat/runner/types.py

@@ -99,6 +99,21 @@ class DailyRunnerArguments(RunnerArguments):
     token: str | None = None
 
 
+@dataclass
+class VonageRunnerArguments(RunnerArguments):
+    """Daily transport session arguments for the runner.
+
+    Parameters:
+        application_id: Vonage application ID
+        session_id: Vonage session ID
+        token: Vonage Session Token
+    """
+
+    application_id: str
+    session_id: str
+    token: str
+
+
 @dataclass
 class WebSocketRunnerArguments(RunnerArguments):
     """WebSocket transport session arguments for the runner.

src/pipecat/runner/utils.py

@@ -33,7 +33,7 @@ import json
 import os
 import re
 from collections.abc import Callable
-from typing import Any
+from typing import Any, cast
 
 from fastapi import WebSocket
 from loguru import logger
@@ -42,9 +42,10 @@ from pipecat.runner.types import (
     DailyRunnerArguments,
     LiveKitRunnerArguments,
     SmallWebRTCRunnerArguments,
+    VonageRunnerArguments,
     WebSocketRunnerArguments,
 )
-from pipecat.transports.base_transport import BaseTransport
+from pipecat.transports.base_transport import BaseTransport, TransportParams
 
 
 def _detect_transport_type_from_message(message_data: dict) -> str:
@@ -271,6 +272,14 @@ def get_transport_client_id(transport: BaseTransport, client: Any) -> str:
     except ImportError:
         pass
 
+    try:
+        from pipecat.transports.vonage.video_connector import VonageVideoConnectorTransport
+
+        if isinstance(transport, VonageVideoConnectorTransport):
+            return client["streamId"]
+    except ImportError:
+        pass
+
     logger.warning(f"Unable to get client id from unsupported transport {type(transport)}")
     return ""
 
@@ -303,6 +312,24 @@ async def maybe_capture_participant_camera(
     except ImportError:
         pass
 
+    try:
+        from pipecat.transports.vonage.video_connector import (
+            SubscribeSettings,
+            VonageVideoConnectorTransport,
+        )
+
+        if isinstance(transport, VonageVideoConnectorTransport):
+            await transport.subscribe_to_stream(
+                client["streamId"],
+                SubscribeSettings(
+                    subscribe_to_audio=True,
+                    subscribe_to_video=True,
+                    preferred_framerate=framerate if framerate != 0 else None,
+                ),
+            )
+    except ImportError:
+        pass
+
 
 async def maybe_capture_participant_screen(
     transport: BaseTransport, client: Any, framerate: int = 0
@@ -534,6 +561,11 @@ async def create_transport(
                 audio_out_enabled=True,
                 # add_wav_header and serializer will be set automatically
             ),
+            "vonage": lambda: VonageVideoConnectorParams(
+                audio_in_enabled=True,
+                audio_out_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(),
+            ),
         }
 
         transport = await create_transport(runner_args, transport_params)
@@ -587,6 +619,31 @@ async def create_transport(
             runner_args.room_name,
             params=params,
         )
+    elif isinstance(runner_args, VonageRunnerArguments):
+        from pipecat.transports.vonage.video_connector import (
+            VonageVideoConnectorTransport,
+            VonageVideoConnectorTransportParams,
+        )
 
+        try:
+            params = cast(
+                VonageVideoConnectorTransportParams,
+                _get_transport_params("vonage", transport_params),
+            )
+        except ValueError:
+            webrtc_params: TransportParams = cast(
+                TransportParams, _get_transport_params("webrtc", transport_params)
+            )
+            params = VonageVideoConnectorTransportParams(
+                **webrtc_params.model_dump(),
+                video_in_auto_subscribe=True,
+            )
+
+        return VonageVideoConnectorTransport(
+            runner_args.application_id,
+            runner_args.session_id,
+            runner_args.token,
+            params=params,
+        )
     else:
         raise ValueError(f"Unsupported runner arguments type: {type(runner_args)}")

src/pipecat/runner/vonage.py

@@ -0,0 +1,52 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Vonage session configuration utilities.
+
+This module extracts the necessary parameters to connect to a Vonage Video session.
+
+Required environment variables:
+
+- VONAGE_APPLICATION_ID - Vonage application ID
+- VONAGE_SESSION_ID - Vonage session ID
+- VONAGE_TOKEN - Vonage token
+
+Example:
+    from pipecat.runner.vonage import configure
+
+    application_id, session_id, token = await configure()
+"""
+
+import os
+
+
+async def configure() -> tuple[str, str, str]:
+    """Configure Vonage application ID, session ID and token from environment.
+
+    Returns:
+        Tuple containing the server application_id, session_id and token.
+
+    Raises:
+        Exception: If required Vonage configuration is not provided.
+    """
+    application_id = os.getenv("VONAGE_APPLICATION_ID")
+    session_id = os.getenv("VONAGE_SESSION_ID")
+    token = os.getenv("VONAGE_TOKEN")
+
+    if not application_id:
+        raise Exception(
+            "No Vonage application ID specified. Use set VONAGE_APPLICATION_ID in your environment."
+        )
+
+    if not session_id:
+        raise Exception(
+            "No Vonage Session ID specified. Use set VONAGE_SESSION_ID in your environment."
+        )
+
+    if not token:
+        raise Exception("No Vonage token specified. Use set VONAGE_TOKEN in your environment.")
+
+    return (application_id, session_id, token)

src/pipecat/transports/vonage/utils.py

@@ -0,0 +1,150 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+"""Vonage Video Connector utils."""
+
+from dataclasses import dataclass, replace
+from enum import StrEnum
+
+import numpy as np
+import numpy.typing as npt
+
+from pipecat.audio.resamplers.base_audio_resampler import BaseAudioResampler
+
+
+@dataclass
+class AudioProps:
+    """Audio properties for normalization.
+
+    Parameters:
+        sample_rate: The sample rate of the audio.
+        is_stereo: Whether the audio is stereo (True) or mono (False).
+    """
+
+    sample_rate: int
+    is_stereo: bool
+
+
+class ImageFormat(StrEnum):
+    """Enum for image formats."""
+
+    PLANAR_YUV420 = "PLANAR_YUV420"
+    PACKED_YUV444 = "PACKED_YUV444"
+    RGB = "RGB"
+    RGBA = "RGBA"
+    BGR = "BGR"
+    BGRA = "BGRA"
+
+
+def check_audio_data(
+    buffer: bytes | memoryview, number_of_frames: int, number_of_channels: int
+) -> None:
+    """Check the audio sample width based on buffer size, number of frames and channels."""
+    if number_of_channels not in (1, 2):
+        raise ValueError(f"We only accept mono or stereo audio, got {number_of_channels}")
+
+    if isinstance(buffer, memoryview):
+        bytes_per_sample = buffer.itemsize
+    else:
+        bytes_per_sample = len(buffer) // (number_of_frames * number_of_channels)
+
+    if bytes_per_sample != 2:
+        raise ValueError(f"We only accept 16 bit PCM audio, got {bytes_per_sample * 8} bit")
+
+
+def process_audio_channels(
+    audio: npt.NDArray[np.int16], current: AudioProps, target: AudioProps
+) -> npt.NDArray[np.int16]:
+    """Normalize audio channels to the target properties."""
+    if current.is_stereo != target.is_stereo:
+        if target.is_stereo:
+            audio = np.repeat(audio, 2)
+        else:
+            audio = audio.reshape(-1, 2).mean(axis=1).astype(np.int16)
+
+    return audio
+
+
+async def process_audio(
+    resampler: BaseAudioResampler,
+    audio: npt.NDArray[np.int16],
+    current: AudioProps,
+    target: AudioProps,
+) -> npt.NDArray[np.int16]:
+    """Normalize audio to the target properties."""
+    res_audio = audio
+    if current.sample_rate != target.sample_rate:
+        # first normalize channels to mono if needed, then resample, then normalize channels to target
+        res_audio = process_audio_channels(res_audio, current, replace(current, is_stereo=False))
+        current = replace(current, is_stereo=False)
+        res_audio_bytes: bytes = await resampler.resample(
+            res_audio.tobytes(), current.sample_rate, target.sample_rate
+        )
+        res_audio = np.frombuffer(res_audio_bytes, dtype=np.int16)
+
+    res_audio = process_audio_channels(res_audio, current, target)
+
+    return res_audio
+
+
+def image_colorspace_conversion(
+    image: bytes, size: tuple[int, int], from_format: ImageFormat, to_format: ImageFormat
+) -> bytes | None:
+    """Convert image colorspace from one format to another."""
+    match (from_format, to_format):
+        case (fmt1, fmt2) if fmt1 == fmt2:
+            return image
+        case (ImageFormat.RGB, ImageFormat.BGR) | (ImageFormat.BGR, ImageFormat.RGB):
+            np_input = np.frombuffer(image, dtype=np.uint8)
+            np_output = np_input.reshape(size[1], size[0], 3)[:, :, ::-1]
+            return np_output.tobytes()
+        case (ImageFormat.RGBA, ImageFormat.BGRA) | (ImageFormat.BGRA, ImageFormat.RGBA):
+            np_input = np.frombuffer(image, dtype=np.uint8)
+            np_output = np_input.reshape(size[1], size[0], 4)[:, :, [2, 1, 0, 3]]
+            return np_output.tobytes()
+        case (ImageFormat.PLANAR_YUV420, ImageFormat.PACKED_YUV444):
+            # YUV420 (I420) has Y plane of size width*height, U and V planes of size (width/2)*(height/2)
+            # Packed YUV444 interleaves Y, U, V values for each pixel (YUVYUVYUV...)
+            width, height = size
+            y_plane_size = width * height
+            uv_plane_size_420 = (width // 2) * (height // 2)
+
+            np_input = np.frombuffer(image, dtype=np.uint8)
+            y_plane = np_input[:y_plane_size].reshape(height, width)
+            u_plane_420 = np_input[y_plane_size : y_plane_size + uv_plane_size_420].reshape(
+                height // 2, width // 2
+            )
+            v_plane_420 = np_input[
+                y_plane_size + uv_plane_size_420 : y_plane_size + 2 * uv_plane_size_420
+            ].reshape(height // 2, width // 2)
+
+            # Upsample U and V planes by repeating each pixel in 2x2 blocks
+            u_plane_444 = np.repeat(np.repeat(u_plane_420, 2, axis=0), 2, axis=1)
+            v_plane_444 = np.repeat(np.repeat(v_plane_420, 2, axis=0), 2, axis=1)
+
+            # Interleave Y, U, V values for packed format (YUVYUVYUV...)
+            np_output = np.stack([y_plane, u_plane_444, v_plane_444], axis=-1)
+            return np_output.tobytes()
+        case (ImageFormat.PACKED_YUV444, ImageFormat.PLANAR_YUV420):
+            # Packed YUV444 has Y, U, V interleaved (YUVYUVYUV...)
+            # YUV420 (I420) has Y plane of size width*height, U and V planes of size (width/2)*(height/2)
+            width, height = size
+
+            np_input = np.frombuffer(image, dtype=np.uint8).reshape(height, width, 3)
+            y_plane = np_input[:, :, 0].reshape(height, width)
+            u_plane_444 = np_input[:, :, 1]
+            v_plane_444 = np_input[:, :, 2]
+
+            # Downsample U and V planes by taking every other pixel (2x2 -> 1 averaging)
+            u_plane_420 = u_plane_444[::2, ::2].reshape(height // 2, width // 2)
+            v_plane_420 = v_plane_444[::2, ::2].reshape(height // 2, width // 2)
+
+            # Concatenate Y, U, V planes
+            np_output = np.concatenate(
+                [y_plane.flatten(), u_plane_420.flatten(), v_plane_420.flatten()]
+            )
+            return np_output.tobytes()
+        case _:
+            return None

src/pipecat/transports/vonage/video_connector.py

@@ -0,0 +1,483 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+"""Vonage Video Connector transport."""
+
+from typing import Optional
+
+from loguru import logger
+
+from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
+    Frame,
+    InputAudioRawFrame,
+    InterruptionFrame,
+    OutputAudioRawFrame,
+    OutputImageRawFrame,
+    StartFrame,
+    UserImageRawFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
+from pipecat.transports.base_input import BaseInputTransport
+from pipecat.transports.base_output import BaseOutputTransport
+from pipecat.transports.base_transport import BaseTransport
+from pipecat.transports.vonage.client import (
+    Session,
+    Stream,
+    Subscriber,
+    VonageClient,
+    VonageClientListener,
+)
+
+# the following "as" imports help to re-export these types and avoid type checking warnings
+# when importing these types from the main transport module
+from pipecat.transports.vonage.client import (
+    SubscribeSettings as SubscribeSettings,
+)
+from pipecat.transports.vonage.client import (
+    VonageException as VonageException,
+)
+from pipecat.transports.vonage.client import (
+    VonageVideoConnectorTransportParams as VonageVideoConnectorTransportParams,
+)
+
+
+class VonageVideoConnectorInputTransport(BaseInputTransport):
+    """Input transport for Vonage, handling audio input from the Vonage session.
+
+    Receives audio from a Vonage Video session and pushes it as input frames.
+    """
+
+    _params: VonageVideoConnectorTransportParams
+
+    def __init__(self, client: VonageClient, params: VonageVideoConnectorTransportParams):
+        """Initialize the Vonage input transport.
+
+        Args:
+            client: The VonageClient instance to use.
+            params: Transport parameters for input configuration.
+        """
+        super().__init__(params)
+        self._initialized: bool = False
+        self._client: VonageClient = client
+        self._listener_id: int = -1
+        self._connected: bool = False
+
+    async def start(self, frame: StartFrame) -> None:
+        """Start the Vonage input transport.
+
+        Args:
+            frame: The StartFrame to initiate the transport.
+        """
+        await super().start(frame)
+
+        if self._initialized:
+            return
+
+        self._initialized = True
+
+        if self._params.audio_in_enabled or self._params.video_in_enabled:
+            self._listener_id = self._client.add_listener(
+                VonageClientListener(
+                    on_audio_in=self._audio_in_cb,
+                    on_video_in=self._video_in_cb,
+                    on_error=self._on_error_cb,
+                )
+            )
+            try:
+                await self._client.connect(frame)
+                self._connected = True
+            except Exception as exc:
+                logger.error(f"Error connecting to Vonage session: {exc}")
+                await self.push_error("Vonage video connector connection error", fatal=True)
+                return
+
+        await self.set_transport_ready(frame)
+
+    async def setup(self, setup: FrameProcessorSetup) -> None:
+        """Set up the processor with required components.
+
+        Args:
+            setup: Configuration object containing setup parameters.
+        """
+        await super().setup(setup)
+        await self._client.setup(setup)
+
+    async def cleanup(self) -> None:
+        """Cleanup input transport."""
+        await super().cleanup()  # type: ignore
+        await self._client.cleanup()
+
+    async def _audio_in_cb(self, _session: Session, audio: InputAudioRawFrame) -> None:
+        if self._connected and self._params.audio_in_enabled:
+            await self.push_audio_frame(audio)
+
+    async def _video_in_cb(self, _subscriber: Subscriber, video: UserImageRawFrame) -> None:
+        if self._connected and self._params.video_in_enabled:
+            await self.push_video_frame(video)
+
+    async def _on_error_cb(self, session: Session, description: str, code: int) -> None:
+        logger.error(
+            f"Vonage input transport error session={session.id} code={code} description={description}"
+        )
+        if self._connected:
+            await self.push_error("Vonage video connector error", fatal=True)
+
+    async def stop(self, frame: EndFrame) -> None:
+        """Stop the Vonage input transport.
+
+        Args:
+            frame: The EndFrame to stop the transport.
+        """
+        await super().stop(frame)
+        await self._stop_client()
+
+    async def cancel(self, frame: CancelFrame) -> None:
+        """Cancel the Vonage input transport.
+
+        Args:
+            frame: The CancelFrame to cancel the transport.
+        """
+        await super().cancel(frame)
+        await self._stop_client()
+
+    async def _stop_client(self) -> None:
+        if self._connected:
+            self._client.remove_listener(self._listener_id)
+            self._connected = False
+            try:
+                await self._client.disconnect()
+            except Exception:
+                pass
+
+    async def subscribe_to_stream(self, stream_id: str, params: SubscribeSettings) -> None:
+        """Subscribe to a participant's stream.
+
+        Args:
+            stream_id: The ID of the participant to subscribe to.
+            params: Subscription parameters for the subscription.
+        """
+        await self._client.subscribe_to_stream(stream_id, params)
+
+
+class VonageVideoConnectorOutputTransport(BaseOutputTransport):
+    """Output transport for Vonage, handling audio output to the Vonage session.
+
+    Sends audio frames to a Vonage Video session as output.
+    """
+
+    _params: VonageVideoConnectorTransportParams
+
+    def __init__(self, client: VonageClient, params: VonageVideoConnectorTransportParams):
+        """Initialize the Vonage output transport.
+
+        Args:
+            client: The VonageClient instance to use.
+            params: Transport parameters for output configuration.
+        """
+        super().__init__(params)
+        self._initialized: bool = False
+        self._client = client
+        self._connected: bool = False
+        self._listener_id: int = -1
+
+    async def start(self, frame: StartFrame) -> None:
+        """Start the Vonage output transport.
+
+        Args:
+            frame: The StartFrame to initiate the transport.
+        """
+        await super().start(frame)
+
+        if self._initialized:
+            return
+
+        self._initialized = True
+
+        if self._params.audio_out_enabled or self._params.video_out_enabled:
+            self._listener_id = self._client.add_listener(
+                VonageClientListener(on_error=self._on_error_cb)
+            )
+            try:
+                await self._client.connect(frame)
+                self._connected = True
+            except Exception as exc:
+                logger.error(f"Error connecting to Vonage session: {exc}")
+                await self.push_error("Vonage video connector connection error", fatal=True)
+                return
+
+        await self.set_transport_ready(frame)
+
+    async def setup(self, setup: FrameProcessorSetup) -> None:
+        """Set up the processor with required components.
+
+        Args:
+            setup: Configuration object containing setup parameters.
+        """
+        await super().setup(setup)
+        await self._client.setup(setup)
+
+    async def cleanup(self) -> None:
+        """Cleanup output transport."""
+        await super().cleanup()  # type: ignore
+        await self._client.cleanup()
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
+        """Process a frame for the Vonage output transport.
+
+        Args:
+            frame: The frame to process.
+            direction: The direction of frame flow in the pipeline.
+        """
+        await super().process_frame(frame, direction)
+
+        # if we get an interruption frame, we need to ensure the buffers inside Vonage Video Connector are cleared
+        if (
+            self._connected
+            and isinstance(frame, InterruptionFrame)
+            and self._params.clear_buffers_on_interruption
+        ):
+            logger.info("Clearing Vonage media buffers due to interruption frame")
+            self._client.clear_media_buffers()
+
+    async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
+        """Write an audio frame to the Vonage session.
+
+        Args:
+            frame: The OutputAudioRawFrame to send.
+        """
+        result = False
+        if self._connected and self._params.audio_out_enabled:
+            result = await self._client.write_audio(frame)
+
+        return result
+
+    async def write_video_frame(self, frame: OutputImageRawFrame) -> bool:
+        """Write a video frame to the transport.
+
+        Args:
+            frame: The output video frame to write.
+        """
+        result = False
+        if self._connected and self._params.video_out_enabled:
+            result = await self._client.write_video(frame)
+
+        return result
+
+    async def stop(self, frame: EndFrame) -> None:
+        """Stop the Vonage output transport.
+
+        Args:
+            frame: The EndFrame to stop the transport.
+        """
+        await super().stop(frame)
+        await self._stop_client()
+
+    async def cancel(self, frame: CancelFrame) -> None:
+        """Cancel the Vonage output transport.
+
+        Args:
+            frame: The CancelFrame to cancel the transport.
+        """
+        await super().cancel(frame)
+        await self._stop_client()
+
+    async def _stop_client(self) -> None:
+        if self._connected:
+            self._client.remove_listener(self._listener_id)
+            self._connected = False
+            try:
+                await self._client.disconnect()
+            except Exception:
+                pass
+
+    async def _on_error_cb(self, session: Session, description: str, code: int) -> None:
+        logger.error(
+            f"Vonage output transport error session={session.id} code={code} description={description}"
+        )
+        if self._connected:
+            await self.push_error("Vonage video connector error", fatal=True)
+
+
+class VonageVideoConnectorTransport(BaseTransport):
+    """Vonage Video Connector transport implementation for Pipecat.
+
+    Provides input and output audio transport for Vonage Video sessions, supporting event handling
+    for session and participant lifecycle.
+
+    Supported features:
+
+    - Audio input and output transport for Vonage Video sessions
+    - Event handler registration for session and participant events
+    - Publisher and subscriber management
+    - Configurable audio and migration parameters
+    """
+
+    _params: VonageVideoConnectorTransportParams
+
+    def __init__(
+        self,
+        application_id: str,
+        session_id: str,
+        token: str,
+        params: VonageVideoConnectorTransportParams,
+    ):
+        """Initialize the Vonage Video Connector transport.
+
+        Args:
+            application_id: The Vonage Video application ID.
+            session_id: The session ID to connect to.
+            token: The authentication token for the session.
+            params: Transport parameters for input/output configuration.
+        """
+        super().__init__()
+        self._params = params
+
+        self._client = VonageClient(application_id, session_id, token, params)
+
+        # Register supported handlers.
+        self._register_event_handler("on_joined")
+        self._register_event_handler("on_left")
+        self._register_event_handler("on_error")
+        self._register_event_handler("on_client_connected", sync=True)
+        self._register_event_handler("on_client_disconnected")
+        self._register_event_handler("on_first_participant_joined", sync=True)
+        self._register_event_handler("on_participant_joined", sync=True)
+        self._register_event_handler("on_participant_left")
+
+        self._client.add_listener(
+            VonageClientListener(
+                on_connected=self._on_connected,
+                on_disconnected=self._on_disconnected,
+                on_error=self._on_error,
+                on_stream_received=self._on_stream_received,
+                on_stream_dropped=self._on_stream_dropped,
+                on_subscriber_connected=self._on_subscriber_connected,
+                on_subscriber_disconnected=self._on_subscriber_disconnected,
+            )
+        )
+
+        self._input: Optional[VonageVideoConnectorInputTransport] = None
+        self._output: Optional[VonageVideoConnectorOutputTransport] = None
+        self._one_stream_received: bool = False
+
+    def input(self) -> FrameProcessor:
+        """Get the input transport for Vonage.
+
+        Returns:
+            The VonageVideoConnectorInputTransport instance.
+        """
+        if not self._input:
+            self._input = VonageVideoConnectorInputTransport(self._client, self._params)
+        return self._input
+
+    def output(self) -> FrameProcessor:
+        """Get the output transport for Vonage.
+
+        Returns:
+            The VonageVideoConnectorOutputTransport instance.
+        """
+        if not self._output:
+            self._output = VonageVideoConnectorOutputTransport(self._client, self._params)
+        return self._output
+
+    async def subscribe_to_stream(self, stream_id: str, params: SubscribeSettings) -> None:
+        """Subscribe to a participant's stream.
+
+        Args:
+            stream_id: The ID of the participant to subscribe to.
+            params: Subscription parameters for the subscription.
+        """
+        if self._input:
+            await self._input.subscribe_to_stream(stream_id, params)
+
+    async def _on_connected(self, session: Session) -> None:
+        """Handle session connected event.
+
+        Args:
+            session: The connected Session object.
+        """
+        await self._call_event_handler("on_joined", {"sessionId": session.id})
+
+    async def _on_disconnected(self, session: Session) -> None:
+        """Handle session disconnected event.
+
+        Args:
+            session: The disconnected Session object.
+        """
+        await self._call_event_handler("on_left", {"sessionId": session.id})
+
+    async def _on_error(self, _session: Session, description: str, _code: int) -> None:
+        """Handle session error event.
+
+        Args:
+            _session: The Session object.
+            description: Error description.
+            _code: Error code.
+        """
+        await self._call_event_handler("on_error", description)
+
+    async def _on_stream_received(self, session: Session, stream: Stream) -> None:
+        """Handle stream received event.
+
+        Args:
+            session: The Session object.
+            stream: The received Stream object.
+        """
+        client = {
+            "sessionId": session.id,
+            "streamId": stream.id,
+            "connectionData": stream.connection.data,
+        }
+        if not self._one_stream_received:
+            self._one_stream_received = True
+            await self._call_event_handler("on_first_participant_joined", client)
+
+        await self._call_event_handler("on_participant_joined", client)
+
+    async def _on_stream_dropped(self, session: Session, stream: Stream) -> None:
+        """Handle stream dropped event.
+
+        Args:
+            session: The Session object.
+            stream: The dropped Stream object.
+        """
+        client = {
+            "sessionId": session.id,
+            "streamId": stream.id,
+            "connectionData": stream.connection.data,
+        }
+        await self._call_event_handler("on_participant_left", client)
+
+    async def _on_subscriber_connected(self, subscriber: Subscriber) -> None:
+        """Handle subscriber connected event.
+
+        Args:
+            subscriber: The connected Subscriber object.
+        """
+        await self._call_event_handler(
+            "on_client_connected",
+            {
+                "subscriberId": subscriber.stream.id,
+                "streamId": subscriber.stream.id,
+                "connectionData": subscriber.stream.connection.data,
+            },
+        )
+
+    async def _on_subscriber_disconnected(self, subscriber: Subscriber) -> None:
+        """Handle subscriber disconnected event.
+
+        Args:
+            subscriber: The disconnected Subscriber object.
+        """
+        await self._call_event_handler(
+            "on_client_disconnected",
+            {
+                "subscriberId": subscriber.stream.id,
+                "streamId": subscriber.stream.id,
+                "connectionData": subscriber.stream.connection.data,
+            },
+        )