Skip to content

refactor: standardize request body handling and error messages across providers #730

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Pratham-Mishra04 wants to merge 1 commit into from
Closed

refactor: standardize request body handling and error messages across providers #730

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
136 changes: 77 additions & 59 deletions core/providers/anthropic.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,13 +138,7 @@ func parseStreamAnthropicError(resp *http.Response, providerType schemas.ModelPr
// completeRequest sends a request to Anthropic's API and handles the response.
// It constructs the API URL, sets up authentication, and processes the response.
// Returns the response body or an error if the request fails.
func (provider *AnthropicProvider) completeRequest(ctx context.Context, requestBody interface{}, url string, key string) ([]byte, time.Duration, *schemas.BifrostError) {
// Marshal the request body
jsonData, err := sonic.Marshal(requestBody)
if err != nil {
return nil, 0, newBifrostOperationError(schemas.ErrProviderJSONMarshaling, err, provider.GetProviderKey())
}

func (provider *AnthropicProvider) completeRequest(ctx context.Context, jsonData []byte, url string, key string) ([]byte, time.Duration, *schemas.BifrostError) {
// Create the request with the JSON body
req := fasthttp.AcquireRequest()
resp := fasthttp.AcquireResponse()
Expand Down Expand Up @@ -184,11 +178,16 @@ func (provider *AnthropicProvider) completeRequest(ctx context.Context, requestB
return nil, latency, bifrostErr
}

respBody, err := checkAndDecodeBody(resp)
if err != nil {
return nil, latency, newBifrostOperationError(schemas.ErrProviderResponseUnmarshal, err, provider.GetProviderKey())
}

// Read the response body and copy it before releasing the response
// to avoid use-after-free since resp.Body() references fasthttp's internal buffer
bodyCopy := append([]byte(nil), resp.Body()...)
// bodyCopy := append([]byte(nil), resp.Body()...)

return bodyCopy, latency, nil
return respBody, latency, nil
}
Comment on lines +181 to 191

@coderabbitai coderabbitai Bot Nov 2, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

Restore the defensive body copy before releasing the fasthttp response.

checkAndDecodeBody returns the slice from fasthttp.Response.Body() for non-gzipped payloads. Once the deferred fasthttp.ReleaseResponse(resp) runs, that buffer is recycled, so propagating respBody upward causes use-after-free corruption in every caller (TextCompletion, ChatCompletion, Responses, etc.). Bring back the copy that existed here previously.

 	respBody, err := checkAndDecodeBody(resp)
 	if err != nil {
 		return nil, latency, newBifrostOperationError(schemas.ErrProviderResponseUnmarshal, err, provider.GetProviderKey())
 	}
 
-	// Read the response body and copy it before releasing the response
-	// to avoid use-after-free since resp.Body() references fasthttp's internal buffer
-
-	return respBody, latency, nil
+	// Read the response body and copy it before releasing the response
+	// to avoid use-after-free since resp.Body() references fasthttp's internal buffer
+	bodyCopy := append([]byte(nil), respBody...)
+	return bodyCopy, latency, nil
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
respBody, err := checkAndDecodeBody(resp)
if err != nil {
return nil, latency, newBifrostOperationError(schemas.ErrProviderResponseUnmarshal, err, provider.GetProviderKey())
}
// Read the response body and copy it before releasing the response
// to avoid use-after-free since resp.Body() references fasthttp's internal buffer
bodyCopy := append([]byte(nil), resp.Body()...)
// bodyCopy := append([]byte(nil), resp.Body()...)
return bodyCopy, latency, nil
return respBody, latency, nil
}
respBody, err := checkAndDecodeBody(resp)
if err != nil {
return nil, latency, newBifrostOperationError(schemas.ErrProviderResponseUnmarshal, err, provider.GetProviderKey())
}
// Read the response body and copy it before releasing the response
// to avoid use-after-free since resp.Body() references fasthttp's internal buffer
bodyCopy := append([]byte(nil), respBody...)
return bodyCopy, latency, nil
}
🤖 Prompt for AI Agents 
In core/providers/anthropic.go around lines 181 to 191, restore a defensive copy
of the response body returned by checkAndDecodeBody before releasing the
fasthttp response to avoid use-after-free: after getting respBody and before
calling fasthttp.ReleaseResponse(resp) (or allowing its deferred call to run),
create a copy (e.g. append([]byte(nil), respBody...)) and return that copy (and
not the slice that points into resp's internal buffer); ensure this is done for
the non-gzipped path so callers receive a safe, independently allocated byte
slice.


// ListModels performs a list models request to Anthropic's API.
Expand All @@ -213,7 +212,9 @@ func (provider *AnthropicProvider) ListModels(ctx context.Context, key schemas.K
req.SetRequestURI(requestURL)
req.Header.SetMethod(http.MethodGet)
req.Header.SetContentType("application/json")
req.Header.Set("x-api-key", key.Value)
if key.Value != "" {
req.Header.Set("x-api-key", key.Value)
}
req.Header.Set("anthropic-version", provider.apiVersion)

// Make request
Expand All @@ -224,8 +225,6 @@ func (provider *AnthropicProvider) ListModels(ctx context.Context, key schemas.K

// Handle error response
if resp.StatusCode() != fasthttp.StatusOK {
provider.logger.Debug(fmt.Sprintf("error from %s provider: %s", provider.GetProviderKey(), string(resp.Body())))

var errorResp anthropic.AnthropicError

bifrostErr := handleProviderAPIError(resp, &errorResp)
Expand All @@ -235,9 +234,14 @@ func (provider *AnthropicProvider) ListModels(ctx context.Context, key schemas.K
return nil, bifrostErr
}

body, err := checkAndDecodeBody(resp)
if err != nil {
return nil, newBifrostOperationError(schemas.ErrProviderResponseUnmarshal, err, provider.GetProviderKey())
}

// Parse Anthropic's response
var anthropicResponse anthropic.AnthropicListModelsResponse
rawResponse, bifrostErr := handleProviderResponse(resp.Body(), &anthropicResponse, provider.sendBackRawResponse)
rawResponse, bifrostErr := handleProviderResponse(body, &anthropicResponse, provider.sendBackRawResponse)
if bifrostErr != nil {
return nil, bifrostErr
}
Expand Down Expand Up @@ -267,9 +271,11 @@ func (provider *AnthropicProvider) TextCompletion(ctx context.Context, key schem
}

// Convert to Anthropic format using the centralized converter
reqBody := anthropic.ToAnthropicTextCompletionRequest(request)
if reqBody == nil {
return nil, newBifrostOperationError("text completion input is not provided", nil, provider.GetProviderKey())
jsonData, err := checkContextAndGetRequestBody(ctx,
func() (any, error) { return anthropic.ToAnthropicTextCompletionRequest(request), nil },
provider.GetProviderKey())
if err != nil {
return nil, err
}

// Use struct directly for JSON marshaling
Expand Down Expand Up @@ -307,7 +313,7 @@ func (provider *AnthropicProvider) TextCompletion(ctx context.Context, key schem
// It formats the request, sends it to Anthropic, and processes the response.
// Returns a channel of BifrostStream objects or an error if the request fails.
func (provider *AnthropicProvider) TextCompletionStream(ctx context.Context, postHookRunner schemas.PostHookRunner, key schemas.Key, request *schemas.BifrostTextCompletionRequest) (chan *schemas.BifrostStream, *schemas.BifrostError) {
return nil, newUnsupportedOperationError("text completion stream", "anthropic")
return nil, newUnsupportedOperationError(schemas.TextCompletionStreamRequest, provider.GetProviderKey())
}

// ChatCompletion performs a chat completion request to Anthropic's API.
Expand All @@ -319,9 +325,11 @@ func (provider *AnthropicProvider) ChatCompletion(ctx context.Context, key schem
}

// Convert to Anthropic format using the centralized converter
reqBody := anthropic.ToAnthropicChatCompletionRequest(request)
if reqBody == nil {
return nil, newBifrostOperationError("chat completion input is not provided", nil, provider.GetProviderKey())
jsonData, err := checkContextAndGetRequestBody(ctx,
func() (any, error) { return anthropic.ToAnthropicChatCompletionRequest(request), nil },
provider.GetProviderKey())
if err != nil {
return nil, err
}

// Use struct directly for JSON marshaling
Expand Down Expand Up @@ -365,27 +373,36 @@ func (provider *AnthropicProvider) ChatCompletionStream(ctx context.Context, pos
}

// Convert to Anthropic format using the centralized converter
reqBody := anthropic.ToAnthropicChatCompletionRequest(request)
if reqBody == nil {
return nil, newBifrostOperationError("failed to convert request", fmt.Errorf("conversion returned nil"), provider.GetProviderKey())
jsonData, err := checkContextAndGetRequestBody(ctx,
func() (any, error) {
reqBody := anthropic.ToAnthropicChatCompletionRequest(request)
if reqBody != nil {
reqBody.Stream = schemas.Ptr(true)
}
return reqBody, nil
},
provider.GetProviderKey())
if err != nil {
return nil, err
}
reqBody.Stream = schemas.Ptr(true)

// Prepare Anthropic headers
headers := map[string]string{
"Content-Type": "application/json",
"x-api-key": key.Value,
"anthropic-version": provider.apiVersion,
"Accept": "text/event-stream",
"Cache-Control": "no-cache",
}
if key.Value != "" {
headers["x-api-key"] = key.Value
}

// Use shared Anthropic streaming logic
return handleAnthropicChatCompletionStreaming(
ctx,
provider.streamClient,
provider.networkConfig.BaseURL+getPathFromContext(ctx, "/v1/messages"),
reqBody,
jsonData,
headers,
provider.networkConfig.ExtraHeaders,
provider.sendBackRawResponse,
Expand All @@ -401,21 +418,16 @@ func handleAnthropicChatCompletionStreaming(
ctx context.Context,
httpClient *http.Client,
url string,
requestBody interface{},
jsonData []byte,
headers map[string]string,
extraHeaders map[string]string,
sendBackRawResponse bool,
providerType schemas.ModelProvider,
postHookRunner schemas.PostHookRunner,
logger schemas.Logger,
) (chan *schemas.BifrostStream, *schemas.BifrostError) {
jsonBody, err := sonic.Marshal(requestBody)
if err != nil {
return nil, newBifrostOperationError(schemas.ErrProviderJSONMarshaling, err, providerType)
}

// Create HTTP request for streaming
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(jsonBody))
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(jsonData))
if err != nil {
if errors.Is(err, context.Canceled) {
return nil, &schemas.BifrostError{
Expand Down Expand Up @@ -598,9 +610,11 @@ func (provider *AnthropicProvider) Responses(ctx context.Context, key schemas.Ke
}

// Convert to Anthropic format using the centralized converter
reqBody := anthropic.ToAnthropicResponsesRequest(request)
if reqBody == nil {
return nil, newBifrostOperationError("responses input is not provided", nil, provider.GetProviderKey())
jsonData, err := checkContextAndGetRequestBody(ctx,
func() (any, error) { return anthropic.ToAnthropicResponsesRequest(request), nil },
provider.GetProviderKey())
if err != nil {
return nil, err
}

// Use struct directly for JSON marshaling
Expand Down Expand Up @@ -642,24 +656,17 @@ func (provider *AnthropicProvider) ResponsesStream(ctx context.Context, postHook
}

// Convert to Anthropic format using the centralized converter
reqBody := anthropic.ToAnthropicResponsesRequest(request)
if reqBody == nil {
return nil, newBifrostOperationError("failed to convert request", fmt.Errorf("conversion returned nil"), provider.GetProviderKey())
}
reqBody.Stream = schemas.Ptr(true)

// Prepare Anthropic headers
headers := map[string]string{
"Content-Type": "application/json",
"x-api-key": key.Value,
"anthropic-version": provider.apiVersion,
"Accept": "text/event-stream",
"Cache-Control": "no-cache",
}

jsonBody, err := sonic.Marshal(reqBody)
if err != nil {
return nil, newBifrostOperationError(schemas.ErrProviderJSONMarshaling, err, provider.GetProviderKey())
jsonData, bifrostErr := checkContextAndGetRequestBody(ctx,
func() (any, error) {
reqBody := anthropic.ToAnthropicResponsesRequest(request)
if reqBody != nil {
reqBody.Stream = schemas.Ptr(true)
}
return reqBody, nil
},
provider.GetProviderKey())
if bifrostErr != nil {
return nil, bifrostErr
}

// Create HTTP request for streaming
Expand All @@ -681,6 +688,17 @@ func (provider *AnthropicProvider) ResponsesStream(ctx context.Context, postHook
return nil, newBifrostOperationError(schemas.ErrProviderRequest, err, provider.GetProviderKey())
}

// Prepare Anthropic headers
headers := map[string]string{
"Content-Type": "application/json",
"anthropic-version": provider.apiVersion,
"Accept": "text/event-stream",
"Cache-Control": "no-cache",
}
if key.Value != "" {
headers["x-api-key"] = key.Value
}

// Set headers
for key, value := range headers {
req.Header.Set(key, value)
Expand Down Expand Up @@ -833,25 +851,25 @@ func (provider *AnthropicProvider) ResponsesStream(ctx context.Context, postHook

// Embedding is not supported by the Anthropic provider.
func (provider *AnthropicProvider) Embedding(ctx context.Context, key schemas.Key, input *schemas.BifrostEmbeddingRequest) (*schemas.BifrostEmbeddingResponse, *schemas.BifrostError) {
return nil, newUnsupportedOperationError("embedding", "anthropic")
return nil, newUnsupportedOperationError(schemas.EmbeddingRequest, provider.GetProviderKey())
}

// Speech is not supported by the Anthropic provider.
func (provider *AnthropicProvider) Speech(ctx context.Context, key schemas.Key, request *schemas.BifrostSpeechRequest) (*schemas.BifrostSpeechResponse, *schemas.BifrostError) {
return nil, newUnsupportedOperationError("speech", "anthropic")
return nil, newUnsupportedOperationError(schemas.SpeechRequest, provider.GetProviderKey())
}

// SpeechStream is not supported by the Anthropic provider.
func (provider *AnthropicProvider) SpeechStream(ctx context.Context, postHookRunner schemas.PostHookRunner, key schemas.Key, request *schemas.BifrostSpeechRequest) (chan *schemas.BifrostStream, *schemas.BifrostError) {
return nil, newUnsupportedOperationError("speech stream", "anthropic")
return nil, newUnsupportedOperationError(schemas.SpeechStreamRequest, provider.GetProviderKey())
}

// Transcription is not supported by the Anthropic provider.
func (provider *AnthropicProvider) Transcription(ctx context.Context, key schemas.Key, request *schemas.BifrostTranscriptionRequest) (*schemas.BifrostTranscriptionResponse, *schemas.BifrostError) {
return nil, newUnsupportedOperationError("transcription", "anthropic")
return nil, newUnsupportedOperationError(schemas.TranscriptionRequest, provider.GetProviderKey())
}

// TranscriptionStream is not supported by the Anthropic provider.
func (provider *AnthropicProvider) TranscriptionStream(ctx context.Context, postHookRunner schemas.PostHookRunner, key schemas.Key, request *schemas.BifrostTranscriptionRequest) (chan *schemas.BifrostStream, *schemas.BifrostError) {
return nil, newUnsupportedOperationError("transcription stream", "anthropic")
return nil, newUnsupportedOperationError(schemas.TranscriptionStreamRequest, provider.GetProviderKey())
}
Loading