chenchenghao/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

[model-gateway] move all responses api event from oai to proto (#14446)

Browse Source 
Co-authored-by: key4ng <rukeyang@gmail.com>
This commit is contained in:
Simo Lin
2025-12-04 13:11:03 -08:00
committed by GitHub
parent 0e6441b434
commit eb85fa6daf
6 changed files with 250 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

175
sgl-router/src/protocols/event_types.rs Normal file
View File

@@ -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
}

1
sgl-router/src/protocols/mod.rs
View File

@@ -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;

52
sgl-router/src/routers/openai/mcp.rs
View File

@@ -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<mcp::McpManager>, 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())

13
sgl-router/src/routers/openai/responses.rs
View File

@@ -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;

99
sgl-router/src/routers/openai/streaming.rs
View File

@@ -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<usize> = 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::<Value>(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 {

38
sgl-router/src/routers/openai/utils.rs
View File

@@ -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
// ============================================================================