openai_tools/live/
client.rs

1//! WebSocket client and active-session types for the OpenAI Live API.
2
3use std::sync::{
4    atomic::{AtomicBool, Ordering},
5    Arc,
6};
7
8use base64::prelude::*;
9use futures_util::{stream::SplitSink, stream::SplitStream, SinkExt, StreamExt};
10use tokio::{net::TcpStream, sync::Mutex};
11use tokio_tungstenite::{
12    connect_async_with_config,
13    tungstenite::{client::IntoClientRequest, Message as WsMessage},
14    MaybeTlsStream, WebSocketStream,
15};
16
17use crate::common::{
18    auth::{AuthProvider, OpenAIAuth},
19    errors::{OpenAIToolError, Result},
20    models::LiveModel,
21};
22
23use super::{
24    events::{ClientEvent, ServerEvent, SessionClosedEvent, SessionResource},
25    session::{
26        Delegation, LiveAudioConfig, LiveAudioEncoding, LiveAudioFormat, LiveAudioOutputConfig, LiveSessionConfig, LiveSessionUpdate, LiveVoice,
27    },
28};
29
30const LIVE_SESSIONS_URL: &str = "wss://api.openai.com/v1/live/sessions";
31
32type LiveWebSocket = WebSocketStream<MaybeTlsStream<TcpStream>>;
33type LiveSink = SplitSink<LiveWebSocket, WsMessage>;
34type LiveStream = SplitStream<LiveWebSocket>;
35
36/// Builder for a primary or forked Live WebSocket session.
37#[derive(Debug, Clone)]
38pub struct LiveClient {
39    auth: AuthProvider,
40    session_config: LiveSessionConfig,
41}
42
43impl LiveClient {
44    /// Creates a client using `OPENAI_API_KEY`.
45    pub fn new() -> Self {
46        let auth = AuthProvider::openai_from_env().expect("OPENAI_API_KEY must be set");
47        Self::with_auth(auth)
48    }
49
50    /// Creates a client using an explicit OpenAI API key.
51    pub fn with_api_key(api_key: impl Into<String>) -> Self {
52        Self::with_auth(AuthProvider::OpenAI(OpenAIAuth::new(api_key)))
53    }
54
55    /// Creates a client with an authentication provider.
56    ///
57    /// The Live API currently supports only [`AuthProvider::OpenAI`]. Passing
58    /// Azure authentication is allowed here so that configuration can be
59    /// shared, but [`connect`](Self::connect) and [`fork`](Self::fork) reject it.
60    pub fn with_auth(auth: AuthProvider) -> Self {
61        Self { auth, session_config: LiveSessionConfig::default() }
62    }
63
64    /// Selects the frontend Live model.
65    pub fn model(&mut self, model: LiveModel) -> &mut Self {
66        self.session_config.model = Some(model);
67        self
68    }
69
70    /// Sets immutable frontend instructions.
71    pub fn instructions(&mut self, instructions: impl Into<String>) -> &mut Self {
72        self.session_config.instructions = Some(instructions.into());
73        self
74    }
75
76    /// Selects the output voice.
77    pub fn voice(&mut self, voice: LiveVoice) -> &mut Self {
78        self.audio_mut().output.get_or_insert_with(LiveAudioOutputConfig::default).voice = Some(voice);
79        self
80    }
81
82    /// Selects the common input/output audio format.
83    pub fn audio_format(&mut self, format: LiveAudioFormat) -> &mut Self {
84        self.audio_mut().format = Some(format);
85        self
86    }
87
88    /// Selects the delegation strategy.
89    pub fn delegation(&mut self, delegation: Delegation) -> &mut Self {
90        self.session_config.delegation = Some(delegation);
91        self
92    }
93
94    /// Enables or disables storage for later forks.
95    pub fn store(&mut self, store: bool) -> &mut Self {
96        self.session_config.store = Some(store);
97        self
98    }
99
100    /// Supplies initial Responses-style input items.
101    pub fn input(&mut self, input: Vec<serde_json::Value>) -> &mut Self {
102        self.session_config.input = Some(input);
103        self
104    }
105
106    /// Starts a new primary Live WebSocket session.
107    pub async fn connect(&self) -> Result<LiveSession> {
108        let mut config = self.session_config.clone();
109        config.model.get_or_insert_default();
110        self.connect_to(LIVE_SESSIONS_URL.to_string(), config).await
111    }
112
113    /// Forks a stored Live session, applying only explicitly configured fields.
114    pub async fn fork(&self, session_id: &str) -> Result<LiveSession> {
115        let url = format!("{LIVE_SESSIONS_URL}/{session_id}/fork");
116        self.connect_to(url, self.session_config.clone()).await
117    }
118
119    fn audio_mut(&mut self) -> &mut LiveAudioConfig {
120        self.session_config.audio.get_or_insert_with(LiveAudioConfig::default)
121    }
122
123    async fn connect_to(&self, url: String, config: LiveSessionConfig) -> Result<LiveSession> {
124        let api_key = match &self.auth {
125            AuthProvider::OpenAI(auth) => auth.api_key(),
126            AuthProvider::Azure(_) => {
127                return Err(OpenAIToolError::Error("Azure authentication is not supported by the OpenAI Live API".to_string()));
128            }
129        };
130
131        let mut request =
132            url.into_client_request().map_err(|error| OpenAIToolError::Error(format!("Failed to build Live WebSocket request: {error}")))?;
133        request.headers_mut().insert(
134            "Authorization",
135            format!("Bearer {api_key}").parse().map_err(|error| OpenAIToolError::Error(format!("Invalid Authorization header: {error}")))?,
136        );
137
138        let (socket, _) = connect_async_with_config(request, None, false)
139            .await
140            .map_err(|error| OpenAIToolError::WebSocketError(format!("Live connection failed: {error}")))?;
141        let (sink, stream) = socket.split();
142        let sender = LiveSender::new(sink);
143        let mut receiver = LiveReceiver::new(sender.clone(), stream);
144
145        sender.send(ClientEvent::SessionStart { event_id: None, session: config }).await?;
146
147        let started = loop {
148            match receiver.recv().await? {
149                Some(ServerEvent::SessionStarted(event)) => break event.session,
150                Some(ServerEvent::Error(event)) => {
151                    return Err(OpenAIToolError::Error(format!("Live session start failed: {} - {}", event.error.code, event.error.message)));
152                }
153                Some(_) => continue,
154                None => {
155                    return Err(OpenAIToolError::Error("Live WebSocket closed before session.started".to_string()));
156                }
157            }
158        };
159
160        let pcm16 =
161            started.audio.as_ref().and_then(|audio| audio.format.as_ref()).is_none_or(|format| matches!(format.encoding, LiveAudioEncoding::Pcm));
162        sender.pcm16.store(pcm16, Ordering::Relaxed);
163
164        Ok(LiveSession { started, sender, receiver })
165    }
166}
167
168impl Default for LiveClient {
169    fn default() -> Self {
170        Self::new()
171    }
172}
173
174/// Sending half of a split Live session.
175#[derive(Clone)]
176pub struct LiveSender {
177    sink: Arc<Mutex<LiveSink>>,
178    pcm16: Arc<AtomicBool>,
179}
180
181impl LiveSender {
182    fn new(sink: LiveSink) -> Self {
183        Self { sink: Arc::new(Mutex::new(sink)), pcm16: Arc::new(AtomicBool::new(true)) }
184    }
185
186    /// Sends a raw typed client event.
187    pub async fn send(&self, event: ClientEvent) -> Result<()> {
188        let json = serde_json::to_string(&event)?;
189        self.send_message(WsMessage::Text(json.into())).await
190    }
191
192    /// Appends base64-encoded audio input.
193    pub async fn append_audio(&self, base64: &str) -> Result<()> {
194        self.send(ClientEvent::InputAudioAppend { event_id: None, audio: base64.to_string() }).await
195    }
196
197    /// Base64-encodes and appends raw audio input.
198    ///
199    /// PCM input must contain complete signed 16-bit samples, so odd byte
200    /// lengths are rejected before anything is sent.
201    pub async fn append_audio_bytes(&self, audio: &[u8]) -> Result<()> {
202        if self.pcm16.load(Ordering::Relaxed) && !audio.len().is_multiple_of(2) {
203            return Err(OpenAIToolError::Error("PCM audio must contain an even number of bytes (16-bit sample boundary)".to_string()));
204        }
205        self.append_audio(&BASE64_STANDARD.encode(audio)).await
206    }
207
208    /// Mutes Live input audio.
209    pub async fn mute(&self) -> Result<()> {
210        self.send(ClientEvent::InputAudioMute { event_id: None }).await
211    }
212
213    /// Unmutes Live input audio.
214    pub async fn unmute(&self) -> Result<()> {
215        self.send(ClientEvent::InputAudioUnmute { event_id: None }).await
216    }
217
218    /// Appends system-level instructions.
219    pub async fn append_instructions(&self, content: impl Into<String>, delegation_id: Option<&str>) -> Result<()> {
220        self.send(ClientEvent::InstructionsAppend { event_id: None, content: content.into(), delegation_id: delegation_id.map(str::to_string) }).await
221    }
222
223    /// Appends silent context for model reasoning.
224    pub async fn append_thinking(&self, content: impl Into<String>, delegation_id: Option<&str>) -> Result<()> {
225        self.send(ClientEvent::ThinkingAppend { event_id: None, content: content.into(), delegation_id: delegation_id.map(str::to_string) }).await
226    }
227
228    /// Appends content that the model should speak.
229    pub async fn append_commentary(&self, content: impl Into<String>, delegation_id: Option<&str>) -> Result<()> {
230        self.send(ClientEvent::CommentaryAppend { event_id: None, content: content.into(), delegation_id: delegation_id.map(str::to_string) }).await
231    }
232
233    /// Adds an input item to a Responses delegation.
234    pub async fn create_response_item(&self, item: serde_json::Value) -> Result<()> {
235        self.send(ClientEvent::ResponseItemCreate { event_id: None, item }).await
236    }
237
238    /// Submits a function-call result to a Responses delegation.
239    pub async fn submit_function_output(&self, call_id: impl Into<String>, output: impl Into<String>) -> Result<()> {
240        self.create_response_item(serde_json::json!({
241            "type": "function_call_output",
242            "call_id": call_id.into(),
243            "output": output.into(),
244        }))
245        .await
246    }
247
248    /// Requests that a Responses delegation continue generating.
249    pub async fn create_response(&self) -> Result<()> {
250        self.send(ClientEvent::ResponseCreate { event_id: None }).await
251    }
252
253    /// Updates settings within the current delegation type.
254    pub async fn update_delegation(&self, delegation: Delegation) -> Result<()> {
255        self.send(ClientEvent::SessionUpdate { event_id: None, session: LiveSessionUpdate { delegation } }).await
256    }
257
258    async fn send_message(&self, message: WsMessage) -> Result<()> {
259        self.sink.lock().await.send(message).await.map_err(|error| OpenAIToolError::WebSocketError(format!("Failed to send Live event: {error}")))
260    }
261
262    async fn close_transport(&self) -> Result<()> {
263        self.sink.lock().await.close().await.map_err(|error| OpenAIToolError::WebSocketError(format!("Failed to close Live WebSocket: {error}")))
264    }
265}
266
267/// Receiving half of a split Live session.
268pub struct LiveReceiver {
269    sender: LiveSender,
270    stream: LiveStream,
271}
272
273impl LiveReceiver {
274    fn new(sender: LiveSender, stream: LiveStream) -> Self {
275        Self { sender, stream }
276    }
277
278    /// Receives the next typed server event.
279    ///
280    /// Returns `None` after a WebSocket Close frame or end of stream. Ping
281    /// frames are answered with Pong and are not surfaced to the caller.
282    pub async fn recv(&mut self) -> Result<Option<ServerEvent>> {
283        loop {
284            match self.stream.next().await {
285                Some(Ok(WsMessage::Text(text))) => return Ok(Some(serde_json::from_str(&text)?)),
286                Some(Ok(WsMessage::Close(_))) | None => return Ok(None),
287                Some(Ok(WsMessage::Ping(payload))) => {
288                    self.sender.send_message(WsMessage::Pong(payload)).await?;
289                }
290                Some(Ok(_)) => {}
291                Some(Err(error)) => {
292                    return Err(OpenAIToolError::WebSocketError(format!("Live WebSocket receive failed: {error}")));
293                }
294            }
295        }
296    }
297
298    /// Gracefully closes the Live session and then the WebSocket.
299    pub async fn close(mut self) -> Result<Option<SessionClosedEvent>> {
300        if let Err(error) = self.sender.send(ClientEvent::SessionClose { event_id: None }).await {
301            let _ = self.sender.close_transport().await;
302            return Err(error);
303        }
304
305        let closed = loop {
306            match self.recv().await {
307                Ok(Some(ServerEvent::SessionClosed(event))) => break Ok(Some(event)),
308                Ok(Some(_)) => continue,
309                Ok(None) => break Ok(None),
310                Err(error) => break Err(error),
311            }
312        };
313
314        let transport_close = self.sender.close_transport().await;
315        match closed {
316            Ok(event) => {
317                transport_close?;
318                Ok(event)
319            }
320            Err(error) => {
321                let _ = transport_close;
322                Err(error)
323            }
324        }
325    }
326}
327
328/// An active primary or forked Live WebSocket session.
329///
330/// The model advances only while input audio is streaming in real time. Keep
331/// sending audio frames (silence is acceptable) even when injecting commentary
332/// or instructions; otherwise the injection can remain incomplete until close.
333pub struct LiveSession {
334    started: SessionResource,
335    sender: LiveSender,
336    receiver: LiveReceiver,
337}
338
339impl LiveSession {
340    /// Returns the opaque Live session identifier.
341    pub fn session_id(&self) -> &str {
342        &self.started.id
343    }
344
345    /// Returns the full resolved session from `session.started`.
346    pub fn started(&self) -> &SessionResource {
347        &self.started
348    }
349
350    /// Splits the session for concurrent sending and receiving.
351    pub fn split(self) -> (LiveSender, LiveReceiver) {
352        (self.sender, self.receiver)
353    }
354
355    /// Sends a raw typed client event.
356    pub async fn send(&self, event: ClientEvent) -> Result<()> {
357        self.sender.send(event).await
358    }
359
360    /// Receives the next typed server event.
361    pub async fn recv(&mut self) -> Result<Option<ServerEvent>> {
362        self.receiver.recv().await
363    }
364
365    /// Appends base64-encoded audio input.
366    pub async fn append_audio(&self, base64: &str) -> Result<()> {
367        self.sender.append_audio(base64).await
368    }
369
370    /// Base64-encodes and appends raw audio input.
371    pub async fn append_audio_bytes(&self, audio: &[u8]) -> Result<()> {
372        self.sender.append_audio_bytes(audio).await
373    }
374
375    /// Mutes input audio.
376    pub async fn mute(&self) -> Result<()> {
377        self.sender.mute().await
378    }
379
380    /// Unmutes input audio.
381    pub async fn unmute(&self) -> Result<()> {
382        self.sender.unmute().await
383    }
384
385    /// Appends system-level instructions.
386    pub async fn append_instructions(&self, content: impl Into<String>, delegation_id: Option<&str>) -> Result<()> {
387        self.sender.append_instructions(content, delegation_id).await
388    }
389
390    /// Appends silent reasoning context.
391    pub async fn append_thinking(&self, content: impl Into<String>, delegation_id: Option<&str>) -> Result<()> {
392        self.sender.append_thinking(content, delegation_id).await
393    }
394
395    /// Appends content that the model should speak.
396    pub async fn append_commentary(&self, content: impl Into<String>, delegation_id: Option<&str>) -> Result<()> {
397        self.sender.append_commentary(content, delegation_id).await
398    }
399
400    /// Adds an input item to a Responses delegation.
401    pub async fn create_response_item(&self, item: serde_json::Value) -> Result<()> {
402        self.sender.create_response_item(item).await
403    }
404
405    /// Submits a function-call result to a Responses delegation.
406    pub async fn submit_function_output(&self, call_id: impl Into<String>, output: impl Into<String>) -> Result<()> {
407        self.sender.submit_function_output(call_id, output).await
408    }
409
410    /// Requests that a Responses delegation continue generating.
411    pub async fn create_response(&self) -> Result<()> {
412        self.sender.create_response().await
413    }
414
415    /// Updates settings within the current delegation type.
416    pub async fn update_delegation(&self, delegation: Delegation) -> Result<()> {
417        self.sender.update_delegation(delegation).await
418    }
419
420    /// Sends `session.close`, waits for `session.closed`, and closes the socket.
421    pub async fn close(self) -> Result<Option<SessionClosedEvent>> {
422        self.receiver.close().await
423    }
424}
openai_tools/live/client.rs

openai_tools/live/
client.rs
