feat: support OpenAI Responses API on Claude upstream channels

relay/channel/claude/adaptor.go

@@ -124,6 +124,9 @@ func (a *Adaptor) DoRequest(c *gin.Context, info *relaycommon.RelayInfo, request
 func (a *Adaptor) DoResponse(c *gin.Context, resp *http.Response, info *relaycommon.RelayInfo) (usage any, err *types.NewAPIError) {
 	info.FinalRequestRelayFormat = types.RelayFormatClaude
 	if info.IsStream {
+		if info.RelayFormat == types.RelayFormatOpenAIResponses {
+			return ClaudeResponsesStreamHandler(c, info, resp)
+		}
 		return ClaudeStreamHandler(c, resp, info)
 	} else {
 		return ClaudeHandler(c, resp, info)

relay/channel/claude/relay-claude.go

@@ -1,6 +1,7 @@
 package claude
 
 import (
+	"fmt"
 	"io"
 	"net/http"
 	"strings"
@@ -214,6 +215,105 @@ func ClaudeStreamHandler(c *gin.Context, resp *http.Response, info *relaycommon.
 	return claudeInfo.Usage, nil
 }
 
+// ClaudeResponsesStreamHandler converts an upstream Claude SSE stream into an
+// OpenAI Responses SSE stream. It is used when info.RelayFormat ==
+// types.RelayFormatOpenAIResponses, i.e. the inbound request was /v1/responses
+// routed to a Claude channel (see ResponsesHelper). Usage and billable-tool
+// accounting follow the same rules as the other Claude stream formats.
+func ClaudeResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*dto.Usage, *types.NewAPIError) {
+	defer service.CloseResponseBodyGracefully(resp)
+
+	claudeInfo := &ClaudeResponseInfo{
+		ResponseId:   helper.GetResponseID(c),
+		Created:      common.GetTimestamp(),
+		Model:        info.UpstreamModelName,
+		ResponseText: strings.Builder{},
+		Usage:        &dto.Usage{},
+	}
+	state, convErr := relayconvert.NewResponseStreamState(types.RelayFormatClaude, types.RelayFormatOpenAIResponses, relayconvert.ResponseStreamOptions{
+		ID:    claudeInfo.ResponseId,
+		Model: info.UpstreamModelName,
+	})
+	if convErr != nil {
+		return nil, types.NewError(convErr, types.ErrorCodeBadResponse)
+	}
+	streamErr := (*types.NewAPIError)(nil)
+
+	sendEvent := func(event relayconvert.ChatToResponsesStreamEvent) bool {
+		data, mErr := common.Marshal(event.Payload)
+		if mErr != nil {
+			streamErr = types.NewError(mErr, types.ErrorCodeJsonMarshalFailed)
+			return false
+		}
+		if wErr := helper.ResponseChunkData(c, dto.ResponsesStreamResponse{Type: event.Type}, string(data)); wErr != nil {
+			streamErr = types.NewError(wErr, types.ErrorCodeBadResponse)
+			return false
+		}
+		return true
+	}
+
+	helper.StreamScannerHandler(c, resp, info, func(data string, sr *helper.StreamResult) {
+		if streamErr != nil {
+			sr.Stop(streamErr)
+			return
+		}
+		var claudeResponse dto.ClaudeResponse
+		if err := common.UnmarshalJsonStr(data, &claudeResponse); err != nil {
+			common.SysLog("error unmarshalling stream response: " + err.Error())
+			sr.Error(err)
+			return
+		}
+		if claudeError := claudeResponse.GetClaudeError(); claudeError != nil && claudeError.Type != "" {
+			streamErr = types.WithClaudeError(*claudeError, http.StatusInternalServerError)
+			sr.Stop(streamErr)
+			return
+		}
+		// keep usage/billing/tool accounting consistent with the other formats
+		FormatClaudeResponseInfo(&claudeResponse, nil, claudeInfo)
+		countClaudeStreamBillableTools(c, info, &claudeResponse)
+
+		results, err := relayconvert.ConvertStreamResponseChunk(c, info, state, &claudeResponse)
+		if err != nil {
+			streamErr = types.NewError(err, types.ErrorCodeBadResponse)
+			sr.Stop(streamErr)
+			return
+		}
+		for _, result := range results {
+			event, ok := result.Value.(relayconvert.ChatToResponsesStreamEvent)
+			if !ok {
+				continue
+			}
+			if !sendEvent(event) {
+				sr.Stop(streamErr)
+				return
+			}
+		}
+	})
+	if streamErr != nil {
+		return nil, streamErr
+	}
+
+	// Emit the terminal response.completed event with the final usage. Without
+	// this the client (e.g. Codex) never sees the stream terminate.
+	if claudeInfo.Usage != nil {
+		state.SetUsage(claudeInfo.Usage)
+	}
+	finalResults, err := relayconvert.FinalizeStreamResponse(c, info, state)
+	if err != nil {
+		return nil, types.NewError(err, types.ErrorCodeBadResponse)
+	}
+	for _, result := range finalResults {
+		event, ok := result.Value.(relayconvert.ChatToResponsesStreamEvent)
+		if !ok {
+			continue
+		}
+		if !sendEvent(event) {
+			return nil, streamErr
+		}
+	}
+	return claudeInfo.Usage, nil
+}
+
 func HandleClaudeResponseData(c *gin.Context, info *relaycommon.RelayInfo, claudeInfo *ClaudeResponseInfo, httpResp *http.Response, data []byte) *types.NewAPIError {
 	var claudeResponse dto.ClaudeResponse
 	err := common.Unmarshal(data, &claudeResponse)
@@ -247,6 +347,22 @@ func HandleClaudeResponseData(c *gin.Context, info *relaycommon.RelayInfo, claud
 		if err != nil {
 			return types.NewError(err, types.ErrorCodeBadResponseBody)
 		}
+	case types.RelayFormatOpenAIResponses:
+		convertResult, convErr := relayconvert.ConvertResponse(c, info, types.RelayFormatOpenAIResponses, &claudeResponse)
+		if convErr != nil {
+			return types.NewError(convErr, types.ErrorCodeBadResponseBody)
+		}
+		responsesResp, ok := convertResult.Value.(*dto.OpenAIResponsesResponse)
+		if !ok {
+			return types.NewError(fmt.Errorf("expected OpenAI responses response, got %T", convertResult.Value), types.ErrorCodeBadResponseBody)
+		}
+		if responseID := helper.GetResponseID(c); responseID != "" {
+			responsesResp.ID = responseID
+		}
+		responseData, err = common.Marshal(responsesResp)
+		if err != nil {
+			return types.NewError(err, types.ErrorCodeBadResponseBody)
+		}
 	case types.RelayFormatClaude:
 		responseData = data
 	}

relay/channel/claude/relay_responses_test.go

@@ -0,0 +1,180 @@
+package claude
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/QuantumNous/new-api/common"
+	"github.com/QuantumNous/new-api/constant"
+	relaycommon "github.com/QuantumNous/new-api/relay/common"
+	"github.com/QuantumNous/new-api/relaykit/dto"
+	"github.com/QuantumNous/new-api/relaykit/types"
+	"github.com/gin-gonic/gin"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// Non-streaming: a Claude Messages response routed from /v1/responses must be
+// converted to OpenAI Responses JSON on the wire.
+func TestHandleClaudeResponseData_ConvertsToResponsesFormat(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil)
+
+	info := &relaycommon.RelayInfo{
+		RelayFormat: types.RelayFormatOpenAIResponses,
+		ChannelMeta: &relaycommon.ChannelMeta{UpstreamModelName: "claude-test"},
+	}
+	claudeInfo := &ClaudeResponseInfo{Usage: &dto.Usage{}}
+
+	data := []byte(`{
+		"id": "msg_123",
+		"type": "message",
+		"role": "assistant",
+		"model": "claude-test",
+		"content": [{"type": "text", "text": "ok"}],
+		"stop_reason": "end_turn",
+		"usage": {"input_tokens": 10, "output_tokens": 5}
+	}`)
+
+	httpResp := &http.Response{StatusCode: http.StatusOK, Header: http.Header{}}
+	err := HandleClaudeResponseData(c, info, claudeInfo, httpResp, data)
+	require.Nil(t, err)
+
+	var resp dto.OpenAIResponsesResponse
+	require.NoError(t, common.Unmarshal(w.Body.Bytes(), &resp), "body should be Responses JSON, got: %s", w.Body.String())
+	assert.Equal(t, "response", resp.Object)
+	assert.NotEmpty(t, resp.ID, "response ID should be set")
+
+	// converted output items: the Claude text block becomes an output_text message
+	var outputText string
+	for _, item := range resp.Output {
+		for _, content := range item.Content {
+			if content.Type == "output_text" {
+				outputText += content.Text
+			}
+		}
+	}
+	assert.Equal(t, "ok", outputText, "Claude text should be converted to output_text")
+
+	// usage: input 10 + output 5 = 15 total tokens
+	require.NotNil(t, resp.Usage, "usage should be populated")
+	assert.Equal(t, 10, resp.Usage.PromptTokens)
+	assert.Equal(t, 5, resp.Usage.CompletionTokens)
+	assert.Equal(t, 15, resp.Usage.TotalTokens)
+}
+
+// Streaming: a Claude SSE stream routed from /v1/responses must be converted to
+// OpenAI Responses SSE events.
+func TestClaudeResponsesStreamHandler_EmitsResponsesSSE(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	oldStreamingTimeout := constant.StreamingTimeout
+	constant.StreamingTimeout = 300
+	t.Cleanup(func() { constant.StreamingTimeout = oldStreamingTimeout })
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil)
+
+	sse := "event: message_start\n" +
+		"data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_1\",\"model\":\"claude-test\",\"usage\":{\"input_tokens\":10,\"output_tokens\":0}}}\n\n" +
+		"event: content_block_start\n" +
+		"data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n" +
+		"event: content_block_delta\n" +
+		"data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"ok\"}}\n\n" +
+		"event: content_block_stop\n" +
+		"data: {\"type\":\"content_block_stop\",\"index\":0}\n\n" +
+		"event: message_delta\n" +
+		"data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"},\"usage\":{\"input_tokens\":10,\"output_tokens\":5}}\n\n" +
+		"event: message_stop\n" +
+		"data: {\"type\":\"message_stop\"}\n\n"
+
+	info := &relaycommon.RelayInfo{
+		RelayFormat: types.RelayFormatOpenAIResponses,
+		IsStream:    true,
+		ChannelMeta: &relaycommon.ChannelMeta{UpstreamModelName: "claude-test"},
+	}
+	httpResp := &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     http.Header{},
+		Body:       io.NopCloser(strings.NewReader(sse)),
+	}
+
+	usage, err := ClaudeResponsesStreamHandler(c, info, httpResp)
+	require.Nil(t, err)
+	require.NotNil(t, usage)
+
+	// Parse the emitted Responses SSE stream and assert the converted contract:
+	// a response.created carrying the response ID, output_text deltas that join to
+	// the Claude text, and a terminal response.completed with final usage. Event
+	// positions are recorded so ordering can be asserted as well.
+	var (
+		createdID    string
+		deltaText    strings.Builder
+		finalUsage   *dto.Usage
+		createdIndex = -1
+		deltaIndex   = -1
+		doneIndex    = -1
+	)
+	eventIndex := 0
+	for _, block := range strings.Split(w.Body.String(), "\n\n") {
+		var eventType, dataLine string
+		for _, line := range strings.Split(block, "\n") {
+			if rest, ok := strings.CutPrefix(line, "event: "); ok {
+				eventType = rest
+			}
+			if rest, ok := strings.CutPrefix(line, "data: "); ok {
+				dataLine = rest
+			}
+		}
+		if dataLine == "" {
+			continue
+		}
+		var payload map[string]any
+		require.NoError(t, common.Unmarshal([]byte(dataLine), &payload), "SSE data should be JSON: %s", dataLine)
+
+		switch eventType {
+		case "response.created":
+			if createdIndex == -1 {
+				createdIndex = eventIndex
+			}
+			if respObj, ok := payload["response"].(map[string]any); ok {
+				createdID, _ = respObj["id"].(string)
+			}
+		case "response.output_text.delta":
+			deltaIndex = eventIndex
+			if d, ok := payload["delta"].(string); ok {
+				deltaText.WriteString(d)
+			}
+		case "response.completed":
+			doneIndex = eventIndex
+			if respObj, ok := payload["response"].(map[string]any); ok {
+				if u, ok := respObj["usage"].(map[string]any); ok {
+					b, mErr := common.Marshal(u)
+					require.NoError(t, mErr)
+					var uu dto.Usage
+					require.NoError(t, common.Unmarshal(b, &uu))
+					finalUsage = &uu
+				}
+			}
+		}
+		eventIndex++
+	}
+
+	assert.NotEqual(t, -1, createdIndex, "should emit response.created")
+	assert.NotEmpty(t, createdID, "response.created should carry the response ID")
+	assert.NotEqual(t, -1, deltaIndex, "should emit response.output_text.delta")
+	assert.Equal(t, "ok", deltaText.String(), "text deltas should join to the Claude text")
+	assert.NotEqual(t, -1, doneIndex, "should emit a terminal response.completed")
+	require.NotNil(t, finalUsage, "response.completed should carry usage")
+	assert.Equal(t, 10, finalUsage.PromptTokens)
+	assert.Equal(t, 5, finalUsage.CompletionTokens)
+	assert.Equal(t, 15, finalUsage.TotalTokens)
+
+	// ordering: created precedes the output deltas; completed follows them
+	assert.Less(t, createdIndex, deltaIndex, "response.created should precede output events")
+	assert.Greater(t, doneIndex, deltaIndex, "response.completed should follow output events")
+}

relay/responses_handler.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 
 	"github.com/QuantumNous/new-api/common"
+	"github.com/QuantumNous/new-api/constant"
 	"github.com/QuantumNous/new-api/logger"
 	relaycommon "github.com/QuantumNous/new-api/relay/common"
 	relayconstant "github.com/QuantumNous/new-api/relay/constant"
@@ -77,21 +78,47 @@ func ResponsesHelper(c *gin.Context, info *relaycommon.RelayInfo) (newAPIError *
 	}
 	adaptor.Init(info)
 	var requestBody io.Reader
-	if model_setting.GetGlobalSettings().PassThroughRequestEnabled || info.ChannelSetting.PassThroughBodyEnabled {
+	// Claude (Anthropic) channels must never take the pass-through path: a raw
+	// /v1/responses body is not valid for the Claude Messages endpoint, so the
+	// request always has to go through the conversion branch below.
+	if info.ApiType != constant.APITypeAnthropic &&
+		(model_setting.GetGlobalSettings().PassThroughRequestEnabled || info.ChannelSetting.PassThroughBodyEnabled) {
 		storage, err := common.GetBodyStorage(c)
 		if err != nil {
 			return types.NewError(err, types.ErrorCodeReadRequestBodyFailed, types.ErrOptionWithSkipRetry())
 		}
 		requestBody = common.NewReplayableBodyReader(storage)
 	} else {
-		convertedRequest, err := adaptor.ConvertOpenAIResponsesRequest(c, info, *request)
-		if err != nil {
-			return types.NewError(err, types.ErrorCodeConvertRequestFailed, types.ErrOptionWithSkipRetry())
-		}
-		relaycommon.AppendRequestConversionFromRequest(info, convertedRequest)
-		jsonData, err := common.Marshal(convertedRequest)
-		if err != nil {
-			return types.NewError(err, types.ErrorCodeConvertRequestFailed, types.ErrOptionWithSkipRetry())
+		var jsonData []byte
+		if info.ApiType == constant.APITypeAnthropic {
+			// Claude upstream: convert the Responses request to Claude Messages and
+			// forward it to /v1/messages. info.RelayFormat stays
+			// RelayFormatOpenAIResponses so the Claude response pipeline
+			// (HandleClaudeResponseData / ClaudeResponsesStreamHandler) converts the
+			// upstream response back to OpenAI Responses format.
+			result, convErr := service.ConvertRequest(c, info, types.RelayFormatClaude, request)
+			if convErr != nil {
+				return types.NewError(convErr, types.ErrorCodeConvertRequestFailed, types.ErrOptionWithSkipRetry())
+			}
+			claudeRequest, ok := result.Value.(*dto.ClaudeRequest)
+			if !ok {
+				return types.NewError(fmt.Errorf("expected Claude request, got %T", result.Value), types.ErrorCodeConvertRequestFailed, types.ErrOptionWithSkipRetry())
+			}
+			relaycommon.AppendRequestConversionFromRequest(info, claudeRequest)
+			jsonData, err = common.Marshal(claudeRequest)
+			if err != nil {
+				return types.NewError(err, types.ErrorCodeConvertRequestFailed, types.ErrOptionWithSkipRetry())
+			}
+		} else {
+			convertedRequest, err := adaptor.ConvertOpenAIResponsesRequest(c, info, *request)
+			if err != nil {
+				return types.NewError(err, types.ErrorCodeConvertRequestFailed, types.ErrOptionWithSkipRetry())
+			}
+			relaycommon.AppendRequestConversionFromRequest(info, convertedRequest)
+			jsonData, err = common.Marshal(convertedRequest)
+			if err != nil {
+				return types.NewError(err, types.ErrorCodeConvertRequestFailed, types.ErrOptionWithSkipRetry())
+			}
 		}
 
 		// remove disabled fields for OpenAI Responses API