feat(zai): native Anthropic Messages and Responses API passthrough

This commit is contained in:
togear 2026-09-07 18:36:40 +08:00
parent eeb7732fc1
commit c96fe090fd
7 changed files with 285 additions and 0 deletions

View file

@ -0,0 +1 @@

View file

@ -0,0 +1,85 @@
"""
Z.AI Anthropic-compatible messages transformation config.
"""
from typing import Any, Final
import litellm
from litellm.llms.anthropic.experimental_pass_through.messages.transformation import (
AnthropicMessagesConfig,
)
from litellm.secret_managers.main import get_secret_str
class ZAIAnthropicMessagesConfig(AnthropicMessagesConfig):
"""
Z.AI exposes an Anthropic-compatible Messages API at
https://api.z.ai/api/anthropic (see
https://docs.z.ai/guides/llm/glm-5.3).
The endpoint accepts the native Anthropic Messages conversation shape
and authenticates with the Z.AI API key sent as the Anthropic
``x-api-key`` header.
"""
@property
def custom_llm_provider(self) -> str | None:
return "zai"
@staticmethod
def get_api_key(api_key: str | None = None) -> str | None:
return api_key or get_secret_str("ZAI_API_KEY") or litellm.api_key
@staticmethod
def get_api_base(api_base: str | None = None) -> str:
return api_base or get_secret_str("ZAI_ANTHROPIC_API_BASE") or "https://api.z.ai/api/anthropic"
def validate_anthropic_messages_environment(
self,
headers: dict,
model: str,
messages: list[Any],
optional_params: dict,
litellm_params: dict,
api_key: str | None = None,
api_base: str | None = None,
) -> tuple[dict, str | None]:
dynamic_api_key: Final = self.get_api_key(api_key=api_key)
if "x-api-key" not in headers and "authorization" not in headers and dynamic_api_key is not None:
headers["x-api-key"] = dynamic_api_key
if "anthropic-version" not in headers:
headers["anthropic-version"] = "2023-06-01"
if "content-type" not in headers:
headers["content-type"] = "application/json"
headers = self._update_headers_with_anthropic_beta(
headers=headers,
optional_params=optional_params,
custom_llm_provider=self.custom_llm_provider or "zai",
)
return headers, api_base
def get_complete_url(
self,
api_base: str | None,
api_key: str | None,
model: str,
optional_params: dict,
litellm_params: dict,
stream: bool | None = None,
) -> str:
base_url = self.get_api_base(api_base=api_base).rstrip("/")
if base_url.endswith("/v1/messages"):
return base_url
base_url = base_url.removesuffix("/v1/messages")
base_url = base_url.removesuffix("/v1")
base_url = base_url.removesuffix("/beta")
if not base_url.endswith("/anthropic") and "/anthropic/" not in base_url:
base_url = f"{base_url}/anthropic"
return f"{base_url}/v1/messages"

View file

@ -0,0 +1 @@

View file

@ -0,0 +1,52 @@
"""
Z.AI OpenAI-compatible Responses API transformation config.
"""
from typing import Final
import litellm
from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
from litellm.secret_managers.main import get_secret_str
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import LlmProviders
class ZAIResponsesAPIConfig(OpenAIResponsesAPIConfig):
"""
Z.AI exposes an OpenAI-compatible Responses API at
https://api.z.ai/api/v1 (see https://docs.z.ai/guides/llm/glm-5.3).
The endpoint authenticates with the Z.AI API key sent as an
``Authorization: Bearer`` header.
"""
@property
def custom_llm_provider(self) -> LlmProviders:
return LlmProviders.ZAI
def validate_environment(
self,
headers: dict,
model: str,
litellm_params: GenericLiteLLMParams | None,
) -> dict:
litellm_params = litellm_params or GenericLiteLLMParams()
api_key: Final = litellm_params.api_key or litellm.api_key or get_secret_str("ZAI_API_KEY")
headers.setdefault("Content-Type", "application/json")
if api_key is not None:
headers["Authorization"] = f"Bearer {api_key}"
return headers
def get_complete_url(
self,
api_base: str | None,
litellm_params: dict,
) -> str:
base_url = api_base or get_secret_str("ZAI_RESPONSES_API_BASE") or "https://api.z.ai/api/v1"
base_url = base_url.rstrip("/")
if base_url.endswith("/responses"):
return base_url
return f"{base_url}/responses"

View file

@ -8482,6 +8482,12 @@ class ProviderConfigManager:
)
return TencentAnthropicMessagesConfig()
elif litellm.LlmProviders.ZAI == provider:
from litellm.llms.zai.messages.transformation import (
ZAIAnthropicMessagesConfig,
)
return ZAIAnthropicMessagesConfig()
elif litellm.LlmProviders.GITHUB_COPILOT == provider:
if "claude" in model_lower:
from litellm.llms.github_copilot.messages.transformation import (
@ -8703,6 +8709,12 @@ class ProviderConfigManager:
return litellm.BedrockMantleResponsesAPIConfig(
use_openai_path=mantle_base_segment(model, litellm.model_cost) == "openai/v1"
)
elif litellm.LlmProviders.ZAI == provider:
from litellm.llms.zai.responses.transformation import (
ZAIResponsesAPIConfig,
)
return ZAIResponsesAPIConfig()
return None
@staticmethod

View file

@ -0,0 +1,77 @@
import litellm
from litellm.llms.anthropic.experimental_pass_through.messages.transformation import (
AnthropicMessagesConfig,
)
from litellm.llms.zai.messages.transformation import ZAIAnthropicMessagesConfig
from litellm.utils import ProviderConfigManager
def test_zai_provider_uses_anthropic_messages_config():
config = ProviderConfigManager.get_provider_anthropic_messages_config(
model="glm-5.3",
provider=litellm.LlmProviders.ZAI,
)
assert isinstance(config, ZAIAnthropicMessagesConfig)
assert config.custom_llm_provider == "zai"
def test_anthropic_provider_keeps_default_config_for_zai_named_model():
config = ProviderConfigManager.get_provider_anthropic_messages_config(
model="glm-5.3",
provider=litellm.LlmProviders.ANTHROPIC,
)
assert isinstance(config, AnthropicMessagesConfig)
assert not isinstance(config, ZAIAnthropicMessagesConfig)
def test_zai_anthropic_messages_config_defaults(monkeypatch):
monkeypatch.delenv("ZAI_ANTHROPIC_API_BASE", raising=False)
config = ZAIAnthropicMessagesConfig()
assert config.custom_llm_provider == "zai"
assert config.get_api_base() == "https://api.z.ai/api/anthropic"
def test_zai_anthropic_messages_url_defaults_to_anthropic_endpoint():
config = ZAIAnthropicMessagesConfig()
url_cases = {
None: "https://api.z.ai/api/anthropic/v1/messages",
"https://api.z.ai/api/anthropic": "https://api.z.ai/api/anthropic/v1/messages",
"https://api.z.ai/api/anthropic/v1": "https://api.z.ai/api/anthropic/v1/messages",
"https://api.z.ai/api/anthropic/v1/messages": "https://api.z.ai/api/anthropic/v1/messages",
"https://api.z.ai/api": "https://api.z.ai/api/anthropic/v1/messages",
}
for api_base, expected_url in url_cases.items():
assert (
config.get_complete_url(
api_base=api_base,
api_key=None,
model="glm-5.3",
optional_params={},
litellm_params={},
)
== expected_url
)
def test_zai_anthropic_messages_headers_use_zai_key():
config = ZAIAnthropicMessagesConfig()
headers, api_base = config.validate_anthropic_messages_environment(
headers={},
model="glm-5.3",
messages=[],
optional_params={},
litellm_params={},
api_key="sk-zai",
api_base="https://example.test/anthropic",
)
assert api_base == "https://example.test/anthropic"
assert headers["x-api-key"] == "sk-zai"
assert headers["anthropic-version"] == "2023-06-01"
assert headers["content-type"] == "application/json"

View file

@ -0,0 +1,57 @@
import litellm
from litellm.llms.zai.responses.transformation import ZAIResponsesAPIConfig
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import LlmProviders
from litellm.utils import ProviderConfigManager
def test_zai_provider_uses_responses_api_config():
config = ProviderConfigManager.get_provider_responses_api_config(
model="glm-5.3",
provider=litellm.LlmProviders.ZAI,
)
assert isinstance(config, ZAIResponsesAPIConfig)
assert config.custom_llm_provider == LlmProviders.ZAI
def test_zai_responses_url_defaults_to_responses_endpoint(monkeypatch):
monkeypatch.delenv("ZAI_RESPONSES_API_BASE", raising=False)
config = ZAIResponsesAPIConfig()
url_cases = {
None: "https://api.z.ai/api/v1/responses",
"https://api.z.ai/api/v1": "https://api.z.ai/api/v1/responses",
"https://api.z.ai/api/v1/": "https://api.z.ai/api/v1/responses",
"https://api.z.ai/api/v1/responses": "https://api.z.ai/api/v1/responses",
}
for api_base, expected_url in url_cases.items():
assert config.get_complete_url(api_base=api_base, litellm_params={}) == expected_url
def test_zai_responses_headers_use_bearer_token():
config = ZAIResponsesAPIConfig()
litellm_params = GenericLiteLLMParams(api_key="sk-zai")
headers = config.validate_environment(
headers={},
model="glm-5.3",
litellm_params=litellm_params,
)
assert headers["Authorization"] == "Bearer sk-zai"
assert headers["Content-Type"] == "application/json"
def test_zai_responses_headers_fall_back_to_environment_key(monkeypatch):
monkeypatch.setenv("ZAI_API_KEY", "sk-zai-env")
config = ZAIResponsesAPIConfig()
headers = config.validate_environment(
headers={},
model="glm-5.3",
litellm_params=GenericLiteLLMParams(),
)
assert headers["Authorization"] == "Bearer sk-zai-env"