Source code for kani.ext.realtime.events.server
from typing import Literal
from pydantic import Field
from .base import ServerEvent as BaseEvent
from ..models import (
ConversationDetails,
ConversationItem,
ErrorDetails,
RateLimitInfo,
RealtimeResponse,
SessionDetails,
)
[docs]
class SessionCreated(BaseEvent):
type: Literal["session.created"] = "session.created"
session: SessionDetails
[docs]
class SessionUpdated(BaseEvent):
type: Literal["session.updated"] = "session.updated"
session: SessionDetails
[docs]
class ConversationCreated(BaseEvent):
type: Literal["conversation.created"] = "conversation.created"
conversation: ConversationDetails
[docs]
class ConversationItemCreated(BaseEvent):
type: Literal["conversation.item.created"] = "conversation.item.created"
previous_item_id: str | None
item: ConversationItem
[docs]
class ConversationItemInputAudioTranscriptionCompleted(BaseEvent):
type: Literal["conversation.item.input_audio_transcription.completed"] = (
"conversation.item.input_audio_transcription.completed"
)
item_id: str
content_index: int
transcript: str
[docs]
class ConversationItemInputAudioTranscriptionFailed(BaseEvent):
type: Literal["conversation.item.input_audio_transcription.failed"] = (
"conversation.item.input_audio_transcription.failed"
)
item_id: str
content_index: int
error: ErrorDetails
[docs]
class ConversationItemTruncated(BaseEvent):
type: Literal["conversation.item.truncated"] = "conversation.item.truncated"
item_id: str
content_index: int
audio_end_ms: int
[docs]
class ConversationItemDeleted(BaseEvent):
type: Literal["conversation.item.deleted"] = "conversation.item.deleted"
item_id: str
[docs]
class InputAudioBufferCommitted(BaseEvent):
type: Literal["input_audio_buffer.committed"] = "input_audio_buffer.committed"
previous_item_id: str | None
item_id: str
[docs]
class InputAudioBufferCleared(BaseEvent):
type: Literal["input_audio_buffer.cleared"] = "input_audio_buffer.cleared"
[docs]
class InputAudioBufferSpeechStarted(BaseEvent):
type: Literal["input_audio_buffer.speech_started"] = "input_audio_buffer.speech_started"
audio_start_ms: int
item_id: str
[docs]
class InputAudioBufferSpeechStopped(BaseEvent):
type: Literal["input_audio_buffer.speech_stopped"] = "input_audio_buffer.speech_stopped"
audio_end_ms: int
item_id: str
[docs]
class ResponseCreated(BaseEvent):
type: Literal["response.created"] = "response.created"
response: RealtimeResponse
[docs]
class ResponseDone(BaseEvent):
type: Literal["response.done"] = "response.done"
response: RealtimeResponse
[docs]
class RateLimitsUpdated(BaseEvent):
type: Literal["rate_limits.updated"] = "rate_limits.updated"
rate_limits: list[RateLimitInfo]
# ---- streaming ----
# TODO:
# response.output_item.added
# response.output_item.done
[docs]
class ResponseOutputItemDone(BaseEvent):
type: Literal["response.output_item.done"] = "response.output_item.done"
response_id: str
output_index: int
item: ConversationItem
# response.content_part.added
# response.content_part.done
# response.text.delta
[docs]
class ResponseTextDelta(BaseEvent):
type: Literal["response.text.delta"] = "response.text.delta"
response_id: str
item_id: str
output_index: int
content_index: int
delta: str
# response.text.done
# response.audio_transcript.delta
[docs]
class ResponseAudioTranscriptDelta(BaseEvent):
type: Literal["response.audio_transcript.delta"] = "response.audio_transcript.delta"
response_id: str
item_id: str
output_index: int
content_index: int
delta: str
# response.audio_transcript.done
# response.audio.delta
[docs]
class ResponseAudioDelta(BaseEvent):
type: Literal["response.audio.delta"] = "response.audio.delta"
response_id: str
item_id: str
output_index: int
content_index: int
delta: str = Field(repr=False)
# response.audio.done
# response.function_call_arguments.delta
# response.function_call_arguments.done