use crate::codex::Session; use crate::codex::TurnContext; use crate::function_tool::FunctionCallError; use crate::tools::context::FunctionToolOutput; use crate::tools::context::ToolInvocation; use crate::tools::context::ToolPayload; use crate::tools::handlers::parse_arguments; use crate::tools::registry::ToolHandler; use crate::tools::registry::ToolKind; use codex_protocol::dynamic_tools::DynamicToolCallRequest; use codex_protocol::dynamic_tools::DynamicToolResponse; use codex_protocol::models::FunctionCallOutputContentItem; use codex_protocol::protocol::DynamicToolCallResponseEvent; use codex_protocol::protocol::EventMsg; use serde_json::Value; use std::time::Instant; use tokio::sync::oneshot; use tracing::warn; pub struct DynamicToolHandler; impl ToolHandler for DynamicToolHandler { type Output = FunctionToolOutput; fn kind(&self) -> ToolKind { ToolKind::Function } async fn is_mutating(&self, _invocation: &ToolInvocation) -> bool { true } async fn handle(&self, invocation: ToolInvocation) -> Result { let ToolInvocation { session, turn, call_id, tool_name, payload, .. } = invocation; let arguments = match payload { ToolPayload::Function { arguments } => arguments, _ => { return Err(FunctionCallError::RespondToModel( "dynamic tool handler received unsupported payload".to_string(), )); } }; let args: Value = parse_arguments(&arguments)?; let response = request_dynamic_tool(&session, turn.as_ref(), call_id, tool_name, args) .await .ok_or_else(|| { FunctionCallError::RespondToModel( "dynamic tool call was cancelled before receiving a response".to_string(), ) })?; let DynamicToolResponse { content_items, success, } = response; let body = content_items .into_iter() .map(FunctionCallOutputContentItem::from) .collect::>(); Ok(FunctionToolOutput::from_content(body, Some(success))) } } async fn request_dynamic_tool( session: &Session, turn_context: &TurnContext, call_id: String, tool: String, arguments: Value, ) -> Option { let turn_id = turn_context.sub_id.clone(); let (tx_response, rx_response) = oneshot::channel(); let event_id = call_id.clone(); let prev_entry = { let mut active = session.active_turn.lock().await; match active.as_mut() { Some(at) => { let mut ts = at.turn_state.lock().await; ts.insert_pending_dynamic_tool(call_id.clone(), tx_response) } None => None, } }; if prev_entry.is_some() { warn!("Overwriting existing pending dynamic tool call for call_id: {event_id}"); } let started_at = Instant::now(); let event = EventMsg::DynamicToolCallRequest(DynamicToolCallRequest { call_id: call_id.clone(), turn_id: turn_id.clone(), tool: tool.clone(), arguments: arguments.clone(), }); session.send_event(turn_context, event).await; let response = rx_response.await.ok(); let response_event = match &response { Some(response) => EventMsg::DynamicToolCallResponse(DynamicToolCallResponseEvent { call_id, turn_id, tool, arguments, content_items: response.content_items.clone(), success: response.success, error: None, duration: started_at.elapsed(), }), None => EventMsg::DynamicToolCallResponse(DynamicToolCallResponseEvent { call_id, turn_id, tool, arguments, content_items: Vec::new(), success: false, error: Some("dynamic tool call was cancelled before receiving a response".to_string()), duration: started_at.elapsed(), }), }; session.send_event(turn_context, response_event).await; response }