feat: route bedrock messages through mantle (#3820)

Signed-off-by: wydream <yaodiwu618@gmail.com> Signed-off-by: woody <yaodiwu618@gmail.com>
2026-05-21 03:07:27 +08:00 · 2026-05-20 18:13:04 +08:00
parent 739d47ba9c
commit 37b038f797
5 changed files with 327 additions and 21 deletions
--- a/plugins/wasm-go/extensions/ai-proxy/README.md
+++ b/plugins/wasm-go/extensions/ai-proxy/README.md
@@ -2325,6 +2325,8 @@ Vertex AI 支持的分辨率（imageSize）：`1k`、`2k`、`4k`

 ### 使用 OpenAI 协议代理 AWS Bedrock 服务

+对于 Bedrock，`/v1/chat/completions` 会继续转换为 Bedrock Runtime Converse API；`/v1/messages` 会直接转发到 Bedrock Mantle 的 Anthropic Messages API：`https://bedrock-mantle.{awsRegion}.api.aws/anthropic/v1/messages`，请求体、响应体和流式 SSE 都保持 Anthropic 原生格式，仅执行模型映射和认证处理。使用 `apiTokens` 访问 Mantle 时，插件会写入 `x-api-key` 请求头。
+
 AWS Bedrock 支持两种认证方式：

 #### 方式一：使用 AWS Access Key/Secret Key 认证（AWS Signature V4）
--- a/plugins/wasm-go/extensions/ai-proxy/README_EN.md
+++ b/plugins/wasm-go/extensions/ai-proxy/README_EN.md
@@ -2101,6 +2101,8 @@ Vertex AI supported resolutions (imageSize): `1k`, `2k`, `4k`

 ### Utilizing OpenAI Protocol Proxy for AWS Bedrock Services

+For Bedrock, `/v1/chat/completions` continues to be converted to the Bedrock Runtime Converse API. `/v1/messages` is forwarded directly to the Bedrock Mantle Anthropic Messages API: `https://bedrock-mantle.{awsRegion}.api.aws/anthropic/v1/messages`. The request body, response body, and streaming SSE keep the native Anthropic format; the plugin only applies model mapping and authentication handling. When `apiTokens` are used with Mantle, the plugin sends the token in the `x-api-key` request header.
+
 AWS Bedrock supports two authentication methods:

 #### Method 1: Using AWS Access Key/Secret Key Authentication (AWS Signature V4)
--- a/plugins/wasm-go/extensions/ai-proxy/provider/bedrock.go
+++ b/plugins/wasm-go/extensions/ai-proxy/provider/bedrock.go
@@ -23,26 +23,32 @@ import (
 	"github.com/higress-group/proxy-wasm-go-sdk/proxywasm/types"
 	"github.com/higress-group/wasm-go/pkg/log"
 	"github.com/higress-group/wasm-go/pkg/wrapper"
+	"github.com/tidwall/gjson"
+	"github.com/tidwall/sjson"
 )

 const (
-	httpPostMethod = "POST"
-	awsService     = "bedrock"
+	httpPostMethod          = "POST"
+	awsServiceBedrock       = "bedrock"
+	awsServiceBedrockMantle = "bedrock-mantle"
 	// bedrock-runtime.{awsRegion}.amazonaws.com
 	bedrockDefaultDomain = "bedrock-runtime.%s.amazonaws.com"
+	// bedrock-mantle.{awsRegion}.api.aws
+	bedrockMantleDomain = "bedrock-mantle.%s.api.aws"
 	// converse路径 /model/{modelId}/converse
 	bedrockChatCompletionPath = "/model/%s/converse"
 	// converseStream路径 /model/{modelId}/converse-stream
 	bedrockStreamChatCompletionPath = "/model/%s/converse-stream"
 	// invoke_model 路径 /model/{modelId}/invoke
-	bedrockInvokeModelPath   = "/model/%s/invoke"
-	bedrockSignedHeaders     = "host;x-amz-date"
-	requestIdHeader          = "X-Amzn-Requestid"
-	bedrockCacheTypeDefault  = "default"
-	bedrockCacheTTL5m        = "5m"
-	bedrockCacheTTL1h        = "1h"
-	bedrockPromptCacheNova   = "amazon.nova"
-	bedrockPromptCacheClaude = "anthropic.claude"
+	bedrockInvokeModelPath    = "/model/%s/invoke"
+	bedrockMantleMessagesPath = "/anthropic/v1/messages"
+	bedrockSignedHeaders      = "host;x-amz-date"
+	requestIdHeader           = "X-Amzn-Requestid"
+	bedrockCacheTypeDefault   = "default"
+	bedrockCacheTTL5m         = "5m"
+	bedrockCacheTTL1h         = "1h"
+	bedrockPromptCacheNova    = "amazon.nova"
+	bedrockPromptCacheClaude  = "anthropic.claude"

 	bedrockCachePointPositionSystemPrompt    = "systemPrompt"
 	bedrockCachePointPositionLastUserMessage = "lastUserMessage"
@@ -73,8 +79,9 @@ func (b *bedrockProviderInitializer) ValidateConfig(config *ProviderConfig) erro

 func (b *bedrockProviderInitializer) DefaultCapabilities() map[string]string {
 	return map[string]string{
-		string(ApiNameChatCompletion):  bedrockChatCompletionPath,
-		string(ApiNameImageGeneration): bedrockInvokeModelPath,
+		string(ApiNameChatCompletion):    bedrockChatCompletionPath,
+		string(ApiNameAnthropicMessages): bedrockMantleMessagesPath,
+		string(ApiNameImageGeneration):   bedrockInvokeModelPath,
 	}
 }

@@ -92,6 +99,10 @@ type bedrockProvider struct {
 }

 func (b *bedrockProvider) OnStreamingResponseBody(ctx wrapper.HttpContext, name ApiName, chunk []byte, isLastChunk bool) ([]byte, error) {
+	if name == ApiNameAnthropicMessages {
+		return chunk, nil
+	}
+
 	events := extractAmazonEventStreamEvents(ctx, chunk)
 	if len(events) == 0 {
 		if isLastChunk {
@@ -718,6 +729,18 @@ func (b *bedrockProvider) OnRequestHeaders(ctx wrapper.HttpContext, apiName ApiN
 }

 func (b *bedrockProvider) TransformRequestHeaders(ctx wrapper.HttpContext, apiName ApiName, headers http.Header) {
+	if apiName == ApiNameAnthropicMessages {
+		util.OverwriteRequestHostHeader(headers, fmt.Sprintf(bedrockMantleDomain, strings.TrimSpace(b.config.awsRegion)))
+		util.OverwriteRequestPathHeaderByCapability(headers, string(apiName), b.config.capabilities)
+		headers.Set("anthropic-version", b.anthropicVersion())
+
+		if len(b.config.apiTokens) > 0 {
+			headers.Set("x-api-key", b.config.GetApiTokenInUse(ctx))
+			headers.Del(util.HeaderAuthorization)
+		}
+		return
+	}
+
 	util.OverwriteRequestHostHeader(headers, fmt.Sprintf(bedrockDefaultDomain, strings.TrimSpace(b.config.awsRegion)))

 	// If apiTokens is configured, set Bearer token authentication here
@@ -733,7 +756,7 @@ func (b *bedrockProvider) OnRequestBody(ctx wrapper.HttpContext, apiName ApiName
 	// and only apply auth headers.
 	if b.config.IsOriginal() {
 		headers := util.GetRequestHeaders()
-		b.setAuthHeaders(body, headers)
+		b.setAuthHeaders(apiName, body, headers)
 		util.ReplaceRequestHeaders(headers)
 		return types.ActionContinue, replaceRequestBody(body)
 	}
@@ -750,6 +773,8 @@ func (b *bedrockProvider) TransformRequestBodyHeaders(ctx wrapper.HttpContext, a
 	switch apiName {
 	case ApiNameChatCompletion:
 		transformedBody, err = b.onChatCompletionRequestBody(ctx, body, headers)
+	case ApiNameAnthropicMessages:
+		transformedBody, err = b.onAnthropicMessagesRequestBody(ctx, body, headers)
 	case ApiNameImageGeneration:
 		transformedBody, err = b.onImageGenerationRequestBody(ctx, body, headers)
 	default:
@@ -762,7 +787,7 @@ func (b *bedrockProvider) TransformRequestBodyHeaders(ctx wrapper.HttpContext, a

 	// Always apply auth after request body/path are finalized.
 	// For Bearer token mode this is a no-op; for AK/SK mode this generates SigV4 headers.
-	b.setAuthHeaders(transformedBody, headers)
+	b.setAuthHeaders(apiName, transformedBody, headers)
 	return transformedBody, nil
 }

@@ -770,6 +795,8 @@ func (b *bedrockProvider) TransformResponseBody(ctx wrapper.HttpContext, apiName
 	switch apiName {
 	case ApiNameChatCompletion:
 		return b.onChatCompletionResponseBody(ctx, body)
+	case ApiNameAnthropicMessages:
+		return body, nil
 	case ApiNameImageGeneration:
 		return b.onImageGenerationResponseBody(body)
 	}
@@ -797,6 +824,28 @@ func (b *bedrockProvider) onImageGenerationRequestBody(ctx wrapper.HttpContext,
 	return b.buildBedrockImageGenerationRequest(request, headers)
 }

+func (b *bedrockProvider) onAnthropicMessagesRequestBody(ctx wrapper.HttpContext, body []byte, headers http.Header) ([]byte, error) {
+	if gjson.GetBytes(body, "stream").Bool() {
+		headers.Set("Accept", "text/event-stream")
+		ctx.SetContext(ctxKeyIsStreaming, true)
+	} else {
+		ctx.SetContext(ctxKeyIsStreaming, false)
+	}
+
+	model := gjson.GetBytes(body, "model").String()
+	if err := b.config.mapModel(ctx, &model); err != nil {
+		return nil, err
+	}
+	return sjson.SetBytes(body, "model", model)
+}
+
+func (b *bedrockProvider) anthropicVersion() string {
+	if b.config.apiVersion != "" {
+		return b.config.apiVersion
+	}
+	return claudeDefaultVersion
+}
+
 func (b *bedrockProvider) buildBedrockImageGenerationRequest(origRequest *imageGenerationRequest, headers http.Header) ([]byte, error) {
 	width, height := 1024, 1024
 	pairs := strings.Split(origRequest.Size, "x")
@@ -1587,7 +1636,7 @@ func claudeToolResultBlockToBedrock(block claudeChatMessageContent) *toolResultB
 	return result
 }

-func (b *bedrockProvider) setAuthHeaders(body []byte, headers http.Header) {
+func (b *bedrockProvider) setAuthHeaders(apiName ApiName, body []byte, headers http.Header) {
 	// Bearer token authentication is already set in TransformRequestHeaders
 	// This function only handles AWS SigV4 authentication which requires the request body
 	if len(b.config.apiTokens) > 0 {
@@ -1601,32 +1650,51 @@ func (b *bedrockProvider) setAuthHeaders(body []byte, headers http.Header) {
 	amzDate := t.Format("20060102T150405Z")
 	dateStamp := t.Format("20060102")
 	path := headers.Get(":path")
-	signature := b.generateSignature(path, amzDate, dateStamp, body)
+	service := bedrockAWSService(apiName)
+	signature := b.generateSignatureWithService(path, amzDate, dateStamp, body, service)
 	headers.Set("X-Amz-Date", amzDate)
-	util.OverwriteRequestAuthorizationHeader(headers, fmt.Sprintf("AWS4-HMAC-SHA256 Credential=%s/%s/%s/%s/aws4_request, SignedHeaders=%s, Signature=%s", accessKey, dateStamp, region, awsService, bedrockSignedHeaders, signature))
+	util.OverwriteRequestAuthorizationHeader(headers, fmt.Sprintf("AWS4-HMAC-SHA256 Credential=%s/%s/%s/%s/aws4_request, SignedHeaders=%s, Signature=%s", accessKey, dateStamp, region, service, bedrockSignedHeaders, signature))
 }

 func (b *bedrockProvider) generateSignature(path, amzDate, dateStamp string, body []byte) string {
+	return b.generateSignatureWithService(path, amzDate, dateStamp, body, awsServiceBedrock)
+}
+
+func (b *bedrockProvider) generateSignatureWithService(path, amzDate, dateStamp string, body []byte, service string) string {
 	canonicalURI := encodeSigV4Path(path)
 	hashedPayload := sha256Hex(body)
 	region := strings.TrimSpace(b.config.awsRegion)
 	secretKey := strings.TrimSpace(b.config.awsSecretKey)

-	endpoint := fmt.Sprintf(bedrockDefaultDomain, region)
+	endpoint := bedrockAWSEndpoint(service, region)
 	canonicalHeaders := fmt.Sprintf("host:%s\nx-amz-date:%s\n", endpoint, amzDate)
 	canonicalRequest := fmt.Sprintf("%s\n%s\n\n%s\n%s\n%s",
 		httpPostMethod, canonicalURI, canonicalHeaders, bedrockSignedHeaders, hashedPayload)

-	credentialScope := fmt.Sprintf("%s/%s/%s/aws4_request", dateStamp, region, awsService)
+	credentialScope := fmt.Sprintf("%s/%s/%s/aws4_request", dateStamp, region, service)
 	hashedCanonReq := sha256Hex([]byte(canonicalRequest))
 	stringToSign := fmt.Sprintf("AWS4-HMAC-SHA256\n%s\n%s\n%s",
 		amzDate, credentialScope, hashedCanonReq)

-	signingKey := getSignatureKey(secretKey, dateStamp, region, awsService)
+	signingKey := getSignatureKey(secretKey, dateStamp, region, service)
 	signature := hmacHex(signingKey, stringToSign)
 	return signature
 }

+func bedrockAWSService(apiName ApiName) string {
+	if apiName == ApiNameAnthropicMessages {
+		return awsServiceBedrockMantle
+	}
+	return awsServiceBedrock
+}
+
+func bedrockAWSEndpoint(service, region string) string {
+	if service == awsServiceBedrockMantle {
+		return fmt.Sprintf(bedrockMantleDomain, region)
+	}
+	return fmt.Sprintf(bedrockDefaultDomain, region)
+}
+
 func encodeSigV4Path(path string) string {
 	// Keep only the URI path for canonical URI. Query string is handled separately in SigV4,
 	// and this implementation uses an empty canonical query string.
--- a/plugins/wasm-go/extensions/ai-proxy/provider/provider.go
+++ b/plugins/wasm-go/extensions/ai-proxy/provider/provider.go
@@ -1166,7 +1166,10 @@ func ExtractStreamingEvents(ctx wrapper.HttpContext, chunk []byte) []StreamEvent

 func (c *ProviderConfig) isSupportedAPI(apiName ApiName) bool {
 	_, exist := c.capabilities[string(apiName)]
-	return exist
+	if exist {
+		return true
+	}
+	return c.typ == providerTypeBedrock && apiName == ApiNameAnthropicMessages
 }

 func (c *ProviderConfig) IsSupportedAPI(apiName ApiName) bool {
--- a/plugins/wasm-go/extensions/ai-proxy/test/bedrock.go
+++ b/plugins/wasm-go/extensions/ai-proxy/test/bedrock.go
@@ -11,6 +11,7 @@ import (
 	"github.com/higress-group/proxy-wasm-go-sdk/proxywasm/types"
 	"github.com/higress-group/wasm-go/pkg/test"
 	"github.com/stretchr/testify/require"
+	"github.com/tidwall/gjson"
 )

 // Test config: Basic Bedrock config with AWS Access Key/Secret Key (AWS Signature V4)
@@ -116,6 +117,37 @@ var bedrockApiTokenConfig = func() json.RawMessage {
 	return data
 }()

+var bedrockMantleApiTokenConfig = func() json.RawMessage {
+	data, _ := json.Marshal(map[string]interface{}{
+		"provider": map[string]interface{}{
+			"type": "bedrock",
+			"apiTokens": []string{
+				"test-token-for-unit-test",
+			},
+			"awsRegion": "us-east-1",
+			"modelMapping": map[string]string{
+				"*": "anthropic.claude-opus-4-7",
+			},
+		},
+	})
+	return data
+}()
+
+var bedrockMantleAkSkConfig = func() json.RawMessage {
+	data, _ := json.Marshal(map[string]interface{}{
+		"provider": map[string]interface{}{
+			"type":         "bedrock",
+			"awsAccessKey": "test-ak-for-unit-test",
+			"awsSecretKey": "test-sk-for-unit-test",
+			"awsRegion":    "us-east-1",
+			"modelMapping": map[string]string{
+				"*": "anthropic.claude-opus-4-7",
+			},
+		},
+	})
+	return data
+}()
+
 func bedrockApiTokenConfigWithCachePointPositions(positions map[string]bool) json.RawMessage {
 	data, _ := json.Marshal(map[string]interface{}{
 		"provider": map[string]interface{}{
@@ -425,6 +457,136 @@ func RunBedrockOnHttpRequestBodyTests(t *testing.T) {
 			require.Contains(t, pathValue, "/converse", "Path should contain converse endpoint")
 		})

+		t.Run("bedrock anthropic messages request should use mantle endpoint and preserve native body", func(t *testing.T) {
+			host, status := test.NewTestHost(bedrockMantleApiTokenConfig)
+			defer host.Reset()
+			require.Equal(t, types.OnPluginStartStatusOK, status)
+
+			action := host.CallOnHttpRequestHeaders([][2]string{
+				{":authority", "example.com"},
+				{":path", "/v1/messages"},
+				{":method", "POST"},
+				{"Content-Type", "application/json"},
+			})
+			require.Equal(t, types.HeaderStopIteration, action)
+
+			requestHeaders := host.GetRequestHeaders()
+			hostValue, hasHost := test.GetHeaderValue(requestHeaders, ":authority")
+			require.True(t, hasHost, "Host header should exist")
+			require.Equal(t, "bedrock-mantle.us-east-1.api.aws", hostValue)
+
+			pathValue, hasPath := test.GetHeaderValue(requestHeaders, ":path")
+			require.True(t, hasPath, "Path header should exist")
+			require.Equal(t, "/anthropic/v1/messages", pathValue)
+
+			apiKeyValue, hasAPIKey := test.GetHeaderValue(requestHeaders, "x-api-key")
+			require.True(t, hasAPIKey, "x-api-key header should exist")
+			require.Equal(t, "test-token-for-unit-test", apiKeyValue)
+
+			_, hasAuth := test.GetHeaderValue(requestHeaders, "Authorization")
+			require.False(t, hasAuth, "Authorization header should not be sent for Mantle token auth")
+
+			anthropicVersion, hasAnthropicVersion := test.GetHeaderValue(requestHeaders, "anthropic-version")
+			require.True(t, hasAnthropicVersion, "anthropic-version header should exist")
+			require.Equal(t, "2023-06-01", anthropicVersion)
+
+			requestBody := `{
+					"model": "claude-request-model",
+					"max_tokens": 1024,
+					"messages": [{
+						"role": "assistant",
+						"content": [{
+							"type": "tool_use",
+							"id": "toolu_1",
+							"name": "list_items",
+							"input": {}
+						}]
+					}]
+				}`
+
+			action = host.CallOnHttpRequestBody([]byte(requestBody))
+			require.Equal(t, types.ActionContinue, action)
+
+			processedBody := host.GetRequestBody()
+			require.Equal(t, "anthropic.claude-opus-4-7", gjson.GetBytes(processedBody, "model").String())
+			require.Equal(t, "{}", gjson.GetBytes(processedBody, "messages.0.content.0.input").Raw)
+			require.False(t, gjson.GetBytes(processedBody, "inferenceConfig").Exists(), "native Anthropic body should not be converted to Bedrock Converse")
+			require.False(t, gjson.GetBytes(processedBody, "toolConfig").Exists(), "native Anthropic tools should not be converted to Bedrock Converse")
+
+			requestHeaders = host.GetRequestHeaders()
+			pathValue, hasPath = test.GetHeaderValue(requestHeaders, ":path")
+			require.True(t, hasPath, "Path header should exist")
+			require.Equal(t, "/anthropic/v1/messages", pathValue)
+		})
+
+		t.Run("bedrock anthropic messages request with ak sk should sign for mantle service", func(t *testing.T) {
+			host, status := test.NewTestHost(bedrockMantleAkSkConfig)
+			defer host.Reset()
+			require.Equal(t, types.OnPluginStartStatusOK, status)
+
+			action := host.CallOnHttpRequestHeaders([][2]string{
+				{":authority", "example.com"},
+				{":path", "/v1/messages"},
+				{":method", "POST"},
+				{"Content-Type", "application/json"},
+			})
+			require.Equal(t, types.HeaderStopIteration, action)
+
+			requestBody := `{
+					"model": "claude-request-model",
+					"max_tokens": 1024,
+					"messages": [{"role": "user", "content": "Hello"}]
+				}`
+			action = host.CallOnHttpRequestBody([]byte(requestBody))
+			require.Equal(t, types.ActionContinue, action)
+
+			requestHeaders := host.GetRequestHeaders()
+			hostValue, hasHost := test.GetHeaderValue(requestHeaders, ":authority")
+			require.True(t, hasHost, "Host header should exist")
+			require.Equal(t, "bedrock-mantle.us-east-1.api.aws", hostValue)
+
+			pathValue, hasPath := test.GetHeaderValue(requestHeaders, ":path")
+			require.True(t, hasPath, "Path header should exist")
+			require.Equal(t, "/anthropic/v1/messages", pathValue)
+
+			authValue, hasAuth := test.GetHeaderValue(requestHeaders, "Authorization")
+			require.True(t, hasAuth, "Authorization header should exist")
+			require.Contains(t, authValue, "AWS4-HMAC-SHA256")
+			require.Contains(t, authValue, "/bedrock-mantle/aws4_request")
+		})
+
+		t.Run("bedrock anthropic messages streaming request should keep mantle path and accept sse", func(t *testing.T) {
+			host, status := test.NewTestHost(bedrockMantleApiTokenConfig)
+			defer host.Reset()
+			require.Equal(t, types.OnPluginStartStatusOK, status)
+
+			action := host.CallOnHttpRequestHeaders([][2]string{
+				{":authority", "example.com"},
+				{":path", "/v1/messages"},
+				{":method", "POST"},
+				{"Content-Type", "application/json"},
+			})
+			require.Equal(t, types.HeaderStopIteration, action)
+
+			requestBody := `{
+					"model": "claude-request-model",
+					"max_tokens": 1024,
+					"messages": [{"role": "user", "content": "Hello"}],
+					"stream": true
+				}`
+			action = host.CallOnHttpRequestBody([]byte(requestBody))
+			require.Equal(t, types.ActionContinue, action)
+
+			requestHeaders := host.GetRequestHeaders()
+			pathValue, hasPath := test.GetHeaderValue(requestHeaders, ":path")
+			require.True(t, hasPath, "Path header should exist")
+			require.Equal(t, "/anthropic/v1/messages", pathValue)
+
+			acceptValue, hasAccept := test.GetHeaderValue(requestHeaders, "Accept")
+			require.True(t, hasAccept, "Accept header should exist")
+			require.Equal(t, "text/event-stream", acceptValue)
+		})
+
 		t.Run("bedrock request body prompt cache in-memory should inject system cache point only by default", func(t *testing.T) {
 			host, status := test.NewTestHost(bedrockApiTokenConfig)
 			defer host.Reset()
@@ -1821,6 +1983,40 @@ func RunBedrockOnHttpResponseBodyTests(t *testing.T) {
 			_, hasCacheWriteTokens := promptTokensDetails["cache_write_tokens"]
 			require.False(t, hasCacheWriteTokens, "cache_write_tokens should not exist in OpenAI-compatible usage")
 		})
+
+		t.Run("bedrock anthropic messages response body should pass through", func(t *testing.T) {
+			host, status := test.NewTestHost(bedrockMantleApiTokenConfig)
+			defer host.Reset()
+			require.Equal(t, types.OnPluginStartStatusOK, status)
+
+			action := host.CallOnHttpRequestHeaders([][2]string{
+				{":authority", "example.com"},
+				{":path", "/v1/messages"},
+				{":method", "POST"},
+				{"Content-Type", "application/json"},
+			})
+			require.Equal(t, types.HeaderStopIteration, action)
+
+			requestBody := `{
+					"model": "claude-request-model",
+					"max_tokens": 1024,
+					"messages": [{"role": "user", "content": "Hello"}]
+				}`
+			action = host.CallOnHttpRequestBody([]byte(requestBody))
+			require.Equal(t, types.ActionContinue, action)
+
+			host.SetProperty([]string{"response", "code_details"}, []byte("via_upstream"))
+			action = host.CallOnHttpResponseHeaders([][2]string{
+				{":status", "200"},
+				{"Content-Type", "application/json"},
+			})
+			require.Equal(t, types.ActionContinue, action)
+
+			responseBody := `{"id":"msg_01","type":"message","role":"assistant","model":"anthropic.claude-opus-4-7","content":[{"type":"text","text":"Hello"}],"stop_reason":"end_turn","usage":{"input_tokens":10,"output_tokens":2}}`
+			action = host.CallOnHttpResponseBody([]byte(responseBody))
+			require.Equal(t, types.ActionContinue, action)
+			require.JSONEq(t, responseBody, string(host.GetResponseBody()))
+		})
 	})
 }

@@ -1840,6 +2036,41 @@ func RunBedrockOnStreamingResponseBodyTests(t *testing.T) {
 			require.Equal(t, "", payload)
 		})

+		t.Run("bedrock anthropic messages streaming response should pass through mantle sse", func(t *testing.T) {
+			host, status := test.NewTestHost(bedrockMantleApiTokenConfig)
+			defer host.Reset()
+			require.Equal(t, types.OnPluginStartStatusOK, status)
+
+			action := host.CallOnHttpRequestHeaders([][2]string{
+				{":authority", "example.com"},
+				{":path", "/v1/messages"},
+				{":method", "POST"},
+				{"Content-Type", "application/json"},
+			})
+			require.Equal(t, types.HeaderStopIteration, action)
+
+			requestBody := `{
+					"model": "claude-request-model",
+					"max_tokens": 1024,
+					"messages": [{"role": "user", "content": "Hello"}],
+					"stream": true
+				}`
+			action = host.CallOnHttpRequestBody([]byte(requestBody))
+			require.Equal(t, types.ActionContinue, action)
+
+			host.SetProperty([]string{"response", "code_details"}, []byte("via_upstream"))
+			action = host.CallOnHttpResponseHeaders([][2]string{
+				{":status", "200"},
+				{"Content-Type", "text/event-stream; charset=utf-8"},
+			})
+			require.Equal(t, types.ActionContinue, action)
+
+			chunk := []byte("event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"Hi\"}}\n\n")
+			action = host.CallOnHttpStreamingResponseBody(chunk, false)
+			require.Equal(t, types.ActionContinue, action)
+			require.Equal(t, chunk, host.GetResponseBody())
+		})
+
 		t.Run("bedrock streaming parallel tool calls should use dense OpenAI indexes", func(t *testing.T) {
 			host, status := test.NewTestHost(bedrockApiTokenConfig)
 			defer host.Reset()