scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-07 02:59:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

Add Bedrock Claude Platform route (#27678)

Browse source 
* Add Claude Platform AWS Bedrock route

Co-authored-by: ishaan-berri <ishaan-berri@users.noreply.github.com>

* Use Bedrock Claude Platform route

Co-authored-by: ishaan-berri <ishaan-berri@users.noreply.github.com>

* Move Claude Platform route under Bedrock

Co-authored-by: ishaan-berri <ishaan-berri@users.noreply.github.com>

* Split Claude Platform messages config

Co-authored-by: ishaan-berri <ishaan-berri@users.noreply.github.com>

* Centralize Claude Platform Bedrock route

Co-authored-by: ishaan-berri <ishaan-berri@users.noreply.github.com>

* Address Claude Platform review feedback

Co-authored-by: ishaan-berri <ishaan-berri@users.noreply.github.com>

---------

Co-authored-by: oss-agent-shin <279349115+oss-agent-shin@users.noreply.github.com>
Co-authored-by: ishaan-berri <ishaan-berri@users.noreply.github.com>
This commit is contained in:
oss-agent-shin 2026-05-11 15:50:54 -07:00 • committed by GitHub
parent be84d5cd7d
commit 473cfca969
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 688 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
litellm/__init__.py
View file

@ -1426,6 +1426,12 @@ if TYPE_CHECKING:
)
from .llms.datarobot.chat.transformation import DataRobotConfig as DataRobotConfig
from .llms.anthropic.chat.transformation import AnthropicConfig as AnthropicConfig
from .llms.bedrock.claude_platform.transformation import (
BedrockClaudePlatformConfig as BedrockClaudePlatformConfig,
)
from .llms.bedrock.claude_platform.messages_transformation import (
BedrockClaudePlatformMessagesConfig as BedrockClaudePlatformMessagesConfig,
)
from .llms.anthropic.completion.transformation import (
AnthropicTextConfig as AnthropicTextConfig,
)

10
litellm/_lazy_imports_registry.py
View file

@ -131,6 +131,7 @@ LLM_CONFIG_NAMES = (
"OpenrouterConfig",
"DataRobotConfig",
"AnthropicConfig",
"BedrockClaudePlatformConfig",
"AnthropicTextConfig",
"GroqSTTConfig",
"TritonConfig",
@ -170,6 +171,7 @@ LLM_CONFIG_NAMES = (
"SagemakerNovaConfig",
"CohereChatConfig",
"AnthropicMessagesConfig",
"BedrockClaudePlatformMessagesConfig",
"AmazonAnthropicClaudeMessagesConfig",
"AmazonMantleMessagesConfig",
"TogetherAIConfig",
@ -610,6 +612,10 @@ _LLM_CONFIGS_IMPORT_MAP = {
"OpenrouterConfig": (".llms.openrouter.chat.transformation", "OpenrouterConfig"),
"DataRobotConfig": (".llms.datarobot.chat.transformation", "DataRobotConfig"),
"AnthropicConfig": (".llms.anthropic.chat.transformation", "AnthropicConfig"),
"BedrockClaudePlatformConfig": (
".llms.bedrock.claude_platform.transformation",
"BedrockClaudePlatformConfig",
),
"AnthropicTextConfig": (
".llms.anthropic.completion.transformation",
"AnthropicTextConfig",
@ -712,6 +718,10 @@ _LLM_CONFIGS_IMPORT_MAP = {
".llms.anthropic.experimental_pass_through.messages.transformation",
"AnthropicMessagesConfig",
),
"BedrockClaudePlatformMessagesConfig": (
".llms.bedrock.claude_platform.messages_transformation",
"BedrockClaudePlatformMessagesConfig",
),
"AmazonAnthropicClaudeMessagesConfig": (
".llms.bedrock.messages.invoke_transformations.anthropic_claude3_transformation",
"AmazonAnthropicClaudeMessagesConfig",

8
litellm/llms/bedrock/base_aws_llm.py
View file

@ -1428,7 +1428,13 @@ class BaseAWSLLM:
def _sign_request(
self,
service_name: Literal["bedrock", "sagemaker", "bedrock-agentcore", "s3vectors"],
service_name: Literal[
"bedrock",
"sagemaker",
"bedrock-agentcore",
"s3vectors",
"aws-external-anthropic",
],
headers: dict,
optional_params: dict,
request_data: dict,

8
litellm/llms/bedrock/claude_platform/__init__.py Normal file
View file

@ -0,0 +1,8 @@
from .transformation import (
BedrockClaudePlatformConfig,
)
from .messages_transformation import (
BedrockClaudePlatformMessagesConfig,
)
__all__ = ["BedrockClaudePlatformConfig", "BedrockClaudePlatformMessagesConfig"]

107
litellm/llms/bedrock/claude_platform/common_utils.py Normal file
View file

@ -0,0 +1,107 @@
from typing import Literal, Optional, Tuple
import litellm
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
from litellm.secret_managers.main import get_secret_str
CLAUDE_PLATFORM_SERVICE_NAME: Literal["aws-external-anthropic"] = (
"aws-external-anthropic"
)
CLAUDE_PLATFORM_BEDROCK_ROUTE = "claude_platform/"
def strip_claude_platform_route(model: str) -> str:
if model.startswith(CLAUDE_PLATFORM_BEDROCK_ROUTE):
return model.replace(CLAUDE_PLATFORM_BEDROCK_ROUTE, "", 1)
return model
class BedrockClaudePlatformMixin(BaseAWSLLM):
@staticmethod
def _get_workspace_id(optional_params: dict, litellm_params: dict) -> Optional[str]:
workspace_id = (
optional_params.get("workspace_id")
or litellm_params.get("workspace_id")
or optional_params.get("aws_workspace_id")
or litellm_params.get("aws_workspace_id")
or optional_params.get("anthropic-workspace-id")
or litellm_params.get("anthropic-workspace-id")
)
if workspace_id is None:
workspace_id = optional_params.get(
"anthropic_workspace_id"
) or litellm_params.get("anthropic_workspace_id")
if workspace_id is not None:
return str(workspace_id)
return get_secret_str("ANTHROPIC_AWS_WORKSPACE_ID") or get_secret_str(
"ANTHROPIC_WORKSPACE_ID"
)
def _get_required_aws_region_name(self, optional_params: dict) -> str:
aws_region_name = (
optional_params.get("aws_region_name")
or get_secret_str("AWS_REGION_NAME")
or get_secret_str("AWS_REGION")
or get_secret_str("AWS_DEFAULT_REGION")
)
if aws_region_name is None:
raise litellm.AuthenticationError(
message=(
"Missing AWS region for Claude Platform on AWS. Pass "
"`aws_region_name` or set a standard AWS region environment value."
),
llm_provider="bedrock",
model="",
)
self._validate_aws_region_name(str(aws_region_name))
return str(aws_region_name)
def get_complete_url(
self,
api_base: Optional[str],
api_key: Optional[str],
model: str,
optional_params: dict,
litellm_params: dict,
stream: Optional[bool] = None,
) -> str:
api_base = (
api_base
or litellm.api_base
or get_secret_str("ANTHROPIC_AWS_BASE_URL")
or get_secret_str("ANTHROPIC_AWS_API_BASE")
)
if api_base is None:
aws_region_name = self._get_required_aws_region_name(optional_params)
api_base = (
f"https://{CLAUDE_PLATFORM_SERVICE_NAME}.{aws_region_name}.api.aws"
)
if not api_base.endswith("/v1/messages"):
api_base = f"{api_base.rstrip('/')}/v1/messages"
return api_base
def sign_request(
self,
headers: dict,
optional_params: dict,
request_data: dict,
api_base: str,
api_key: Optional[str] = None,
model: Optional[str] = None,
stream: Optional[bool] = None,
fake_stream: Optional[bool] = None,
) -> Tuple[dict, Optional[bytes]]:
if api_key or get_secret_str("ANTHROPIC_AWS_API_KEY"):
return headers, None
return self._sign_request(
service_name=CLAUDE_PLATFORM_SERVICE_NAME,
headers=headers,
optional_params=optional_params,
request_data=request_data,
api_base=api_base,
model=model,
stream=stream,
fake_stream=fake_stream,
)

71
litellm/llms/bedrock/claude_platform/messages_transformation.py Normal file
View file

@ -0,0 +1,71 @@
from typing import Any, Dict, List, Optional, Tuple
import litellm
from litellm.llms.anthropic.experimental_pass_through.messages.transformation import (
DEFAULT_ANTHROPIC_API_VERSION,
AnthropicMessagesConfig,
)
from litellm.secret_managers.main import get_secret_str
from litellm.types.router import GenericLiteLLMParams
from .common_utils import BedrockClaudePlatformMixin, strip_claude_platform_route
class BedrockClaudePlatformMessagesConfig(
BedrockClaudePlatformMixin, AnthropicMessagesConfig
):
def validate_anthropic_messages_environment(
self,
headers: dict,
model: str,
messages: List[Any],
optional_params: dict,
litellm_params: dict,
api_key: Optional[str] = None,
api_base: Optional[str] = None,
) -> Tuple[dict, Optional[str]]:
workspace_id = self._get_workspace_id(optional_params, litellm_params)
if workspace_id is None:
raise litellm.AuthenticationError(
message=(
"Missing workspace ID for Claude Platform on AWS. Pass "
"`workspace_id` or configure the provider workspace setting."
),
llm_provider="bedrock",
model=model,
)
resolved_api_key = api_key or get_secret_str("ANTHROPIC_AWS_API_KEY")
headers = {
**headers,
"anthropic-version": headers.get(
"anthropic-version", DEFAULT_ANTHROPIC_API_VERSION
),
"content-type": headers.get("content-type", "application/json"),
"anthropic-workspace-id": workspace_id,
}
if resolved_api_key and "x-api-key" not in headers:
headers["x-api-key"] = resolved_api_key
headers = self._update_headers_with_anthropic_beta(
headers=headers,
optional_params=optional_params,
)
return headers, api_base
def transform_anthropic_messages_request(
self,
model: str,
messages: List[Dict],
anthropic_messages_optional_request_params: Dict,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> Dict:
return super().transform_anthropic_messages_request(
model=strip_claude_platform_route(model),
messages=messages,
anthropic_messages_optional_request_params=anthropic_messages_optional_request_params,
litellm_params=litellm_params,
headers=headers,
)

94
litellm/llms/bedrock/claude_platform/transformation.py Normal file
View file

@ -0,0 +1,94 @@
from typing import Any, Dict, List, Optional
import litellm
from litellm.llms.anthropic.chat.transformation import AnthropicConfig
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.openai import AllMessageValues
from .common_utils import BedrockClaudePlatformMixin
class BedrockClaudePlatformConfig(BedrockClaudePlatformMixin, AnthropicConfig):
"""
Bedrock Claude Platform uses Anthropic's Messages API with AWS gateway auth.
"""
@property
def custom_llm_provider(self) -> Optional[str]:
return "bedrock"
def validate_environment(
self,
headers: dict,
model: str,
messages: List[AllMessageValues],
optional_params: dict,
litellm_params: dict,
api_key: Optional[str] = None,
api_base: Optional[str] = None,
) -> Dict:
workspace_id = self._get_workspace_id(optional_params, litellm_params)
if workspace_id is None:
raise litellm.AuthenticationError(
message=(
"Missing workspace ID for Claude Platform on AWS. Pass "
"`workspace_id` or configure the provider workspace setting."
),
llm_provider="bedrock",
model=model,
)
api_key = api_key or get_secret_str("ANTHROPIC_AWS_API_KEY")
anthropic_headers = self.get_anthropic_headers(
api_key=api_key,
auth_token=None,
computer_tool_used=self.is_computer_tool_used(
tools=optional_params.get("tools")
),
prompt_caching_set=self.is_cache_control_set(messages=messages),
pdf_used=self.is_pdf_used(messages=messages),
file_id_used=self.is_file_id_used(messages=messages),
mcp_server_used=self.is_mcp_server_used(
mcp_servers=optional_params.get("mcp_servers")
),
web_search_tool_used=self.is_web_search_tool_used(
tools=optional_params.get("tools")
),
tool_search_used=self.is_tool_search_used(
tools=optional_params.get("tools")
),
programmatic_tool_calling_used=self.is_programmatic_tool_calling_used(
tools=optional_params.get("tools")
),
input_examples_used=self.is_input_examples_used(
tools=optional_params.get("tools")
),
effort_used=self.is_effort_used(
optional_params=optional_params, model=model
),
user_anthropic_beta_headers=self._get_user_anthropic_beta_headers(
anthropic_beta_header=headers.get("anthropic-beta")
),
code_execution_tool_used=self.is_code_execution_tool_used(
tools=optional_params.get("tools")
),
container_with_skills_used=self.is_container_with_skills_used(
optional_params=optional_params
),
)
anthropic_headers["anthropic-workspace-id"] = workspace_id
return {**headers, **anthropic_headers}
def get_model_response_iterator(
self,
streaming_response: Any,
sync_stream: bool,
json_mode: Optional[bool] = False,
) -> Any:
from litellm.llms.anthropic.chat.handler import ModelResponseIterator
return ModelResponseIterator(
streaming_response=streaming_response,
sync_stream=sync_stream,
json_mode=bool(json_mode),
)

43
litellm/llms/bedrock/common_utils.py
View file

@ -692,6 +692,7 @@ class BedrockModelInfo(BaseLLMModelInfo):
) -> Literal[
"converse",
"invoke",
"claude_platform",
"converse_like",
"agent",
"agentcore",
@ -706,6 +707,7 @@ class BedrockModelInfo(BaseLLMModelInfo):
str,
Literal[
"invoke",
"claude_platform",
"converse_like",
"converse",
"agent",
@ -716,6 +718,7 @@ class BedrockModelInfo(BaseLLMModelInfo):
],
] = {
"invoke/": "invoke",
"claude_platform/": "claude_platform",
"converse_like/": "converse_like",
"converse/": "converse",
"agent/": "agent",
@ -753,6 +756,36 @@ class BedrockModelInfo(BaseLLMModelInfo):
"""
return "converse/" in model
@staticmethod
def _explicit_claude_platform_route(model: str) -> bool:
"""
Check if the model is an explicit Claude Platform on AWS route.
"""
return "claude_platform/" in model
@staticmethod
def get_claude_platform_model(model: str) -> str:
"""
Strip the Claude Platform route prefix from a Bedrock model name.
"""
return model.replace("claude_platform/", "", 1)
@staticmethod
def map_claude_platform_auth_params(
passed_params: dict, optional_params: dict
) -> dict:
"""
Map Claude Platform route auth params that are not OpenAI request params.
"""
for key in (
"workspace_id",
"aws_workspace_id",
"anthropic_workspace_id",
):
if key in passed_params:
optional_params[key] = passed_params[key]
return optional_params
@staticmethod
def _explicit_invoke_route(model: str) -> bool:
"""
@ -815,6 +848,12 @@ class BedrockModelInfo(BaseLLMModelInfo):
All other routes should return None since they will go through litellm.completion
"""
#########################################################
# Claude Platform route uses Anthropic Messages API via the AWS gateway.
#########################################################
if BedrockModelInfo._explicit_claude_platform_route(model):
return litellm.BedrockClaudePlatformMessagesConfig()
#########################################################
# Converse routes should go through litellm.completion()
if BedrockModelInfo._explicit_converse_route(model):
@ -860,7 +899,9 @@ def get_bedrock_chat_config(model: str):
base_model = BedrockModelInfo.get_base_model(model)
# Handle explicit routes first
if bedrock_route == "converse" or bedrock_route == "converse_like":
if bedrock_route == "claude_platform":
return litellm.BedrockClaudePlatformConfig()
elif bedrock_route == "converse" or bedrock_route == "converse_like":
return litellm.AmazonConverseConfig()
elif bedrock_route == "openai":
return litellm.AmazonBedrockOpenAIConfig()

28
litellm/main.py
View file

@ -3858,7 +3858,33 @@ def completion( # type: ignore # noqa: PLR0915
)
bedrock_route = BedrockModelInfo.get_bedrock_route(model)
if bedrock_route == "converse":
if bedrock_route == "claude_platform":
provider_config = ProviderConfigManager.get_provider_chat_config(
model=model,
provider=LlmProviders.BEDROCK,
)
model = BedrockModelInfo.get_claude_platform_model(model)
response = base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
custom_llm_provider="bedrock",
timeout=timeout,
headers=headers,
encoding=_get_encoding(),
api_key=api_key,
logging_obj=logging,
client=client,
provider_config=provider_config,
)
return response
elif bedrock_route == "converse":
model = model.replace("converse/", "")
response = bedrock_converse_chat_completion.completion(
model=model,

4
litellm/utils.py
View file

@ -4407,6 +4407,10 @@ def get_optional_params( # noqa: PLR0915
else False
),
)
if bedrock_route == "claude_platform":
optional_params = BedrockModelInfo.map_claude_platform_auth_params(
passed_params=passed_params, optional_params=optional_params
)
elif custom_llm_provider == "cloudflare":
optional_params = litellm.CloudflareChatConfig().map_openai_params(
model=model,

312
tests/test_litellm/llms/bedrock/test_claude_platform_provider.py Normal file
View file

@ -0,0 +1,312 @@
import json
from unittest.mock import patch
import httpx
import pytest
def _anthropic_response(url: str) -> httpx.Response:
return httpx.Response(
status_code=200,
json={
"id": "msg_test",
"type": "message",
"role": "assistant",
"model": "claude-sonnet-4-6",
"content": [{"type": "text", "text": "ok"}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {"input_tokens": 1, "output_tokens": 1},
},
request=httpx.Request("POST", url),
)
def _capture_request(url: str, headers: dict, data: bytes | str | None) -> dict:
raw_body = data.decode("utf-8") if isinstance(data, bytes) else data or "{}"
return {
"path": httpx.URL(url).path,
"headers": headers,
"body": json.loads(raw_body),
}
def test_claude_platform_builds_default_messages_url_from_region():
from litellm.llms.bedrock.claude_platform.transformation import (
BedrockClaudePlatformConfig,
)
config = BedrockClaudePlatformConfig()
assert (
config.get_complete_url(
api_base=None,
api_key=None,
model="claude-sonnet-4-6",
optional_params={"aws_region_name": "us-west-2"},
litellm_params={},
)
== "https://aws-external-anthropic.us-west-2.api.aws/v1/messages"
)
def test_claude_platform_ignores_standard_anthropic_base_url(monkeypatch):
from litellm.llms.bedrock.claude_platform.transformation import (
BedrockClaudePlatformConfig,
)
monkeypatch.setenv("ANTHROPIC_BASE_URL", "https://api.anthropic.example")
monkeypatch.setenv("ANTHROPIC_API_BASE", "https://api.anthropic-api.example")
config = BedrockClaudePlatformConfig()
assert (
config.get_complete_url(
api_base=None,
api_key=None,
model="claude-sonnet-4-6",
optional_params={"aws_region_name": "us-west-2"},
litellm_params={},
)
== "https://aws-external-anthropic.us-west-2.api.aws/v1/messages"
)
def test_claude_platform_uses_bedrock_subroute():
import litellm
from litellm.llms.bedrock.common_utils import BedrockModelInfo
model, provider, _, _ = litellm.get_llm_provider(
model="bedrock/claude_platform/claude-sonnet-4-6"
)
assert provider == "bedrock"
assert model == "claude_platform/claude-sonnet-4-6"
assert BedrockModelInfo.get_bedrock_route(model) == "claude_platform"
assert BedrockModelInfo.get_claude_platform_model(model) == "claude-sonnet-4-6"
def test_claude_platform_requires_workspace_header():
from litellm import AuthenticationError
from litellm.llms.bedrock.claude_platform.transformation import (
BedrockClaudePlatformConfig,
)
config = BedrockClaudePlatformConfig()
with pytest.raises(AuthenticationError) as exc_info:
config.validate_environment(
api_key="fake-platform-key",
headers={},
model="claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
optional_params={},
litellm_params={},
)
assert "workspace" in str(exc_info.value).lower()
def test_claude_platform_api_key_auth_sets_workspace_and_key_headers():
from litellm.llms.bedrock.claude_platform.transformation import (
BedrockClaudePlatformConfig,
)
config = BedrockClaudePlatformConfig()
headers = config.validate_environment(
api_key="fake-platform-key",
headers={"anthropic-beta": "skills-2025-10-02"},
model="claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
optional_params={"workspace_id": "wrkspc_test"},
litellm_params={},
)
assert headers["x-api-key"] == "fake-platform-key"
assert headers["anthropic-workspace-id"] == "wrkspc_test"
assert headers["anthropic-beta"] == "skills-2025-10-02"
def test_claude_platform_does_not_use_standard_anthropic_api_key(monkeypatch):
from litellm.llms.bedrock.claude_platform.transformation import (
BedrockClaudePlatformConfig,
)
monkeypatch.setenv("ANTHROPIC_API_KEY", "standard-anthropic-key")
config = BedrockClaudePlatformConfig()
headers = config.validate_environment(
api_key=None,
headers={},
model="claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
optional_params={"workspace_id": "wrkspc_test"},
litellm_params={},
)
assert "x-api-key" not in headers
def test_claude_platform_sigv4_signs_transformed_request_body():
from litellm.llms.bedrock.claude_platform.transformation import (
BedrockClaudePlatformConfig,
)
config = BedrockClaudePlatformConfig()
request_body = {
"model": "claude-sonnet-4-6",
"messages": [{"role": "user", "content": "hello"}],
"max_tokens": 10,
}
with patch.object(
config,
"_sign_request",
return_value=({"Authorization": "signed"}, json.dumps(request_body).encode()),
) as mock_sign_request:
headers, signed_body = config.sign_request(
headers={"anthropic-workspace-id": "wrkspc_test"},
optional_params={"aws_region_name": "us-west-2"},
request_data=request_body,
api_base="https://aws-external-anthropic.us-west-2.api.aws/v1/messages",
api_key=None,
model="claude-sonnet-4-6",
)
assert signed_body == json.dumps(request_body).encode()
assert headers["Authorization"] == "signed"
mock_sign_request.assert_called_once()
assert (
mock_sign_request.call_args.kwargs["service_name"] == "aws-external-anthropic"
)
assert mock_sign_request.call_args.kwargs["request_data"] == request_body
def test_claude_platform_standard_anthropic_api_key_does_not_skip_sigv4(monkeypatch):
from litellm.llms.bedrock.claude_platform.transformation import (
BedrockClaudePlatformConfig,
)
monkeypatch.setenv("ANTHROPIC_API_KEY", "standard-anthropic-key")
config = BedrockClaudePlatformConfig()
request_body = {
"model": "claude-sonnet-4-6",
"messages": [{"role": "user", "content": "hello"}],
"max_tokens": 10,
}
with patch.object(
config,
"_sign_request",
return_value=({"Authorization": "signed"}, json.dumps(request_body).encode()),
) as mock_sign_request:
headers, signed_body = config.sign_request(
headers={"anthropic-workspace-id": "wrkspc_test"},
optional_params={"aws_region_name": "us-west-2"},
request_data=request_body,
api_base="https://aws-external-anthropic.us-west-2.api.aws/v1/messages",
api_key=None,
model="claude-sonnet-4-6",
)
assert signed_body == json.dumps(request_body).encode()
assert headers["Authorization"] == "signed"
mock_sign_request.assert_called_once()
def test_bedrock_claude_platform_messages_config_round_trips_native_body():
import litellm
from litellm.types.utils import LlmProviders
config = litellm.ProviderConfigManager.get_provider_anthropic_messages_config(
model="claude_platform/claude-sonnet-4-6",
provider=LlmProviders.BEDROCK,
)
assert config is not None
headers, _ = config.validate_anthropic_messages_environment(
api_key="fake-platform-key",
headers={},
model="claude_platform/claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
optional_params={"max_tokens": 10},
litellm_params={"workspace_id": "wrkspc_test"},
)
request_body = config.transform_anthropic_messages_request(
model="claude_platform/claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
anthropic_messages_optional_request_params={"max_tokens": 10},
litellm_params={},
headers=headers,
)
assert headers["anthropic-workspace-id"] == "wrkspc_test"
assert headers["x-api-key"] == "fake-platform-key"
assert request_body == {
"model": "claude-sonnet-4-6",
"messages": [{"role": "user", "content": "hello"}],
"max_tokens": 10,
}
def test_chat_completion_routes_bedrock_claude_platform_to_messages_api():
import litellm
requests = []
def mock_post(self, url, data=None, headers=None, **kwargs):
requests.append(_capture_request(url=url, headers=headers or {}, data=data))
return _anthropic_response(url)
with patch("litellm.llms.custom_httpx.http_handler.HTTPHandler.post", mock_post):
response = litellm.completion(
model="bedrock/claude_platform/claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
max_tokens=10,
api_base="https://aws-external-anthropic.us-west-2.api.aws",
api_key="fake-platform-key",
workspace_id="wrkspc_test",
)
assert response.choices[0].message.content == "ok"
assert len(requests) == 1
assert requests[0]["path"] == "/v1/messages"
assert requests[0]["headers"]["x-api-key"] == "fake-platform-key"
assert requests[0]["headers"]["anthropic-workspace-id"] == "wrkspc_test"
assert requests[0]["body"]["model"] == "claude-sonnet-4-6"
@pytest.mark.asyncio
async def test_anthropic_messages_routes_bedrock_claude_platform_to_messages_api():
import litellm
requests = []
async def mock_post(self, url, data=None, headers=None, **kwargs):
requests.append(_capture_request(url=url, headers=headers or {}, data=data))
return _anthropic_response(url)
try:
with patch(
"litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
new=mock_post,
):
response = await litellm.anthropic_messages(
model="bedrock/claude_platform/claude-sonnet-4-6",
messages=[{"role": "user", "content": "hello"}],
max_tokens=10,
api_base="https://aws-external-anthropic.us-west-2.api.aws",
api_key="fake-platform-key",
workspace_id="wrkspc_test",
)
finally:
await litellm.close_litellm_async_clients()
assert response["content"][0]["text"] == "ok"
assert len(requests) == 1
assert requests[0]["path"] == "/v1/messages"
assert requests[0]["headers"]["x-api-key"] == "fake-platform-key"
assert requests[0]["headers"]["anthropic-workspace-id"] == "wrkspc_test"
assert requests[0]["body"]["messages"] == [{"role": "user", "content": "hello"}]
assert requests[0]["body"]["max_tokens"] == 10
assert requests[0]["body"]["model"] == "claude-sonnet-4-6"