From 84c4320083111b116fff7f0e5a27cba0caebe183 Mon Sep 17 00:00:00 2001 From: Jarvis Date: Tue, 12 May 2026 09:37:39 +0800 Subject: [PATCH 1/2] fix(proxy): translate Anthropic tools/tool_choice to OpenAI shape in cross-provider dispatch (#236) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a client sends an Anthropic Messages API request with tools and tool_choice to a non-Anthropic upstream, the gateway now translates these fields from Anthropic shape to OpenAI shape before dispatching. Previously, tools ({name, description, input_schema}) and tool_choice ({type: "any"}) were passed through verbatim in ChatFormat.extra, which OpenAI-compatible upstreams silently ignored or rejected. Added: - translate_anthropic_tools_to_openai(): converts Anthropic tool defs to OpenAI {type:"function", function:{name, description, parameters}} - translate_anthropic_tool_choice_to_openai(): converts Anthropic tool_choice ({type:"any"} → "required", {type:"tool",name} → {type:"function",function:{name}}, etc.) - Called in cross_provider_dispatch() after parse_inbound_request() - Unit tests for both translators - E2E test for Anthropic client → OpenAI upstream tool round-trip Closes #236 --- crates/aisix-provider-anthropic/src/lib.rs | 5 +- crates/aisix-provider-anthropic/src/wire.rs | 244 +++++++++++++++++- crates/aisix-provider-openai/src/wire.rs | 71 ++++- crates/aisix-proxy/src/messages.rs | 18 +- ...anthropic-tools-cross-provider-e2e.test.ts | 217 ++++++++++++++++ 5 files changed, 540 insertions(+), 15 deletions(-) create mode 100644 tests/e2e/src/cases/anthropic-tools-cross-provider-e2e.test.ts diff --git a/crates/aisix-provider-anthropic/src/lib.rs b/crates/aisix-provider-anthropic/src/lib.rs index 6d6e92a9..123e1d83 100644 --- a/crates/aisix-provider-anthropic/src/lib.rs +++ b/crates/aisix-provider-anthropic/src/lib.rs @@ -29,6 +29,7 @@ pub use bridge::{AnthropicBridge, ANTHROPIC_DEFAULT_BASE, ANTHROPIC_VERSION}; /// - [`AnthropicSseEncoder`] re-encodes the bridge's `ChatChunk` /// stream as Anthropic typed SSE events. pub use wire::{ - chat_response_into_anthropic_json, parse_inbound_request, AnthropicInboundError, - AnthropicSseEncoder, AnthropicSseEvent, + chat_response_into_anthropic_json, parse_inbound_request, + translate_anthropic_tool_choice_to_openai, translate_anthropic_tools_to_openai, + AnthropicInboundError, AnthropicSseEncoder, AnthropicSseEvent, }; diff --git a/crates/aisix-provider-anthropic/src/wire.rs b/crates/aisix-provider-anthropic/src/wire.rs index 9a593db2..4bc116a0 100644 --- a/crates/aisix-provider-anthropic/src/wire.rs +++ b/crates/aisix-provider-anthropic/src/wire.rs @@ -294,6 +294,71 @@ pub(crate) fn translate_openai_tool_choice_to_anthropic( } } +/// Translate Anthropic-shape `tools` array into OpenAI's tools-spec shape. +/// +/// Anthropic OpenAI +/// {name, {type: "function", +/// description, function: {name, description, +/// input_schema} parameters}} +/// +/// Returns `None` when the input isn't an array or when no entries +/// translated — keeping the field absent from the outbound request. +pub fn translate_anthropic_tools_to_openai(tools: serde_json::Value) -> Option { + let arr = tools.as_array()?; + let translated: Vec = arr + .iter() + .filter_map(|t| { + let name = t.get("name")?.as_str()?; + let mut function = serde_json::Map::new(); + function.insert("name".into(), name.into()); + if let Some(desc) = t.get("description") { + function.insert("description".into(), desc.clone()); + } + if let Some(schema) = t.get("input_schema") { + function.insert("parameters".into(), schema.clone()); + } + Some(serde_json::json!({ + "type": "function", + "function": serde_json::Value::Object(function), + })) + }) + .collect(); + if translated.is_empty() { + None + } else { + Some(serde_json::Value::Array(translated)) + } +} + +/// Translate Anthropic-shape `tool_choice` to OpenAI's. +/// +/// Anthropic OpenAI +/// {"type":"auto"} → "auto" +/// {"type":"none"} → "none" (Anthropic doesn't officially +/// document this but clients may send it) +/// {"type":"any"} → "required" +/// {"type":"tool", "name":"X"} → {type:"function", function:{name:"X"}} +/// +/// Returns `None` for unrecognised shapes. +pub fn translate_anthropic_tool_choice_to_openai( + v: serde_json::Value, +) -> Option { + let obj = v.as_object()?; + let typ = obj.get("type").and_then(|t| t.as_str())?; + match typ { + "auto" | "none" => Some(serde_json::Value::String(typ.to_string())), + "any" => Some(serde_json::Value::String("required".to_string())), + "tool" => { + let name = obj.get("name").and_then(|n| n.as_str())?; + Some(serde_json::json!({ + "type": "function", + "function": {"name": name} + })) + } + _ => None, + } +} + /// Non-streaming response shape from `/v1/messages`. #[derive(Debug, Deserialize)] pub(crate) struct AnthropicResponse { @@ -719,12 +784,61 @@ pub fn chat_response_into_anthropic_json( FinishReason::ToolCalls => "tool_use", FinishReason::Other(_) => "end_turn", }; + + let mut content: Vec = Vec::new(); + + if !resp.message.content.is_empty() { + content.push(serde_json::json!({"type": "text", "text": resp.message.content})); + } + + // Translate OpenAI-shape tool_calls from message.extra into + // Anthropic tool_use content blocks so Anthropic clients see + // the tool invocations the model requested. + if let Some(tool_calls) = resp + .message + .extra + .get("tool_calls") + .and_then(|v| v.as_array()) + { + for tc in tool_calls { + let id = match tc.get("id").and_then(|v| v.as_str()) { + Some(s) if !s.is_empty() => s, + _ => continue, + }; + let name = match tc + .get("function") + .and_then(|f| f.get("name")) + .and_then(|n| n.as_str()) + { + Some(s) if !s.is_empty() => s, + _ => continue, + }; + let input = tc + .get("function") + .and_then(|f| f.get("arguments")) + .and_then(|a| a.as_str()) + .and_then(|s| serde_json::from_str::(s).ok()) + .filter(|v| v.is_object()) + .unwrap_or(serde_json::json!({})); + content.push(serde_json::json!({ + "type": "tool_use", + "id": id, + "name": name, + "input": input, + })); + } + } + + if content.is_empty() { + content.push(serde_json::json!({"type": "text", "text": ""})); + } + serde_json::json!({ "id": resp.id, "type": "message", "role": "assistant", "model": model_display_name, - "content": [{"type": "text", "text": resp.message.content}], + "content": content, "stop_reason": stop_reason, "stop_sequence": serde_json::Value::Null, "usage": { @@ -1249,6 +1363,102 @@ mod tests { ); } + // ─── Anthropic → OpenAI tool translation (#236) ────────────── + + #[test] + fn anthropic_tools_translate_to_openai_function_shape() { + let anthropic = serde_json::json!([ + { + "name": "get_weather", + "description": "Get current weather", + "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}} + } + ]); + let result = translate_anthropic_tools_to_openai(anthropic).unwrap(); + let arr = result.as_array().unwrap(); + assert_eq!(arr.len(), 1); + assert_eq!(arr[0]["type"], "function"); + assert_eq!(arr[0]["function"]["name"], "get_weather"); + assert_eq!(arr[0]["function"]["description"], "Get current weather"); + assert_eq!(arr[0]["function"]["parameters"]["type"], "object"); + } + + #[test] + fn anthropic_tool_without_description_or_schema_still_translates() { + let anthropic = serde_json::json!([{"name": "noop"}]); + let result = translate_anthropic_tools_to_openai(anthropic).unwrap(); + let tool = &result.as_array().unwrap()[0]; + assert_eq!(tool["type"], "function"); + assert_eq!(tool["function"]["name"], "noop"); + assert!(tool["function"].get("description").is_none()); + assert!(tool["function"].get("parameters").is_none()); + } + + #[test] + fn anthropic_tools_non_array_returns_none() { + assert!(translate_anthropic_tools_to_openai(serde_json::json!("not_array")).is_none()); + } + + #[test] + fn anthropic_tools_empty_array_returns_none() { + assert!(translate_anthropic_tools_to_openai(serde_json::json!([])).is_none()); + } + + #[test] + fn anthropic_tools_entries_without_name_are_skipped() { + let anthropic = serde_json::json!([ + {"description": "no name field"}, + {"name": "valid", "description": "ok"} + ]); + let result = translate_anthropic_tools_to_openai(anthropic).unwrap(); + let arr = result.as_array().unwrap(); + assert_eq!(arr.len(), 1); + assert_eq!(arr[0]["function"]["name"], "valid"); + } + + #[test] + fn anthropic_tool_choice_auto_translates() { + assert_eq!( + translate_anthropic_tool_choice_to_openai(serde_json::json!({"type": "auto"})), + Some(serde_json::json!("auto")), + ); + } + + #[test] + fn anthropic_tool_choice_any_translates_to_required() { + assert_eq!( + translate_anthropic_tool_choice_to_openai(serde_json::json!({"type": "any"})), + Some(serde_json::json!("required")), + ); + } + + #[test] + fn anthropic_tool_choice_none_translates() { + assert_eq!( + translate_anthropic_tool_choice_to_openai(serde_json::json!({"type": "none"})), + Some(serde_json::json!("none")), + ); + } + + #[test] + fn anthropic_tool_choice_specific_tool_translates() { + let anthropic = serde_json::json!({"type": "tool", "name": "get_weather"}); + assert_eq!( + translate_anthropic_tool_choice_to_openai(anthropic), + Some(serde_json::json!({"type": "function", "function": {"name": "get_weather"}})), + ); + } + + #[test] + fn anthropic_tool_choice_unrecognised_returns_none() { + assert!( + translate_anthropic_tool_choice_to_openai(serde_json::json!({"type": "unknown"})) + .is_none() + ); + assert!(translate_anthropic_tool_choice_to_openai(serde_json::json!("auto")).is_none()); + assert!(translate_anthropic_tool_choice_to_openai(serde_json::json!(42)).is_none()); + } + #[test] fn build_request_strips_tool_choice_from_extra() { // Even when the value is unrecognised, tool_choice MUST NOT @@ -1570,6 +1780,38 @@ mod tests { // ─── AnthropicSseEncoder ────────────────────────────────────── + #[test] + fn render_anthropic_response_translates_openai_tool_calls_to_tool_use() { + let mut msg = ChatMessage::assistant(""); + msg.extra.insert( + "tool_calls".to_string(), + serde_json::json!([{ + "id": "call_abc", + "type": "function", + "function": { + "name": "get_time", + "arguments": "{\"timezone\":\"UTC\"}" + } + }]), + ); + let resp = ChatResponse { + id: "cmpl-tc".into(), + model: "gpt-4o".into(), + message: msg, + finish_reason: FinishReason::ToolCalls, + usage: UsageStats::new(10, 5), + }; + let json = chat_response_into_anthropic_json(&resp, "my-model"); + assert_eq!(json["stop_reason"], "tool_use"); + let content = json["content"].as_array().unwrap(); + let tool_block = content.iter().find(|b| b["type"] == "tool_use"); + assert!(tool_block.is_some(), "tool_use block must be present"); + let tb = tool_block.unwrap(); + assert_eq!(tb["id"], "call_abc"); + assert_eq!(tb["name"], "get_time"); + assert_eq!(tb["input"]["timezone"], "UTC"); + } + fn delta_chunk(text: &str) -> ChatChunk { ChatChunk { id: "cmpl-1".into(), diff --git a/crates/aisix-provider-openai/src/wire.rs b/crates/aisix-provider-openai/src/wire.rs index 0bde98db..e0ba7b20 100644 --- a/crates/aisix-provider-openai/src/wire.rs +++ b/crates/aisix-provider-openai/src/wire.rs @@ -147,6 +147,8 @@ pub(crate) struct OpenAiResponseMessage { pub role: String, #[serde(default)] pub content: Option, + #[serde(default)] + pub tool_calls: Option>, } #[derive(Debug, Default, Deserialize)] @@ -184,17 +186,28 @@ pub(crate) struct OpenAiCompletionDetails { pub(crate) fn response_into_chat_response(mut raw: OpenAiResponse) -> ChatResponse { let first = raw.choices.drain(..).next(); let (message, finish) = match first { - Some(c) => ( - ChatMessage { - role: role_from_str(&c.message.role), - content: c.message.content.unwrap_or_default(), - content_blocks: None, - name: None, - tool_call_id: None, - extra: serde_json::Map::new(), - }, - finish_reason(c.finish_reason.as_deref()), - ), + Some(c) => { + let mut extra = serde_json::Map::new(); + if let Some(tool_calls) = c.message.tool_calls { + if !tool_calls.is_empty() { + extra.insert( + "tool_calls".to_string(), + serde_json::Value::Array(tool_calls), + ); + } + } + ( + ChatMessage { + role: role_from_str(&c.message.role), + content: c.message.content.unwrap_or_default(), + content_blocks: None, + name: None, + tool_call_id: None, + extra, + }, + finish_reason(c.finish_reason.as_deref()), + ) + } None => (ChatMessage::assistant(""), FinishReason::Stop), }; @@ -426,6 +439,42 @@ mod tests { assert_eq!(out.usage.reasoning_tokens, 0); } + #[test] + fn response_with_tool_calls_propagates_to_message_extra() { + let body = r#"{ + "id": "cmpl-tc", + "object": "chat.completion", + "model": "gpt-4o", + "choices": [{ + "index": 0, + "message": { + "role": "assistant", + "content": null, + "tool_calls": [{ + "id": "call_abc", + "type": "function", + "function": {"name": "get_time", "arguments": "{\"tz\":\"UTC\"}"} + }] + }, + "finish_reason": "tool_calls" + }], + "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15} + }"#; + let raw: OpenAiResponse = serde_json::from_str(body).unwrap(); + let out = response_into_chat_response(raw); + assert_eq!(out.finish_reason, FinishReason::ToolCalls); + let tc = out + .message + .extra + .get("tool_calls") + .expect("tool_calls in extra") + .as_array() + .unwrap(); + assert_eq!(tc.len(), 1); + assert_eq!(tc[0]["id"], "call_abc"); + assert_eq!(tc[0]["function"]["name"], "get_time"); + } + #[test] fn cache_and_reasoning_details_populate_when_present() { // Verified shape from diff --git a/crates/aisix-proxy/src/messages.rs b/crates/aisix-proxy/src/messages.rs index f9bce53a..24e07046 100644 --- a/crates/aisix-proxy/src/messages.rs +++ b/crates/aisix-proxy/src/messages.rs @@ -380,7 +380,9 @@ async fn cross_provider_dispatch( ) -> Result { use aisix_gateway::{Bridge, BridgeContext}; use aisix_provider_anthropic::{ - chat_response_into_anthropic_json, parse_inbound_request, AnthropicSseEncoder, + chat_response_into_anthropic_json, parse_inbound_request, + translate_anthropic_tool_choice_to_openai, translate_anthropic_tools_to_openai, + AnthropicSseEncoder, }; use std::sync::Arc; @@ -402,6 +404,20 @@ async fn cross_provider_dispatch( // through `ctx.model.upstream_model()` exactly like chat.rs does. chat.model = model_name.to_string(); + // Translate Anthropic-shape tools/tool_choice in `extra` to + // OpenAI shape so the non-Anthropic bridge receives the format + // it expects. Without this, tools are silently dropped (#236). + if let Some(tools) = chat.extra.remove("tools") { + if let Some(translated) = translate_anthropic_tools_to_openai(tools) { + chat.extra.insert("tools".to_string(), translated); + } + } + if let Some(tc) = chat.extra.remove("tool_choice") { + if let Some(translated) = translate_anthropic_tool_choice_to_openai(tc) { + chat.extra.insert("tool_choice".to_string(), translated); + } + } + let is_stream = chat.is_streaming(); let model_arc = Arc::new(model.clone()); let pk_arc = Arc::new(provider_key.clone()); diff --git a/tests/e2e/src/cases/anthropic-tools-cross-provider-e2e.test.ts b/tests/e2e/src/cases/anthropic-tools-cross-provider-e2e.test.ts new file mode 100644 index 00000000..47324e5b --- /dev/null +++ b/tests/e2e/src/cases/anthropic-tools-cross-provider-e2e.test.ts @@ -0,0 +1,217 @@ +import { createHash } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, test } from "vitest"; +import { + AdminClient, + EtcdClient, + spawnApp, + startOpenAiUpstream, + waitConfigPropagation, + type OpenAiUpstream, + type SpawnedApp, +} from "../harness/index.js"; + +// E2E: Anthropic Messages client → OpenAI-compatible upstream — tool +// translation (#236). +// +// When a caller sends an Anthropic Messages request (`POST /v1/messages`) +// with `tools` and `tool_choice`, and the upstream Model is +// OpenAI-compatible, the gateway must: +// +// 1. Translate Anthropic `tools` → OpenAI `tools` on the way out +// (`{name, description, input_schema}` → `{type:"function", +// function:{name, description, parameters}}`). +// 2. Translate Anthropic `tool_choice` → OpenAI `tool_choice` +// (`{type:"any"}` → `"required"`, etc.). +// 3. Translate OpenAI `tool_calls` in the response back to Anthropic +// `content: [{type:"tool_use", …}]` on the way back. +// +// Prior to this fix (#236), tools/tool_choice were passed through +// verbatim in Anthropic shape, which OpenAI upstreams silently ignore +// or reject. + +const CALLER_PLAINTEXT = "sk-anth-tools-xprov-caller"; +const CALLER_KEY_HASH = createHash("sha256") + .update(CALLER_PLAINTEXT) + .digest("hex"); + +describe("Anthropic Messages client → OpenAI upstream: tools translation (#236)", () => { + let app: SpawnedApp | undefined; + let upstream: OpenAiUpstream | undefined; + let admin: AdminClient | undefined; + let etcdReachable = false; + + beforeAll(async () => { + etcdReachable = await new EtcdClient().ping(); + if (!etcdReachable) return; + + // Mock OpenAI upstream returns a tool_calls response. + upstream = await startOpenAiUpstream({ + nonStreamBody: { + id: "chatcmpl-tool-01", + object: "chat.completion", + created: Math.floor(Date.now() / 1000), + model: "gpt-4o", + choices: [ + { + index: 0, + message: { + role: "assistant", + content: null, + tool_calls: [ + { + id: "call_abc123", + type: "function", + function: { + name: "get_time", + arguments: '{"timezone":"UTC"}', + }, + }, + ], + }, + finish_reason: "tool_calls", + }, + ], + usage: { prompt_tokens: 10, completion_tokens: 5, total_tokens: 15 }, + }, + }); + + app = await spawnApp(); + admin = new AdminClient(app.adminUrl, app.adminKey); + + const pk = await admin.createProviderKey({ + display_name: "anth-tools-xprov-pk", + secret: "sk-openai-mock", + api_base: `${upstream.baseUrl}/v1`, + }); + await admin.createModel({ + display_name: "anth-tools-xprov", + provider: "openai", + model_name: "gpt-4o", + provider_key_id: pk.id, + }); + await admin.createApiKey({ + key_hash: CALLER_KEY_HASH, + allowed_models: ["anth-tools-xprov"], + }); + }); + + afterAll(async () => { + await app?.exit(); + await upstream?.close(); + }); + + test("Anthropic tools/tool_choice translate to OpenAI shape on upstream request", async (ctx) => { + if (!etcdReachable || !app || !upstream) { + ctx.skip(); + return; + } + + // Wait for config propagation using a simple probe via /v1/messages + await waitConfigPropagation(async () => { + try { + const res = await fetch(`${app!.proxyUrl}/v1/messages`, { + method: "POST", + headers: { + "content-type": "application/json", + "x-api-key": CALLER_PLAINTEXT, + }, + body: JSON.stringify({ + model: "anth-tools-xprov", + max_tokens: 100, + messages: [{ role: "user", content: "probe" }], + }), + }); + return res.ok; + } catch { + return false; + } + }); + + const baseline = upstream.receivedRequests.length; + + // Send Anthropic-shaped request with tools + tool_choice + const res = await fetch(`${app.proxyUrl}/v1/messages`, { + method: "POST", + headers: { + "content-type": "application/json", + "x-api-key": CALLER_PLAINTEXT, + }, + body: JSON.stringify({ + model: "anth-tools-xprov", + max_tokens: 200, + tools: [ + { + name: "get_time", + description: "Get the current time", + input_schema: { + type: "object", + properties: { + timezone: { type: "string" }, + }, + required: ["timezone"], + }, + }, + ], + tool_choice: { type: "any" }, + messages: [ + { role: "user", content: "What time is it? Use get_time." }, + ], + }), + }); + + expect(res.ok).toBe(true); + const body = (await res.json()) as { + type?: string; + content?: Array<{ + type?: string; + id?: string; + name?: string; + input?: Record; + }>; + stop_reason?: string; + }; + + // Response should be Anthropic-shaped with tool_use content block + expect(body.type).toBe("message"); + expect(body.stop_reason).toBe("tool_use"); + expect(body.content).toBeDefined(); + const toolBlock = body.content?.find((b) => b.type === "tool_use"); + expect(toolBlock).toBeDefined(); + expect(toolBlock?.id).toBe("call_abc123"); + expect(toolBlock?.name).toBe("get_time"); + expect(toolBlock?.input).toEqual({ timezone: "UTC" }); + + // Verify upstream received OpenAI-shaped tools + const upstreamReq = upstream.receivedRequests + .slice(baseline) + .find((r) => r.path === "/v1/chat/completions"); + expect(upstreamReq).toBeDefined(); + + const sentBody = JSON.parse(upstreamReq!.body) as { + tools?: Array<{ + type?: string; + function?: { + name?: string; + description?: string; + parameters?: { type?: string; required?: string[] }; + }; + }>; + tool_choice?: string | { type?: string }; + }; + + // tools: Anthropic shape must be translated to OpenAI shape + expect(sentBody.tools).toHaveLength(1); + expect(sentBody.tools?.[0]?.type).toBe("function"); + expect(sentBody.tools?.[0]?.function?.name).toBe("get_time"); + expect(sentBody.tools?.[0]?.function?.description).toBe( + "Get the current time", + ); + expect(sentBody.tools?.[0]?.function?.parameters?.type).toBe("object"); + expect(sentBody.tools?.[0]?.function?.parameters?.required).toEqual([ + "timezone", + ]); + + // tool_choice: Anthropic {type:"any"} → OpenAI "required" + expect(sentBody.tool_choice).toBe("required"); + }); +}); From 8067ff521049da22d0cd82f2d99bbb7496e150c7 Mon Sep 17 00:00:00 2001 From: Jarvis Date: Tue, 12 May 2026 11:01:20 +0800 Subject: [PATCH 2/2] feat(proxy): add streaming tool_calls support for Anthropic cross-provider dispatch Extends #236 fix to handle streaming mode: - Add tool_calls field to ChatDelta (gateway core) and OpenAiStreamDelta - Propagate streaming tool_calls through stream_chunk_into_chat_chunk - Update AnthropicSseEncoder with per-tool-call state machine that emits content_block_start (tool_use), content_block_delta (input_json_delta), and content_block_stop events - Update RenderedDelta to forward tool_calls in OpenAI streaming path - Add unit tests for streaming encoder (single tool, multi-tool, mixed text+tool, force_finish with open tool blocks) - Add E2E test for streaming tool_calls round-trip --- crates/aisix-gateway/src/chat.rs | 3 + crates/aisix-provider-anthropic/src/wire.rs | 273 +++++++++++++++++- crates/aisix-provider-openai/src/wire.rs | 29 ++ crates/aisix-proxy/src/render.rs | 4 + ...anthropic-tools-cross-provider-e2e.test.ts | 242 ++++++++++++++++ 5 files changed, 536 insertions(+), 15 deletions(-) diff --git a/crates/aisix-gateway/src/chat.rs b/crates/aisix-gateway/src/chat.rs index 445c4474..24d18b63 100644 --- a/crates/aisix-gateway/src/chat.rs +++ b/crates/aisix-gateway/src/chat.rs @@ -352,6 +352,8 @@ pub struct ChatDelta { pub role: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub content: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tool_calls: Option>, } // ─── Embeddings ────────────────────────────────────────────────────────────── @@ -675,6 +677,7 @@ mod tests { delta: ChatDelta { role: None, content: Some("hello".into()), + tool_calls: None, }, finish_reason: None, usage: None, diff --git a/crates/aisix-provider-anthropic/src/wire.rs b/crates/aisix-provider-anthropic/src/wire.rs index 4bc116a0..90f58b44 100644 --- a/crates/aisix-provider-anthropic/src/wire.rs +++ b/crates/aisix-provider-anthropic/src/wire.rs @@ -592,6 +592,7 @@ impl StreamState { delta: ChatDelta { role: None, content: Some(text.clone()), + tool_calls: None, }, finish_reason: None, usage: None, @@ -882,6 +883,16 @@ impl AnthropicSseEvent { } } +/// Per-tool-call accumulator used by the SSE encoder to track which +/// tool_use blocks have been started and at which content-block index. +#[derive(Debug)] +struct ToolCallState { + id: String, + name: String, + content_block_index: usize, + started: bool, +} + /// State machine for re-encoding a stream of internal `ChatChunk`s as /// Anthropic SSE events. #[derive(Debug)] @@ -890,8 +901,13 @@ pub struct AnthropicSseEncoder { model_display_name: String, initial_input_tokens: u32, sent_message_start: bool, - sent_content_block_start: bool, + /// Index assigned to the text content block (if any). + text_block_index: Option, finished: bool, + /// Next content-block index to assign (shared across text + tool_use blocks). + next_block_index: usize, + /// Per-OpenAI-delta-index tool call state. + tool_calls: std::collections::BTreeMap, } impl AnthropicSseEncoder { @@ -910,8 +926,10 @@ impl AnthropicSseEncoder { model_display_name: model_display_name.into(), initial_input_tokens, sent_message_start: false, - sent_content_block_start: false, + text_block_index: None, finished: false, + next_block_index: 0, + tool_calls: std::collections::BTreeMap::new(), } } @@ -929,24 +947,33 @@ impl AnthropicSseEncoder { .content .as_deref() .is_some_and(|s| !s.is_empty()); + let has_tool_calls = chunk + .delta + .tool_calls + .as_ref() + .is_some_and(|v| !v.is_empty()); let has_finish = chunk.finish_reason.is_some(); - if !self.sent_message_start && (has_content || has_finish) { + if !self.sent_message_start && (has_content || has_tool_calls || has_finish) { events.push(self.message_start_event()); self.sent_message_start = true; } - if !self.sent_content_block_start && has_content { - events.push(content_block_start_event()); - self.sent_content_block_start = true; + // ── Text content block ── + if self.text_block_index.is_none() && has_content { + let idx = self.next_block_index; + self.next_block_index += 1; + self.text_block_index = Some(idx); + events.push(content_block_start_event(idx)); } if has_content { + let idx = self.text_block_index.unwrap_or(0); events.push(AnthropicSseEvent { event: "content_block_delta", data: serde_json::json!({ "type": "content_block_delta", - "index": 0, + "index": idx, "delta": { "type": "text_delta", "text": chunk.delta.content.clone().unwrap_or_default(), @@ -955,10 +982,88 @@ impl AnthropicSseEncoder { }); } + // ── Tool-use content blocks ── + if let Some(tool_calls) = &chunk.delta.tool_calls { + for tc in tool_calls { + let oai_index = tc.get("index").and_then(|v| v.as_u64()).unwrap_or(0); + + let id = tc.get("id").and_then(|v| v.as_str()).unwrap_or(""); + let name = tc + .get("function") + .and_then(|f| f.get("name")) + .and_then(|n| n.as_str()) + .unwrap_or(""); + let arguments = tc + .get("function") + .and_then(|f| f.get("arguments")) + .and_then(|a| a.as_str()) + .unwrap_or(""); + + let state = self.tool_calls.entry(oai_index).or_insert_with(|| { + let block_idx = self.next_block_index; + self.next_block_index += 1; + ToolCallState { + id: String::new(), + name: String::new(), + content_block_index: block_idx, + started: false, + } + }); + + if !id.is_empty() { + state.id = id.to_string(); + } + if !name.is_empty() { + state.name = name.to_string(); + } + + // Emit content_block_start once id and name are known. + if !state.started && !state.id.is_empty() && !state.name.is_empty() { + state.started = true; + events.push(AnthropicSseEvent { + event: "content_block_start", + data: serde_json::json!({ + "type": "content_block_start", + "index": state.content_block_index, + "content_block": { + "type": "tool_use", + "id": state.id, + "name": state.name, + "input": {}, + }, + }), + }); + } + + if state.started && !arguments.is_empty() { + events.push(AnthropicSseEvent { + event: "content_block_delta", + data: serde_json::json!({ + "type": "content_block_delta", + "index": state.content_block_index, + "delta": { + "type": "input_json_delta", + "partial_json": arguments, + }, + }), + }); + } + } + } + + // ── Finish ── if let Some(fr) = &chunk.finish_reason { - if self.sent_content_block_start { - events.push(content_block_stop_event()); + // Close text block if open. + if let Some(text_idx) = self.text_block_index { + events.push(content_block_stop_event(text_idx)); + } + // Close all open tool_use blocks. + for state in self.tool_calls.values() { + if state.started { + events.push(content_block_stop_event(state.content_block_index)); + } } + let stop_reason = match fr { FinishReason::Stop => "end_turn", FinishReason::Length => "max_tokens", @@ -1009,8 +1114,13 @@ impl AnthropicSseEncoder { events.push(self.message_start_event()); self.sent_message_start = true; } - if self.sent_content_block_start { - events.push(content_block_stop_event()); + if let Some(text_idx) = self.text_block_index { + events.push(content_block_stop_event(text_idx)); + } + for state in self.tool_calls.values() { + if state.started { + events.push(content_block_stop_event(state.content_block_index)); + } } events.push(AnthropicSseEvent { event: "message_delta", @@ -1054,21 +1164,21 @@ impl AnthropicSseEncoder { } } -fn content_block_start_event() -> AnthropicSseEvent { +fn content_block_start_event(index: usize) -> AnthropicSseEvent { AnthropicSseEvent { event: "content_block_start", data: serde_json::json!({ "type": "content_block_start", - "index": 0, + "index": index, "content_block": {"type": "text", "text": ""}, }), } } -fn content_block_stop_event() -> AnthropicSseEvent { +fn content_block_stop_event(index: usize) -> AnthropicSseEvent { AnthropicSseEvent { event: "content_block_stop", - data: serde_json::json!({"type": "content_block_stop", "index": 0}), + data: serde_json::json!({"type": "content_block_stop", "index": index}), } } @@ -1819,6 +1929,7 @@ mod tests { delta: ChatDelta { role: None, content: Some(text.into()), + tool_calls: None, }, finish_reason: None, usage: None, @@ -1930,4 +2041,136 @@ mod tests { let s = ev.to_sse_string(); assert_eq!(s, "event: content_block_delta\ndata: {\"x\":1}\n\n"); } + + // ─── Streaming tool_calls ────────────────────────────────────── + + fn tool_call_chunk(index: u64, id: &str, name: &str, arguments: &str) -> ChatChunk { + let mut tc = serde_json::json!({"index": index}); + if !id.is_empty() { + tc["id"] = serde_json::json!(id); + tc["type"] = serde_json::json!("function"); + } + let mut func = serde_json::Map::new(); + if !name.is_empty() { + func.insert("name".into(), serde_json::json!(name)); + } + if !arguments.is_empty() { + func.insert("arguments".into(), serde_json::json!(arguments)); + } + if !func.is_empty() { + tc["function"] = serde_json::Value::Object(func); + } + ChatChunk { + id: "cmpl-1".into(), + model: "u".into(), + delta: ChatDelta { + role: None, + content: None, + tool_calls: Some(vec![tc]), + }, + finish_reason: None, + usage: None, + } + } + + fn tool_finish_chunk() -> ChatChunk { + ChatChunk { + id: "cmpl-1".into(), + model: "u".into(), + delta: ChatDelta::default(), + finish_reason: Some(FinishReason::ToolCalls), + usage: Some(UsageStats::new(10, 5)), + } + } + + #[test] + fn sse_encoder_tool_call_emits_block_start_and_argument_deltas() { + let mut enc = AnthropicSseEncoder::new("msg_01", "m", 0); + // First chunk: tool header with id+name and initial args. + let events = enc.next_events(&tool_call_chunk(0, "call_1", "get_weather", "{\"loc")); + let kinds: Vec<_> = events.iter().map(|e| e.event).collect(); + assert_eq!( + kinds, + vec![ + "message_start", + "content_block_start", + "content_block_delta" + ] + ); + // content_block_start should be tool_use + assert_eq!(events[1].data["content_block"]["type"], "tool_use"); + assert_eq!(events[1].data["content_block"]["id"], "call_1"); + assert_eq!(events[1].data["content_block"]["name"], "get_weather"); + // content_block_delta should be input_json_delta + assert_eq!(events[2].data["delta"]["type"], "input_json_delta"); + assert_eq!(events[2].data["delta"]["partial_json"], "{\"loc"); + } + + #[test] + fn sse_encoder_tool_call_subsequent_args_emit_delta_only() { + let mut enc = AnthropicSseEncoder::new("msg_01", "m", 0); + enc.next_events(&tool_call_chunk(0, "call_1", "get_weather", "")); + let events = enc.next_events(&tool_call_chunk(0, "", "", "ation\"}")); + let kinds: Vec<_> = events.iter().map(|e| e.event).collect(); + assert_eq!(kinds, vec!["content_block_delta"]); + assert_eq!(events[0].data["delta"]["partial_json"], "ation\"}"); + } + + #[test] + fn sse_encoder_tool_finish_closes_all_blocks() { + let mut enc = AnthropicSseEncoder::new("msg_01", "m", 0); + enc.next_events(&tool_call_chunk(0, "call_1", "fn_a", "{}")); + enc.next_events(&tool_call_chunk(1, "call_2", "fn_b", "{}")); + let events = enc.next_events(&tool_finish_chunk()); + let kinds: Vec<_> = events.iter().map(|e| e.event).collect(); + // Should close both tool blocks, then message_delta + message_stop + assert_eq!( + kinds, + vec![ + "content_block_stop", + "content_block_stop", + "message_delta", + "message_stop" + ] + ); + assert_eq!(events[2].data["delta"]["stop_reason"], "tool_use"); + } + + #[test] + fn sse_encoder_mixed_text_and_tool_call() { + let mut enc = AnthropicSseEncoder::new("msg_01", "m", 0); + // Text first + enc.next_events(&delta_chunk("thinking...")); + // Then a tool call + let events = enc.next_events(&tool_call_chunk(0, "call_1", "search", "{\"q\":\"x\"}")); + let kinds: Vec<_> = events.iter().map(|e| e.event).collect(); + assert_eq!(kinds, vec!["content_block_start", "content_block_delta"]); + // Tool block should be at index 1 (text was 0) + assert_eq!(events[0].data["index"], 1); + // Finish + let events = enc.next_events(&tool_finish_chunk()); + let kinds: Vec<_> = events.iter().map(|e| e.event).collect(); + // Close text block (0), tool block (1), then message_delta + stop + assert_eq!( + kinds, + vec![ + "content_block_stop", + "content_block_stop", + "message_delta", + "message_stop" + ] + ); + } + + #[test] + fn sse_encoder_force_finish_closes_tool_blocks() { + let mut enc = AnthropicSseEncoder::new("msg_01", "m", 0); + enc.next_events(&tool_call_chunk(0, "call_1", "fn_a", "{}")); + let events = enc.force_finish(); + let kinds: Vec<_> = events.iter().map(|e| e.event).collect(); + assert_eq!( + kinds, + vec!["content_block_stop", "message_delta", "message_stop"] + ); + } } diff --git a/crates/aisix-provider-openai/src/wire.rs b/crates/aisix-provider-openai/src/wire.rs index e0ba7b20..770466ea 100644 --- a/crates/aisix-provider-openai/src/wire.rs +++ b/crates/aisix-provider-openai/src/wire.rs @@ -277,6 +277,8 @@ pub(crate) struct OpenAiStreamDelta { pub role: Option, #[serde(default)] pub content: Option, + #[serde(default)] + pub tool_calls: Option>, } pub(crate) fn stream_chunk_into_chat_chunk(mut raw: OpenAiStreamChunk) -> ChatChunk { @@ -286,6 +288,7 @@ pub(crate) fn stream_chunk_into_chat_chunk(mut raw: OpenAiStreamChunk) -> ChatCh ChatDelta { role: c.delta.role.as_deref().map(role_from_str), content: c.delta.content, + tool_calls: c.delta.tool_calls, }, c.finish_reason .as_deref() @@ -569,6 +572,32 @@ mod tests { assert_eq!(chunk.finish_reason, Some(FinishReason::Stop)); } + #[test] + fn stream_chunk_with_tool_calls_propagates_to_delta() { + let body = r#"{ + "id": "cmpl-t", + "model": "gpt-4o", + "choices": [{ + "index": 0, + "delta": { + "tool_calls": [{ + "index": 0, + "id": "call_abc", + "type": "function", + "function": { "name": "get_weather", "arguments": "" } + }] + }, + "finish_reason": null + }] + }"#; + let raw: OpenAiStreamChunk = serde_json::from_str(body).unwrap(); + let chunk = stream_chunk_into_chat_chunk(raw); + let tc = chunk.delta.tool_calls.expect("tool_calls in delta"); + assert_eq!(tc.len(), 1); + assert_eq!(tc[0]["id"], "call_abc"); + assert_eq!(tc[0]["function"]["name"], "get_weather"); + } + #[test] fn build_request_sets_stream_flag_and_propagates_params() { let req = ChatFormat { diff --git a/crates/aisix-proxy/src/render.rs b/crates/aisix-proxy/src/render.rs index d2e379d4..8089fe0a 100644 --- a/crates/aisix-proxy/src/render.rs +++ b/crates/aisix-proxy/src/render.rs @@ -74,6 +74,8 @@ pub struct RenderedDelta { pub role: Option<&'static str>, #[serde(skip_serializing_if = "Option::is_none")] pub content: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub tool_calls: Option>, } pub fn render_response(created_unix_ts: i64, resp: ChatResponse) -> ChatCompletion { @@ -112,6 +114,7 @@ pub fn render_chunk(created_unix_ts: i64, chunk: ChatChunk) -> ChatCompletionChu delta: RenderedDelta { role: chunk.delta.role.map(role_to_str), content: chunk.delta.content, + tool_calls: chunk.delta.tool_calls, }, finish_reason: chunk .finish_reason @@ -232,6 +235,7 @@ mod tests { delta: aisix_gateway::ChatDelta { role: None, content: Some("hi".into()), + tool_calls: None, }, finish_reason: None, usage: None, diff --git a/tests/e2e/src/cases/anthropic-tools-cross-provider-e2e.test.ts b/tests/e2e/src/cases/anthropic-tools-cross-provider-e2e.test.ts index 47324e5b..e2d644b9 100644 --- a/tests/e2e/src/cases/anthropic-tools-cross-provider-e2e.test.ts +++ b/tests/e2e/src/cases/anthropic-tools-cross-provider-e2e.test.ts @@ -215,3 +215,245 @@ describe("Anthropic Messages client → OpenAI upstream: tools translation (#236 expect(sentBody.tool_choice).toBe("required"); }); }); + +// ─── Streaming test ───────────────────────────────────────────────── + +const STREAM_CALLER_PLAINTEXT = "sk-anth-tools-stream-xprov"; +const STREAM_CALLER_KEY_HASH = createHash("sha256") + .update(STREAM_CALLER_PLAINTEXT) + .digest("hex"); + +describe("Anthropic Messages client → OpenAI upstream: streaming tool_calls (#236)", () => { + let app: SpawnedApp | undefined; + let upstream: OpenAiUpstream | undefined; + let admin: AdminClient | undefined; + let etcdReachable = false; + + beforeAll(async () => { + etcdReachable = await new EtcdClient().ping(); + if (!etcdReachable) return; + + // Mock OpenAI upstream returns streaming tool_calls chunks. + const streamEvents = [ + JSON.stringify({ + id: "cmpl-stream-tool", + object: "chat.completion.chunk", + model: "gpt-4o", + choices: [ + { + index: 0, + delta: { + role: "assistant", + content: null, + tool_calls: [ + { + index: 0, + id: "call_stream_1", + type: "function", + function: { name: "get_time", arguments: "" }, + }, + ], + }, + finish_reason: null, + }, + ], + }), + JSON.stringify({ + id: "cmpl-stream-tool", + object: "chat.completion.chunk", + model: "gpt-4o", + choices: [ + { + index: 0, + delta: { + tool_calls: [ + { + index: 0, + function: { arguments: '{"timezone"' }, + }, + ], + }, + finish_reason: null, + }, + ], + }), + JSON.stringify({ + id: "cmpl-stream-tool", + object: "chat.completion.chunk", + model: "gpt-4o", + choices: [ + { + index: 0, + delta: { + tool_calls: [ + { + index: 0, + function: { arguments: ':"UTC"}' }, + }, + ], + }, + finish_reason: null, + }, + ], + }), + JSON.stringify({ + id: "cmpl-stream-tool", + object: "chat.completion.chunk", + model: "gpt-4o", + choices: [ + { + index: 0, + delta: {}, + finish_reason: "tool_calls", + }, + ], + usage: { prompt_tokens: 10, completion_tokens: 5, total_tokens: 15 }, + }), + "[DONE]", + ]; + + upstream = await startOpenAiUpstream({ streamEvents }); + + app = await spawnApp(); + admin = new AdminClient(app.adminUrl, app.adminKey); + + const pk = await admin.createProviderKey({ + display_name: "anth-tools-stream-xprov-pk", + secret: "sk-openai-stream-mock", + api_base: `${upstream.baseUrl}/v1`, + }); + await admin.createModel({ + display_name: "anth-tools-stream-xprov", + provider: "openai", + model_name: "gpt-4o", + provider_key_id: pk.id, + }); + await admin.createApiKey({ + key_hash: STREAM_CALLER_KEY_HASH, + allowed_models: ["anth-tools-stream-xprov"], + }); + }); + + afterAll(async () => { + await app?.exit(); + await upstream?.close(); + }); + + test("Streaming tool_calls are translated to Anthropic SSE tool_use events", async (ctx) => { + if (!etcdReachable || !app || !upstream) { + ctx.skip(); + return; + } + + await waitConfigPropagation(async () => { + try { + const res = await fetch(`${app!.proxyUrl}/v1/messages`, { + method: "POST", + headers: { + "content-type": "application/json", + "x-api-key": STREAM_CALLER_PLAINTEXT, + }, + body: JSON.stringify({ + model: "anth-tools-stream-xprov", + max_tokens: 100, + stream: true, + messages: [{ role: "user", content: "probe" }], + }), + }); + return res.ok; + } catch { + return false; + } + }); + + const baseline = upstream.receivedRequests.length; + + const res = await fetch(`${app.proxyUrl}/v1/messages`, { + method: "POST", + headers: { + "content-type": "application/json", + "x-api-key": STREAM_CALLER_PLAINTEXT, + }, + body: JSON.stringify({ + model: "anth-tools-stream-xprov", + max_tokens: 200, + stream: true, + tools: [ + { + name: "get_time", + description: "Get the current time", + input_schema: { + type: "object", + properties: { timezone: { type: "string" } }, + required: ["timezone"], + }, + }, + ], + tool_choice: { type: "any" }, + messages: [ + { role: "user", content: "What time is it?" }, + ], + }), + }); + + expect(res.ok).toBe(true); + expect(res.headers.get("content-type")).toContain("text/event-stream"); + + const text = await res.text(); + const lines = text.split("\n").filter((l) => l.startsWith("data: ")); + const events = lines.map((l) => JSON.parse(l.slice(6)) as Record); + + // Should contain message_start, content_block_start (tool_use), + // content_block_delta (input_json_delta), content_block_stop, + // message_delta, message_stop + const types = events.map((e) => e.type); + expect(types).toContain("message_start"); + expect(types).toContain("message_stop"); + + // Find the tool_use content_block_start + const toolStart = events.find( + (e) => + e.type === "content_block_start" && + (e.content_block as Record)?.type === "tool_use", + ); + expect(toolStart).toBeDefined(); + const cb = toolStart!.content_block as Record; + expect(cb.id).toBe("call_stream_1"); + expect(cb.name).toBe("get_time"); + + // Find input_json_delta events + const jsonDeltas = events.filter( + (e) => + e.type === "content_block_delta" && + (e.delta as Record)?.type === "input_json_delta", + ); + expect(jsonDeltas.length).toBeGreaterThanOrEqual(1); + const fullJson = jsonDeltas + .map((d) => (d.delta as Record).partial_json as string) + .join(""); + expect(fullJson).toBe('{"timezone":"UTC"}'); + + // Find message_delta with stop_reason tool_use + const msgDelta = events.find( + (e) => + e.type === "message_delta" && + (e.delta as Record)?.stop_reason === "tool_use", + ); + expect(msgDelta).toBeDefined(); + + // Verify upstream received OpenAI-shaped tools + const upstreamReq = upstream.receivedRequests + .slice(baseline) + .find((r) => r.path === "/v1/chat/completions"); + expect(upstreamReq).toBeDefined(); + const sentBody = JSON.parse(upstreamReq!.body) as { + tools?: Array<{ type?: string; function?: { name?: string } }>; + tool_choice?: string; + stream?: boolean; + }; + expect(sentBody.tools?.[0]?.type).toBe("function"); + expect(sentBody.tools?.[0]?.function?.name).toBe("get_time"); + expect(sentBody.tool_choice).toBe("required"); + expect(sentBody.stream).toBe(true); + }); +});