From c1006fd8a14094197f088fc70bb4cdb738668005 Mon Sep 17 00:00:00 2001 From: Simo Lin Date: Thu, 4 Dec 2025 14:30:19 -0800 Subject: [PATCH] [model-gateway] grpc to leverage event type (#14450) Co-authored-by: Chang Su --- sgl-router/src/protocols/event_types.rs | 53 +++++++++++++++++++ .../grpc/common/responses/streaming.rs | 40 +++++++------- 2 files changed, 75 insertions(+), 18 deletions(-) diff --git a/sgl-router/src/protocols/event_types.rs b/sgl-router/src/protocols/event_types.rs index 00c7ffd58..c36dc5d45 100644 --- a/sgl-router/src/protocols/event_types.rs +++ b/sgl-router/src/protocols/event_types.rs @@ -81,6 +81,56 @@ impl fmt::Display for FunctionCallEvent { } } +/// Content part streaming events +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum ContentPartEvent { + Added, + Done, +} + +impl ContentPartEvent { + pub const ADDED: &'static str = "response.content_part.added"; + pub const DONE: &'static str = "response.content_part.done"; + + pub const fn as_str(&self) -> &'static str { + match self { + Self::Added => Self::ADDED, + Self::Done => Self::DONE, + } + } +} + +impl fmt::Display for ContentPartEvent { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +/// Output text streaming events +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum OutputTextEvent { + Delta, + Done, +} + +impl OutputTextEvent { + pub const DELTA: &'static str = "response.output_text.delta"; + pub const DONE: &'static str = "response.output_text.done"; + + pub const fn as_str(&self) -> &'static str { + match self { + Self::Delta => Self::DELTA, + Self::Done => Self::DONE, + } + } +} + +impl fmt::Display for OutputTextEvent { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + // ============================================================================ // MCP Events // ============================================================================ @@ -92,6 +142,7 @@ pub enum McpEvent { CallArgumentsDone, CallInProgress, CallCompleted, + CallFailed, ListToolsInProgress, ListToolsCompleted, } @@ -101,6 +152,7 @@ impl McpEvent { 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 CALL_FAILED: &'static str = "response.mcp_call.failed"; 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"; @@ -110,6 +162,7 @@ impl McpEvent { Self::CallArgumentsDone => Self::CALL_ARGUMENTS_DONE, Self::CallInProgress => Self::CALL_IN_PROGRESS, Self::CallCompleted => Self::CALL_COMPLETED, + Self::CallFailed => Self::CALL_FAILED, Self::ListToolsInProgress => Self::LIST_TOOLS_IN_PROGRESS, Self::ListToolsCompleted => Self::LIST_TOOLS_COMPLETED, } diff --git a/sgl-router/src/routers/grpc/common/responses/streaming.rs b/sgl-router/src/routers/grpc/common/responses/streaming.rs index 47bea0d91..cad8c731f 100644 --- a/sgl-router/src/routers/grpc/common/responses/streaming.rs +++ b/sgl-router/src/routers/grpc/common/responses/streaming.rs @@ -14,6 +14,10 @@ use crate::{ protocols::{ chat::ChatCompletionStreamResponse, common::{Usage, UsageInfo}, + event_types::{ + ContentPartEvent, FunctionCallEvent, McpEvent, OutputItemEvent, OutputTextEvent, + ResponseEvent, + }, responses::{ ResponseOutputItem, ResponseStatus, ResponsesRequest, ResponsesResponse, ResponsesUsage, }, @@ -159,7 +163,7 @@ impl ResponseStreamEventEmitter { pub fn emit_created(&mut self) -> serde_json::Value { self.has_emitted_created = true; json!({ - "type": "response.created", + "type": ResponseEvent::CREATED, "sequence_number": self.next_sequence(), "response": { "id": self.response_id, @@ -175,7 +179,7 @@ impl ResponseStreamEventEmitter { pub fn emit_in_progress(&mut self) -> serde_json::Value { self.has_emitted_in_progress = true; json!({ - "type": "response.in_progress", + "type": ResponseEvent::IN_PROGRESS, "sequence_number": self.next_sequence(), "response": { "id": self.response_id, @@ -193,7 +197,7 @@ impl ResponseStreamEventEmitter { ) -> serde_json::Value { self.has_emitted_content_part_added = true; json!({ - "type": "response.content_part.added", + "type": ContentPartEvent::ADDED, "sequence_number": self.next_sequence(), "output_index": output_index, "item_id": item_id, @@ -214,7 +218,7 @@ impl ResponseStreamEventEmitter { ) -> serde_json::Value { self.accumulated_text.push_str(delta); json!({ - "type": "response.output_text.delta", + "type": OutputTextEvent::DELTA, "sequence_number": self.next_sequence(), "output_index": output_index, "item_id": item_id, @@ -230,7 +234,7 @@ impl ResponseStreamEventEmitter { content_index: usize, ) -> serde_json::Value { json!({ - "type": "response.output_text.done", + "type": OutputTextEvent::DONE, "sequence_number": self.next_sequence(), "output_index": output_index, "item_id": item_id, @@ -246,7 +250,7 @@ impl ResponseStreamEventEmitter { content_index: usize, ) -> serde_json::Value { json!({ - "type": "response.content_part.done", + "type": ContentPartEvent::DONE, "sequence_number": self.next_sequence(), "output_index": output_index, "item_id": item_id, @@ -336,7 +340,7 @@ impl ResponseStreamEventEmitter { } json!({ - "type": "response.completed", + "type": ResponseEvent::COMPLETED, "sequence_number": self.next_sequence(), "response": response_obj }) @@ -359,7 +363,7 @@ impl ResponseStreamEventEmitter { pub fn emit_mcp_list_tools_in_progress(&mut self, output_index: usize) -> serde_json::Value { json!({ - "type": "response.mcp_list_tools.in_progress", + "type": McpEvent::LIST_TOOLS_IN_PROGRESS, "sequence_number": self.next_sequence(), "output_index": output_index }) @@ -382,7 +386,7 @@ impl ResponseStreamEventEmitter { .collect(); json!({ - "type": "response.mcp_list_tools.completed", + "type": McpEvent::LIST_TOOLS_COMPLETED, "sequence_number": self.next_sequence(), "output_index": output_index, "tools": tool_items @@ -395,7 +399,7 @@ impl ResponseStreamEventEmitter { item_id: &str, ) -> serde_json::Value { json!({ - "type": "response.mcp_call.in_progress", + "type": McpEvent::CALL_IN_PROGRESS, "sequence_number": self.next_sequence(), "output_index": output_index, "item_id": item_id @@ -415,7 +419,7 @@ impl ResponseStreamEventEmitter { .push_str(delta); json!({ - "type": "response.mcp_call_arguments.delta", + "type": McpEvent::CALL_ARGUMENTS_DELTA, "sequence_number": self.next_sequence(), "output_index": output_index, "item_id": item_id, @@ -430,7 +434,7 @@ impl ResponseStreamEventEmitter { arguments: &str, ) -> serde_json::Value { json!({ - "type": "response.mcp_call_arguments.done", + "type": McpEvent::CALL_ARGUMENTS_DONE, "sequence_number": self.next_sequence(), "output_index": output_index, "item_id": item_id, @@ -444,7 +448,7 @@ impl ResponseStreamEventEmitter { item_id: &str, ) -> serde_json::Value { json!({ - "type": "response.mcp_call.completed", + "type": McpEvent::CALL_COMPLETED, "sequence_number": self.next_sequence(), "output_index": output_index, "item_id": item_id @@ -458,7 +462,7 @@ impl ResponseStreamEventEmitter { error: &str, ) -> serde_json::Value { json!({ - "type": "response.mcp_call.failed", + "type": McpEvent::CALL_FAILED, "sequence_number": self.next_sequence(), "output_index": output_index, "item_id": item_id, @@ -477,7 +481,7 @@ impl ResponseStreamEventEmitter { delta: &str, ) -> serde_json::Value { json!({ - "type": "response.function_call_arguments.delta", + "type": FunctionCallEvent::ARGUMENTS_DELTA, "sequence_number": self.next_sequence(), "output_index": output_index, "item_id": item_id, @@ -492,7 +496,7 @@ impl ResponseStreamEventEmitter { arguments: &str, ) -> serde_json::Value { json!({ - "type": "response.function_call_arguments.done", + "type": FunctionCallEvent::ARGUMENTS_DONE, "sequence_number": self.next_sequence(), "output_index": output_index, "item_id": item_id, @@ -511,7 +515,7 @@ impl ResponseStreamEventEmitter { item: &serde_json::Value, ) -> serde_json::Value { json!({ - "type": "response.output_item.added", + "type": OutputItemEvent::ADDED, "sequence_number": self.next_sequence(), "output_index": output_index, "item": item @@ -528,7 +532,7 @@ impl ResponseStreamEventEmitter { self.store_output_item_data(output_index, item.clone()); json!({ - "type": "response.output_item.done", + "type": OutputItemEvent::DONE, "sequence_number": self.next_sequence(), "output_index": output_index, "item": item