Skip to content

fix: add callback_interruption to Conversation/AsyncConversation #843

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
SomSamantray wants to merge 3 commits into
base: main
Choose a base branch
from
Open

fix: add callback_interruption to Conversation/AsyncConversation #843

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0a2d3c3
feat(conversational_ai): add callback_interruption to Conversation/As…
Aug 11, 2026
ac2e5b1
refactor(conversational_ai): use typed InterruptionEvent instead of r…
Aug 11, 2026
795971e
test(conversational_ai): cover InterruptionEvent.reason defaulting to…
Aug 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 32 additions & 4 deletions src/elevenlabs/conversational_ai/conversation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -413,6 +413,14 @@ class AudioEventAlignment:
char_durations_ms: List[int]


@dataclass
class InterruptionEvent:
"""Data describing why and where an interruption occurred."""

event_id: int
reason: Optional[str] = None


class BaseConversation:
"""Base class for conversation implementations with shared parameters and logic."""

Expand Down Expand Up @@ -572,7 +580,8 @@ def _handle_message_core(self, message, message_handler):
elif message["type"] == "interruption":
event = message["interruption_event"]
self._last_interrupt_id = int(event["event_id"])
message_handler.handle_interruption()
interruption = InterruptionEvent(event_id=self._last_interrupt_id, reason=event.get("reason"))
message_handler.handle_interruption(interruption)

elif message["type"] == "ping":
event = message["ping_event"]
Expand Down Expand Up @@ -642,7 +651,8 @@ async def _handle_message_core_async(self, message, message_handler):
elif message["type"] == "interruption":
event = message["interruption_event"]
self._last_interrupt_id = int(event["event_id"])
await message_handler.handle_interruption()
interruption = InterruptionEvent(event_id=self._last_interrupt_id, reason=event.get("reason"))
await message_handler.handle_interruption(interruption)

elif message["type"] == "ping":
event = message["ping_event"]
Expand All @@ -667,6 +677,7 @@ class Conversation(BaseConversation):
callback_user_transcript: Optional[Callable[[str], None]]
callback_latency_measurement: Optional[Callable[[int], None]]
callback_audio_alignment: Optional[Callable[[AudioEventAlignment], None]]
callback_interruption: Optional[Callable[[InterruptionEvent], None]]
callback_end_session: Optional[Callable]

_thread: Optional[threading.Thread]
Expand All @@ -689,6 +700,7 @@ def __init__(
callback_user_transcript: Optional[Callable[[str], None]] = None,
callback_latency_measurement: Optional[Callable[[int], None]] = None,
callback_audio_alignment: Optional[Callable[[AudioEventAlignment], None]] = None,
callback_interruption: Optional[Callable[[InterruptionEvent], None]] = None,
callback_end_session: Optional[Callable] = None,
on_prem_config: Optional[OnPremInitiationData] = None,
environment: Optional[str] = None,
Expand All @@ -714,6 +726,9 @@ def __init__(
callback_user_transcript: Callback for user transcripts.
callback_latency_measurement: Callback for latency measurements (in milliseconds).
callback_audio_alignment: Callback for audio alignment data with character-level timing.
callback_interruption: Callback for interruption events, invoked with an
InterruptionEvent. Fires in addition to interrupting the audio
interface (if one is attached).
environment: The environment to use. Defaults to "production" on the server.
"""

Expand All @@ -736,6 +751,7 @@ def __init__(
self.callback_user_transcript = callback_user_transcript
self.callback_latency_measurement = callback_latency_measurement
self.callback_audio_alignment = callback_audio_alignment
self.callback_interruption = callback_interruption
self.callback_end_session = callback_end_session

self._thread = None
Expand Down Expand Up @@ -912,6 +928,7 @@ def __init__(self, conversation, ws):
self.callback_user_transcript = conversation.callback_user_transcript
self.callback_latency_measurement = conversation.callback_latency_measurement
self.callback_audio_alignment = conversation.callback_audio_alignment
self.callback_interruption = conversation.callback_interruption

def handle_audio_output(self, audio):
if self.conversation.audio_interface is not None:
Expand All @@ -932,9 +949,11 @@ def handle_agent_chat_response_part(self, text, part_type):
def handle_user_transcript(self, transcript):
self.conversation.callback_user_transcript(transcript)

def handle_interruption(self):
def handle_interruption(self, event):
if self.conversation.audio_interface is not None:
self.conversation.audio_interface.interrupt()
if self.callback_interruption:
self.callback_interruption(event)

def handle_ping(self, event):
self.ws.send(
Expand Down Expand Up @@ -968,6 +987,7 @@ class AsyncConversation(BaseConversation):
callback_user_transcript: Optional[Callable[[str], Awaitable[None]]]
callback_latency_measurement: Optional[Callable[[int], Awaitable[None]]]
callback_audio_alignment: Optional[Callable[[AudioEventAlignment], Awaitable[None]]]
callback_interruption: Optional[Callable[[InterruptionEvent], Awaitable[None]]]
callback_end_session: Optional[Callable[[], Awaitable[None]]]

_task: Optional[asyncio.Task]
Expand All @@ -990,6 +1010,7 @@ def __init__(
callback_user_transcript: Optional[Callable[[str], Awaitable[None]]] = None,
callback_latency_measurement: Optional[Callable[[int], Awaitable[None]]] = None,
callback_audio_alignment: Optional[Callable[[AudioEventAlignment], Awaitable[None]]] = None,
callback_interruption: Optional[Callable[[InterruptionEvent], Awaitable[None]]] = None,
callback_end_session: Optional[Callable[[], Awaitable[None]]] = None,
on_prem_config: Optional[OnPremInitiationData] = None,
environment: Optional[str] = None,
Expand All @@ -1015,6 +1036,9 @@ def __init__(
callback_user_transcript: Async callback for user transcripts.
callback_latency_measurement: Async callback for latency measurements (in milliseconds).
callback_audio_alignment: Async callback for audio alignment data with character-level timing.
callback_interruption: Async callback for interruption events, invoked with an
InterruptionEvent. Fires in addition to interrupting the audio
interface (if one is attached).
callback_end_session: Async callback for when session ends.
environment: The environment to use. Defaults to "production" on the server.
"""
Expand All @@ -1038,6 +1062,7 @@ def __init__(
self.callback_user_transcript = callback_user_transcript
self.callback_latency_measurement = callback_latency_measurement
self.callback_audio_alignment = callback_audio_alignment
self.callback_interruption = callback_interruption
self.callback_end_session = callback_end_session

self._task = None
Expand Down Expand Up @@ -1220,6 +1245,7 @@ def __init__(self, conversation, ws):
self.callback_user_transcript = conversation.callback_user_transcript
self.callback_latency_measurement = conversation.callback_latency_measurement
self.callback_audio_alignment = conversation.callback_audio_alignment
self.callback_interruption = conversation.callback_interruption

async def handle_audio_output(self, audio):
if self.conversation.audio_interface is not None:
Expand All @@ -1240,9 +1266,11 @@ async def handle_agent_chat_response_part(self, text, part_type):
async def handle_user_transcript(self, transcript):
await self.conversation.callback_user_transcript(transcript)

async def handle_interruption(self):
async def handle_interruption(self, event):
if self.conversation.audio_interface is not None:
await self.conversation.audio_interface.interrupt()
if self.callback_interruption:
await self.callback_interruption(event)

async def handle_ping(self, event):
await self.ws.send(
Expand Down
138 changes: 138 additions & 0 deletions tests/test_async_convai.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,10 +9,14 @@
AsyncAudioInterface,
AsyncConversation,
ConversationInitiationData,
InterruptionEvent,
)


class MockAsyncAudioInterface(AsyncAudioInterface):
def __init__(self):
self.interrupt_count = 0

async def start(self, input_callback):
print("Async audio interface started")
self.input_callback = input_callback
Expand All @@ -25,6 +29,7 @@ async def output(self, audio):

async def interrupt(self):
print("Async audio interrupted")
self.interrupt_count += 1


# Add test constants and helpers at module level
Expand Down Expand Up @@ -465,3 +470,136 @@ async def test_async_websocket_url_construction_edge_cases():
# Ensure no double slashes in the path (except after the protocol)
url_path = conv_url.split("://", 1)[1] # Remove protocol
assert "//" not in url_path, f"Async conversation URL should not contain double slashes in path: {conv_url}"


ASYNC_INTERRUPTION_MESSAGES = [
{
"type": "conversation_initiation_metadata",
"conversation_initiation_metadata_event": {"conversation_id": TEST_CONVERSATION_ID},
},
{
"type": "interruption",
"interruption_event": {"reason": "user_interrupted", "event_id": 1},
},
]


@pytest.mark.asyncio
async def test_async_callback_interruption_invoked_with_audio_interface():
"""callback_interruption fires with the raw event, alongside audio_interface.interrupt()."""
mock_ws = create_mock_async_websocket(ASYNC_INTERRUPTION_MESSAGES)
mock_client = MagicMock()
mock_client._client_wrapper.get_base_url.return_value = "https://api.elevenlabs.io"
interruption_callback = AsyncMock()
audio_interface = MockAsyncAudioInterface()

conversation = AsyncConversation(
client=mock_client,
agent_id=TEST_AGENT_ID,
requires_auth=False,
audio_interface=audio_interface,
callback_interruption=interruption_callback,
)

with patch("elevenlabs.conversational_ai.conversation.websockets.connect") as mock_connect:
mock_connect.return_value.__aenter__.return_value = mock_ws

await conversation.start_session()
await asyncio.sleep(0.1)

await conversation.end_session()
await conversation.wait_for_session_end()

interruption_callback.assert_called_once_with(InterruptionEvent(event_id=1, reason="user_interrupted"))
assert audio_interface.interrupt_count == 1


@pytest.mark.asyncio
async def test_async_callback_interruption_invoked_without_audio_interface():
"""callback_interruption still fires in text-only mode (no audio_interface attached)."""
mock_ws = create_mock_async_websocket(ASYNC_INTERRUPTION_MESSAGES)
mock_client = MagicMock()
mock_client._client_wrapper.get_base_url.return_value = "https://api.elevenlabs.io"
interruption_callback = AsyncMock()

conversation = AsyncConversation(
client=mock_client,
agent_id=TEST_AGENT_ID,
requires_auth=False,
audio_interface=None,
callback_interruption=interruption_callback,
)

with patch("elevenlabs.conversational_ai.conversation.websockets.connect") as mock_connect:
mock_connect.return_value.__aenter__.return_value = mock_ws

await conversation.start_session()
await asyncio.sleep(0.1)

await conversation.end_session()
await conversation.wait_for_session_end()

interruption_callback.assert_called_once_with(InterruptionEvent(event_id=1, reason="user_interrupted"))


@pytest.mark.asyncio
async def test_async_interruption_without_callback_does_not_raise():
"""With callback_interruption left as the default None, interruption handling is unchanged."""
mock_ws = create_mock_async_websocket(ASYNC_INTERRUPTION_MESSAGES)
mock_client = MagicMock()
mock_client._client_wrapper.get_base_url.return_value = "https://api.elevenlabs.io"
audio_interface = MockAsyncAudioInterface()

conversation = AsyncConversation(
client=mock_client,
agent_id=TEST_AGENT_ID,
requires_auth=False,
audio_interface=audio_interface,
)

with patch("elevenlabs.conversational_ai.conversation.websockets.connect") as mock_connect:
mock_connect.return_value.__aenter__.return_value = mock_ws

await conversation.start_session()
await asyncio.sleep(0.1)

await conversation.end_session()
await conversation.wait_for_session_end()

assert audio_interface.interrupt_count == 1


@pytest.mark.asyncio
async def test_async_callback_interruption_defaults_reason_to_none():
"""When the raw event has no 'reason' key, InterruptionEvent.reason falls back to None."""
mock_ws = create_mock_async_websocket(
[
{
"type": "conversation_initiation_metadata",
"conversation_initiation_metadata_event": {"conversation_id": TEST_CONVERSATION_ID},
},
{"type": "interruption", "interruption_event": {"event_id": 1}},
]
)
mock_client = MagicMock()
mock_client._client_wrapper.get_base_url.return_value = "https://api.elevenlabs.io"
interruption_callback = AsyncMock()

conversation = AsyncConversation(
client=mock_client,
agent_id=TEST_AGENT_ID,
requires_auth=False,
audio_interface=None,
callback_interruption=interruption_callback,
)

with patch("elevenlabs.conversational_ai.conversation.websockets.connect") as mock_connect:
mock_connect.return_value.__aenter__.return_value = mock_ws

await conversation.start_session()
await asyncio.sleep(0.1)

await conversation.end_session()
await conversation.wait_for_session_end()

interruption_callback.assert_called_once_with(InterruptionEvent(event_id=1, reason=None))
Loading