Speech-to-Speech

• Turn detection: let the provider API decide when the user has finished, or plug in a local VAD and decide yourself
• Barge-in: the user speaking cuts off the current reply, and the context keeps only what the user actually heard; your code can interrupt as well
• Tool calling with user confirmation: works with Toolkit and the permission system, and the audio stream keeps running while the user decides
• Text input: send text during a voice conversation, on models that accept text
• Automatic reconnection: after the provider API closes a session on idle or timeout, the next input reconnects and restores the conversation
• Turn aggregation: merge sentences split by a pause, and drop acknowledgements that carry no content

​

Core Concepts

• Audio transport (TransportBase): where sound comes from and goes to, such as a local sound card or a browser
• Realtime speech model (RealtimeModelBase): the session with the provider API, translating protocol messages into uniform model events
• RealtimeAgent: sits between the two, tracking conversation turns and handling barge-in, tool calls, and event output

​

Quick Start

pip install "agentscope[realtime]"

import os

from agentscope.credential import DashScopeCredential
from agentscope.realtime import DashScopeAudioRealtimeModel

model = DashScopeAudioRealtimeModel(
    model="qwen-audio-3.0-realtime-plus",
    credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
    parameters=DashScopeAudioRealtimeModel.Parameters(voice="longanqian"),
)

import os

from agentscope.credential import DashScopeCredential
from agentscope.realtime import DashScopeRealtimeModel

model = DashScopeRealtimeModel(
    model="qwen3.5-omni-plus-realtime",
    credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
    parameters=DashScopeRealtimeModel.Parameters(voice="Tina"),
)

import os

from agentscope.credential import OpenAICredential
from agentscope.realtime import OpenAIRealtimeModel

model = OpenAIRealtimeModel(
    model="gpt-realtime-2.1",
    credential=OpenAICredential(api_key=os.environ["OPENAI_API_KEY"]),
    parameters=OpenAIRealtimeModel.Parameters(voice="marin"),
)

import os

from agentscope.credential import GeminiCredential
from agentscope.realtime import GeminiRealtimeModel

model = GeminiRealtimeModel(
    model="gemini-3.1-flash-live-preview",
    credential=GeminiCredential(api_key=os.environ["GEMINI_API_KEY"]),
    parameters=GeminiRealtimeModel.Parameters(voice="Puck"),
)

import os

from agentscope.credential import XAICredential
from agentscope.realtime import XAIRealtimeModel

model = XAIRealtimeModel(
    model="grok-voice-latest",
    credential=XAICredential(api_key=os.environ["XAI_API_KEY"]),
    # Setting reasoning_effort to "none" buys faster but shallower answers
    parameters=XAIRealtimeModel.Parameters(voice="eve", reasoning_effort="high"),
)

from agentscope.agent import RealtimeAgent

agent = RealtimeAgent(
    name="Friday",
    system_prompt="You are a voice assistant. Keep your answers short.",
    model=model,
)

from agentscope.realtime import LocalAudioTransport

transport = LocalAudioTransport(
    input_sample_rate=model.input_sample_rate,
    output_sample_rate=model.output_sample_rate,
)

from agentscope.event import ReplyEndEvent, ReplyStartEvent, TextBlockDeltaEvent

user_turns: set[str] = set()

async with agent, transport:
    async for event in agent.reply_stream(transport):
        match event:
            case ReplyStartEvent(role="user"):
                user_turns.add(event.reply_id)
            case ReplyStartEvent():
                print(f"[{agent.name}] ", end="", flush=True)
            case TextBlockDeltaEvent() if event.reply_id in user_turns:
                print(f"[user] {event.delta}")
            case TextBlockDeltaEvent():
                print(event.delta, end="", flush=True)
            case ReplyEndEvent() if event.reply_id not in user_turns:
                print(f"  ({event.finished_reason})")

​

Use the Agent

​

Run and Handle Events

​

Send Input

await agent.send("Check today's weather for me")

​

Barge-In

await agent.interrupt()

​

Call Tools

from agentscope.event import RequireUserConfirmEvent
from agentscope.tool import Bash, Read, Toolkit

agent = RealtimeAgent(
    name="Friday",
    system_prompt="...",
    model=model,
    toolkit=Toolkit(tools=[Bash(), Read()]),
)

async with agent, transport:
    async for event in agent.reply_stream(transport):
        if isinstance(event, RequireUserConfirmEvent):
            # Hand the request to the UI, and do not block the event stream here
            show_confirm_dialog(event)

from agentscope.event import ConfirmResult, UserConfirmResultEvent


async def on_confirm_clicked(event: RequireUserConfirmEvent, allowed: bool) -> None:
    # Triggered by a button click or similar, unrelated to the reply_stream() loop
    await agent.send(
        UserConfirmResultEvent(
            reply_id=event.reply_id,
            confirm_results=[
                ConfirmResult(tool_call=call, confirmed=allowed)
                for call in event.tool_calls
            ],
        ),
    )

import asyncio

from agentscope.event import ConfirmResult, UserConfirmResultEvent


async def confirm_in_terminal(event: RequireUserConfirmEvent) -> None:
    loop = asyncio.get_running_loop()
    results = []
    for call in event.tool_calls:
        # Terminal input blocks the thread, so run it in a thread pool and keep the audio going
        answer = await loop.run_in_executor(
            None, input, f"Allow {call.name}({call.input})? [y/N] ",
        )
        results.append(
            ConfirmResult(tool_call=call, confirmed=answer.lower() == "y"),
        )
    await agent.send(
        UserConfirmResultEvent(reply_id=event.reply_id, confirm_results=results),
    )

• Only models whose model card sets supports_tools receive the tool list. Among DashScope’s Qwen-Omni models, that is the qwen3.5 series only; every model on the other provider APIs supports tools.
• A confirmation request that goes unanswered for five minutes is treated as a rejection.

​

Turn Detection

from agentscope.realtime import SpeechTransition, VADBase


class MyVAD(VADBase):
    sample_rate = 16000  # Must match the transport's input sample rate

    def push(self, pcm: bytes) -> SpeechTransition | None:
        # Return STARTED on the chunk where speech begins, ENDED on the one where it stops
        ...

    def reset(self) -> None:
        ...


agent = RealtimeAgent(name="Friday", system_prompt="...", model=model, vad=MyVAD())

from agentscope.agent import RealtimeAgent, TurnAggregator

agent = RealtimeAgent(
    name="Friday",
    system_prompt="...",
    model=model,
    aggregator=TurnAggregator(
        merge_window_ms=800,                       # Transcripts within 800ms of the last turn merge into it
        backchannels=frozenset({"uh-huh", "ok"}),  # These acknowledgements do not form a turn
        min_chars=1,                               # Transcripts shorter than this are dropped
    ),
)

​

Automatic Reconnection

​

Audio Transport

​

Local Sound Card

python -m sounddevice

• Wear headphones. On speakers, the microphone picks up the agent’s own voice, turn detection reads it as the user speaking, and the agent interrupts itself. LocalAudioTransport does no echo cancellation.
• Do not let one Bluetooth headset handle both input and output. macOS switches devices such as AirPods into hands-free mode, which often ends up silent. Pair the headset microphone with the built-in speaker instead, for example LocalAudioTransport(input_device=3, output_device=2).

​

Custom Transport