From eb85fa6daf4a3922823b74ca024d788792fd2fb4 Mon Sep 17 00:00:00 2001 From: Simo Lin Date: Thu, 4 Dec 2025 13:11:03 -0800 Subject: [PATCH] [model-gateway] move all responses api event from oai to proto (#14446) Co-authored-by: key4ng --- sgl-router/src/protocols/event_types.rs | 175 +++++++++++++++++++++ sgl-router/src/protocols/mod.rs | 1 + sgl-router/src/routers/openai/mcp.rs | 52 +++--- sgl-router/src/routers/openai/responses.rs | 13 +- sgl-router/src/routers/openai/streaming.rs | 99 ++++++------ sgl-router/src/routers/openai/utils.rs | 38 +---- 6 files changed, 250 insertions(+), 128 deletions(-) create mode 100644 sgl-router/src/protocols/event_types.rs diff --git a/sgl-router/src/protocols/event_types.rs b/sgl-router/src/protocols/event_types.rs new file mode 100644 index 000000000..00c7ffd58 --- /dev/null +++ b/sgl-router/src/protocols/event_types.rs @@ -0,0 +1,175 @@ +use std::fmt; + +/// Response lifecycle events +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum ResponseEvent { + Created, + InProgress, + Completed, +} + +impl ResponseEvent { + pub const CREATED: &'static str = "response.created"; + pub const IN_PROGRESS: &'static str = "response.in_progress"; + pub const COMPLETED: &'static str = "response.completed"; + + pub const fn as_str(&self) -> &'static str { + match self { + Self::Created => Self::CREATED, + Self::InProgress => Self::IN_PROGRESS, + Self::Completed => Self::COMPLETED, + } + } +} + +impl fmt::Display for ResponseEvent { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +/// Output item events for streaming +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum OutputItemEvent { + Added, + Done, + Delta, +} + +impl OutputItemEvent { + pub const ADDED: &'static str = "response.output_item.added"; + pub const DONE: &'static str = "response.output_item.done"; + pub const DELTA: &'static str = "response.output_item.delta"; + + pub const fn as_str(&self) -> &'static str { + match self { + Self::Added => Self::ADDED, + Self::Done => Self::DONE, + Self::Delta => Self::DELTA, + } + } +} + +impl fmt::Display for OutputItemEvent { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +/// Function call argument streaming events +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum FunctionCallEvent { + ArgumentsDelta, + ArgumentsDone, +} + +impl FunctionCallEvent { + pub const ARGUMENTS_DELTA: &'static str = "response.function_call_arguments.delta"; + pub const ARGUMENTS_DONE: &'static str = "response.function_call_arguments.done"; + + pub const fn as_str(&self) -> &'static str { + match self { + Self::ArgumentsDelta => Self::ARGUMENTS_DELTA, + Self::ArgumentsDone => Self::ARGUMENTS_DONE, + } + } +} + +impl fmt::Display for FunctionCallEvent { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +// ============================================================================ +// MCP Events +// ============================================================================ + +/// MCP (Model Context Protocol) call events +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum McpEvent { + CallArgumentsDelta, + CallArgumentsDone, + CallInProgress, + CallCompleted, + ListToolsInProgress, + ListToolsCompleted, +} + +impl McpEvent { + pub const CALL_ARGUMENTS_DELTA: &'static str = "response.mcp_call_arguments.delta"; + pub const CALL_ARGUMENTS_DONE: &'static str = "response.mcp_call_arguments.done"; + pub const CALL_IN_PROGRESS: &'static str = "response.mcp_call.in_progress"; + pub const CALL_COMPLETED: &'static str = "response.mcp_call.completed"; + pub const LIST_TOOLS_IN_PROGRESS: &'static str = "response.mcp_list_tools.in_progress"; + pub const LIST_TOOLS_COMPLETED: &'static str = "response.mcp_list_tools.completed"; + + pub const fn as_str(&self) -> &'static str { + match self { + Self::CallArgumentsDelta => Self::CALL_ARGUMENTS_DELTA, + Self::CallArgumentsDone => Self::CALL_ARGUMENTS_DONE, + Self::CallInProgress => Self::CALL_IN_PROGRESS, + Self::CallCompleted => Self::CALL_COMPLETED, + Self::ListToolsInProgress => Self::LIST_TOOLS_IN_PROGRESS, + Self::ListToolsCompleted => Self::LIST_TOOLS_COMPLETED, + } + } +} + +impl fmt::Display for McpEvent { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +/// Item type discriminators used in output items +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum ItemType { + FunctionCall, + FunctionToolCall, + McpCall, + Function, + McpListTools, +} + +impl ItemType { + pub const FUNCTION_CALL: &'static str = "function_call"; + pub const FUNCTION_TOOL_CALL: &'static str = "function_tool_call"; + pub const MCP_CALL: &'static str = "mcp_call"; + pub const FUNCTION: &'static str = "function"; + pub const MCP_LIST_TOOLS: &'static str = "mcp_list_tools"; + + pub const fn as_str(&self) -> &'static str { + match self { + Self::FunctionCall => Self::FUNCTION_CALL, + Self::FunctionToolCall => Self::FUNCTION_TOOL_CALL, + Self::McpCall => Self::MCP_CALL, + Self::Function => Self::FUNCTION, + Self::McpListTools => Self::MCP_LIST_TOOLS, + } + } + + /// Check if this is a function call variant (FunctionCall or FunctionToolCall) + pub const fn is_function_call(&self) -> bool { + matches!(self, Self::FunctionCall | Self::FunctionToolCall) + } +} + +impl fmt::Display for ItemType { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +/// Check if an event type string matches any response lifecycle event +pub fn is_response_event(event_type: &str) -> bool { + matches!( + event_type, + ResponseEvent::CREATED | ResponseEvent::IN_PROGRESS | ResponseEvent::COMPLETED + ) +} + +/// Check if an item type string is a function call variant +pub fn is_function_call_type(item_type: &str) -> bool { + item_type == ItemType::FUNCTION_CALL || item_type == ItemType::FUNCTION_TOOL_CALL +} diff --git a/sgl-router/src/protocols/mod.rs b/sgl-router/src/protocols/mod.rs index 813ff48df..7b3b4b695 100644 --- a/sgl-router/src/protocols/mod.rs +++ b/sgl-router/src/protocols/mod.rs @@ -7,6 +7,7 @@ pub mod classify; pub mod common; pub mod completion; pub mod embedding; +pub mod event_types; pub mod generate; pub mod rerank; pub mod responses; diff --git a/sgl-router/src/routers/openai/mcp.rs b/sgl-router/src/routers/openai/mcp.rs index 8087bffa1..6c2dd6b07 100644 --- a/sgl-router/src/routers/openai/mcp.rs +++ b/sgl-router/src/routers/openai/mcp.rs @@ -16,11 +16,11 @@ use serde_json::{json, to_value, Value}; use tokio::sync::mpsc; use tracing::{debug, info, warn}; -use super::utils::event_types; use crate::{ mcp, - protocols::responses::{ - generate_id, ResponseInput, ResponseTool, ResponseToolType, ResponsesRequest, + protocols::{ + event_types::{is_function_call_type, ItemType, McpEvent, OutputItemEvent}, + responses::{generate_id, ResponseInput, ResponseTool, ResponseToolType, ResponsesRequest}, }, routers::header_utils::apply_request_headers, }; @@ -76,7 +76,7 @@ impl ToolLoopState { ) { // Add function_call item to history let func_item = json!({ - "type": event_types::ITEM_TYPE_FUNCTION_CALL, + "type": ItemType::FUNCTION_CALL, "call_id": call_id, "name": tool_name, "arguments": args_json_str @@ -285,7 +285,7 @@ pub(super) fn prepare_mcp_payload_for_streaming( arr.retain(|item| { item.get("type") .and_then(|v| v.as_str()) - .map(|s| s == event_types::ITEM_TYPE_FUNCTION) + .map(|s| s == ItemType::FUNCTION) .unwrap_or(false) }); } @@ -297,7 +297,7 @@ pub(super) fn prepare_mcp_payload_for_streaming( for t in tools { let parameters = Value::Object((*t.input_schema).clone()); let tool = serde_json::json!({ - "type": event_types::ITEM_TYPE_FUNCTION, + "type": ItemType::FUNCTION, "name": t.name, "description": t.description, "parameters": parameters @@ -399,7 +399,7 @@ pub(super) fn send_mcp_list_tools_events( // Event 1: response.output_item.added with empty tools let event1_payload = json!({ - "type": event_types::OUTPUT_ITEM_ADDED, + "type": OutputItemEvent::ADDED, "sequence_number": *sequence_number, "output_index": output_index, "item": tools_item_empty @@ -407,7 +407,7 @@ pub(super) fn send_mcp_list_tools_events( *sequence_number += 1; let event1 = format!( "event: {}\ndata: {}\n\n", - event_types::OUTPUT_ITEM_ADDED, + OutputItemEvent::ADDED, event1_payload ); if tx.send(Ok(Bytes::from(event1))).is_err() { @@ -416,7 +416,7 @@ pub(super) fn send_mcp_list_tools_events( // Event 2: response.mcp_list_tools.in_progress let event2_payload = json!({ - "type": event_types::MCP_LIST_TOOLS_IN_PROGRESS, + "type": McpEvent::LIST_TOOLS_IN_PROGRESS, "sequence_number": *sequence_number, "output_index": output_index, "item_id": item_id @@ -424,7 +424,7 @@ pub(super) fn send_mcp_list_tools_events( *sequence_number += 1; let event2 = format!( "event: {}\ndata: {}\n\n", - event_types::MCP_LIST_TOOLS_IN_PROGRESS, + McpEvent::LIST_TOOLS_IN_PROGRESS, event2_payload ); if tx.send(Ok(Bytes::from(event2))).is_err() { @@ -433,7 +433,7 @@ pub(super) fn send_mcp_list_tools_events( // Event 3: response.mcp_list_tools.completed let event3_payload = json!({ - "type": event_types::MCP_LIST_TOOLS_COMPLETED, + "type": McpEvent::LIST_TOOLS_COMPLETED, "sequence_number": *sequence_number, "output_index": output_index, "item_id": item_id @@ -441,7 +441,7 @@ pub(super) fn send_mcp_list_tools_events( *sequence_number += 1; let event3 = format!( "event: {}\ndata: {}\n\n", - event_types::MCP_LIST_TOOLS_COMPLETED, + McpEvent::LIST_TOOLS_COMPLETED, event3_payload ); if tx.send(Ok(Bytes::from(event3))).is_err() { @@ -450,7 +450,7 @@ pub(super) fn send_mcp_list_tools_events( // Event 4: response.output_item.done with full tools list let event4_payload = json!({ - "type": event_types::OUTPUT_ITEM_DONE, + "type": OutputItemEvent::DONE, "sequence_number": *sequence_number, "output_index": output_index, "item": tools_item_full @@ -458,7 +458,7 @@ pub(super) fn send_mcp_list_tools_events( *sequence_number += 1; let event4 = format!( "event: {}\ndata: {}\n\n", - event_types::OUTPUT_ITEM_DONE, + OutputItemEvent::DONE, event4_payload ); tx.send(Ok(Bytes::from(event4))).is_ok() @@ -495,7 +495,7 @@ pub(super) fn send_mcp_call_completion_events_with_error( // Event 1: response.mcp_call.completed let completed_payload = json!({ - "type": event_types::MCP_CALL_COMPLETED, + "type": McpEvent::CALL_COMPLETED, "sequence_number": *sequence_number, "output_index": effective_output_index, "item_id": item_id @@ -504,7 +504,7 @@ pub(super) fn send_mcp_call_completion_events_with_error( let completed_event = format!( "event: {}\ndata: {}\n\n", - event_types::MCP_CALL_COMPLETED, + McpEvent::CALL_COMPLETED, completed_payload ); if tx.send(Ok(Bytes::from(completed_event))).is_err() { @@ -513,7 +513,7 @@ pub(super) fn send_mcp_call_completion_events_with_error( // Event 2: response.output_item.done (with completed mcp_call) let done_payload = json!({ - "type": event_types::OUTPUT_ITEM_DONE, + "type": OutputItemEvent::DONE, "sequence_number": *sequence_number, "output_index": effective_output_index, "item": mcp_call_item @@ -522,7 +522,7 @@ pub(super) fn send_mcp_call_completion_events_with_error( let done_event = format!( "event: {}\ndata: {}\n\n", - event_types::OUTPUT_ITEM_DONE, + OutputItemEvent::DONE, done_payload ); tx.send(Ok(Bytes::from(done_event))).is_ok() @@ -541,7 +541,7 @@ pub(super) fn inject_mcp_metadata_streaming( ) { if let Some(output_array) = response.get_mut("output").and_then(|v| v.as_array_mut()) { output_array.retain(|item| { - item.get("type").and_then(|t| t.as_str()) != Some(event_types::ITEM_TYPE_MCP_LIST_TOOLS) + item.get("type").and_then(|t| t.as_str()) != Some(ItemType::MCP_LIST_TOOLS) }); let list_tools_item = build_mcp_list_tools_item(mcp, server_label); @@ -782,9 +782,7 @@ pub(super) fn build_incomplete_response( let mut mcp_call_items = Vec::new(); for item in output_array.iter() { let item_type = item.get("type").and_then(|t| t.as_str()); - if item_type == Some(event_types::ITEM_TYPE_FUNCTION_TOOL_CALL) - || item_type == Some(event_types::ITEM_TYPE_FUNCTION_CALL) - { + if item_type.is_some_and(is_function_call_type) { let tool_name = item.get("name").and_then(|v| v.as_str()).unwrap_or(""); let args = item .get("arguments") @@ -870,7 +868,7 @@ pub(super) fn build_mcp_list_tools_item(mcp: &Arc, server_label json!({ "id": generate_id("mcpl"), - "type": event_types::ITEM_TYPE_MCP_LIST_TOOLS, + "type": ItemType::MCP_LIST_TOOLS, "server_label": server_label, "tools": tools_json }) @@ -887,7 +885,7 @@ pub(super) fn build_mcp_call_item( ) -> Value { json!({ "id": generate_id("mcp"), - "type": event_types::ITEM_TYPE_MCP_CALL, + "type": ItemType::MCP_CALL, "status": if success { "completed" } else { "failed" }, "approval_request_id": Value::Null, "arguments": arguments, @@ -906,7 +904,7 @@ pub(super) fn build_executed_mcp_call_items( let mut mcp_call_items = Vec::new(); for item in conversation_history { - if item.get("type").and_then(|t| t.as_str()) == Some(event_types::ITEM_TYPE_FUNCTION_CALL) { + if item.get("type").and_then(|t| t.as_str()) == Some(ItemType::FUNCTION_CALL) { let call_id = item.get("call_id").and_then(|v| v.as_str()).unwrap_or(""); let tool_name = item.get("name").and_then(|v| v.as_str()).unwrap_or(""); let args = item @@ -958,9 +956,7 @@ pub(super) fn extract_function_call(resp: &Value) -> Option<(String, String, Str for item in output { let obj = item.as_object()?; let t = obj.get("type")?.as_str()?; - if t == event_types::ITEM_TYPE_FUNCTION_TOOL_CALL - || t == event_types::ITEM_TYPE_FUNCTION_CALL - { + if is_function_call_type(t) { let call_id = obj .get("call_id") .and_then(|v| v.as_str()) diff --git a/sgl-router/src/routers/openai/responses.rs b/sgl-router/src/routers/openai/responses.rs index b9d6a925c..c48134920 100644 --- a/sgl-router/src/routers/openai/responses.rs +++ b/sgl-router/src/routers/openai/responses.rs @@ -5,10 +5,12 @@ use std::collections::HashMap; use serde_json::{json, Value}; use tracing::warn; -use super::utils::event_types; use crate::{ data_connector::{ResponseId, StoredResponse}, - protocols::responses::{ResponseToolType, ResponsesRequest}, + protocols::{ + event_types::is_response_event, + responses::{ResponseToolType, ResponsesRequest}, + }, }; // ============================================================================ @@ -201,12 +203,7 @@ pub(super) fn rewrite_streaming_block( .and_then(|v| v.as_str()) .unwrap_or_default(); - let should_patch = matches!( - event_type, - event_types::RESPONSE_CREATED - | event_types::RESPONSE_IN_PROGRESS - | event_types::RESPONSE_COMPLETED - ); + let should_patch = is_response_event(event_type); if !should_patch { return None; diff --git a/sgl-router/src/routers/openai/streaming.rs b/sgl-router/src/routers/openai/streaming.rs index a1608931a..5b5c547ec 100644 --- a/sgl-router/src/routers/openai/streaming.rs +++ b/sgl-router/src/routers/openai/streaming.rs @@ -31,10 +31,16 @@ use super::{ send_mcp_list_tools_events, McpLoopConfig, ToolLoopState, }, responses::{mask_tools_as_mcp, patch_streaming_response_json, rewrite_streaming_block}, - utils::{event_types, FunctionCallInProgress, OutputIndexMapper, StreamAction}, + utils::{FunctionCallInProgress, OutputIndexMapper, StreamAction}, }; use crate::{ - protocols::responses::{ResponseToolType, ResponsesRequest}, + protocols::{ + event_types::{ + is_function_call_type, is_response_event, FunctionCallEvent, ItemType, McpEvent, + OutputItemEvent, ResponseEvent, + }, + responses::{ResponseToolType, ResponsesRequest}, + }, routers::header_utils::{apply_request_headers, preserve_response_headers}, }; @@ -161,19 +167,19 @@ impl StreamingResponseAccumulator { .unwrap_or_default(); match event_type.as_str() { - event_types::RESPONSE_CREATED => { + ResponseEvent::CREATED => { if self.initial_response.is_none() { if let Some(response) = parsed.get("response") { self.initial_response = Some(response.clone()); } } } - event_types::RESPONSE_COMPLETED => { + ResponseEvent::COMPLETED => { if let Some(response) = parsed.get("response") { self.completed_response = Some(response.clone()); } } - event_types::OUTPUT_ITEM_DONE => { + OutputItemEvent::DONE => { if let (Some(index), Some(item)) = ( parsed .get("output_index") @@ -292,7 +298,7 @@ impl StreamingToolHandler { .unwrap_or_default(); match event_type.as_str() { - event_types::RESPONSE_CREATED => { + ResponseEvent::CREATED => { if self.original_response_id.is_none() { if let Some(response_obj) = parsed.get("response").and_then(|v| v.as_object()) { if let Some(id) = response_obj.get("id").and_then(|v| v.as_str()) { @@ -302,8 +308,8 @@ impl StreamingToolHandler { } StreamAction::Forward } - event_types::RESPONSE_COMPLETED => StreamAction::Forward, - event_types::OUTPUT_ITEM_ADDED => { + ResponseEvent::COMPLETED => StreamAction::Forward, + OutputItemEvent::ADDED => { if let Some(idx) = parsed.get("output_index").and_then(|v| v.as_u64()) { self.ensure_output_index(idx as usize); } @@ -311,9 +317,7 @@ impl StreamingToolHandler { // Check if this is a function_call item being added if let Some(item) = parsed.get("item") { if let Some(item_type) = item.get("type").and_then(|v| v.as_str()) { - if item_type == event_types::ITEM_TYPE_FUNCTION_CALL - || item_type == event_types::ITEM_TYPE_FUNCTION_TOOL_CALL - { + if is_function_call_type(item_type) { match parsed.get("output_index").and_then(|v| v.as_u64()) { Some(idx) => { let output_index = idx as usize; @@ -343,7 +347,7 @@ impl StreamingToolHandler { } StreamAction::Forward } - event_types::FUNCTION_CALL_ARGUMENTS_DELTA => { + FunctionCallEvent::ARGUMENTS_DELTA => { // Accumulate arguments for the function call if let Some(output_index) = parsed .get("output_index") @@ -371,7 +375,7 @@ impl StreamingToolHandler { } StreamAction::Forward } - event_types::FUNCTION_CALL_ARGUMENTS_DONE => { + FunctionCallEvent::ARGUMENTS_DONE => { // Function call arguments complete - check if ready to execute if let Some(output_index) = parsed .get("output_index") @@ -396,8 +400,8 @@ impl StreamingToolHandler { StreamAction::Forward } } - event_types::OUTPUT_ITEM_DELTA => self.process_output_delta(&parsed), - event_types::OUTPUT_ITEM_DONE => { + OutputItemEvent::DELTA => self.process_output_delta(&parsed), + OutputItemEvent::DONE => { // Check if we have complete function calls ready to execute if let Some(output_index) = parsed .get("output_index") @@ -435,9 +439,7 @@ impl StreamingToolHandler { // Check if this is a function call delta let item_type = delta.get("type").and_then(|v| v.as_str()); - if item_type == Some(event_types::ITEM_TYPE_FUNCTION_TOOL_CALL) - || item_type == Some(event_types::ITEM_TYPE_FUNCTION_CALL) - { + if item_type.is_some_and(is_function_call_type) { self.in_function_call = true; // Get or create function call for this output index @@ -561,12 +563,7 @@ pub(super) fn apply_event_transformations_inplace( .map(|s| s.to_string()) .unwrap_or_default(); - let should_patch = matches!( - event_type.as_str(), - event_types::RESPONSE_CREATED - | event_types::RESPONSE_IN_PROGRESS - | event_types::RESPONSE_COMPLETED - ); + let should_patch = is_response_event(event_type.as_str()); if should_patch { if let Some(response_obj) = parsed_data @@ -622,13 +619,11 @@ pub(super) fn apply_event_transformations_inplace( // 2. Apply transform_streaming_event logic (function_call → mcp_call) match event_type.as_str() { - event_types::OUTPUT_ITEM_ADDED | event_types::OUTPUT_ITEM_DONE => { + OutputItemEvent::ADDED | OutputItemEvent::DONE => { if let Some(item) = parsed_data.get_mut("item") { if let Some(item_type) = item.get("type").and_then(|v| v.as_str()) { - if item_type == event_types::ITEM_TYPE_FUNCTION_CALL - || item_type == event_types::ITEM_TYPE_FUNCTION_TOOL_CALL - { - item["type"] = json!(event_types::ITEM_TYPE_MCP_CALL); + if is_function_call_type(item_type) { + item["type"] = json!(ItemType::MCP_CALL); item["server_label"] = json!(ctx.server_label); // Transform ID from fc_* to mcp_* @@ -644,8 +639,8 @@ pub(super) fn apply_event_transformations_inplace( } } } - event_types::FUNCTION_CALL_ARGUMENTS_DONE => { - parsed_data["type"] = json!(event_types::MCP_CALL_ARGUMENTS_DONE); + FunctionCallEvent::ARGUMENTS_DONE => { + parsed_data["type"] = json!(McpEvent::CALL_ARGUMENTS_DONE); // Transform item_id from fc_* to mcp_* if let Some(item_id) = parsed_data.get("item_id").and_then(|v| v.as_str()) { @@ -691,7 +686,7 @@ pub(super) fn forward_streaming_event( sequence_number: &mut u64, ) -> bool { // Skip individual function_call_arguments.delta events - we'll send them as one - if event_name == Some(event_types::FUNCTION_CALL_ARGUMENTS_DELTA) { + if event_name == Some(FunctionCallEvent::ARGUMENTS_DELTA) { return true; } @@ -709,14 +704,14 @@ pub(super) fn forward_streaming_event( .or_else(|| parsed_data.get("type").and_then(|v| v.as_str())) .unwrap_or(""); - if event_type == event_types::RESPONSE_COMPLETED { + if event_type == ResponseEvent::COMPLETED { return true; } // Check if this is function_call_arguments.done - need to send buffered args first let mut mapped_output_index: Option = None; - if event_name == Some(event_types::FUNCTION_CALL_ARGUMENTS_DONE) { + if event_name == Some(FunctionCallEvent::ARGUMENTS_DONE) { if let Some(output_index) = parsed_data .get("output_index") .and_then(|v| v.as_u64()) @@ -754,7 +749,7 @@ pub(super) fn forward_streaming_event( // Emit a synthetic MCP arguments delta event before the done event let mut delta_event = json!({ - "type": event_types::MCP_CALL_ARGUMENTS_DELTA, + "type": McpEvent::CALL_ARGUMENTS_DELTA, "sequence_number": *sequence_number, "output_index": assigned_index, "item_id": mcp_item_id, @@ -776,7 +771,7 @@ pub(super) fn forward_streaming_event( let delta_block = format!( "event: {}\ndata: {}\n\n", - event_types::MCP_CALL_ARGUMENTS_DELTA, + McpEvent::CALL_ARGUMENTS_DELTA, delta_event ); if tx.send(Ok(Bytes::from(delta_block))).is_err() { @@ -835,16 +830,10 @@ pub(super) fn forward_streaming_event( let mut final_block = String::new(); if let Some(evt) = event_name { // Update event name for function_call_arguments events - if evt == event_types::FUNCTION_CALL_ARGUMENTS_DELTA { - final_block.push_str(&format!( - "event: {}\n", - event_types::MCP_CALL_ARGUMENTS_DELTA - )); - } else if evt == event_types::FUNCTION_CALL_ARGUMENTS_DONE { - final_block.push_str(&format!( - "event: {}\n", - event_types::MCP_CALL_ARGUMENTS_DONE - )); + if evt == FunctionCallEvent::ARGUMENTS_DELTA { + final_block.push_str(&format!("event: {}\n", McpEvent::CALL_ARGUMENTS_DELTA)); + } else if evt == FunctionCallEvent::ARGUMENTS_DONE { + final_block.push_str(&format!("event: {}\n", McpEvent::CALL_ARGUMENTS_DONE)); } else { final_block.push_str(&format!("event: {}\n", evt)); } @@ -857,16 +846,16 @@ pub(super) fn forward_streaming_event( } // After sending output_item.added for mcp_call, inject mcp_call.in_progress event - if event_name == Some(event_types::OUTPUT_ITEM_ADDED) { + if event_name == Some(OutputItemEvent::ADDED) { if let Some(item) = parsed_data.get("item") { - if item.get("type").and_then(|v| v.as_str()) == Some(event_types::ITEM_TYPE_MCP_CALL) { + if item.get("type").and_then(|v| v.as_str()) == Some(ItemType::MCP_CALL) { // Already transformed to mcp_call if let (Some(item_id), Some(output_index)) = ( item.get("id").and_then(|v| v.as_str()), parsed_data.get("output_index").and_then(|v| v.as_u64()), ) { let in_progress_event = json!({ - "type": event_types::MCP_CALL_IN_PROGRESS, + "type": McpEvent::CALL_IN_PROGRESS, "sequence_number": *sequence_number, "output_index": output_index, "item_id": item_id @@ -874,7 +863,7 @@ pub(super) fn forward_streaming_event( *sequence_number += 1; let in_progress_block = format!( "event: {}\ndata: {}\n\n", - event_types::MCP_CALL_IN_PROGRESS, + McpEvent::CALL_IN_PROGRESS, in_progress_event ); if tx.send(Ok(Bytes::from(in_progress_block))).is_err() { @@ -928,7 +917,7 @@ pub(super) fn send_final_response_event( } let completed_payload = json!({ - "type": event_types::RESPONSE_COMPLETED, + "type": ResponseEvent::COMPLETED, "sequence_number": *sequence_number, "response": final_response }); @@ -936,7 +925,7 @@ pub(super) fn send_final_response_event( let completed_event = format!( "event: {}\ndata: {}\n\n", - event_types::RESPONSE_COMPLETED, + ResponseEvent::COMPLETED, completed_payload ); tx.send(Ok(Bytes::from(completed_event))).is_ok() @@ -1241,8 +1230,8 @@ pub(super) async fn handle_streaming_with_tool_interception( { matches!( parsed.get("type").and_then(|v| v.as_str()), - Some(event_types::RESPONSE_CREATED) - | Some(event_types::RESPONSE_IN_PROGRESS) + Some(ResponseEvent::CREATED) + | Some(ResponseEvent::IN_PROGRESS) ) } else { false @@ -1273,7 +1262,7 @@ pub(super) async fn handle_streaming_with_tool_interception( serde_json::from_str::(data.as_ref()) { if parsed.get("type").and_then(|v| v.as_str()) - == Some(event_types::RESPONSE_IN_PROGRESS) + == Some(ResponseEvent::IN_PROGRESS) { seen_in_progress = true; if !mcp_list_tools_sent { diff --git a/sgl-router/src/routers/openai/utils.rs b/sgl-router/src/routers/openai/utils.rs index 0a3f79329..b9262c772 100644 --- a/sgl-router/src/routers/openai/utils.rs +++ b/sgl-router/src/routers/openai/utils.rs @@ -1,43 +1,7 @@ -//! Utility types and constants for OpenAI router +//! Utility types for OpenAI router use std::collections::HashMap; -// ============================================================================ -// SSE Event Type Constants -// ============================================================================ - -/// SSE event type constants - single source of truth for event type strings -pub(crate) mod event_types { - // Response lifecycle events - pub const RESPONSE_CREATED: &str = "response.created"; - pub const RESPONSE_IN_PROGRESS: &str = "response.in_progress"; - pub const RESPONSE_COMPLETED: &str = "response.completed"; - - // Output item events - pub const OUTPUT_ITEM_ADDED: &str = "response.output_item.added"; - pub const OUTPUT_ITEM_DONE: &str = "response.output_item.done"; - pub const OUTPUT_ITEM_DELTA: &str = "response.output_item.delta"; - - // Function call events - pub const FUNCTION_CALL_ARGUMENTS_DELTA: &str = "response.function_call_arguments.delta"; - pub const FUNCTION_CALL_ARGUMENTS_DONE: &str = "response.function_call_arguments.done"; - - // MCP call events - pub const MCP_CALL_ARGUMENTS_DELTA: &str = "response.mcp_call_arguments.delta"; - pub const MCP_CALL_ARGUMENTS_DONE: &str = "response.mcp_call_arguments.done"; - pub const MCP_CALL_IN_PROGRESS: &str = "response.mcp_call.in_progress"; - pub const MCP_CALL_COMPLETED: &str = "response.mcp_call.completed"; - pub const MCP_LIST_TOOLS_IN_PROGRESS: &str = "response.mcp_list_tools.in_progress"; - pub const MCP_LIST_TOOLS_COMPLETED: &str = "response.mcp_list_tools.completed"; - - // Item types - pub const ITEM_TYPE_FUNCTION_CALL: &str = "function_call"; - pub const ITEM_TYPE_FUNCTION_TOOL_CALL: &str = "function_tool_call"; - pub const ITEM_TYPE_MCP_CALL: &str = "mcp_call"; - pub const ITEM_TYPE_FUNCTION: &str = "function"; - pub const ITEM_TYPE_MCP_LIST_TOOLS: &str = "mcp_list_tools"; -} - // ============================================================================ // Stream Action Enum // ============================================================================