adk_realtime/
agent.rs

1//! RealtimeAgent - an Agent implementation for real-time voice interactions.
2//!
3//! This module provides `RealtimeAgent`, which implements the `adk_core::Agent` trait
4//! and provides the same callback/tool/instruction features as `LlmAgent`, but uses
5//! real-time bidirectional audio streaming instead of text-based LLM calls.
6//!
7//! # Architecture
8//!
9//! ```text
10//!                     ┌─────────────────────────────────────────┐
11//!                     │              Agent Trait                │
12//!                     │  (name, description, run, sub_agents)   │
13//!                     └────────────────┬────────────────────────┘
14//!                                      │
15//!              ┌───────────────────────┼───────────────────────┐
16//!              │                       │                       │
17//!     ┌────────▼────────┐    ┌─────────▼─────────┐   ┌─────────▼─────────┐
18//!     │    LlmAgent     │    │  RealtimeAgent    │   │  SequentialAgent  │
19//!     │  (text-based)   │    │  (voice-based)    │   │   (workflow)      │
20//!     └─────────────────┘    └───────────────────┘   └───────────────────┘
21//! ```
22//!
23//! # Shared Features with LlmAgent
24//!
25//! - **Tools**: Function tools that can be called during conversation
26//! - **Callbacks**: before_agent, after_agent, before_tool, after_tool
27//! - **Instructions**: Static or dynamic instruction providers
28//! - **Sub-agents**: Agent handoff/transfer support
29//! - **Context**: Full access to InvocationContext (session, memory, artifacts)
30//!
31//! # Example
32//!
33//! ```rust,ignore
34//! use adk_realtime::RealtimeAgent;
35//! use adk_realtime::openai::OpenAIRealtimeModel;
36//!
37//! let model = OpenAIRealtimeModel::new(api_key, "gpt-4o-realtime-preview-2024-12-17");
38//!
39//! let agent = RealtimeAgent::builder("voice_assistant")
40//!     .model(Box::new(model))
41//!     .instruction("You are a helpful voice assistant.")
42//!     .voice("alloy")
43//!     .tool(Arc::new(weather_tool))
44//!     .before_agent_callback(|ctx| async move {
45//!         println!("Starting voice session for user: {}", ctx.user_id());
46//!         Ok(None)
47//!     })
48//!     .build()?;
49//!
50//! // Run through standard ADK runner
51//! let runner = Runner::new(agent);
52//! runner.run(session, user_content).await?;
53//! ```
54
55use crate::config::{RealtimeConfig, ToolDefinition, VadConfig, VadMode};
56use crate::events::{ServerEvent, ToolResponse};
57use adk_core::{
58    AdkError, AfterAgentCallback, AfterToolCallback, Agent, BeforeAgentCallback,
59    BeforeToolCallback, CallbackContext, Content, Event, EventActions, EventStream,
60    GlobalInstructionProvider, InstructionProvider, InvocationContext, MemoryEntry, Part,
61    ReadonlyContext, Result, Tool, ToolCallbackContext, ToolContext, Toolset,
62};
63use async_stream::stream;
64use async_trait::async_trait;
65
66use std::sync::{Arc, Mutex};
67
68/// Shared realtime model type (thread-safe for async usage).
69pub type BoxedRealtimeModel = Arc<dyn crate::model::RealtimeModel>;
70
71/// A real-time voice agent that implements the ADK Agent trait.
72///
73/// `RealtimeAgent` provides bidirectional audio streaming while maintaining
74/// compatibility with the standard ADK agent ecosystem. It supports the same
75/// callbacks, tools, and instruction patterns as `LlmAgent`.
76pub struct RealtimeAgent {
77    name: String,
78    description: String,
79    model: BoxedRealtimeModel,
80
81    // Instructions (same as LlmAgent)
82    instruction: Option<String>,
83    instruction_provider: Option<Arc<InstructionProvider>>,
84    global_instruction: Option<String>,
85    global_instruction_provider: Option<Arc<GlobalInstructionProvider>>,
86
87    // Voice-specific settings
88    voice: Option<String>,
89    vad_config: Option<VadConfig>,
90    modalities: Vec<String>,
91
92    // Tools (same as LlmAgent)
93    tools: Vec<Arc<dyn Tool>>,
94    toolsets: Vec<Arc<dyn Toolset>>,
95    sub_agents: Vec<Arc<dyn Agent>>,
96
97    // Callbacks (same as LlmAgent)
98    before_callbacks: Arc<Vec<BeforeAgentCallback>>,
99    after_callbacks: Arc<Vec<AfterAgentCallback>>,
100    before_tool_callbacks: Arc<Vec<BeforeToolCallback>>,
101    after_tool_callbacks: Arc<Vec<AfterToolCallback>>,
102
103    // Realtime-specific callbacks
104    on_audio: Option<AudioCallback>,
105    on_transcript: Option<TranscriptCallback>,
106    on_speech_started: Option<SpeechCallback>,
107    on_speech_stopped: Option<SpeechCallback>,
108}
109
110/// Callback for audio output events (receives raw PCM bytes).
111pub type AudioCallback = Arc<
112    dyn Fn(&[u8], &str) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>>
113        + Send
114        + Sync,
115>;
116
117/// Callback for transcript events.
118pub type TranscriptCallback = Arc<
119    dyn Fn(&str, &str) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>>
120        + Send
121        + Sync,
122>;
123
124/// Callback for speech detection events.
125pub type SpeechCallback = Arc<
126    dyn Fn(u64) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>> + Send + Sync,
127>;
128
129impl std::fmt::Debug for RealtimeAgent {
130    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
131        f.debug_struct("RealtimeAgent")
132            .field("name", &self.name)
133            .field("description", &self.description)
134            .field("model", &self.model.model_id())
135            .field("voice", &self.voice)
136            .field("tools_count", &self.tools.len())
137            .field("toolsets_count", &self.toolsets.len())
138            .field("sub_agents_count", &self.sub_agents.len())
139            .finish()
140    }
141}
142
143/// Builder for RealtimeAgent.
144pub struct RealtimeAgentBuilder {
145    name: String,
146    description: Option<String>,
147    model: Option<BoxedRealtimeModel>,
148    instruction: Option<String>,
149    instruction_provider: Option<Arc<InstructionProvider>>,
150    global_instruction: Option<String>,
151    global_instruction_provider: Option<Arc<GlobalInstructionProvider>>,
152    voice: Option<String>,
153    vad_config: Option<VadConfig>,
154    modalities: Vec<String>,
155    tools: Vec<Arc<dyn Tool>>,
156    toolsets: Vec<Arc<dyn Toolset>>,
157    sub_agents: Vec<Arc<dyn Agent>>,
158    before_callbacks: Vec<BeforeAgentCallback>,
159    after_callbacks: Vec<AfterAgentCallback>,
160    before_tool_callbacks: Vec<BeforeToolCallback>,
161    after_tool_callbacks: Vec<AfterToolCallback>,
162    on_audio: Option<AudioCallback>,
163    on_transcript: Option<TranscriptCallback>,
164    on_speech_started: Option<SpeechCallback>,
165    on_speech_stopped: Option<SpeechCallback>,
166}
167
168impl RealtimeAgentBuilder {
169    /// Create a new builder with the given agent name.
170    pub fn new(name: impl Into<String>) -> Self {
171        Self {
172            name: name.into(),
173            description: None,
174            model: None,
175            instruction: None,
176            instruction_provider: None,
177            global_instruction: None,
178            global_instruction_provider: None,
179            voice: None,
180            vad_config: None,
181            modalities: vec!["text".to_string(), "audio".to_string()],
182            tools: Vec::new(),
183            toolsets: Vec::new(),
184            sub_agents: Vec::new(),
185            before_callbacks: Vec::new(),
186            after_callbacks: Vec::new(),
187            before_tool_callbacks: Vec::new(),
188            after_tool_callbacks: Vec::new(),
189            on_audio: None,
190            on_transcript: None,
191            on_speech_started: None,
192            on_speech_stopped: None,
193        }
194    }
195
196    /// Set the agent description.
197    pub fn description(mut self, desc: impl Into<String>) -> Self {
198        self.description = Some(desc.into());
199        self
200    }
201
202    /// Set the realtime model.
203    pub fn model(mut self, model: BoxedRealtimeModel) -> Self {
204        self.model = Some(model);
205        self
206    }
207
208    /// Set a static instruction.
209    pub fn instruction(mut self, instruction: impl Into<String>) -> Self {
210        self.instruction = Some(instruction.into());
211        self
212    }
213
214    /// Set a dynamic instruction provider.
215    pub fn instruction_provider(mut self, provider: InstructionProvider) -> Self {
216        self.instruction_provider = Some(Arc::new(provider));
217        self
218    }
219
220    /// Set a static global instruction.
221    pub fn global_instruction(mut self, instruction: impl Into<String>) -> Self {
222        self.global_instruction = Some(instruction.into());
223        self
224    }
225
226    /// Set a dynamic global instruction provider.
227    pub fn global_instruction_provider(mut self, provider: GlobalInstructionProvider) -> Self {
228        self.global_instruction_provider = Some(Arc::new(provider));
229        self
230    }
231
232    /// Set the voice for audio output.
233    pub fn voice(mut self, voice: impl Into<String>) -> Self {
234        self.voice = Some(voice.into());
235        self
236    }
237
238    /// Set voice activity detection configuration.
239    pub fn vad(mut self, config: VadConfig) -> Self {
240        self.vad_config = Some(config);
241        self
242    }
243
244    /// Enable server-side VAD with default settings.
245    pub fn server_vad(mut self) -> Self {
246        self.vad_config = Some(VadConfig {
247            mode: VadMode::ServerVad,
248            threshold: Some(0.5),
249            prefix_padding_ms: Some(300),
250            silence_duration_ms: Some(500),
251            interrupt_response: Some(true),
252            eagerness: None,
253        });
254        self
255    }
256
257    /// Set output modalities (e.g., ["text", "audio"]).
258    pub fn modalities(mut self, modalities: Vec<String>) -> Self {
259        self.modalities = modalities;
260        self
261    }
262
263    /// Add a tool.
264    pub fn tool(mut self, tool: Arc<dyn Tool>) -> Self {
265        self.tools.push(tool);
266        self
267    }
268
269    /// Register a dynamic toolset for per-invocation tool resolution.
270    ///
271    /// Toolsets are resolved at the start of each `run()` call using the
272    /// invocation's `ReadonlyContext`. This enables context-dependent tools
273    /// like per-user browser sessions from a pool.
274    pub fn toolset(mut self, toolset: Arc<dyn Toolset>) -> Self {
275        self.toolsets.push(toolset);
276        self
277    }
278
279    /// Add a sub-agent for handoffs.
280    pub fn sub_agent(mut self, agent: Arc<dyn Agent>) -> Self {
281        self.sub_agents.push(agent);
282        self
283    }
284
285    /// Add a before-agent callback.
286    pub fn before_agent_callback(mut self, callback: BeforeAgentCallback) -> Self {
287        self.before_callbacks.push(callback);
288        self
289    }
290
291    /// Add an after-agent callback.
292    pub fn after_agent_callback(mut self, callback: AfterAgentCallback) -> Self {
293        self.after_callbacks.push(callback);
294        self
295    }
296
297    /// Add a before-tool callback.
298    pub fn before_tool_callback(mut self, callback: BeforeToolCallback) -> Self {
299        self.before_tool_callbacks.push(callback);
300        self
301    }
302
303    /// Add an after-tool callback.
304    pub fn after_tool_callback(mut self, callback: AfterToolCallback) -> Self {
305        self.after_tool_callbacks.push(callback);
306        self
307    }
308
309    /// Set callback for audio output events.
310    pub fn on_audio(mut self, callback: AudioCallback) -> Self {
311        self.on_audio = Some(callback);
312        self
313    }
314
315    /// Set callback for transcript events.
316    pub fn on_transcript(mut self, callback: TranscriptCallback) -> Self {
317        self.on_transcript = Some(callback);
318        self
319    }
320
321    /// Set callback for speech started events.
322    pub fn on_speech_started(mut self, callback: SpeechCallback) -> Self {
323        self.on_speech_started = Some(callback);
324        self
325    }
326
327    /// Set callback for speech stopped events.
328    pub fn on_speech_stopped(mut self, callback: SpeechCallback) -> Self {
329        self.on_speech_stopped = Some(callback);
330        self
331    }
332
333    /// Build the RealtimeAgent.
334    pub fn build(self) -> Result<RealtimeAgent> {
335        let model =
336            self.model.ok_or_else(|| AdkError::agent("RealtimeModel is required".to_string()))?;
337
338        Ok(RealtimeAgent {
339            name: self.name,
340            description: self.description.unwrap_or_default(),
341            model,
342            instruction: self.instruction,
343            instruction_provider: self.instruction_provider,
344            global_instruction: self.global_instruction,
345            global_instruction_provider: self.global_instruction_provider,
346            voice: self.voice,
347            vad_config: self.vad_config,
348            modalities: self.modalities,
349            tools: self.tools,
350            toolsets: self.toolsets,
351            sub_agents: self.sub_agents,
352            before_callbacks: Arc::new(self.before_callbacks),
353            after_callbacks: Arc::new(self.after_callbacks),
354            before_tool_callbacks: Arc::new(self.before_tool_callbacks),
355            after_tool_callbacks: Arc::new(self.after_tool_callbacks),
356            on_audio: self.on_audio,
357            on_transcript: self.on_transcript,
358            on_speech_started: self.on_speech_started,
359            on_speech_stopped: self.on_speech_stopped,
360        })
361    }
362}
363
364impl RealtimeAgent {
365    /// Create a new builder.
366    pub fn builder(name: impl Into<String>) -> RealtimeAgentBuilder {
367        RealtimeAgentBuilder::new(name)
368    }
369
370    /// Get the static instruction, if set.
371    pub fn instruction(&self) -> Option<&String> {
372        self.instruction.as_ref()
373    }
374
375    /// Get the voice setting, if set.
376    pub fn voice(&self) -> Option<&String> {
377        self.voice.as_ref()
378    }
379
380    /// Get the VAD configuration, if set.
381    pub fn vad_config(&self) -> Option<&VadConfig> {
382        self.vad_config.as_ref()
383    }
384
385    /// Get the list of tools.
386    pub fn tools(&self) -> &[Arc<dyn Tool>] {
387        &self.tools
388    }
389
390    /// Build the realtime configuration from agent settings.
391    async fn build_config(
392        &self,
393        ctx: &Arc<dyn InvocationContext>,
394        resolved_tools: &[Arc<dyn Tool>],
395    ) -> Result<RealtimeConfig> {
396        let mut config = RealtimeConfig::default();
397
398        // Build instruction from providers or static value
399        if let Some(provider) = &self.global_instruction_provider {
400            let global_inst = provider(ctx.clone() as Arc<dyn ReadonlyContext>).await?;
401            if !global_inst.is_empty() {
402                config.instruction = Some(global_inst);
403            }
404        } else if let Some(ref template) = self.global_instruction {
405            let processed = adk_core::inject_session_state(ctx.as_ref(), template).await?;
406            config.instruction = Some(processed);
407        }
408
409        // Add agent-specific instruction
410        if let Some(provider) = &self.instruction_provider {
411            let inst = provider(ctx.clone() as Arc<dyn ReadonlyContext>).await?;
412            if !inst.is_empty() {
413                if let Some(existing) = &mut config.instruction {
414                    existing.push_str("\n\n");
415                    existing.push_str(&inst);
416                } else {
417                    config.instruction = Some(inst);
418                }
419            }
420        } else if let Some(ref template) = self.instruction {
421            let processed = adk_core::inject_session_state(ctx.as_ref(), template).await?;
422            if let Some(existing) = &mut config.instruction {
423                existing.push_str("\n\n");
424                existing.push_str(&processed);
425            } else {
426                config.instruction = Some(processed);
427            }
428        }
429
430        // Voice settings
431        config.voice = self.voice.clone();
432        config.turn_detection = self.vad_config.clone();
433        config.modalities = Some(self.modalities.clone());
434
435        // Convert ADK tools to realtime tool definitions
436        let tool_defs: Vec<ToolDefinition> = resolved_tools
437            .iter()
438            .map(|t| ToolDefinition {
439                name: t.name().to_string(),
440                description: Some(t.enhanced_description().to_string()),
441                parameters: t.parameters_schema(),
442            })
443            .collect();
444
445        if !tool_defs.is_empty() {
446            config.tools = Some(tool_defs);
447        }
448
449        // Add transfer_to_agent tool if sub-agents exist
450        if !self.sub_agents.is_empty() {
451            let mut tools = config.tools.unwrap_or_default();
452            tools.push(ToolDefinition {
453                name: "transfer_to_agent".to_string(),
454                description: Some("Transfer execution to another agent.".to_string()),
455                parameters: Some(serde_json::json!({
456                    "type": "object",
457                    "properties": {
458                        "agent_name": {
459                            "type": "string",
460                            "description": "The name of the agent to transfer to."
461                        }
462                    },
463                    "required": ["agent_name"]
464                })),
465            });
466            config.tools = Some(tools);
467        }
468
469        Ok(config)
470    }
471
472    /// Execute a tool call.
473    #[allow(dead_code)]
474    async fn execute_tool(
475        &self,
476        ctx: &Arc<dyn InvocationContext>,
477        call_id: &str,
478        name: &str,
479        arguments: &str,
480    ) -> (serde_json::Value, EventActions) {
481        // Find the tool
482        let tool = self.tools.iter().find(|t| t.name() == name);
483
484        if let Some(tool) = tool {
485            let args: serde_json::Value =
486                serde_json::from_str(arguments).unwrap_or(serde_json::json!({}));
487
488            // Create tool context
489            let tool_ctx: Arc<dyn ToolContext> =
490                Arc::new(RealtimeToolContext::new(ctx.clone(), call_id.to_string()));
491
492            // Execute before_tool callbacks
493            let tool_cb_ctx =
494                Arc::new(ToolCallbackContext::new(ctx.clone(), name.to_string(), args.clone()));
495            for callback in self.before_tool_callbacks.as_ref() {
496                if let Err(e) = callback(tool_cb_ctx.clone() as Arc<dyn CallbackContext>).await {
497                    return (
498                        serde_json::json!({ "error": e.to_string() }),
499                        EventActions::default(),
500                    );
501                }
502            }
503
504            // Execute the tool
505            let result = match tool.execute(tool_ctx.clone(), args.clone()).await {
506                Ok(result) => result,
507                Err(e) => serde_json::json!({ "error": e.to_string() }),
508            };
509
510            let actions = tool_ctx.actions();
511
512            // Execute after_tool callbacks
513            let tool_cb_ctx =
514                Arc::new(ToolCallbackContext::new(ctx.clone(), name.to_string(), args.clone()));
515            for callback in self.after_tool_callbacks.as_ref() {
516                if let Err(e) = callback(tool_cb_ctx.clone() as Arc<dyn CallbackContext>).await {
517                    return (serde_json::json!({ "error": e.to_string() }), actions);
518                }
519            }
520
521            (result, actions)
522        } else {
523            (
524                serde_json::json!({ "error": format!("Tool {} not found", name) }),
525                EventActions::default(),
526            )
527        }
528    }
529}
530
531#[async_trait]
532impl Agent for RealtimeAgent {
533    fn name(&self) -> &str {
534        &self.name
535    }
536
537    fn description(&self) -> &str {
538        &self.description
539    }
540
541    fn sub_agents(&self) -> &[Arc<dyn Agent>] {
542        &self.sub_agents
543    }
544
545    async fn run(&self, ctx: Arc<dyn InvocationContext>) -> Result<EventStream> {
546        let agent_name = self.name.clone();
547        let invocation_id = ctx.invocation_id().to_string();
548        let model = self.model.clone();
549        let _sub_agents = self.sub_agents.clone();
550
551        // Clone callback refs
552        let before_callbacks = self.before_callbacks.clone();
553        let after_callbacks = self.after_callbacks.clone();
554        let before_tool_callbacks = self.before_tool_callbacks.clone();
555        let after_tool_callbacks = self.after_tool_callbacks.clone();
556        let tools = self.tools.clone();
557        let toolsets = self.toolsets.clone();
558
559        // Clone realtime callbacks
560        let on_audio = self.on_audio.clone();
561        let on_transcript = self.on_transcript.clone();
562        let on_speech_started = self.on_speech_started.clone();
563        let on_speech_stopped = self.on_speech_stopped.clone();
564
565        // ===== RESOLVE TOOLSETS =====
566        let mut resolved_tools: Vec<Arc<dyn Tool>> = tools.clone();
567        let static_tool_names: std::collections::HashSet<String> =
568            tools.iter().map(|t| t.name().to_string()).collect();
569        let mut toolset_source: std::collections::HashMap<String, String> =
570            std::collections::HashMap::new();
571
572        for toolset in &toolsets {
573            let toolset_tools = toolset.tools(ctx.clone() as Arc<dyn ReadonlyContext>).await?;
574            for tool in &toolset_tools {
575                let name = tool.name().to_string();
576                if static_tool_names.contains(&name) {
577                    return Err(AdkError::agent(format!(
578                        "Duplicate tool name '{}': conflict between static tool and toolset '{}'",
579                        name,
580                        toolset.name()
581                    )));
582                }
583                if let Some(other_toolset_name) = toolset_source.get(&name) {
584                    return Err(AdkError::agent(format!(
585                        "Duplicate tool name '{}': conflict between toolset '{}' and toolset '{}'",
586                        name,
587                        other_toolset_name,
588                        toolset.name()
589                    )));
590                }
591                toolset_source.insert(name, toolset.name().to_string());
592                resolved_tools.push(tool.clone());
593            }
594        }
595
596        // Build config with resolved tools
597        let config = self.build_config(&ctx, &resolved_tools).await?;
598
599        let s = stream! {
600            // ===== BEFORE AGENT CALLBACKS =====
601            for callback in before_callbacks.as_ref() {
602                match callback(ctx.clone() as Arc<dyn CallbackContext>).await {
603                    Ok(Some(content)) => {
604                        let mut early_event = Event::new(&invocation_id);
605                        early_event.author = agent_name.clone();
606                        early_event.llm_response.content = Some(content);
607                        yield Ok(early_event);
608                        return;
609                    }
610                    Ok(None) => continue,
611                    Err(e) => {
612                        yield Err(e);
613                        return;
614                    }
615                }
616            }
617
618            // ===== CONNECT TO REALTIME SESSION =====
619            let session = match model.connect(config).await {
620                Ok(s) => s,
621                Err(e) => {
622                    yield Err(AdkError::model(format!("Failed to connect: {}", e)));
623                    return;
624                }
625            };
626
627            // Yield session started event
628            let mut start_event = Event::new(&invocation_id);
629            start_event.author = agent_name.clone();
630            start_event.llm_response.content = Some(Content {
631                role: "system".to_string(),
632                parts: vec![Part::Text {
633                    text: format!("Realtime session started: {}", session.session_id()),
634                }],
635            });
636            yield Ok(start_event);
637
638            // ===== SEND INITIAL USER CONTENT =====
639            // If user provided text input, send it to start the conversation
640            let user_content = ctx.user_content();
641            for part in &user_content.parts {
642                if let Part::Text { text } = part {
643                    if let Err(e) = session.send_text(text).await {
644                        yield Err(AdkError::model(format!("Failed to send text: {}", e)));
645                        return;
646                    }
647                    // Request a response
648                    if let Err(e) = session.create_response().await {
649                        yield Err(AdkError::model(format!("Failed to create response: {}", e)));
650                        return;
651                    }
652                }
653            }
654
655            // ===== PROCESS REALTIME EVENTS =====
656            loop {
657                let event = session.next_event().await;
658
659                match event {
660                    Some(Ok(server_event)) => {
661                        match server_event {
662                            ServerEvent::AudioDelta { delta, item_id, .. } => {
663                                // Call audio callback if set
664                                if let Some(ref cb) = on_audio {
665                                    cb(&delta, &item_id).await;
666                                }
667
668                                // Yield audio event (delta is already raw bytes)
669                                let mut audio_event = Event::new(&invocation_id);
670                                audio_event.author = agent_name.clone();
671                                audio_event.llm_response.content = Some(Content {
672                                    role: "model".to_string(),
673                                    parts: vec![Part::InlineData {
674                                        mime_type: "audio/pcm".to_string(),
675                                        data: delta,
676                                    }],
677                                });
678                                yield Ok(audio_event);
679                            }
680
681                            ServerEvent::TextDelta { delta, .. } => {
682                                let mut text_event = Event::new(&invocation_id);
683                                text_event.author = agent_name.clone();
684                                text_event.llm_response.content = Some(Content {
685                                    role: "model".to_string(),
686                                    parts: vec![Part::Text { text: delta.clone() }],
687                                });
688                                yield Ok(text_event);
689                            }
690
691                            ServerEvent::TranscriptDelta { delta, item_id, .. } => {
692                                if let Some(ref cb) = on_transcript {
693                                    cb(&delta, &item_id).await;
694                                }
695                            }
696
697                            ServerEvent::SpeechStarted { audio_start_ms, .. } => {
698                                if let Some(ref cb) = on_speech_started {
699                                    cb(audio_start_ms).await;
700                                }
701                            }
702
703                            ServerEvent::SpeechStopped { audio_end_ms, .. } => {
704                                if let Some(ref cb) = on_speech_stopped {
705                                    cb(audio_end_ms).await;
706                                }
707                            }
708
709                            ServerEvent::FunctionCallDone {
710                                call_id,
711                                name,
712                                arguments,
713                                ..
714                            } => {
715                                // Handle transfer_to_agent
716                                if name == "transfer_to_agent" {
717                                    let args: serde_json::Value = serde_json::from_str(&arguments)
718                                        .unwrap_or(serde_json::json!({}));
719                                    let target = args.get("agent_name")
720                                        .and_then(|v| v.as_str())
721                                        .unwrap_or_default()
722                                        .to_string();
723
724                                    let mut transfer_event = Event::new(&invocation_id);
725                                    transfer_event.author = agent_name.clone();
726                                    transfer_event.actions.transfer_to_agent = Some(target);
727                                    yield Ok(transfer_event);
728
729                                    let _ = session.close().await;
730                                    return;
731                                }
732
733                                // Execute tool
734                                let tool = resolved_tools.iter().find(|t| t.name() == name);
735
736                                let (result, actions) = if let Some(tool) = tool {
737                                    let args: serde_json::Value = serde_json::from_str(&arguments)
738                                        .unwrap_or(serde_json::json!({}));
739
740                                    let tool_ctx: Arc<dyn ToolContext> = Arc::new(
741                                        RealtimeToolContext::new(ctx.clone(), call_id.clone())
742                                    );
743
744                                    // Execute before_tool callbacks
745                                    let tool_cb_ctx = Arc::new(ToolCallbackContext::new(
746                                        ctx.clone(),
747                                        name.clone(),
748                                        args.clone(),
749                                    ));
750                                    for callback in before_tool_callbacks.as_ref() {
751                                        if let Err(e) = callback(tool_cb_ctx.clone() as Arc<dyn CallbackContext>).await {
752                                            let error_result = serde_json::json!({ "error": e.to_string() });
753                                            (error_result, EventActions::default())
754                                        } else {
755                                            continue;
756                                        };
757                                    }
758
759                                    let result = match tool.execute(tool_ctx.clone(), args.clone()).await {
760                                        Ok(r) => r,
761                                        Err(e) => serde_json::json!({ "error": e.to_string() }),
762                                    };
763
764                                    let actions = tool_ctx.actions();
765
766                                    // Execute after_tool callbacks
767                                    let tool_cb_ctx = Arc::new(ToolCallbackContext::new(
768                                        ctx.clone(),
769                                        name.clone(),
770                                        args.clone(),
771                                    ));
772                                    for callback in after_tool_callbacks.as_ref() {
773                                        let _ = callback(tool_cb_ctx.clone() as Arc<dyn CallbackContext>).await;
774                                    }
775
776                                    (result, actions)
777                                } else {
778                                    (
779                                        serde_json::json!({ "error": format!("Tool {} not found", name) }),
780                                        EventActions::default(),
781                                    )
782                                };
783
784                                // Yield tool event
785                                let mut tool_event = Event::new(&invocation_id);
786                                tool_event.author = agent_name.clone();
787                                tool_event.actions = actions.clone();
788                                tool_event.llm_response.content = Some(Content {
789                                    role: "function".to_string(),
790                                    parts: vec![Part::FunctionResponse {
791                                        function_response: adk_core::FunctionResponseData::new(name.clone(), result.clone()),
792                                        id: Some(call_id.clone()),
793                                    }],
794                                });
795                                yield Ok(tool_event);
796
797                                // Check for escalation
798                                if actions.escalate || actions.skip_summarization {
799                                    let _ = session.close().await;
800                                    return;
801                                }
802
803                                // Send tool response back to session
804                                let response = ToolResponse {
805                                    call_id,
806                                    output: result,
807                                };
808                                if let Err(e) = session.send_tool_response(response).await {
809                                    yield Err(AdkError::model(format!("Failed to send tool response: {}", e)));
810                                    let _ = session.close().await;
811                                    return;
812                                }
813                            }
814
815                            ServerEvent::ResponseDone { .. } => {
816                                // Response complete, continue listening
817                            }
818
819                            ServerEvent::Error { error, .. } => {
820                                yield Err(AdkError::model(format!(
821                                    "Realtime error: {} - {}",
822                                    error.code.unwrap_or_default(),
823                                    error.message
824                                )));
825                            }
826
827
828                            _ => {
829                                // Ignore other events
830                            }
831                        }
832                    }
833                    Some(Err(e)) => {
834                        yield Err(AdkError::model(format!("Session error: {}", e)));
835                        break;
836                    }
837                    None => {
838                        // Session closed
839                        break;
840                    }
841                }
842            }
843
844            // ===== AFTER AGENT CALLBACKS =====
845            for callback in after_callbacks.as_ref() {
846                match callback(ctx.clone() as Arc<dyn CallbackContext>).await {
847                    Ok(Some(content)) => {
848                        let mut after_event = Event::new(&invocation_id);
849                        after_event.author = agent_name.clone();
850                        after_event.llm_response.content = Some(content);
851                        yield Ok(after_event);
852                        break;
853                    }
854                    Ok(None) => continue,
855                    Err(e) => {
856                        yield Err(e);
857                        return;
858                    }
859                }
860            }
861        };
862
863        Ok(Box::pin(s))
864    }
865}
866
867/// Tool context for realtime agent tool execution.
868struct RealtimeToolContext {
869    parent_ctx: Arc<dyn InvocationContext>,
870    function_call_id: String,
871    actions: Mutex<EventActions>,
872}
873
874impl RealtimeToolContext {
875    fn new(parent_ctx: Arc<dyn InvocationContext>, function_call_id: String) -> Self {
876        Self { parent_ctx, function_call_id, actions: Mutex::new(EventActions::default()) }
877    }
878}
879
880#[async_trait]
881impl ReadonlyContext for RealtimeToolContext {
882    fn invocation_id(&self) -> &str {
883        self.parent_ctx.invocation_id()
884    }
885
886    fn agent_name(&self) -> &str {
887        self.parent_ctx.agent_name()
888    }
889
890    fn user_id(&self) -> &str {
891        self.parent_ctx.user_id()
892    }
893
894    fn app_name(&self) -> &str {
895        self.parent_ctx.app_name()
896    }
897
898    fn session_id(&self) -> &str {
899        self.parent_ctx.session_id()
900    }
901
902    fn branch(&self) -> &str {
903        self.parent_ctx.branch()
904    }
905
906    fn user_content(&self) -> &Content {
907        self.parent_ctx.user_content()
908    }
909}
910
911#[async_trait]
912impl CallbackContext for RealtimeToolContext {
913    fn artifacts(&self) -> Option<Arc<dyn adk_core::Artifacts>> {
914        self.parent_ctx.artifacts()
915    }
916}
917
918#[async_trait]
919impl ToolContext for RealtimeToolContext {
920    fn function_call_id(&self) -> &str {
921        &self.function_call_id
922    }
923
924    fn actions(&self) -> EventActions {
925        self.actions.lock().unwrap().clone()
926    }
927
928    fn set_actions(&self, actions: EventActions) {
929        *self.actions.lock().unwrap() = actions;
930    }
931
932    async fn search_memory(&self, query: &str) -> Result<Vec<MemoryEntry>> {
933        if let Some(memory) = self.parent_ctx.memory() {
934            memory.search(query).await
935        } else {
936            Ok(vec![])
937        }
938    }
939}
adk_realtime/agent.rs

adk_realtime/
agent.rs
