Codex++ Responses<->Anthropic compatibility fixes

Namespace tool flatten/restore, array function_call_output, omit empty input_schema for native tools, lift additional_tools; scoped to ForwardAsResponses.
This commit is contained in:
Ricardo-binZzz
2026-07-27 08:19:19 +08:00
parent 2730c1c43b
commit 7dde9370e4
8 changed files with 324 additions and 24 deletions
@@ -586,7 +586,7 @@ func (r *ResponsesClientToolStreamRestorer) restoreNamespaceEvent(event Response
event.Item.Name, event.Item.Namespace = name.Name, name.Namespace
}
}
if event.Type == "response.function_call_arguments.done" {
if event.Type == "response.function_call_arguments.delta" || event.Type == "response.function_call_arguments.done" {
if name, ok := r.adapter.NamespaceTools[event.Name]; ok {
event.Name = name.Name
}
@@ -146,7 +146,13 @@ func TestResponsesClientToolStreamRestorer_RestoresNamespaceLifecycle(t *testing
require.Equal(t, "open", gjson.GetBytes(added[0], "item.name").String())
require.Equal(t, "browser", gjson.GetBytes(added[0], "item.namespace").String())
done, changed, err := restorer.RestoreEvent([]byte(`{"type":"response.function_call_arguments.done","sequence_number":5,"output_index":0,"item_id":"i1","name":"browser__open","arguments":"{}"}`))
delta, changed, err := restorer.RestoreEvent([]byte(`{"type":"response.function_call_arguments.delta","sequence_number":5,"output_index":0,"item_id":"i1","name":"browser__open","delta":"{\"url\":"}`))
require.NoError(t, err)
require.True(t, changed)
require.Len(t, delta, 1)
require.Equal(t, "open", gjson.GetBytes(delta[0], "name").String())
done, changed, err := restorer.RestoreEvent([]byte(`{"type":"response.function_call_arguments.done","sequence_number":6,"output_index":0,"item_id":"i1","name":"browser__open","arguments":"{}"}`))
require.NoError(t, err)
require.True(t, changed)
require.Len(t, done, 1)
@@ -152,11 +152,7 @@ func convertResponsesInputToAnthropic(instructions string, inputRaw json.RawMess
case item.Type == "function_call_output":
// function_call_output → user message with tool_result block
outputContent := item.Output
if outputContent == "" {
outputContent = "(empty)"
}
contentJSON, _ := json.Marshal(outputContent)
contentJSON := responsesFunctionOutputToAnthropicContent(item)
block := AnthropicContentBlock{
Type: "tool_result",
ToolUseID: fromResponsesCallIDToAnthropic(item.CallID),
@@ -216,6 +212,45 @@ func convertResponsesInputToAnthropic(instructions string, inputRaw json.RawMess
return system, messages, nil
}
func responsesFunctionOutputToAnthropicContent(item ResponsesInputItem) json.RawMessage {
if len(item.outputRaw) == 0 {
output := item.Output
if output == "" {
output = "(empty)"
}
content, _ := json.Marshal(output)
return content
}
var parts []ResponsesContentPart
if err := json.Unmarshal(item.outputRaw, &parts); err == nil {
blocks := make([]AnthropicContentBlock, 0, len(parts))
for _, part := range parts {
switch part.Type {
case "input_text", "output_text", "text":
if part.Text != "" {
blocks = append(blocks, AnthropicContentBlock{Type: "text", Text: part.Text})
}
case "input_image":
if source := dataURIToAnthropicImageSource(part.ImageURL); source != nil {
blocks = append(blocks, AnthropicContentBlock{Type: "image", Source: source})
}
}
}
if len(blocks) > 0 {
content, _ := json.Marshal(blocks)
return content
}
if len(parts) == 0 {
content, _ := json.Marshal("(empty)")
return content
}
}
content, _ := json.Marshal(item.Output)
return content
}
// normalizeAnthropicToolPairing rebuilds the message sequence so it satisfies
// Anthropic's tool_use/tool_result invariants, which the naive item-by-item
// conversion violates whenever the Responses history interleaves anything
@@ -163,3 +163,28 @@ func TestAnthropicPairing_SingleCall(t *testing.T) {
require.True(t, hasToolUse(parseContentBlocks(msgs[1].Content), "call_A"))
require.True(t, hasToolResult(parseContentBlocks(msgs[2].Content), "call_A"))
}
func TestResponsesToAnthropic_FunctionOutputContentArray(t *testing.T) {
msgs := convertAnthropic(t, `[
{"type":"function_call","call_id":"call_A","name":"view_image","arguments":"{}"},
{"type":"function_call_output","call_id":"call_A","output":[
{"type":"input_text","text":"image loaded"},
{"type":"input_image","image_url":"data:image/png;base64,YQ=="}
]}
]`)
require.Len(t, msgs, 2)
resultBlocks := parseContentBlocks(msgs[1].Content)
require.Len(t, resultBlocks, 1)
require.Equal(t, "tool_result", resultBlocks[0].Type)
var content []AnthropicContentBlock
require.NoError(t, json.Unmarshal(resultBlocks[0].Content, &content))
require.Len(t, content, 2)
require.Equal(t, "text", content[0].Type)
require.Equal(t, "image loaded", content[0].Text)
require.Equal(t, "image", content[1].Type)
require.NotNil(t, content[1].Source)
require.Equal(t, "image/png", content[1].Source.MediaType)
require.Equal(t, "YQ==", content[1].Source.Data)
}
@@ -122,6 +122,9 @@ func TestResponsesToAnthropic_MixedToolsProduceValidAnthropicTools(t *testing.T)
assert.Equal(t, "web_search_20250305", tools[2].Type)
assert.Equal(t, "web_search", tools[2].Name)
assert.Empty(t, tools[2].InputSchema)
serverToolWire, err := json.Marshal(tools[2])
require.NoError(t, err)
assert.NotContains(t, string(serverToolWire), `"input_schema"`)
}
func TestResponsesToAnthropic_DefaultToolNormalizesInputSchema(t *testing.T) {
+33 -3
View File
@@ -4,7 +4,10 @@
// formats can be served through a unified gateway.
package apicompat
import "encoding/json"
import (
"bytes"
"encoding/json"
)
// ---------------------------------------------------------------------------
// Anthropic Messages API types
@@ -112,7 +115,7 @@ type AnthropicTool struct {
Type string `json:"type,omitempty"` // e.g. "web_search_20250305" for server tools
Name string `json:"name"`
Description string `json:"description,omitempty"`
InputSchema json.RawMessage `json:"input_schema"` // JSON Schema object
InputSchema json.RawMessage `json:"input_schema,omitempty"` // JSON Schema object
CacheControl *AnthropicCacheControl `json:"cache_control,omitempty"`
}
@@ -260,7 +263,34 @@ type ResponsesInputItem struct {
ID string `json:"id,omitempty"`
// type=function_call_output
Output string `json:"output,omitempty"`
Output string `json:"output,omitempty"`
outputRaw json.RawMessage
}
func (i *ResponsesInputItem) UnmarshalJSON(data []byte) error {
type alias ResponsesInputItem
var wire struct {
*alias
Output json.RawMessage `json:"output"`
}
*i = ResponsesInputItem{}
wire.alias = (*alias)(i)
if err := json.Unmarshal(data, &wire); err != nil {
return err
}
output := bytes.TrimSpace(wire.Output)
if len(output) == 0 || bytes.Equal(output, []byte("null")) {
return nil
}
if err := json.Unmarshal(output, &i.Output); err == nil {
return nil
}
i.outputRaw = append(i.outputRaw[:0], output...)
i.Output = string(output)
return nil
}
// ResponsesContentPart is a typed content part in a Responses message.
@@ -37,15 +37,21 @@ func (s *GatewayService) ForwardAsResponses(
) (*ForwardResult, error) {
startTime := time.Now()
// 1. Parse Responses request
// 1. Lower Codex client-side tools to function tools understood by Anthropic.
adaptedBody, clientToolMapping, err := adaptResponsesClientToolsForAnthropic(body)
if err != nil {
return nil, fmt.Errorf("adapt responses client tools: %w", err)
}
// 2. Parse Responses request
var responsesReq apicompat.ResponsesRequest
if err := json.Unmarshal(body, &responsesReq); err != nil {
if err := json.Unmarshal(adaptedBody, &responsesReq); err != nil {
return nil, fmt.Errorf("parse responses request: %w", err)
}
originalModel := responsesReq.Model
clientStream := responsesReq.Stream
// 2. Convert Responses → Anthropic
// 3. Convert Responses → Anthropic
anthropicReq, err := apicompat.ResponsesToAnthropicRequest(&responsesReq)
if err != nil {
return nil, fmt.Errorf("convert responses to anthropic: %w", err)
@@ -182,14 +188,71 @@ func (s *GatewayService) ForwardAsResponses(
var result *ForwardResult
var handleErr error
if clientStream {
result, handleErr = s.handleResponsesStreamingResponse(resp, c, originalModel, mappedModel, reasoningEffort, startTime)
result, handleErr = s.handleResponsesStreamingResponse(resp, c, originalModel, mappedModel, reasoningEffort, startTime, clientToolMapping)
} else {
result, handleErr = s.handleResponsesBufferedStreamingResponse(resp, c, originalModel, mappedModel, reasoningEffort, startTime)
result, handleErr = s.handleResponsesBufferedStreamingResponse(resp, c, originalModel, mappedModel, reasoningEffort, startTime, clientToolMapping)
}
return result, handleErr
}
func adaptResponsesClientToolsForAnthropic(body []byte) ([]byte, apicompat.ResponsesClientToolMapping, error) {
decoder := json.NewDecoder(bytes.NewReader(body))
decoder.UseNumber()
var requestBody map[string]any
if err := decoder.Decode(&requestBody); err != nil {
return body, apicompat.ResponsesClientToolMapping{}, err
}
additionalToolsChanged, err := liftResponsesAdditionalTools(requestBody)
if err != nil {
return body, apicompat.ResponsesClientToolMapping{}, err
}
mapping, changed, err := apicompat.AdaptResponsesClientTools(requestBody)
if err != nil {
return body, apicompat.ResponsesClientToolMapping{}, err
}
changed = changed || additionalToolsChanged
if !changed {
return body, mapping, nil
}
rebuilt, err := json.Marshal(requestBody)
if err != nil {
return body, apicompat.ResponsesClientToolMapping{}, err
}
return rebuilt, mapping, nil
}
func liftResponsesAdditionalTools(requestBody map[string]any) (bool, error) {
input, ok := requestBody["input"].([]any)
if !ok {
return false, nil
}
tools, _ := requestBody["tools"].([]any)
kept := make([]any, 0, len(input))
changed := false
for _, raw := range input {
item, ok := raw.(map[string]any)
if !ok || strings.TrimSpace(fmt.Sprint(item["type"])) != "additional_tools" {
kept = append(kept, raw)
continue
}
additional, ok := item["tools"].([]any)
if !ok {
return false, fmt.Errorf("additional_tools.tools must be an array")
}
tools = append(tools, additional...)
changed = true
}
if !changed {
return false, nil
}
requestBody["tools"] = tools
requestBody["input"] = kept
return true, nil
}
// ExtractResponsesReasoningEffortFromBody reads Responses API reasoning.effort
// and normalizes it for usage logging.
func ExtractResponsesReasoningEffortFromBody(body []byte) *string {
@@ -243,6 +306,7 @@ func (s *GatewayService) handleResponsesBufferedStreamingResponse(
mappedModel string,
reasoningEffort *string,
startTime time.Time,
clientToolMapping apicompat.ResponsesClientToolMapping,
) (*ForwardResult, error) {
requestID := resp.Header.Get("x-request-id")
@@ -358,6 +422,10 @@ func (s *GatewayService) handleResponsesBufferedStreamingResponse(
c.Writer.Header().Set("Content-Type", "application/json; charset=utf-8")
if respBytes, err := json.Marshal(responsesResp); err == nil {
respBytes = reverseToolNamesIfPresent(c, respBytes)
respBytes, _, err = apicompat.RestoreResponsesClientToolPayload(respBytes, clientToolMapping)
if err != nil {
return nil, fmt.Errorf("restore responses client tools: %w", err)
}
c.Data(http.StatusOK, "application/json; charset=utf-8", respBytes)
} else {
c.JSON(http.StatusOK, responsesResp)
@@ -383,6 +451,7 @@ func (s *GatewayService) handleResponsesStreamingResponse(
mappedModel string,
reasoningEffort *string,
startTime time.Time,
clientToolMapping apicompat.ResponsesClientToolMapping,
) (*ForwardResult, error) {
requestID := resp.Header.Get("x-request-id")
@@ -397,6 +466,7 @@ func (s *GatewayService) handleResponsesStreamingResponse(
state := apicompat.NewAnthropicEventToResponsesState()
state.Model = originalModel
clientToolRestorer := apicompat.NewResponsesClientToolStreamRestorer(clientToolMapping)
var usage ClaudeUsage
var firstTokenMs *int
firstChunk := true
@@ -441,7 +511,7 @@ func (s *GatewayService) handleResponsesStreamingResponse(
// Convert to Responses events
events := apicompat.AnthropicEventToResponsesEvents(event, state)
for _, evt := range events {
sse, err := apicompat.ResponsesEventToSSE(evt)
payload, err := json.Marshal(evt)
if err != nil {
logger.L().Warn("forward_as_responses stream: failed to marshal event",
zap.Error(err),
@@ -449,12 +519,23 @@ func (s *GatewayService) handleResponsesStreamingResponse(
)
continue
}
out := string(reverseToolNamesIfPresent(c, []byte(sse)))
if _, err := fmt.Fprint(c.Writer, out); err != nil {
logger.L().Info("forward_as_responses stream: client disconnected",
payload = reverseToolNamesIfPresent(c, payload)
payloads, _, err := clientToolRestorer.RestoreEvent(payload)
if err != nil {
logger.L().Warn("forward_as_responses stream: failed to restore client tools",
zap.Error(err),
zap.String("request_id", requestID),
)
return true // client disconnected
continue
}
for _, restored := range payloads {
eventType := gjson.GetBytes(restored, "type").String()
if _, err := fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", eventType, restored); err != nil {
logger.L().Info("forward_as_responses stream: client disconnected",
zap.String("request_id", requestID),
)
return true // client disconnected
}
}
}
if len(events) > 0 {
@@ -3,6 +3,7 @@
package service
import (
"encoding/json"
"io"
"net/http"
"net/http/httptest"
@@ -10,10 +11,129 @@ import (
"testing"
"time"
"github.com/Wei-Shaw/sub2api/internal/pkg/apicompat"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
)
func TestAdaptResponsesClientToolsForAnthropic_FlattensNamespace(t *testing.T) {
t.Parallel()
body := []byte(`{
"model":"claude-fable-5",
"input":[{"type":"function_call","call_id":"call_1","namespace":"codex_app","name":"read_thread","arguments":"{}"}],
"tools":[{"type":"namespace","name":"codex_app","tools":[{"type":"function","name":"read_thread","description":"Read a task","parameters":{"type":"object","properties":{}}}]}]
}`)
adapted, mapping, err := adaptResponsesClientToolsForAnthropic(body)
require.NoError(t, err)
require.Equal(t, apicompat.ResponsesNamespaceName{Namespace: "codex_app", Name: "read_thread"}, mapping.NamespaceTools["codex_app__read_thread"])
var request map[string]any
require.NoError(t, json.Unmarshal(adapted, &request))
tools := request["tools"].([]any)
require.Len(t, tools, 1)
tool := tools[0].(map[string]any)
require.Equal(t, "function", tool["type"])
require.Equal(t, "codex_app__read_thread", tool["name"])
input := request["input"].([]any)
call := input[0].(map[string]any)
require.Equal(t, "codex_app__read_thread", call["name"])
require.NotContains(t, call, "namespace")
}
func TestAdaptResponsesClientToolsForAnthropic_LiftsAdditionalTools(t *testing.T) {
body := []byte(`{
"model":"claude-fable-5",
"input":[
{"type":"additional_tools","tools":[
{"type":"custom","name":"exec","description":"Run a command"},
{"type":"namespace","name":"codex_app","tools":[
{"type":"function","name":"read_thread","parameters":{"type":"object"}}
]}
]},
{"type":"message","role":"user","content":[{"type":"input_text","text":"inspect"}]}
]
}`)
adapted, mapping, err := adaptResponsesClientToolsForAnthropic(body)
require.NoError(t, err)
require.True(t, mapping.CustomTools["exec"])
require.Equal(t, apicompat.ResponsesNamespaceName{Namespace: "codex_app", Name: "read_thread"}, mapping.NamespaceTools["codex_app__read_thread"])
var request map[string]any
require.NoError(t, json.Unmarshal(adapted, &request))
tools := request["tools"].([]any)
require.Len(t, tools, 2)
require.Equal(t, "function", tools[0].(map[string]any)["type"])
require.Equal(t, "codex_app__read_thread", tools[1].(map[string]any)["name"])
input := request["input"].([]any)
require.Len(t, input, 1)
require.Equal(t, "message", input[0].(map[string]any)["type"])
}
func namespaceToolAnthropicStream() string {
return strings.Join([]string{
`event: message_start`,
`data: {"type":"message_start","message":{"id":"msg_namespace","type":"message","role":"assistant","content":[],"model":"claude-fable-5","stop_reason":"","usage":{"input_tokens":10}}}`,
``,
`event: content_block_start`,
`data: {"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"toolu_namespace","name":"codex_app__read_thread","input":{"thread_id":"123"}}}`,
``,
`event: content_block_stop`,
`data: {"type":"content_block_stop","index":0}`,
``,
`event: message_delta`,
`data: {"type":"message_delta","delta":{"stop_reason":"tool_use"},"usage":{"output_tokens":5}}`,
``,
`event: message_stop`,
`data: {"type":"message_stop"}`,
``,
}, "\n")
}
func namespaceToolMapping() apicompat.ResponsesClientToolMapping {
return apicompat.ResponsesClientToolMapping{NamespaceTools: map[string]apicompat.ResponsesNamespaceName{
"codex_app__read_thread": {Namespace: "codex_app", Name: "read_thread"},
}}
}
func TestHandleResponsesBufferedStreamingResponse_RestoresNamespaceTool(t *testing.T) {
t.Parallel()
gin.SetMode(gin.TestMode)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
resp := &http.Response{Body: io.NopCloser(strings.NewReader(namespaceToolAnthropicStream()))}
svc := &GatewayService{}
_, err := svc.handleResponsesBufferedStreamingResponse(resp, c, "claude-fable-5", "claude-fable-5", nil, time.Now(), namespaceToolMapping())
require.NoError(t, err)
require.Contains(t, rec.Body.String(), `"type":"function_call"`)
require.Contains(t, rec.Body.String(), `"name":"read_thread"`)
require.Contains(t, rec.Body.String(), `"namespace":"codex_app"`)
require.NotContains(t, rec.Body.String(), `"name":"codex_app__read_thread"`)
}
func TestHandleResponsesStreamingResponse_RestoresNamespaceTool(t *testing.T) {
t.Parallel()
gin.SetMode(gin.TestMode)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
resp := &http.Response{Body: io.NopCloser(strings.NewReader(namespaceToolAnthropicStream()))}
svc := &GatewayService{}
_, err := svc.handleResponsesStreamingResponse(resp, c, "claude-fable-5", "claude-fable-5", nil, time.Now(), namespaceToolMapping())
require.NoError(t, err)
require.Contains(t, rec.Body.String(), `response.output_item.added`)
require.Contains(t, rec.Body.String(), `"name":"read_thread"`)
require.Contains(t, rec.Body.String(), `"namespace":"codex_app"`)
require.NotContains(t, rec.Body.String(), `"name":"codex_app__read_thread"`)
}
func TestExtractResponsesReasoningEffortFromBody(t *testing.T) {
t.Parallel()
@@ -51,7 +171,7 @@ func TestHandleResponsesBufferedStreamingResponse_PreservesMessageStartCacheUsag
}
svc := &GatewayService{}
result, err := svc.handleResponsesBufferedStreamingResponse(resp, c, "claude-sonnet-4.5", "claude-sonnet-4.5", nil, time.Now())
result, err := svc.handleResponsesBufferedStreamingResponse(resp, c, "claude-sonnet-4.5", "claude-sonnet-4.5", nil, time.Now(), apicompat.ResponsesClientToolMapping{})
require.NoError(t, err)
require.NotNil(t, result)
require.Equal(t, 12, result.Usage.InputTokens)
@@ -87,7 +207,7 @@ func TestHandleResponsesStreamingResponse_PreservesMessageStartCacheUsage(t *tes
}
svc := &GatewayService{}
result, err := svc.handleResponsesStreamingResponse(resp, c, "claude-sonnet-4.5", "claude-sonnet-4.5", nil, time.Now())
result, err := svc.handleResponsesStreamingResponse(resp, c, "claude-sonnet-4.5", "claude-sonnet-4.5", nil, time.Now(), apicompat.ResponsesClientToolMapping{})
require.NoError(t, err)
require.NotNil(t, result)
require.Equal(t, 20, result.Usage.InputTokens)
@@ -198,7 +318,7 @@ func TestHandleResponsesBufferedStreamingResponse_CompactSSEFormat(t *testing.T)
}
svc := &GatewayService{}
result, err := svc.handleResponsesBufferedStreamingResponse(resp, c, "claude-sonnet-4.5", "claude-sonnet-4.5", nil, time.Now())
result, err := svc.handleResponsesBufferedStreamingResponse(resp, c, "claude-sonnet-4.5", "claude-sonnet-4.5", nil, time.Now(), apicompat.ResponsesClientToolMapping{})
require.NoError(t, err)
require.NotNil(t, result)
require.Equal(t, 10, result.Usage.InputTokens)
@@ -232,7 +352,7 @@ func TestHandleResponsesStreamingResponse_CompactSSEFormat(t *testing.T) {
}
svc := &GatewayService{}
result, err := svc.handleResponsesStreamingResponse(resp, c, "claude-sonnet-4.5", "claude-sonnet-4.5", nil, time.Now())
result, err := svc.handleResponsesStreamingResponse(resp, c, "claude-sonnet-4.5", "claude-sonnet-4.5", nil, time.Now(), apicompat.ResponsesClientToolMapping{})
require.NoError(t, err)
require.NotNil(t, result)
require.Equal(t, 15, result.Usage.InputTokens)