scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-08 03:08:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 115

Merge remote-tracking branch 'origin/main' into litellm_lens_rust_isolation

Browse source
This commit is contained in:
moe-berri 2026-10-07 11:36:10 -07:00
commit 8cf78a13fd
274 changed files with 8167 additions and 527 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

37
litellm-rust/crates/cache/src/semantic.rs vendored
View file

@ -83,17 +83,16 @@ impl Embedder for PreparedEmbedding {
}
}
/// `get_str_from_messages`: every message's text content followed by its search results.
/// `get_semantic_cache_prompt_from_messages`: every message's text content, including the text of
/// Messages API `tool_result` blocks, followed by its search results.
pub fn str_from_messages(messages: &[Value]) -> String {
let mut text = String::new();
for message in messages.iter().filter_map(Value::as_object) {
match message.get("content") {
Some(Value::String(content)) => text.push_str(content),
Some(Value::Array(parts)) => {
for part in parts {
if let Some(part_text) = part.get("text").and_then(Value::as_str) {
text.push_str(part_text);
}
Some(Value::Array(blocks)) => {
for block in blocks {
push_block_text(&mut text, block);
}
}
_ => {}
@ -103,6 +102,28 @@ pub fn str_from_messages(messages: &[Value]) -> String {
text
}
fn push_block_text(text: &mut String, block: &Value) {
if block.get("type").and_then(Value::as_str) != Some("tool_result") {
push_text_field(text, block);
return;
}
match block.get("content") {
Some(Value::String(result)) => text.push_str(result),
Some(Value::Array(blocks)) => {
for inner in blocks {
push_text_field(text, inner);
}
}
_ => {}
}
}
fn push_text_field(text: &mut String, block: &Value) {
if let Some(block_text) = block.get("text").and_then(Value::as_str) {
text.push_str(block_text);
}
}
/// The messages prompt Qdrant embeds: `None` when the request carries no messages.
pub fn prompt_from_messages(context: &SemanticCacheContext) -> Option<String> {
let messages = context.messages.as_ref()?.as_array()?;
@ -163,6 +184,10 @@ fn collect_input_text(value: &Value, parts: &mut Vec<String>) {
collect_input_text(content, parts);
return;
}
if let Some(output) = map.get("output").filter(|output| output.is_array()) {
collect_input_text(output, parts);
return;
}
for key in ["text", "output", "input_text", "output_text"] {
if let Some(Value::String(text)) = map.get(key)
&& push_trimmed(text, parts)

34
litellm-rust/crates/cache/tests/semantic.rs vendored
View file

@ -30,6 +30,31 @@ fn context(messages: Option<Value>, input: Option<Value>) -> SemanticCacheContex
]}]),
"What is this?",
)]
#[case::tool_result_string(
json!([
{"role": "user", "content": "list the files"},
{"role": "assistant", "content": [
{"type": "tool_use", "id": "toolu_1", "name": "Bash", "input": {"command": "ls"}},
]},
{"role": "user", "content": [
{"type": "tool_result", "tool_use_id": "toolu_1", "content": "calc.py test_calc.py"},
]},
]),
"list the filescalc.py test_calc.py",
)]
#[case::tool_result_blocks(
json!([{"role": "user", "content": [
{"type": "tool_result", "tool_use_id": "toolu_1", "content": [
{"type": "text", "text": "x = 1"},
{"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": ""}},
]},
]}]),
"x = 1",
)]
#[case::tool_result_without_content(
json!([{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "toolu_1"}]}]),
"",
)]
#[case::missing_null_and_empty_content(
json!([{"role": "assistant"}, {"role": "assistant", "content": null}, {"role": "user", "content": ""}]),
"",
@ -166,6 +191,15 @@ fn prompt_from_messages_reads_messages_only(
])),
Some("model dump prompt\ndict prompt\ninline prompt"),
)]
#[case::function_call_output_blocks(
None,
Some(json!([
{"role": "user", "content": "update the config"},
{"type": "function_call", "call_id": "c1", "name": "write_file", "arguments": "{\"path\": \"a\"}"},
{"type": "function_call_output", "call_id": "c1", "output": [{"type": "input_text", "text": "wrote a"}]},
])),
Some("update the config\nwrote a"),
)]
#[case::object_content(
None,
Some(json!({"content": [{"text": "object content prompt"}]})),

17
litellm/caching/caching.py
View file

@ -71,6 +71,17 @@ class CacheMode(str, Enum):
#### LiteLLM.Completion / Embedding Cache ####
def _request_message_count(kwargs: Mapping[str, object]) -> int:
"""Chat and Messages API `messages`, else Responses API `input` items; embedding `input` strings count as none"""
messages: Final = kwargs.get("messages")
if isinstance(messages, list):
return len(messages)
input_items: Final = kwargs.get("input")
if not isinstance(input_items, list):
return 0
return sum(1 for item in input_items if isinstance(item, (Mapping, BaseModel)))
class Cache:
def __init__(
self,
@ -119,6 +130,7 @@ class Cache:
semantic_cache_embedding_max_input_tokens: int | None = None,
semantic_cache_embedding_timeout: float | None = None,
semantic_cache_scope: str = SemanticCacheScope.KEY.value,
max_messages: int | None = 4,
# GCP IAM authentication parameters
gcp_service_account: str | None = None,
gcp_ssl_ca_certs: str | None = None,
@ -148,6 +160,7 @@ class Cache:
semantic_cache_embedding_max_input_tokens (int, optional): Truncate prompts to this many tokens before embedding them for semantic caching. Defaults to the embedding deployment's configured max_input_tokens.
semantic_cache_embedding_timeout (float, optional): Seconds a semantic-cache lookup may spend embedding the prompt before it gives up and lets the request continue to the LLM. Defaults to SEMANTIC_CACHE_EMBEDDING_TIMEOUT_SECONDS.
semantic_cache_scope (str, optional): "key" isolates semantic-cache buckets per key/team/org. "end_user" additionally isolates per end user (falls back to the key scope when the request carries no end-user id). Defaults to "key".
max_messages (int, optional): Requests with more `messages` (or Responses API `input` items) than this are neither looked up nor stored, so long agent conversations never serve or create a cache entry. None disables the limit. Defaults to 4.
# Disk Cache Args
disk_cache_dir (str, optional): The directory for the disk cache. Defaults to None.
@ -298,6 +311,7 @@ class Cache:
self.ttl = ttl
self.mode: CacheMode = mode or CacheMode.default_on
self.semantic_cache_scope: str = SemanticCacheScope(semantic_cache_scope).value
self.max_messages: int | None = max_messages
if self.type == LiteLLMCacheType.LOCAL and default_in_memory_ttl is not None:
self.ttl = default_in_memory_ttl
@ -933,7 +947,10 @@ class Cache:
If cache is default_on then this is True
If cache is default_off then this is only true when user has opted in to use cache
Always False once the request carries more than `max_messages` messages
"""
if self.max_messages is not None and _request_message_count(kwargs) > self.max_messages:
return False
if self.mode == CacheMode.default_on:
return True

10
litellm/caching/qdrant_semantic_cache.py
View file

@ -24,7 +24,7 @@ from litellm.constants import (
)
from litellm.litellm_core_utils.asyncify import asyncify
from litellm.litellm_core_utils.prompt_templates.common_utils import (
get_str_from_messages,
get_semantic_cache_prompt_from_messages,
)
from litellm.types.utils import EmbeddingResponse
@ -286,7 +286,7 @@ class QdrantSemanticCache(BaseCache):
# get the prompt
messages: Final = kwargs["messages"]
prompt: Final = get_str_from_messages(messages)
prompt: Final = get_semantic_cache_prompt_from_messages(messages)
# create an embedding for prompt
embedding_response: Final = cast(
@ -325,7 +325,7 @@ class QdrantSemanticCache(BaseCache):
# get the messages
messages: Final = kwargs["messages"]
prompt: Final = get_str_from_messages(messages)
prompt: Final = get_semantic_cache_prompt_from_messages(messages)
# convert to embedding
embedding_response: Final = cast(
@ -400,7 +400,7 @@ class QdrantSemanticCache(BaseCache):
# get the prompt
messages: Final = kwargs["messages"]
prompt: Final = get_str_from_messages(messages)
prompt: Final = get_semantic_cache_prompt_from_messages(messages)
embedding_response: Final = await self._get_async_embedding(prompt, metadata=kwargs.get("metadata"))
# get the embedding
@ -435,7 +435,7 @@ class QdrantSemanticCache(BaseCache):
# get the messages
messages: Final = kwargs["messages"]
prompt: Final = get_str_from_messages(messages)
prompt: Final = get_semantic_cache_prompt_from_messages(messages)
embedding_response: Final = await self._get_async_embedding(prompt, metadata=kwargs.get("metadata"))

11
litellm/caching/redis_semantic_cache.py
View file

@ -21,7 +21,7 @@ from litellm._logging import print_verbose, verbose_logger
from litellm.constants import SEMANTIC_CACHE_EMBEDDING_TIMEOUT_SECONDS
from litellm.litellm_core_utils.asyncify import asyncify
from litellm.litellm_core_utils.prompt_templates.common_utils import (
get_str_from_messages,
get_semantic_cache_prompt_from_messages,
)
from litellm.types.utils import EmbeddingResponse
@ -263,7 +263,7 @@ class RedisSemanticCache(BaseCache):
"""
messages: Final = kwargs.get("messages")
if messages:
return get_str_from_messages(messages)
return get_semantic_cache_prompt_from_messages(messages)
if "input" not in kwargs:
return None
@ -274,7 +274,7 @@ class RedisSemanticCache(BaseCache):
return prompt or None
@classmethod
def _collect_responses_input_text(cls, value: object, prompt_parts: list[str]) -> None:
def _collect_responses_input_text(cls, value: object, prompt_parts: list[str]) -> None: # noqa: C901 # one branch per Responses input shape
value = cls._coerce_response_input_value(value)
if value is None:
return
@ -296,6 +296,11 @@ class RedisSemanticCache(BaseCache):
cls._collect_responses_input_text(content, prompt_parts)
return
output = value.get("output")
if isinstance(output, list):
cls._collect_responses_input_text(output, prompt_parts)
return
for text_key in ("text", "output", "input_text", "output_text"):
text_value = value.get(text_key)
if isinstance(text_value, str):

18
litellm/integrations/custom_logger.py
View file

@ -909,7 +909,8 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
"""
import litellm
from litellm import Choices, Message, ModelResponse
from litellm.litellm_core_utils.classifier_logging import CLASSIFIER_AUDIT_FIELDS, without_classifier_audit
from litellm.litellm_core_utils.classifier_logging import CLASSIFIER_AUDIT_FIELDS
from litellm.litellm_core_utils.redact_messages import redacted_litellm_params
turn_off_message_logging: Final[bool] = getattr(self, "turn_off_message_logging", False)
excluded_fields: Final[list[str] | None] = getattr(litellm, "standard_logging_payload_excluded_fields", None)
@ -918,9 +919,15 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
if turn_off_message_logging is False and not excluded_fields:
return model_call_details
params: Final = model_call_details.get("litellm_params")
redacted_params: Final = (
MappingProxyType({"litellm_params": redacted_litellm_params(params)})
if turn_off_message_logging and isinstance(params, Mapping)
else EMPTY_MAPPING
)
standard_logging_object: Final = model_call_details.get("standard_logging_object")
if standard_logging_object is None:
return model_call_details.copy()
return {**model_call_details, **redacted_params}
# Make a copy of just the standard_logging_object to avoid modifying the original
standard_logging_object_copy: Final = {
@ -960,13 +967,6 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
model_response_dict: Final = model_response.model_dump()
standard_logging_object_copy["response"] = model_response_dict
params: Final = model_call_details.get("litellm_params")
request: Final = params.get("proxy_server_request") if isinstance(params, dict) else None
redacted_params: Final = (
MappingProxyType({"litellm_params": {**params, "proxy_server_request": without_classifier_audit(request)}})
if turn_off_message_logging and isinstance(params, dict) and isinstance(request, dict)
else EMPTY_MAPPING
)
return {
**model_call_details,
**redacted_params,

27
litellm/litellm_core_utils/logging_worker.py
View file

@ -23,6 +23,19 @@ from litellm.constants import (
MAX_TIME_TO_CLEAR_QUEUE,
)
_CALLBACK_DEADLINE: Final[contextvars.ContextVar[float | None]] = contextvars.ContextVar(
"logging_callback_deadline", default=None
)
def optional_callback_budget(maximum: float, *, fraction: float = 0.25) -> float:
deadline: Final = _CALLBACK_DEADLINE.get()
return (
maximum
if deadline is None
else max(0.0, min(maximum, (deadline - asyncio.get_running_loop().time()) * fraction))
)
def _coroutine_name(coroutine: Coroutine) -> str:
return getattr(coroutine, "__qualname__", None) or getattr(coroutine, "__name__", None) or type(coroutine).__name__
@ -100,12 +113,20 @@ class LoggingWorker:
return len(revived)
def _run_coroutine_silently(self, loop: asyncio.AbstractEventLoop, coroutine: Coroutine) -> bool:
token: Final = _CALLBACK_DEADLINE.set(loop.time() + self.timeout)
try:
loop.run_until_complete(asyncio.wait_for(coroutine, timeout=self.timeout))
except (Exception, asyncio.CancelledError): # noqa: BLE001 # atexit flush must never break the user's program
return False
finally:
_CALLBACK_DEADLINE.reset(token)
return True
def _create_callback_task(self, task: LoggingTask) -> asyncio.Task[object]:
context: Final = task["context"].copy()
context.run(_CALLBACK_DEADLINE.set, asyncio.get_running_loop().time() + self.timeout)
return context.run(asyncio.create_task, task["coroutine"])
@staticmethod
def _drain_pending(queue: "asyncio.Queue[LoggingTask]") -> tuple[LoggingTask, ...]:
"""Pop every task still queued, without awaiting them, so they can be moved to another queue."""
@ -172,7 +193,7 @@ class LoggingWorker:
try:
if self._queue is not None:
# Run the coroutine in its original context
callback_task: Final = task["context"].run(asyncio.create_task, task["coroutine"])
callback_task: Final = self._create_callback_task(task)
try:
await asyncio.wait_for(callback_task, timeout=self.timeout)
except asyncio.TimeoutError as e:
@ -424,7 +445,7 @@ class LoggingWorker:
try:
await asyncio.wait_for(
task["context"].run(asyncio.create_task, task["coroutine"]),
self._create_callback_task(task),
timeout=self.timeout,
)
except Exception:
@ -517,7 +538,7 @@ class LoggingWorker:
# Await the coroutine to properly execute and avoid "never awaited" warnings
try:
await asyncio.wait_for(
task["context"].run(asyncio.create_task, task["coroutine"]),
self._create_callback_task(task),
timeout=self.timeout,
)
except Exception:

27
litellm/litellm_core_utils/prompt_templates/common_utils.py
View file

@ -192,6 +192,33 @@ def get_str_from_messages(messages: list[AllMessageValues]) -> str:
return text
def get_semantic_cache_prompt_from_messages(messages: Sequence[Mapping[str, object]]) -> str:
"""
The text a semantic cache embeds for a request: `get_str_from_messages` plus the text inside
Messages API `tool_result` blocks, so a tool turn does not embed identically to the turn before it
"""
return "".join(
_semantic_cache_content_text(message.get("content"))
+ extract_search_results_text(message.get("search_results"))
for message in messages
)
def _semantic_cache_content_text(content: object) -> str:
if isinstance(content, str):
return content
if not isinstance(content, list):
return ""
return "".join(_semantic_cache_block_text(block) for block in content if isinstance(block, Mapping))
def _semantic_cache_block_text(block: Mapping[str, object]) -> str:
if block.get("type") == "tool_result":
return _semantic_cache_content_text(block.get("content"))
text: Final = block.get("text")
return text if isinstance(text, str) else ""
def is_non_content_values_set(message: AllMessageValues) -> bool:
ignore_keys: Final = ["content", "role", "name"]
return any(message.get(key, None) is not None for key in message if key not in ignore_keys)

27
litellm/litellm_core_utils/redact_messages.py
View file

@ -11,6 +11,7 @@ import asyncio
import copy
import inspect
from collections.abc import Mapping
from dataclasses import replace
from typing import TYPE_CHECKING, Any, Final
import litellm
@ -26,6 +27,7 @@ from litellm.llms.vertex_ai.common_utils import (
redact_vertex_ai_metadata_from_logged_object,
)
from litellm.secret_managers.main import str_to_bool
from litellm.types.router import BaselineRouteStamp
from litellm.types.utils import StandardCallbackDynamicParams
if TYPE_CHECKING:
@ -252,6 +254,26 @@ def _redact_model_response_dict_choices(choices, redacted_str: str):
_redact_choice_content(choice)
def _redacted_baseline_metadata(metadata: Mapping[str, object]) -> Mapping[str, object]:
route: Final = metadata.get("_autorouter_baseline_route")
if not isinstance(route, BaselineRouteStamp):
return metadata
return {**metadata, "_autorouter_baseline_route": replace(route, request_parameters=None)}
def redacted_litellm_params(params: Mapping[str, object]) -> dict[str, object]:
request: Final = params.get("proxy_server_request")
return {
**params,
**{
key: _redacted_baseline_metadata(value)
for key, value in params.items()
if key in ("metadata", "litellm_metadata") and isinstance(value, Mapping)
},
**({"proxy_server_request": without_classifier_audit(request)} if isinstance(request, Mapping) else {}),
}
def perform_redaction(model_call_details: dict, result, redact_streaming_responses: bool = True):
"""
Performs the actual redaction on the logging object and result.
@ -262,9 +284,8 @@ def perform_redaction(model_call_details: dict, result, redact_streaming_respons
"""
# Redact model_call_details
params: Final = model_call_details.get("litellm_params")
request: Final = params.get("proxy_server_request") if isinstance(params, dict) else None
if isinstance(params, dict) and isinstance(request, Mapping):
model_call_details["litellm_params"] = {**params, "proxy_server_request": without_classifier_audit(request)}
if isinstance(params, Mapping):
model_call_details["litellm_params"] = redacted_litellm_params(params)
model_call_details["messages"] = [{"role": "user", "content": REDACTED_BY_LITELLM}]
model_call_details["prompt"] = ""
model_call_details["input"] = ""

63
litellm/llms/anthropic/pass_through/messages/handler.py
View file

@ -15,9 +15,6 @@ import litellm
from litellm.litellm_core_utils.exception_mapping_utils import exception_type
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.anthropic.common_utils import (
flatten_unencrypted_web_search_results_in_anthropic_messages,
sanitize_tool_use_ids_in_anthropic_messages,
strip_empty_content_blocks_from_anthropic_messages,
strip_provider_specific_fields_from_anthropic_messages,
)
from litellm.llms.base_llm.anthropic_messages.transformation import (
@ -36,9 +33,8 @@ from litellm.utils import ProviderConfigManager, client
from ..adapters.handler import LiteLLMMessagesToCompletionTransformationHandler
from ..responses_adapters.handler import LiteLLMMessagesToResponsesAPIHandler
from ..utils import is_reasoning_auto_summary_enabled
from .interceptors import get_messages_interceptors
from .utils import AnthropicMessagesRequestUtils, mock_response
from .utils import AnthropicMessagesRequestUtils, mock_response, prepare_native_messages
__all__ = ("anthropic_messages", "anthropic_messages_handler")
@ -251,28 +247,7 @@ async def anthropic_messages(
Runs the empty-content-block sanitizer before any backend dispatch.
"""
# Anthropic's API rejects requests containing empty / whitespace-only
# text content blocks ("messages: text content blocks must be
# non-empty") and empty thinking blocks ("each thinking block must
# contain thinking"). Multi-turn tool-use clients (e.g. Claude Code)
# routinely loop assistant responses that contain such blocks — an empty
# text block alongside tool_use, or an empty thinking block from a turn
# a non-Anthropic reasoning model served through the bridge — back as
# conversation history, which then causes the next /v1/messages call to
# 400. /v1/chat/completions already handles this in
# anthropic_messages_pt; sanitize the native Anthropic Messages path
# here for the same guarantee. See #22930.
messages = strip_empty_content_blocks_from_anthropic_messages(messages)
# Replay of cross-provider tool history (e.g. kimi -> Anthropic) may carry
# ids like ``functions.Bash:0`` that violate Anthropic's id pattern.
messages = sanitize_tool_use_ids_in_anthropic_messages(messages)
messages = flatten_unencrypted_web_search_results_in_anthropic_messages(messages)
from litellm.integrations.anthropic_cache_control_hook import (
AnthropicCacheControlHook,
)
messages, system = AnthropicCacheControlHook.maybe_inject_cache_control(
messages, system = prepare_native_messages(
messages, system, kwargs, model=model, custom_llm_provider=custom_llm_provider, tools=tools, api_base=api_base
)
@ -454,23 +429,15 @@ def anthropic_messages_handler(
"""
from litellm.types.utils import LlmProviders
# Sanitize empty text blocks so the sync entry point
# (litellm.messages.create -> anthropic_messages_handler) gets the same
# protection as the async wrapper. The async wrapper already sanitized and
# does not reassign messages before dispatch, so it sets
# ``_litellm_messages_presanitized`` to skip this redundant second
# full-messages scan. Pop it so it never leaks into provider params.
if not kwargs.pop("_litellm_messages_presanitized", False):
messages = strip_empty_content_blocks_from_anthropic_messages(messages)
messages = sanitize_tool_use_ids_in_anthropic_messages(messages)
messages = flatten_unencrypted_web_search_results_in_anthropic_messages(messages)
from litellm.integrations.anthropic_cache_control_hook import (
AnthropicCacheControlHook,
)
messages, system = AnthropicCacheControlHook.maybe_inject_cache_control(
messages, system, kwargs, model=model, custom_llm_provider=custom_llm_provider, tools=tools, api_base=api_base
messages, system = prepare_native_messages(
messages,
system,
kwargs,
model=model,
custom_llm_provider=custom_llm_provider,
tools=tools,
api_base=api_base,
presanitized=bool(kwargs.pop("_litellm_messages_presanitized", False)),
)
metadata = validate_anthropic_api_metadata(metadata)
@ -645,14 +612,6 @@ def anthropic_messages_handler(
custom_llm_provider=custom_llm_provider,
)
)
if is_reasoning_auto_summary_enabled():
thinking_param: Final = anthropic_messages_optional_request_params.get("thinking")
if isinstance(thinking_param, dict) and thinking_param.get("type") != "disabled":
anthropic_messages_optional_request_params["thinking"] = {
**thinking_param,
"display": "summarized",
}
resolved_api_base: Final = (
dynamic_api_base
if dynamic_api_base is not None and anthropic_messages_provider_config.uses_get_llm_provider_api_base()

52
litellm/llms/anthropic/pass_through/messages/utils.py
View file

@ -2,6 +2,15 @@ from collections.abc import Iterable, Mapping, Sequence
from functools import lru_cache
from typing import TYPE_CHECKING, Any, Final, cast, get_type_hints
from pydantic import JsonValue
from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook
from litellm.llms.anthropic.common_utils import (
flatten_unencrypted_web_search_results_in_anthropic_messages,
sanitize_tool_use_ids_in_anthropic_messages,
strip_empty_content_blocks_from_anthropic_messages,
)
from litellm.llms.anthropic.pass_through.utils import is_reasoning_auto_summary_enabled
from litellm.types.llms.anthropic import (
AnthropicMessagesRequestOptionalParams,
AnthropicStopDetails,
@ -119,8 +128,40 @@ def anthropic_system_to_openai_message(system: object) -> ChatCompletionSystemMe
return ChatCompletionSystemMessage(role="system", content=system)
def prepare_native_messages(
messages: list[dict[str, JsonValue]],
system: str | list[dict[str, JsonValue]] | None,
kwargs: dict[str, object],
*,
model: str,
custom_llm_provider: str | None = None,
tools: list[dict[str, JsonValue]] | None = None,
api_base: str | None = None,
presanitized: bool = False,
) -> tuple[list[dict[str, JsonValue]], str | list[dict[str, JsonValue]] | None]:
normalized: Final = (
messages
if presanitized
else flatten_unencrypted_web_search_results_in_anthropic_messages(
sanitize_tool_use_ids_in_anthropic_messages(strip_empty_content_blocks_from_anthropic_messages(messages))
)
)
return cast( # cast-ok: legacy normalizers and injection preserve the JSON message and system shapes
tuple[list[dict[str, JsonValue]], str | list[dict[str, JsonValue]] | None],
AnthropicCacheControlHook.maybe_inject_cache_control(
normalized,
system,
kwargs,
model=model,
custom_llm_provider=custom_llm_provider,
tools=tools,
api_base=api_base,
),
)
@lru_cache(maxsize=1)
def _anthropic_messages_optional_param_keys() -> frozenset[str]:
def anthropic_messages_optional_param_keys() -> frozenset[str]:
"""
Valid AnthropicMessagesRequestOptionalParams keys.
@ -152,7 +193,7 @@ class AnthropicMessagesRequestUtils:
Returns:
AnthropicMessagesRequestOptionalParams instance with only the valid parameters
"""
valid_keys: Final = _anthropic_messages_optional_param_keys()
valid_keys: Final = anthropic_messages_optional_param_keys()
filtered_params: Final = {k: v for k, v in params.items() if k in valid_keys and v is not None}
if model is not None:
from litellm.llms.anthropic.chat.transformation import AnthropicConfig
@ -174,6 +215,13 @@ class AnthropicMessagesRequestUtils:
drop_params=drop_params,
output_key=param,
)
if is_reasoning_auto_summary_enabled():
thinking_param: Final = filtered_params.get("thinking")
if isinstance(thinking_param, dict) and thinking_param.get("type") != "disabled":
return cast(
AnthropicMessagesRequestOptionalParams,
{**filtered_params, "thinking": {**thinking_param, "display": "summarized"}},
)
return cast(AnthropicMessagesRequestOptionalParams, filtered_params)

64
litellm/llms/anthropic/prompt_cache_prediction.py
View file

@ -5,6 +5,7 @@ import hashlib
import json
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from functools import reduce
from itertools import accumulate, groupby
from types import MappingProxyType
from typing import Annotated, Final, Literal, Protocol, TypeAlias
@ -13,20 +14,32 @@ import httpx
from pydantic import ConfigDict, Field, JsonValue, StrictInt, TypeAdapter, ValidationError
import litellm
from litellm.llms.anthropic.common_utils import AnthropicModelInfo, is_anthropic_oauth_key
from litellm.litellm_core_utils.dot_notation_indexing import delete_nested_value
from litellm.llms.anthropic.common_utils import (
AnthropicModelInfo,
is_anthropic_oauth_key,
strip_provider_specific_fields_from_anthropic_messages,
)
from litellm.llms.anthropic.count_tokens.handler import AnthropicCountTokensHandler
from litellm.llms.anthropic.count_tokens.transformation import COUNT_TOKEN_OPTION_NAMES
from litellm.llms.anthropic.pass_through.messages.transformation import (
DEFAULT_ANTHROPIC_API_VERSION,
AnthropicMessagesConfig,
)
from litellm.llms.anthropic.pass_through.messages.utils import AnthropicMessagesRequestUtils, prepare_native_messages
from litellm.router_utils.baseline_request import (
BASELINE_PARAMETERS,
capture_baseline_parameters,
)
from litellm.types.llms.base import LiteLLMBaseModel
from litellm.types.router import LiteLLM_Params
from litellm.types.router import GenericLiteLLMParams, LiteLLM_Params
from litellm.types.utils import ModelResponse
from litellm.utils import supports_thinking_cache_preservation
_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
_HEADERS: Final = TypeAdapter(dict[str, str])
_MESSAGES: Final = TypeAdapter(list[dict[str, JsonValue]])
_SYSTEM: Final = TypeAdapter(str | list[dict[str, JsonValue]] | None)
_counter: Final = AnthropicCountTokensHandler()
@ -325,7 +338,7 @@ def _entry_fingerprint(fingerprint: str, ttl_seconds: int) -> str:
def parse_cache_plan(body: Mapping[str, JsonValue]) -> PromptCachePlan | UnsupportedCachePlan:
try:
request: Final = _PlanRequest.model_validate(body)
request: Final = _PlanRequest.model_validate(dict(body))
positions: Final = _positions(body)
except ValidationError:
return UnsupportedCachePlan("unsupported_prompt_shape")
@ -618,13 +631,56 @@ def resolve_baseline_prediction_target(params: LiteLLM_Params) -> NativePredicti
return _resolve_prediction_target(params, allow_configured_endpoint=True)
def prepare_native_baseline_body(request: Mapping[str, object], model: str) -> Mapping[str, JsonValue] | None:
parameters: Final = capture_baseline_parameters(request)
if parameters is None:
return None
source: Final = {**parameters, "messages": request.get("messages"), "stream": request.get("stream", False)}
try:
owned: Final = _JSON_OBJECT.validate_python(source)
context: Final = {**{k: v for k, v in request.items() if k not in ("metadata", "litellm_metadata")}, **owned}
resolved_model: Final = litellm.get_llm_provider(model=model, custom_llm_provider="anthropic")[0]
messages, system = prepare_native_messages(
_MESSAGES.validate_python(owned.get("messages")),
_SYSTEM.validate_python(owned.get("system")),
context,
model=resolved_model,
custom_llm_provider="anthropic",
tools=_MESSAGES.validate_python(owned.get("tools") or []),
)
options: Final = AnthropicMessagesRequestUtils.get_requested_anthropic_messages_optional_param(
{**owned, "system": system},
model=resolved_model,
custom_llm_provider="anthropic",
drop_params=owned.get("drop_params") is True,
)
filtered: Final = reduce(
delete_nested_value,
TypeAdapter(tuple[str, ...]).validate_python(owned.get("additional_drop_params") or ()),
dict(options),
)
body: Final = AnthropicMessagesConfig().transform_anthropic_messages_request(
model=resolved_model,
messages=strip_provider_specific_fields_from_anthropic_messages(messages),
anthropic_messages_optional_request_params=filtered,
litellm_params=GenericLiteLLMParams(),
headers={},
)
return MappingProxyType(_JSON_OBJECT.validate_python(body))
except Exception: # noqa: BLE001 # an unsupported hypothetical request is unavailable, never an inference failure
return None
def _resolve_prediction_target(
params: LiteLLM_Params,
*,
allow_configured_endpoint: bool,
) -> NativePredictionTarget | UnsupportedPredictionTarget:
configured_options: Final = frozenset(params.model_dump(exclude_defaults=True, exclude_none=True))
if configured_options - _DEPLOYMENT_OPTIONS:
allowed: Final = (
_DEPLOYMENT_OPTIONS | frozenset(BASELINE_PARAMETERS) if allow_configured_endpoint else _DEPLOYMENT_OPTIONS
)
if configured_options - allowed:
return UnsupportedPredictionTarget("unsupported_deployment_configuration")
api_base: Final = AnthropicModelInfo.get_api_base(params.api_base)
if not allow_configured_endpoint and api_base not in (

3
litellm/proxy/db/baseline_accounting.py
View file

@ -223,7 +223,8 @@ ON CONFLICT (request_id) DO NOTHING
_MARK_CONFLICT: Final = """
UPDATE "LiteLLM_AutoRouterBaselineObservation"
SET conflicted = TRUE, revision = $4::bigint
WHERE request_id = $1 AND scope = $2 AND data <> $3 AND NOT conflicted
WHERE request_id = $1 AND scope = $2 AND NOT conflicted
AND (data::jsonb #- '{turn,turn_at}') <> ($3::jsonb #- '{turn,turn_at}')
"""
_READ_PAGE: Final = """
WITH times AS (

229
litellm/proxy/hooks/autorouter_baseline_cache.py
View file

@ -5,7 +5,7 @@ import hashlib
import json
import time
from collections.abc import Callable, Mapping
from dataclasses import dataclass, replace
from dataclasses import dataclass, field, replace
from datetime import datetime
from types import MappingProxyType
from typing import TYPE_CHECKING, Final
@ -16,9 +16,8 @@ from pydantic import ConfigDict, Field, JsonValue, TypeAdapter
from litellm._logging import verbose_proxy_logger
from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.core_helpers import (
get_litellm_metadata_from_kwargs, # pyright: ignore[reportUnknownVariableType] # legacy metadata boundary validated below
)
from litellm.litellm_core_utils.core_helpers import get_metadata_variable_name_from_kwargs
from litellm.litellm_core_utils.logging_worker import optional_callback_budget
from litellm.llms.anthropic.prompt_cache_prediction import (
CountedPromptCachePlan,
NativePredictionTarget,
@ -28,6 +27,7 @@ from litellm.llms.anthropic.prompt_cache_prediction import (
count_cache_plan,
count_prompt_tokens,
parse_cache_plan,
prepare_native_baseline_body,
resolve_baseline_prediction_target,
supported_baseline_recipient,
supported_prediction_headers,
@ -37,6 +37,8 @@ from litellm.proxy.spend_tracking.savings import (
_effective_model_info, # pyright: ignore[reportPrivateUsage] # existing deployment-price owner
_proxy_llm_router, # pyright: ignore[reportPrivateUsage] # existing optional proxy-router owner
)
from litellm.router_strategy.complexity_router.context_compaction import compaction_applied
from litellm.router_utils.baseline_request import baseline_request
from litellm.types.llms.base import LiteLLMBaseModel
from litellm.types.router import BaselineRouteStamp
from litellm.types.utils import CallTypes, ModelInfo, Usage
@ -66,6 +68,9 @@ class CapturedBaselineObservation(LiteLLMBaseModel):
prices: ModelInfo | None
observation: BaselineObservation
def with_observation(self, observation: BaselineObservation) -> CapturedBaselineObservation:
return self.model_copy(update={"observation": observation})
@dataclass(frozen=True, slots=True)
class BaselineCacheContext:
@ -73,7 +78,10 @@ class BaselineCacheContext:
capture: CapturedBaselineObservation
target: NativePredictionTarget | UnsupportedPredictionTarget
baseline_deployment_id: str
baseline_body: Mapping[str, JsonValue] | None = field(default=None, repr=False)
selected_body_digest: str | None = field(default=None, repr=False)
invalidated: str | None = None
finalization: asyncio.Task[CapturedBaselineObservation] | None = field(default=None, repr=False, compare=False)
class _Metadata(LiteLLMBaseModel):
@ -102,6 +110,10 @@ def _digest(value: object) -> str:
return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
def _native_body_digest(body: Mapping[str, JsonValue]) -> str:
return _digest({key: value for key, value in body.items() if key not in ("metadata", "stream")})
class AutoRouterBaselineCache(CustomLogger):
def __init__(
self,
@ -124,12 +136,15 @@ class AutoRouterBaselineCache(CustomLogger):
if not isinstance(logging_obj, Logging) or call_type != CallTypes.anthropic_messages:
return
try:
metadata: Final = _METADATA.validate_python(get_litellm_metadata_from_kwargs({"litellm_params": kwargs}))
raw_metadata: Final = kwargs.get(get_metadata_variable_name_from_kwargs(kwargs))
metadata: Final = _METADATA.validate_python(raw_metadata) if isinstance(raw_metadata, Mapping) else {}
if metadata.get(INTERNAL_CALL_ORIGIN_METADATA_KEY):
return
if logging_obj.baseline_cache_context is not None:
await invalidate_baseline_cache(logging_obj, "retried_request")
return
if not isinstance(metadata.get("_autorouter_baseline_route"), BaselineRouteStamp):
return
request: Final = _Metadata.model_validate(metadata)
session: Final = kwargs.get("litellm_session_id") or request.session_id or logging_obj.litellm_session_id
if not isinstance(session, str) or not session or len(session) > 256:
@ -142,13 +157,27 @@ class AutoRouterBaselineCache(CustomLogger):
prices: Final = _PRICES.validate_python(
_effective_model_info(router, request.route.baseline_deployment_id, request.route.baseline_model)
)
params: Final = (
_METADATA.validate_python(deployment.litellm_params.model_dump(mode="json")) if deployment else {}
)
projected: Final = (
baseline_request(
kwargs,
request.route.request_parameters,
params,
include_extra_body=False,
)
if request.route.request_parameters is not None
else None
)
scope: Final = "autorouter-baseline:v3:" + _digest(
(
"baseline_request_v4",
request.user_api_key_hash,
session,
request.route.router_name,
request.route.baseline_deployment_id,
deployment.litellm_params.model_dump(mode="json"),
params,
prices,
)
)
@ -170,8 +199,21 @@ class AutoRouterBaselineCache(CustomLogger):
reason="incomplete_response",
),
)
selected_model: Final = kwargs.get("model")
selected_body: Final = prepare_native_baseline_body(
kwargs, selected_model if isinstance(selected_model, str) else logging_obj.model
)
logging_obj.baseline_cache_context = BaselineCacheContext(
self, capture, target, request.route.baseline_deployment_id
self,
capture,
target,
request.route.baseline_deployment_id,
prepare_native_baseline_body(projected, target.model)
if projected is not None and isinstance(target, NativePredictionTarget)
else None,
_native_body_digest(selected_body)
if selected_body is not None and not compaction_applied(kwargs)
else None,
)
except Exception: # noqa: BLE001 # optional observation cannot fail inference
verbose_proxy_logger.warning("Auto-router baseline observation could not be initialized")
@ -197,14 +239,16 @@ class AutoRouterBaselineCache(CustomLogger):
async def plan(
self, target: NativePredictionTarget, wire: httpx.Request, body: Mapping[str, JsonValue], usage: Usage | None
) -> tuple[CountedPromptCachePlan | None, str | None]:
deadline: Final = asyncio.get_running_loop().time() + optional_callback_budget(_COUNT_TIMEOUT, fraction=0.75)
if not supported_prediction_headers(wire.headers):
return None, "unsupported_request_headers"
plan: Final = parse_cache_plan(body)
if isinstance(plan, UnsupportedCachePlan):
return None, plan.reason
details: Final = usage.prompt_tokens_details if usage is not None else None
selected: Final = parse_cache_plan(_JSON_BODY.validate_json(wire.content))
if (
not plan.breakpoints
(isinstance(selected, UnsupportedCachePlan) or not selected.breakpoints)
and details is not None
and ((details.cached_tokens or 0) + (details.cache_creation_tokens or 0))
):
@ -215,7 +259,8 @@ class AutoRouterBaselineCache(CustomLogger):
try:
counted: Final = await asyncio.wait_for(
count_cache_plan(target.model, target.api_key, plan, token_counter=count), timeout=_COUNT_TIMEOUT
count_cache_plan(target.model, target.api_key, plan, token_counter=count),
timeout=max(0.0, deadline - asyncio.get_running_loop().time()),
)
return (None, counted.reason) if isinstance(counted, UnsupportedCachePlan) else (counted, None)
except TimeoutError:
@ -229,111 +274,121 @@ async def invalidate_baseline_cache(logging_obj: Logging, reason: str, *, comple
if context is not None:
logging_obj.baseline_cache_context = replace(context, invalidated=reason)
logging_obj.baseline_observation = context.capture.model_copy(
update=MappingProxyType(
{
"observation": context.capture.observation.model_copy(
update=MappingProxyType(
{
"available_at": max(context.capture.observation.started_at, context.collector.clock()),
"reason": reason,
}
)
),
}
)
update={
"observation": context.capture.observation.model_copy(
update={
"available_at": max(context.capture.observation.started_at, context.collector.clock()),
"reason": reason,
}
),
}
)
async def finalize_baseline_cache(logging_obj: Logging, response_obj: object) -> None:
context: Final = logging_obj.baseline_cache_context
if context is None:
if context is None or logging_obj.baseline_observation is not None:
return
task: Final = context.finalization or asyncio.create_task(_capture(context, logging_obj, response_obj))
active: Final = context if context.finalization is not None else replace(context, finalization=task)
if context.finalization is None:
task.add_done_callback(_consume_finalization)
logging_obj.baseline_cache_context = active
try:
capture: Final = await _capture(context, logging_obj, response_obj)
if logging_obj.baseline_cache_context is context:
logging_obj.baseline_observation = capture # rebind-ok: attach only to the captured request owner
except Exception: # noqa: BLE001 # observation failures must preserve inference and billing
capture: Final = await asyncio.shield(task)
if logging_obj.baseline_cache_context is active:
logging_obj.baseline_observation = capture # rebind-ok: publish only for the current attempt
except Exception: # noqa: BLE001 # estimation must preserve inference and billing
await invalidate_baseline_cache(logging_obj, "observation_unavailable")
async def _capture(
def _consume_finalization(task: asyncio.Task[CapturedBaselineObservation]) -> None:
if not task.cancelled():
task.exception()
async def _capture_native(
context: BaselineCacheContext, logging_obj: Logging, response_obj: object
) -> CapturedBaselineObservation:
original: Final = context.capture.observation
details: Final = _METADATA.validate_python(logging_obj.model_call_details)
if details.get("cache_hit") is True:
return context.capture.model_copy(
update=MappingProxyType(
{
"observation": original.model_copy(
update=MappingProxyType({"outcome": "response_cache", "reason": "response_cache_hit"})
)
}
)
)
event: Final = _WireEvent.model_validate(details)
capture: Final = context.capture
original: Final = capture.observation
event: Final = _WireEvent.model_validate(logging_obj.model_call_details)
wire: Final = event.httpx_response.request
usage: Final = _ResponseUsage.model_validate(response_obj).usage
available: Final = event.completion_start_time.timestamp()
complete: Final = (
event.custom_llm_provider == "anthropic"
and event.httpx_response.status_code == 200
and (not event.stream or event.prompt_cache_response_complete)
)
started: Final = original.started_at
available: Final = event.completion_start_time.timestamp()
if context.invalidated or not complete or not started <= available <= context.collector.clock():
return context.capture.model_copy(
update=MappingProxyType(
{
"observation": original.model_copy(
update=MappingProxyType(
{
"available_at": max(started, context.collector.clock()),
"reason": context.invalidated or "incomplete_response",
}
)
)
if context.invalidated or not complete or not original.started_at <= available <= context.collector.clock():
return capture.with_observation(
original.model_copy(
update={
"available_at": max(original.started_at, context.collector.clock()),
"reason": context.invalidated or "incomplete_response",
}
)
)
target: Final = context.target
if isinstance(target, UnsupportedPredictionTarget) or not supported_baseline_recipient(target, wire):
return context.capture.model_copy(
update=MappingProxyType(
{
"observation": original.model_copy(
update=MappingProxyType(
{
"available_at": available,
"reason": target.reason
if isinstance(target, UnsupportedPredictionTarget)
else "unsupported_baseline_recipient",
}
)
)
return capture.with_observation(
original.model_copy(
update={
"available_at": available,
"reason": target.reason
if isinstance(target, UnsupportedPredictionTarget)
else "unsupported_baseline_recipient",
}
)
)
body: Final = _JSON_BODY.validate_json(wire.content)
same: Final = (
logging_obj.get_router_model_id() == context.baseline_deployment_id and body.get("model") == target.model
)
plan, reason = await context.collector.plan(target, wire, body, usage)
minimum: Final = get_prompt_cache_min_tokens(target.model)
return context.capture.model_copy(
update=MappingProxyType(
{
"observation": BaselineObservation(
request_id=original.request_id,
started_at=started,
available_at=available,
outcome="complete",
baseline_equivalent=same,
usage=usage,
plan=plan,
minimum_cache_tokens=minimum,
reason=reason,
)
}
projected: Final = context.baseline_body
if projected is None or context.selected_body_digest != _native_body_digest(body):
return capture.with_observation(
original.model_copy(
update={
"available_at": available,
"usage": usage,
"reason": "unsupported_baseline_settings"
if projected is None
else "unsupported_request_transformation",
}
)
)
same: Final = logging_obj.get_router_model_id() == context.baseline_deployment_id and _native_body_digest(
projected
) == _native_body_digest(body)
plan, reason = await context.collector.plan(target, wire, projected, usage)
return capture.with_observation(
BaselineObservation(
request_id=original.request_id,
started_at=original.started_at,
available_at=available,
outcome="complete",
baseline_equivalent=same,
usage=usage.model_copy(update={key: projected.get(key) for key in ("speed", "inference_geo")})
if usage is not None and not same
else usage,
plan=plan,
reason=reason,
minimum_cache_tokens=get_prompt_cache_min_tokens(target.model),
)
)
async def _capture(
context: BaselineCacheContext, logging_obj: Logging, response_obj: object
) -> CapturedBaselineObservation:
if _METADATA.validate_python(logging_obj.model_call_details).get("cache_hit") is True:
return context.capture.model_copy(
update={
"observation": context.capture.observation.model_copy(
update={
"outcome": "response_cache",
"reason": "response_cache_hit",
}
),
}
)
return await _capture_native(context, logging_obj, response_obj)

18
litellm/proxy/spend_tracking/baseline_accounting.py
View file

@ -71,12 +71,12 @@ def _complete_usage(usage: Usage | None) -> bool:
if usage is None or usage.prompt_tokens < 0 or usage.completion_tokens < 0:
return False
details: Final = usage.prompt_tokens_details
if details is None:
if details is None or not hasattr(details, "cache_creation_tokens"):
return False
values: Final = (details.text_tokens, details.cached_tokens, details.cache_creation_tokens)
if any(value is None or value < 0 for value in values):
return False
split: Final = details.cache_creation_token_details
split: Final = details.cache_creation_token_details if hasattr(details, "cache_creation_token_details") else None
writes: Final = details.cache_creation_tokens or 0
return (
usage.total_tokens == usage.prompt_tokens + usage.completion_tokens
@ -133,10 +133,13 @@ def _matches(entry: CacheEntry, markers: tuple[CountedBreakpoint, ...], started:
def _ambiguous(entry: CacheEntry, markers: tuple[CountedBreakpoint, ...], started: float) -> bool:
return entry.available_at <= started < entry.expires_at and any(
entry.content_fingerprint in marker.lookback_content_fingerprints
and (entry.uncertain or entry.ttl_seconds != marker.ttl_seconds)
for marker in markers
matching: Final = tuple(
marker for marker in markers if entry.content_fingerprint in marker.lookback_content_fingerprints
)
return (
entry.available_at <= started < entry.expires_at
and bool(matching)
and (entry.uncertain or all(entry.ttl_seconds != marker.ttl_seconds for marker in matching))
)
@ -261,7 +264,7 @@ def _writes(history: BaselineHistory, observation: BaselineObservation) -> tuple
observation.started_at + hit.ttl_seconds,
),
)
if hit is not None and all(marker.fingerprint != hit.fingerprint for marker in markers)
if hit is not None
else ()
)
return (
@ -277,6 +280,7 @@ def _writes(history: BaselineHistory, observation: BaselineObservation) -> tuple
uncertain=bool(ambiguous),
)
for marker in markers
if hit is None or marker.prefix_tokens > hit.tokens
),
)

6
litellm/router.py
View file

@ -14582,17 +14582,21 @@ class Router:
to the deployment that actually served the request. Every attempt therefore
writes or clears, never just writes.
"""
from litellm.router_utils.baseline_request import capture_baseline_parameters
from litellm.types.router import BaselineRouteStamp
phase_attributes(routing_decision_attributes(routing_decision))
baseline_model: Final = routing_decision.get("savings_baseline_model") if routing_decision else None
baseline_id: Final = routing_decision.get("savings_baseline_deployment_id") if routing_decision else None
router_name: Final = routing_decision.get("router_model_name") if routing_decision else None
caller_parameters: Final = (
capture_baseline_parameters(request_kwargs) if router_name and baseline_model else None
)
Router._stamp_or_clear_metadata_key(
request_kwargs=request_kwargs,
key="_autorouter_baseline_route",
value=(
BaselineRouteStamp(router_name, baseline_model, baseline_id)
BaselineRouteStamp(router_name, baseline_model, baseline_id, caller_parameters)
if router_name and baseline_model and baseline_id
else None
),

5
litellm/router_strategy/complexity_router/context_compaction.py
View file

@ -164,6 +164,11 @@ def compaction_pending(kwargs: Mapping[str, object] | None) -> bool:
return isinstance(state, CompactionState) and state.config is not None and not _client_managed(kwargs or _EMPTY)
def compaction_applied(kwargs: Mapping[str, object]) -> bool:
state: Final = kwargs.get(_STATE_KEY)
return isinstance(state, CompactionState) and state.summary is not None
def _reject(model: str, reason: str) -> NoReturn:
from litellm.exceptions import BadRequestError

153
litellm/router_utils/baseline_request.py Normal file
View file

@ -0,0 +1,153 @@
from __future__ import annotations
from collections.abc import Iterator, Mapping
from itertools import accumulate
from types import MappingProxyType
from typing import Final, cast
from pydantic import JsonValue, TypeAdapter, ValidationError
from litellm.llms.anthropic.pass_through.messages.utils import anthropic_messages_optional_param_keys
CACHE_SETTINGS: Final = (
"system",
"instructions",
"tools",
"tool_choice",
"parallel_tool_calls",
"response_format",
"text",
"reasoning",
"reasoning_effort",
"thinking",
"verbosity",
"output_config",
"output_format",
"speed",
"prompt_cache_key",
"cache_key",
"cached_content",
"previous_response_id",
"conversation",
"context_management",
"compaction",
)
_GENERIC_PARAMETERS: Final = (
*CACHE_SETTINGS,
"prompt_cache_options",
"prompt_cache_retention",
"cache_control",
"max_tokens",
"max_completion_tokens",
"max_output_tokens",
"temperature",
"top_p",
"top_k",
"stop_sequences",
"enable_prompt_caching",
"cache_control_injection_points",
"drop_params",
"additional_drop_params",
)
NATIVE_ONLY_PARAMETERS: Final = tuple(
key
for key in sorted(anthropic_messages_optional_param_keys())
if key not in (*_GENERIC_PARAMETERS, "metadata", "stream")
)
BASELINE_PARAMETERS: Final = (*_GENERIC_PARAMETERS, *NATIVE_ONLY_PARAMETERS)
_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
_MAX_BYTES: Final = 4 * 1024 * 1024
_MAX_NODES: Final = 32768
_MAX_DEPTH: Final = 32
def _json_cost(value: object, depth: int = 0) -> Iterator[int]:
if depth > _MAX_DEPTH:
yield _MAX_BYTES + 1
elif isinstance(value, str):
yield (6 if value.isascii() else 12) * len(value) + 2
elif isinstance(value, dict):
yield 2
for key, item in cast(dict[object, object], value).items():
yield from _json_cost(key, depth + 1)
yield from _json_cost(item, depth + 1)
yield 2
elif isinstance(value, (list, tuple)):
yield 2
for item in cast(list[object] | tuple[object, ...], value):
yield from _json_cost(item, depth + 1)
yield 1
elif isinstance(value, int) and value.bit_length() > 64:
yield _MAX_BYTES + 1
elif value is None or isinstance(value, (bool, int, float)):
yield 32
else:
yield _MAX_BYTES + 1
def within_baseline_budget(value: object) -> bool:
return all(
size <= _MAX_BYTES and nodes <= _MAX_NODES for nodes, size in enumerate(accumulate(_json_cost(value)), 1)
)
def _parameters(value: object, *, envelope: bool = False) -> dict[str, object]:
if not isinstance(value, Mapping):
return {}
mapping: Final = cast(Mapping[str, object], value)
keys: Final = (*BASELINE_PARAMETERS, "messages") if envelope else BASELINE_PARAMETERS
return {key: mapping[key] for key in keys if key in mapping}
def capture_baseline_parameters(
kwargs: Mapping[str, object], *, include_extra_body: bool = True
) -> Mapping[str, JsonValue] | None:
extra: Final = (
{"extra_body": _parameters(kwargs.get("extra_body"), envelope=True)}
if include_extra_body and "extra_body" in kwargs
else {}
)
parameters: Final = {**_parameters(kwargs), **extra}
if not within_baseline_budget(parameters):
return None
try:
return MappingProxyType(_JSON_OBJECT.validate_python(parameters))
except ValidationError:
return None
def baseline_request(
kwargs: Mapping[str, object],
caller: Mapping[str, JsonValue],
deployment: Mapping[str, object],
*,
include_extra_body: bool = True,
) -> Mapping[str, object] | None:
snapshot: Final = capture_baseline_parameters(deployment)
if snapshot is None:
return None
configured: Final = {
**_parameters(snapshot),
**(_parameters(snapshot.get("extra_body")) if include_extra_body else {}),
}
requested: Final = {**_parameters(caller), **(_parameters(caller.get("extra_body")) if include_extra_body else {})}
configured_tools: Final = configured.get("tools") or []
caller_tools: Final = requested.get("tools") or []
merged_tools: Final = (
{"tools": [*configured_tools, *caller_tools]}
if (configured_tools or caller_tools) and isinstance(configured_tools, list) and isinstance(caller_tools, list)
else {}
)
return MappingProxyType(
{
**{key: value for key, value in kwargs.items() if key not in (*BASELINE_PARAMETERS, "extra_body")},
**configured,
**requested,
**merged_tools,
**(
{"extra_body": caller.get("extra_body", snapshot.get("extra_body"))}
if not include_extra_body and ("extra_body" in caller or "extra_body" in snapshot)
else {}
),
}
)

4
litellm/types/integrations/prometheus.py
View file

@ -416,6 +416,7 @@ def _resolve_deployment_and_latency_caller_identity_labels(
class PrometheusMetricLabels:
litellm_llm_api_latency_metric = [
UserAPIKeyLabelNames.MODEL_GROUP.value,
UserAPIKeyLabelNames.v1_LITELLM_MODEL_NAME.value,
UserAPIKeyLabelNames.API_KEY_HASH.value,
UserAPIKeyLabelNames.API_KEY_ALIAS.value,
@ -430,6 +431,7 @@ class PrometheusMetricLabels:
]
litellm_llm_api_time_to_first_token_metric = [
UserAPIKeyLabelNames.MODEL_GROUP.value,
UserAPIKeyLabelNames.v1_LITELLM_MODEL_NAME.value,
UserAPIKeyLabelNames.API_KEY_HASH.value,
UserAPIKeyLabelNames.API_KEY_ALIAS.value,
@ -444,6 +446,7 @@ class PrometheusMetricLabels:
]
litellm_request_total_latency_metric = [
UserAPIKeyLabelNames.MODEL_GROUP.value,
UserAPIKeyLabelNames.END_USER.value,
UserAPIKeyLabelNames.API_KEY_HASH.value,
UserAPIKeyLabelNames.API_KEY_ALIAS.value,
@ -516,6 +519,7 @@ class PrometheusMetricLabels:
]
litellm_deployment_latency_per_output_token = [
UserAPIKeyLabelNames.MODEL_GROUP.value,
UserAPIKeyLabelNames.v2_LITELLM_MODEL_NAME.value,
UserAPIKeyLabelNames.MODEL_ID.value,
UserAPIKeyLabelNames.API_BASE.value,

5
litellm/types/router.py
View file

@ -5,7 +5,7 @@ litellm.Router Types - includes RouterConfig, UpdateRouterConfig, ModelInfo etc
import datetime
import enum
from collections.abc import Container, Mapping, Sequence
from dataclasses import dataclass
from dataclasses import dataclass, field
from typing import (
TYPE_CHECKING,
Annotated,
@ -21,7 +21,7 @@ from typing import (
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
import httpx
from pydantic import ConfigDict, Field, field_validator, model_validator
from pydantic import ConfigDict, Field, JsonValue, field_validator, model_validator
from typing_extensions import Protocol, ReadOnly, Required, TypedDict, runtime_checkable
from litellm._logging import verbose_logger
@ -1223,6 +1223,7 @@ class BaselineRouteStamp:
router_name: str
baseline_model: str
baseline_deployment_id: str
request_parameters: Mapping[str, JsonValue] | None = field(default=None, repr=False)
@dataclass(frozen=True, slots=True)

1
tests/code_coverage_tests/recursive_detector.py
View file

@ -2,6 +2,7 @@ import ast
import os
IGNORE_FUNCTIONS = [
"_json_cost", # bounded at depth 32 and consumed under byte/node limits.
"_format_type",
"remove_additional_properties",
"remove_strict_from_schema",

7
tests/e2e/batches/batch_cleanup.py
View file

@ -8,6 +8,7 @@ from typing import Final, Protocol
from batch_client import BatchObject, FileDeleteResponse
from capabilities import is_cloud_storage_id, is_managed_id
from e2e_http import NetworkError, RateLimitedError, Result, Success, UnknownApiError
from e2e_metadata import STEP_FRAMES, step
from pydantic import BaseModel
CLEANUP_DELAYS: Final = (1.0, 2.0, 4.0)
@ -52,6 +53,7 @@ def _require_cleanup_success[R: BaseModel](result: Result[R], operation: str) ->
raise AssertionError(f"{operation} failed: {result.kind}")
@step("Clean up the uploaded file")
def cleanup_file(client: BatchCleanupClient, file_id: str, *, key: str, provider: str | None = None) -> None:
delete: Final[Callable[[], Result[FileDeleteResponse]]] = (
(lambda: client.delete_file_as_admin(file_id, provider=provider))
@ -65,7 +67,7 @@ def cleanup_file(client: BatchCleanupClient, file_id: str, *, key: str, provider
warnings.warn(
f"Left file {file_id} in place: LiteLLM refused to delete it while a batch still references it",
UserWarning,
stacklevel=2,
stacklevel=2 + STEP_FRAMES,
)
return
deleted: Final = _require_cleanup_success(result, f"Delete file {file_id}")
@ -74,6 +76,7 @@ def cleanup_file(client: BatchCleanupClient, file_id: str, *, key: str, provider
), f"Delete file {file_id} did not confirm deletion"
@step("Cancel the batch if it is still running")
def cleanup_batch(
client: BatchCleanupClient,
batch_id: str,
@ -137,7 +140,7 @@ def cleanup_batch(
warnings.warn(
f"Left batch {batch_id} cancelling after {BATCH_CANCEL_TIMEOUT_SECONDS}s for the provider to finish",
UserWarning,
stacklevel=2,
stacklevel=2 + STEP_FRAMES,
)
return
wait(BATCH_CANCEL_POLL_SECONDS)

12
tests/e2e/batches/batch_client.py
View file

@ -17,6 +17,7 @@ from typing import Final, Literal
from pydantic import BaseModel, Field
from e2e_metadata import step
from proxy_client import ProxyClient
from e2e_http import (
FileUploadForm,
@ -136,12 +137,15 @@ def is_result_access_denied[R: BaseModel](result: Result[R]) -> bool:
class BatchClient:
proxy: ProxyClient
@step("Add a batch deployment named {model_name} that calls {litellm_params.model}")
def create_model(self, model_name: str, litellm_params: LiteLLMParamsBody) -> str:
return self.proxy.create_model(model_name, litellm_params, mode="batch")
@step("Delete the batch deployment")
def delete_model(self, model_id: str) -> None:
self.proxy.delete_model(model_id)
@step("Upload a batch input file to /v1/files")
def upload_file(
self,
*,
@ -161,6 +165,7 @@ class BatchClient:
response_type=FileObject,
)
@step("Retrieve the uploaded file")
def retrieve_file(
self, file_id: str, *, key: str, provider: str | None = None
) -> Result[FileObject]:
@ -171,6 +176,7 @@ class BatchClient:
response_type=FileObject,
)
@step("List the files the key can see from /v1/files")
def list_files(self, *, key: str, provider: str | None = None) -> Result[FileList]:
return self.proxy.transport.get(
_files_path(provider),
@ -179,6 +185,7 @@ class BatchClient:
response_type=FileList,
)
@step("Create a batch of {body.endpoint} requests from the uploaded file")
def create_batch(
self, *, body: BatchCreateBody, key: str, provider: str | None = None
) -> StreamingResponse:
@ -188,6 +195,7 @@ class BatchClient:
json=body,
)
@step("Retrieve the batch")
def retrieve_batch(
self, batch_id: str, *, key: str, provider: str | None = None
) -> Result[BatchObject]:
@ -198,6 +206,7 @@ class BatchClient:
response_type=BatchObject,
)
@step("Cancel the batch")
def cancel_batch(
self, batch_id: str, *, key: str, provider: str | None = None
) -> Result[BatchObject]:
@ -208,6 +217,7 @@ class BatchClient:
response_type=BatchObject,
)
@step("List the batches the key can see from /v1/batches")
def list_batches(
self,
*,
@ -223,6 +233,7 @@ class BatchClient:
response_type=BatchList,
)
@step("Delete the uploaded file")
def delete_file(
self, file_id: str, *, key: str, provider: str | None = None
) -> Result[FileDeleteResponse]:
@ -233,6 +244,7 @@ class BatchClient:
response_type=FileDeleteResponse,
)
@step("Delete the uploaded file as the proxy admin")
def delete_file_as_admin(self, file_id: str, *, provider: str | None = None) -> Result[FileDeleteResponse]:
return self.proxy.transport.delete(
f"{_files_path(provider)}/{file_id}",

42
tests/e2e/batches/capabilities.py
View file

@ -7,7 +7,11 @@ import os
from dataclasses import dataclass
from typing import Final, Literal
import pytest
from e2e_config import provider_edge_base, unique_marker
from e2e_metadata import Domain, Mode, Route, Subject, meta
from e2e_metadata import Provider as MetaProvider
from models import LiteLLMParamsBody
_BATCH_RUN = unique_marker()
@ -18,6 +22,9 @@ def batch_model_name(base: str) -> str:
OPENAI_BATCH_BACKEND: Final = "gpt-4o-mini"
AZURE_BATCH_BACKEND: Final = "gpt-5.4-mini-batch"
VERTEX_BATCH_BACKEND: Final = "gemini-2.5-flash"
BEDROCK_BATCH_BACKEND: Final = "bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0"
def openai_batch_params() -> LiteLLMParamsBody:
@ -65,14 +72,14 @@ class Provider:
return openai_batch_params()
case "azure":
return LiteLLMParamsBody(
model="azure/gpt-5.4-mini-batch",
model=f"azure/{AZURE_BATCH_BACKEND}",
api_base="os.environ/AZURE_API_BASE",
api_key="os.environ/AZURE_API_KEY",
api_version="2025-04-01-preview",
)
case "vertex_ai":
return LiteLLMParamsBody(
model="vertex_ai/gemini-2.5-flash",
model=f"vertex_ai/{VERTEX_BATCH_BACKEND}",
vertex_project="os.environ/VERTEXAI_PROJECT",
vertex_location="us-central1",
vertex_credentials="os.environ/VERTEXAI_CREDENTIALS",
@ -81,7 +88,7 @@ class Provider:
)
case "bedrock":
return LiteLLMParamsBody(
model="bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0",
model=BEDROCK_BATCH_BACKEND,
aws_access_key_id="os.environ/AWS_ACCESS_KEY_ID",
aws_secret_access_key="os.environ/AWS_SECRET_ACCESS_KEY",
aws_region_name="os.environ/AWS_REGION",
@ -132,21 +139,21 @@ PROVIDERS: tuple[Provider, ...] = (
Provider(
"azure",
batch_model_name("azure-batch"),
"gpt-5.4-mini-batch",
AZURE_BATCH_BACKEND,
can_cancel=True,
can_list=True,
),
Provider(
"vertex_ai",
batch_model_name("vertex-batch"),
"gemini-2.5-flash",
VERTEX_BATCH_BACKEND,
can_cancel=True,
can_list=True,
),
Provider(
"bedrock",
batch_model_name("bedrock-batch"),
"bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0",
BEDROCK_BATCH_BACKEND,
can_cancel=True,
can_list=True,
),
@ -181,6 +188,29 @@ CAPABILITIES: tuple[Capability, ...] = tuple(
)
def lifecycle_meta(cap: Capability) -> pytest.MarkDecorator:
return meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.BATCHES,
providers=(MetaProvider(cap.provider),),
models=(cap.raw_model,),
mode=Mode.BATCH,
)
)
def file_content_meta(provider: Provider) -> pytest.MarkDecorator:
return meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.FILES,
providers=(MetaProvider(provider.name),),
models=(provider.raw_model,),
)
)
def raw_id_matches_provider(provider: str, batch_id: str) -> bool:
if provider in ("openai", "azure"):
return batch_id.startswith("batch")

206
tests/e2e/batches/test_batches_e2e.py
View file

@ -44,17 +44,22 @@ from capabilities import (
OPENAI_BATCH_BACKEND,
OPENAI_BATCH_MODEL,
PROVIDERS,
VERTEX_BATCH_BACKEND,
Capability,
Provider,
batch_model_name,
coverage_cells_for_lifecycle,
decoded_model_from_id,
file_content_meta,
is_managed_id,
lifecycle_meta,
matches_id_shape,
openai_batch_params,
raw_id_matches_provider,
)
from e2e_config import MASTER_KEY, PROXY_BASE_URL, unique_marker
from e2e_metadata import Domain, Mode, Route, Subject, meta
from e2e_metadata import Provider as MetaProvider
from e2e_http import (
FileUploadForm,
Result,
@ -249,7 +254,7 @@ def assert_batch_object(batch: BatchObject) -> None:
pytest.param(
cap,
id=cap.id,
marks=pytest.mark.covers(*coverage_cells_for_lifecycle(cap)),
marks=(pytest.mark.covers(*coverage_cells_for_lifecycle(cap)), lifecycle_meta(cap)),
)
for cap in CAPABILITIES
],
@ -350,6 +355,15 @@ def test_batch_lifecycle(
@pytest.mark.covers("llm.batches.openai.key_model_access_denied.nonstream.works")
@meta(
Subject(
domain=Domain.PROXY_AUTH,
route=Route.BATCHES,
providers=(MetaProvider.OPENAI,),
models=(OPENAI_BATCH_BACKEND,),
mode=Mode.BATCH,
)
)
def test_batch_key_model_access_denied(
client: BatchClient, resources: ResourceManager, batch_deployments: None
) -> None:
@ -389,6 +403,14 @@ def test_batch_key_model_access_denied(
"llm.files.openai.upload.nonstream.works",
"llm.files.openai.delete.nonstream.works",
)
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.FILES,
providers=(MetaProvider.OPENAI,),
models=(OPENAI_BATCH_BACKEND,),
)
)
def test_file_upload_and_delete_outputs(
client: BatchClient, resources: ResourceManager, batch_deployments: None
) -> None:
@ -433,6 +455,15 @@ def unattributed_rows(rows: list[SpendLogRow]) -> list[SpendLogRow]:
"once the fetch is bounded."
)
)
@meta(
Subject(
domain=Domain.SPEND_BUDGETS,
route=Route.BATCHES,
providers=(MetaProvider.OPENAI,),
models=(OPENAI_BATCH_BACKEND,),
mode=Mode.BATCH,
)
)
def test_rate_limited_batch_create_leaves_no_unattributed_spend_row(
client: BatchClient, resources: ResourceManager, batch_deployments: None
) -> None:
@ -471,7 +502,7 @@ def test_rate_limited_batch_create_leaves_no_unattributed_spend_row(
file = unwrap(
client.upload_file(
content=render_jsonl("gpt-4o-mini"),
content=render_jsonl(OPENAI_BATCH_BACKEND),
form=FileUploadForm(purpose="batch"),
model=OPENAI_BATCH_MODEL,
key=key,
@ -520,6 +551,14 @@ class TestBatchFileContent:
"llm.files.openai.content.nonstream.works",
exercised_on=["files"],
)
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.FILES,
providers=(MetaProvider.OPENAI,),
models=(OPENAI_BATCH_BACKEND,),
)
)
def test_file_content_matches_upload(
self, client: BatchClient, resources: ResourceManager
) -> None:
@ -558,8 +597,9 @@ class TestBatchFileContent:
pytest.param(
p,
id=p.name,
marks=pytest.mark.covers(
FILE_CONTENT_CELLS[p.name], exercised_on=["files"]
marks=(
pytest.mark.covers(FILE_CONTENT_CELLS[p.name], exercised_on=["files"]),
file_content_meta(p),
),
)
for p in PROVIDERS
@ -632,6 +672,14 @@ class TestOpenAIFiles:
"marker when LIT-4820 is fixed; do not relax the assertion to make it pass."
)
)
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.FILES,
providers=(MetaProvider.OPENAI,),
models=(OPENAI_BATCH_BACKEND,),
)
)
def test_uploaded_file_appears_in_list(
self, client: BatchClient, resources: ResourceManager, batch_deployments: None
) -> None:
@ -662,6 +710,7 @@ class TestOpenAIFiles:
"llm.files.openai.list_isolation.nonstream.works",
exercised_on=["files"],
)
@meta(Subject(domain=Domain.LLM_TRANSLATION, route=Route.FILES, providers=(MetaProvider.OPENAI,)))
def test_list_page_cursors_address_only_the_callers_own_files(
self, client: BatchClient, resources: ResourceManager
) -> None:
@ -697,6 +746,14 @@ class TestOpenAIFiles:
"llm.files.openai.retrieve.nonstream.works",
exercised_on=["files"],
)
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.FILES,
providers=(MetaProvider.OPENAI,),
models=(OPENAI_BATCH_BACKEND,),
)
)
def test_retrieve_round_trips_metadata(
self, client: BatchClient, resources: ResourceManager, batch_deployments: None
) -> None:
@ -760,6 +817,15 @@ class TestBatchRateLimitErrorMapping:
"quota_management.ratelimit.batch_rpm.blocks_over_limit",
exercised_on=["batches"],
)
@meta(
Subject(
domain=Domain.SPEND_BUDGETS,
route=Route.BATCHES,
providers=(MetaProvider.OPENAI,),
models=(OPENAI_BATCH_BACKEND,),
mode=Mode.BATCH,
)
)
def test_batch_create_over_rpm_returns_mapped_429(
self, client: BatchClient, resources: ResourceManager, batch_deployments: None
) -> None:
@ -773,7 +839,7 @@ class TestBatchRateLimitErrorMapping:
file = unwrap(
client.upload_file(
content=_multi_request_jsonl("gpt-4o-mini", BATCH_RL_REQUEST_LINES),
content=_multi_request_jsonl(OPENAI_BATCH_BACKEND, BATCH_RL_REQUEST_LINES),
form=FileUploadForm(purpose="batch"),
model=OPENAI_BATCH_MODEL,
key=key,
@ -826,7 +892,7 @@ class TestBatchEnqueuedTokenLimit:
) -> FileObject:
file = unwrap(
client.upload_file(
content=_multi_request_jsonl("gpt-4o-mini", BATCH_RL_REQUEST_LINES),
content=_multi_request_jsonl(OPENAI_BATCH_BACKEND, BATCH_RL_REQUEST_LINES),
form=FileUploadForm(purpose="batch"),
model=OPENAI_BATCH_MODEL,
key=key,
@ -859,6 +925,15 @@ class TestBatchEnqueuedTokenLimit:
"quota_management.ratelimit.batch_enqueued_tokens.accepts_over_rpm",
exercised_on=["batches"],
)
@meta(
Subject(
domain=Domain.SPEND_BUDGETS,
route=Route.BATCHES,
providers=(MetaProvider.OPENAI,),
models=(OPENAI_BATCH_BACKEND,),
mode=Mode.BATCH,
)
)
def test_enqueued_allowance_accepts_batch_over_key_rpm(
self, client: BatchClient, resources: ResourceManager, batch_deployments: None
) -> None:
@ -890,6 +965,15 @@ class TestBatchEnqueuedTokenLimit:
"quota_management.ratelimit.batch_enqueued_tokens.refunds_on_cancel",
exercised_on=["batches"],
)
@meta(
Subject(
domain=Domain.SPEND_BUDGETS,
route=Route.BATCHES,
providers=(MetaProvider.OPENAI,),
models=(OPENAI_BATCH_BACKEND,),
mode=Mode.BATCH,
)
)
def test_exhausted_allowance_blocks_until_cancel_refunds(
self, client: BatchClient, resources: ResourceManager, batch_deployments: None
) -> None:
@ -985,6 +1069,15 @@ class TestBedrockBatchAssumeRole:
"llm.files.bedrock.upload.nonstream.works",
exercised_on=["batches", "files"],
)
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.BATCHES,
providers=(MetaProvider.BEDROCK,),
models=(ASSUME_ROLE_RAW_MODEL,),
mode=Mode.BATCH,
)
)
def test_unified_batch_create_with_assume_role(
self, client: BatchClient, resources: ResourceManager
) -> None:
@ -1052,6 +1145,14 @@ class TestBedrockBatchSplitS3Credentials:
"llm.files.bedrock.split_s3_credentials.nonstream.works",
exercised_on=["files"],
)
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.FILES,
providers=(MetaProvider.BEDROCK,),
models=(ASSUME_ROLE_RAW_MODEL,),
)
)
def test_file_lifecycle_signs_s3_with_s3_credentials(
self, client: BatchClient, resources: ResourceManager
) -> None:
@ -1123,6 +1224,15 @@ class TestBedrockBatchGovCloud:
"llm.files.bedrock.govcloud_partition.nonstream.works",
exercised_on=["batches", "files"],
)
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.BATCHES,
providers=(MetaProvider.BEDROCK,),
models=(GOVCLOUD_RAW_MODEL,),
mode=Mode.BATCH,
)
)
def test_unified_file_upload_and_batch_create_in_govcloud(
self, client: BatchClient, resources: ResourceManager
) -> None:
@ -1193,6 +1303,14 @@ class TestGeminiFiles:
"llm.files.gemini.upload.nonstream.works",
exercised_on=["files"],
)
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.FILES,
providers=(MetaProvider.GEMINI,),
models=(GEMINI_FILES_RAW_MODEL,),
)
)
def test_gemini_file_upload(
self, client: BatchClient, resources: ResourceManager
) -> None:
@ -1227,7 +1345,7 @@ def _vllm_params(api_base: str, api_key: str | None, model_id: str) -> LiteLLMPa
)
HOSTED_VLLM_DEFAULT_MODEL = "Qwen/Qwen2.5-0.5B-Instruct"
HOSTED_VLLM_MODEL: Final = (os.environ.get("HOSTED_VLLM_MODEL") or "Qwen/Qwen2.5-0.5B-Instruct").strip()
HOSTED_VLLM_BAD_LINE_CUSTOM_ID = "req-bad"
@ -1236,9 +1354,8 @@ def _hosted_vllm_deployment(client: BatchClient, resources: ResourceManager) ->
if api_base is None:
pytest.skip("set HOSTED_VLLM_API_BASE (the live vLLM server this deployment targets)")
api_key = (os.environ.get("HOSTED_VLLM_API_KEY") or "").strip() or None
model_id = (os.environ.get("HOSTED_VLLM_MODEL") or HOSTED_VLLM_DEFAULT_MODEL).strip()
proxy_name = batch_model_name("hosted-vllm-batch")
model_row_id = client.create_model(proxy_name, _vllm_params(api_base, api_key, model_id))
model_row_id = client.create_model(proxy_name, _vllm_params(api_base, api_key, HOSTED_VLLM_MODEL))
resources.defer(lambda: client.delete_model(model_row_id))
return proxy_name
@ -1290,6 +1407,15 @@ class TestHostedVllmBatch:
"llm.files.hosted_vllm.upload.nonstream.works",
exercised_on=["batches", "files"],
)
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.BATCHES,
providers=(MetaProvider.HOSTED_VLLM,),
models=(HOSTED_VLLM_MODEL,),
mode=Mode.BATCH,
)
)
def test_batch_runs_to_completion_with_a_downloadable_output(
self, client: BatchClient, resources: ResourceManager, upload_route: str
) -> None:
@ -1337,6 +1463,15 @@ class TestHostedVllmBatch:
)
@pytest.mark.covers("llm.batches.hosted_vllm.basic.nonstream.works", exercised_on=["batches", "files"])
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.BATCHES,
providers=(MetaProvider.HOSTED_VLLM,),
models=(HOSTED_VLLM_MODEL,),
mode=Mode.BATCH,
)
)
def test_failing_line_lands_in_the_error_file_not_the_batch_status(
self, client: BatchClient, resources: ResourceManager
) -> None:
@ -1417,6 +1552,7 @@ class TestBatchFailurePaths:
"llm.batches.openai.malformed_jsonl.nonstream.works",
exercised_on=["files"],
)
@meta(Subject(domain=Domain.LLM_TRANSLATION, route=Route.FILES))
def test_malformed_jsonl_upload_rejected(
self, client: BatchClient, resources: ResourceManager, batch_deployments: None
) -> None:
@ -1439,13 +1575,22 @@ class TestBatchFailurePaths:
"llm.batches.openai.cancel_terminal.nonstream.works",
exercised_on=["batches", "files"],
)
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.BATCHES,
providers=(MetaProvider.OPENAI,),
models=(OPENAI_BATCH_BACKEND,),
mode=Mode.BATCH,
)
)
def test_endpoint_mismatch_fails_batch_and_cancel_conflicts(
self, client: BatchClient, resources: ResourceManager, batch_deployments: None
) -> None:
key = resources.key()
file = unwrap(
client.upload_file(
content=_mismatched_endpoint_jsonl("gpt-4o-mini"),
content=_mismatched_endpoint_jsonl(OPENAI_BATCH_BACKEND),
form=FileUploadForm(purpose="batch"),
model=OPENAI_BATCH_MODEL,
key=key,
@ -1495,6 +1640,15 @@ class TestBatchFailurePaths:
"llm.batches.openai.foreign_file_id.nonstream.works",
exercised_on=["batches", "files"],
)
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.BATCHES,
providers=(MetaProvider.AZURE,),
models=(AZURE_BATCH_RAW_MODEL,),
mode=Mode.BATCH,
)
)
def test_foreign_encoded_file_id_routes_by_file_model(
self, client: BatchClient, resources: ResourceManager, batch_deployments: None
) -> None:
@ -1544,6 +1698,15 @@ class TestBatchSecondHop:
"llm.batches.openai.second_hop.nonstream.works",
exercised_on=["batches", "files"],
)
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.BATCHES,
providers=(MetaProvider.LITELLM_PROXY, MetaProvider.OPENAI),
models=(OPENAI_BATCH_BACKEND,),
mode=Mode.BATCH,
)
)
def test_unified_create_and_retrieve_via_chained_gateway(
self, client: BatchClient, resources: ResourceManager, batch_deployments: None
) -> None:
@ -1561,7 +1724,7 @@ class TestBatchSecondHop:
file = unwrap(
client.upload_file(
content=render_jsonl("gpt-4o-mini"),
content=render_jsonl(OPENAI_BATCH_BACKEND),
form=FileUploadForm(purpose="batch", target_model_names=hop_name),
key=key,
)
@ -1680,13 +1843,22 @@ class TestBatchTerminalState:
"llm.batches.openai.terminal_state.nonstream.cost_logged",
exercised_on=["batches", "files"],
)
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.BATCHES,
providers=(MetaProvider.OPENAI,),
models=(OPENAI_BATCH_BACKEND,),
mode=Mode.BATCH,
)
)
def test_completed_batch_downloads_output_and_books_cost(
self, client: BatchClient, resources: ResourceManager, batch_deployments: None
) -> None:
key = resources.key()
file = unwrap(
client.upload_file(
content=render_jsonl("gpt-4o-mini"),
content=render_jsonl(OPENAI_BATCH_BACKEND),
form=FileUploadForm(purpose="batch"),
model=OPENAI_BATCH_MODEL,
key=key,
@ -1786,6 +1958,15 @@ class TestVertexNativePassthrough:
"llm.batches.vertex.native_passthrough.nonstream.works",
exercised_on=["files", "batches"],
)
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.BATCHES,
providers=(MetaProvider.VERTEX_AI,),
models=(VERTEX_BATCH_BACKEND,),
mode=Mode.BATCH,
)
)
def test_native_jsonl_round_trips_untouched_and_starts_a_batch(
self, client: BatchClient, resources: ResourceManager, batch_deployments: None
) -> None:
@ -1848,6 +2029,7 @@ class TestVertexNativePassthrough:
),
],
)
@meta(Subject(domain=Domain.LLM_TRANSLATION, route=Route.FILES))
def test_passthrough_upload_is_rejected_outside_a_native_vertex_batch(
self,
content: bytes,

29
tests/e2e/batches/test_managed_files_enforcement_e2e.py
View file

@ -22,9 +22,10 @@ import pytest
from batch_client import BatchClient, FileObject
from batch_cleanup import cleanup_file
from capabilities import batch_model_name, is_managed_id, openai_batch_params
from capabilities import OPENAI_BATCH_BACKEND, batch_model_name, is_managed_id, openai_batch_params
from e2e_config import unique_marker
from e2e_http import FileUploadForm, Result, UnknownApiError, unwrap
from e2e_metadata import Domain, Provider, Route, Subject, meta
from lifecycle import ResourceManager
pytestmark = [pytest.mark.e2e, pytest.mark.managed_files]
@ -64,6 +65,12 @@ def managed_model(client: BatchClient) -> Iterator[str]:
@pytest.mark.covers(UPLOAD_ROW)
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.FILES,
)
)
def test_upload_without_target_model_names_rejected(
client: BatchClient, scoped_key: str, managed_model: str
) -> None:
@ -76,6 +83,12 @@ def test_upload_without_target_model_names_rejected(
@pytest.mark.covers(UPLOAD_ROW)
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.FILES,
)
)
def test_upload_with_model_param_rejected(
client: BatchClient, scoped_key: str, managed_model: str
) -> None:
@ -89,12 +102,26 @@ def test_upload_with_model_param_rejected(
@pytest.mark.covers(ISOLATION_ROW)
@meta(
Subject(
domain=Domain.PROXY_AUTH,
route=Route.FILES,
)
)
def test_raw_provider_file_id_rejected(client: BatchClient, scoped_key: str) -> None:
result = client.retrieve_file("file-e2e-raw-provider-id", key=scoped_key)
expect_api_error(result, 400, "Raw provider file ids cannot be used")
@pytest.mark.covers(ISOLATION_ROW)
@meta(
Subject(
domain=Domain.PROXY_AUTH,
route=Route.FILES,
providers=(Provider.OPENAI,),
models=(OPENAI_BATCH_BACKEND,),
)
)
def test_cross_user_managed_id_denied_owner_allowed(
client: BatchClient, resources: ResourceManager, managed_model: str
) -> None:

3
tests/e2e/claude_code/_basic_messaging.py
View file

@ -31,6 +31,8 @@ from typing import Any, Callable, Mapping, Sequence
import pytest
from e2e_metadata import step
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -74,6 +76,7 @@ def _count_stream_event_deltas(events: Sequence[Mapping[str, Any]]) -> int:
return count
@step("Run Claude Code headless against {models} through the proxy and check every model replies")
def run_basic_messaging_cell(
*,
compat_result,

6
tests/e2e/claude_code/_passthrough.py
View file

@ -58,6 +58,8 @@ from typing import Any, Callable, Dict, Mapping, Optional, Sequence
import pytest
from e2e_metadata import step
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -118,6 +120,10 @@ def foundry_extra_env(proxy_base_url: str) -> Dict[str, str]:
}
@step(
"Run Claude Code headless against {models} through the proxy's native provider passthrough route"
" and check every model replies"
)
def run_passthrough_cell(
*,
compat_result,

10
tests/e2e/claude_code/basic_messaging_non_streaming/test_anthropic.py
View file

@ -21,6 +21,7 @@ the matrix builder still sees three rows for this (feature, provider).
from __future__ import annotations
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._basic_messaging import run_basic_messaging_cell
# Per the PRD: each cell is exercised against three Claude tiers via the
@ -34,6 +35,15 @@ ANTHROPIC_MODELS = [
@pytest.mark.covers("llm.messages.anthropic.basic.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.ANTHROPIC,),
models=tuple(ANTHROPIC_MODELS),
mode=Mode.NONSTREAM,
)
)
def test_basic_messaging_non_streaming_anthropic(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a reply.

10
tests/e2e/claude_code/basic_messaging_non_streaming/test_azure.py
View file

@ -26,6 +26,7 @@ the matrix builder still sees three rows for this (feature, provider).
from __future__ import annotations
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._basic_messaging import run_basic_messaging_cell
# Per-model aliases registered in the LiteLLM proxy's routing config to
@ -40,6 +41,15 @@ AZURE_MODELS = [
@pytest.mark.covers("llm.messages.azure_foundry.basic.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.AZURE_AI,),
models=tuple(AZURE_MODELS),
mode=Mode.NONSTREAM,
)
)
def test_basic_messaging_non_streaming_azure(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a reply.

10
tests/e2e/claude_code/basic_messaging_non_streaming/test_azure_openai.py
View file

@ -25,6 +25,7 @@ green if all three pass.
from __future__ import annotations
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._basic_messaging import run_basic_messaging_cell
AZURE_OPENAI_MODELS = [
@ -34,6 +35,15 @@ AZURE_OPENAI_MODELS = [
]
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.AZURE,),
models=tuple(AZURE_OPENAI_MODELS),
mode=Mode.NONSTREAM,
)
)
def test_basic_messaging_non_streaming_azure_openai(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
non-empty reply from each GPT-5.6 tier."""

10
tests/e2e/claude_code/basic_messaging_non_streaming/test_bedrock_converse.py
View file

@ -21,6 +21,7 @@ the matrix builder still sees three rows for this (feature, provider).
from __future__ import annotations
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._basic_messaging import run_basic_messaging_cell
# Per-model aliases registered in the LiteLLM proxy's routing config to
@ -35,6 +36,15 @@ BEDROCK_CONVERSE_MODELS = [
@pytest.mark.covers("llm.messages.bedrock_converse.basic.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_CONVERSE_MODELS),
mode=Mode.NONSTREAM,
)
)
def test_basic_messaging_non_streaming_bedrock_converse(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a reply."""
run_basic_messaging_cell(

10
tests/e2e/claude_code/basic_messaging_non_streaming/test_bedrock_invoke.py
View file

@ -21,6 +21,7 @@ the matrix builder still sees three rows for this (feature, provider).
from __future__ import annotations
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._basic_messaging import run_basic_messaging_cell
# Per-model aliases registered in the LiteLLM proxy's routing config to
@ -35,6 +36,15 @@ BEDROCK_INVOKE_MODELS = [
@pytest.mark.covers("llm.messages.bedrock_invoke.basic.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_INVOKE_MODELS),
mode=Mode.NONSTREAM,
)
)
def test_basic_messaging_non_streaming_bedrock_invoke(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a reply."""
run_basic_messaging_cell(

10
tests/e2e/claude_code/basic_messaging_non_streaming/test_bedrock_mantle.py
View file

@ -25,6 +25,7 @@ COMPAT_MANTLE_CELLS=1 (see `claude_code._gpt_cells`).
from __future__ import annotations
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._basic_messaging import run_basic_messaging_cell
from claude_code._gpt_cells import skip_unless_mantle_cells_enabled
@ -35,6 +36,15 @@ BEDROCK_MANTLE_MODELS = [
]
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK_MANTLE,),
models=tuple(BEDROCK_MANTLE_MODELS),
mode=Mode.NONSTREAM,
)
)
def test_basic_messaging_non_streaming_bedrock_mantle(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
non-empty reply from each GPT-5.6 tier."""

10
tests/e2e/claude_code/basic_messaging_non_streaming/test_openai.py
View file

@ -22,6 +22,7 @@ green if all three pass.
from __future__ import annotations
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._basic_messaging import run_basic_messaging_cell
from claude_code._gpt_cells import skip_unless_openai_gpt_cells_enabled
@ -32,6 +33,15 @@ OPENAI_MODELS = [
]
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.OPENAI,),
models=tuple(OPENAI_MODELS),
mode=Mode.NONSTREAM,
)
)
def test_basic_messaging_non_streaming_openai(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
non-empty reply from each GPT-5.6 tier."""

10
tests/e2e/claude_code/basic_messaging_non_streaming/test_vertex_ai.py
View file

@ -21,6 +21,7 @@ the matrix builder still sees three rows for this (feature, provider).
from __future__ import annotations
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._basic_messaging import run_basic_messaging_cell
# Per-model aliases registered in the LiteLLM proxy's routing config to
@ -35,6 +36,15 @@ VERTEX_AI_MODELS = [
@pytest.mark.covers("llm.messages.vertex.basic.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.VERTEX_AI,),
models=tuple(VERTEX_AI_MODELS),
mode=Mode.NONSTREAM,
)
)
def test_basic_messaging_non_streaming_vertex_ai(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a reply."""
run_basic_messaging_cell(

2
tests/e2e/claude_code/basic_messaging_non_streaming/test_vertex_ai_gpt.py
View file

@ -16,9 +16,11 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
from e2e_metadata import Domain, Subject, meta
from claude_code._gpt_cells import VERTEX_AI_GPT_NOT_APPLICABLE_REASON
@meta(Subject(domain=Domain.LLM_TRANSLATION))
def test_basic_messaging_non_streaming_vertex_ai_gpt(compat_result):
"""Record the static not_applicable outcome for this cell."""
compat_result.set(

10
tests/e2e/claude_code/basic_messaging_streaming/test_anthropic.py
View file

@ -26,6 +26,7 @@ sees three rows for this (feature, provider).
from __future__ import annotations
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._basic_messaging import run_basic_messaging_cell
ANTHROPIC_MODELS = [
@ -36,6 +37,15 @@ ANTHROPIC_MODELS = [
@pytest.mark.covers("llm.messages.anthropic.basic.stream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.ANTHROPIC,),
models=tuple(ANTHROPIC_MODELS),
mode=Mode.STREAM,
)
)
def test_basic_messaging_streaming_anthropic(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
non-empty streamed reply (one row per Claude tier).

10
tests/e2e/claude_code/basic_messaging_streaming/test_azure.py
View file

@ -20,6 +20,7 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._basic_messaging import run_basic_messaging_cell
AZURE_MODELS = [
@ -30,6 +31,15 @@ AZURE_MODELS = [
@pytest.mark.covers("llm.messages.azure_foundry.basic.stream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.AZURE_AI,),
models=tuple(AZURE_MODELS),
mode=Mode.STREAM,
)
)
def test_basic_messaging_streaming_azure(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
non-empty streamed reply (one row per Claude tier).

10
tests/e2e/claude_code/basic_messaging_streaming/test_azure_openai.py
View file

@ -24,6 +24,7 @@ green if all three pass.
from __future__ import annotations
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._basic_messaging import run_basic_messaging_cell
AZURE_OPENAI_MODELS = [
@ -33,6 +34,15 @@ AZURE_OPENAI_MODELS = [
]
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.AZURE,),
models=tuple(AZURE_OPENAI_MODELS),
mode=Mode.STREAM,
)
)
def test_basic_messaging_streaming_azure_openai(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
non-empty streamed reply from each GPT-5.6 tier."""

10
tests/e2e/claude_code/basic_messaging_streaming/test_bedrock_converse.py
View file

@ -16,6 +16,7 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._basic_messaging import run_basic_messaging_cell
BEDROCK_CONVERSE_MODELS = [
@ -26,6 +27,15 @@ BEDROCK_CONVERSE_MODELS = [
@pytest.mark.covers("llm.messages.bedrock_converse.basic.stream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_CONVERSE_MODELS),
mode=Mode.STREAM,
)
)
def test_basic_messaging_streaming_bedrock_converse(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
non-empty streamed reply (one row per Claude tier).

10
tests/e2e/claude_code/basic_messaging_streaming/test_bedrock_invoke.py
View file

@ -16,6 +16,7 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._basic_messaging import run_basic_messaging_cell
BEDROCK_INVOKE_MODELS = [
@ -26,6 +27,15 @@ BEDROCK_INVOKE_MODELS = [
@pytest.mark.covers("llm.messages.bedrock_invoke.basic.stream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_INVOKE_MODELS),
mode=Mode.STREAM,
)
)
def test_basic_messaging_streaming_bedrock_invoke(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
non-empty streamed reply (one row per Claude tier).

10
tests/e2e/claude_code/basic_messaging_streaming/test_bedrock_mantle.py
View file

@ -25,6 +25,7 @@ COMPAT_MANTLE_CELLS=1 (see `claude_code._gpt_cells`).
from __future__ import annotations
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._basic_messaging import run_basic_messaging_cell
from claude_code._gpt_cells import skip_unless_mantle_cells_enabled
@ -35,6 +36,15 @@ BEDROCK_MANTLE_MODELS = [
]
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK_MANTLE,),
models=tuple(BEDROCK_MANTLE_MODELS),
mode=Mode.STREAM,
)
)
def test_basic_messaging_streaming_bedrock_mantle(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
non-empty streamed reply from each GPT-5.6 tier."""

10
tests/e2e/claude_code/basic_messaging_streaming/test_openai.py
View file

@ -24,6 +24,7 @@ green if all three pass.
from __future__ import annotations
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._basic_messaging import run_basic_messaging_cell
from claude_code._gpt_cells import skip_unless_openai_gpt_cells_enabled
@ -34,6 +35,15 @@ OPENAI_MODELS = [
]
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.OPENAI,),
models=tuple(OPENAI_MODELS),
mode=Mode.STREAM,
)
)
def test_basic_messaging_streaming_openai(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
non-empty streamed reply from each GPT-5.6 tier."""

10
tests/e2e/claude_code/basic_messaging_streaming/test_vertex_ai.py
View file

@ -16,6 +16,7 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._basic_messaging import run_basic_messaging_cell
VERTEX_AI_MODELS = [
@ -26,6 +27,15 @@ VERTEX_AI_MODELS = [
@pytest.mark.covers("llm.messages.vertex.basic.stream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.VERTEX_AI,),
models=tuple(VERTEX_AI_MODELS),
mode=Mode.STREAM,
)
)
def test_basic_messaging_streaming_vertex_ai(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
non-empty streamed reply (one row per Claude tier).

2
tests/e2e/claude_code/basic_messaging_streaming/test_vertex_ai_gpt.py
View file

@ -16,9 +16,11 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
from e2e_metadata import Domain, Subject, meta
from claude_code._gpt_cells import VERTEX_AI_GPT_NOT_APPLICABLE_REASON
@meta(Subject(domain=Domain.LLM_TRANSLATION))
def test_basic_messaging_streaming_vertex_ai_gpt(compat_result):
"""Record the static not_applicable outcome for this cell."""
compat_result.set(

4
tests/e2e/claude_code/cli_driver.py
View file

@ -25,6 +25,8 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Mapping, Optional, Sequence, Tuple, Union
from e2e_metadata import step
from claude_code.rate_limiter import (
RateLimiter,
get_default_limiter,
@ -211,6 +213,7 @@ class DriverResult:
duration_ms: Optional[int] = None
@step("Run Claude Code headless against {model} through the proxy")
def run_claude(
*,
prompt: Optional[str],
@ -395,6 +398,7 @@ def _matches_failure_shape(outcome: ModelResult, pattern: "re.Pattern[str]") ->
return bool(pattern.search(failure_diagnostic(outcome)))
@step("Run Claude Code headless against {models} in parallel through the proxy")
def run_claude_models_parallel(
*,
models: Sequence[str],

10
tests/e2e/claude_code/count_tokens/test_anthropic.py
View file

@ -39,6 +39,7 @@ from __future__ import annotations
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy_client
from claude_code.http_probe import (
assert_count_tokens_shape,
@ -54,6 +55,15 @@ ANTHROPIC_MODELS = [
@pytest.mark.covers("llm.messages.anthropic.count_tokens.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.COUNT_TOKENS,
providers=(Provider.ANTHROPIC,),
models=tuple(ANTHROPIC_MODELS),
mode=Mode.NONSTREAM,
)
)
def test_count_tokens_anthropic(compat_result):
"""Probe `/v1/messages/count_tokens` for each Anthropic tier and
assert the response shape."""

10
tests/e2e/claude_code/count_tokens/test_azure.py
View file

@ -39,6 +39,7 @@ from __future__ import annotations
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy_client
from claude_code.http_probe import (
assert_count_tokens_shape,
@ -54,6 +55,15 @@ AZURE_MODELS = [
@pytest.mark.covers("llm.messages.azure_foundry.count_tokens.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.COUNT_TOKENS,
providers=(Provider.AZURE_AI,),
models=tuple(AZURE_MODELS),
mode=Mode.NONSTREAM,
)
)
def test_count_tokens_azure(compat_result):
"""Probe `/v1/messages/count_tokens` for each Azure (Microsoft Foundry) tier and
assert the response shape."""

10
tests/e2e/claude_code/count_tokens/test_bedrock_converse.py
View file

@ -39,6 +39,7 @@ from __future__ import annotations
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy_client
from claude_code.http_probe import (
assert_count_tokens_shape,
@ -54,6 +55,15 @@ BEDROCK_CONVERSE_MODELS = [
@pytest.mark.covers("llm.messages.bedrock_converse.count_tokens.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.COUNT_TOKENS,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_CONVERSE_MODELS),
mode=Mode.NONSTREAM,
)
)
def test_count_tokens_bedrock_converse(compat_result):
"""Probe `/v1/messages/count_tokens` for each Bedrock (Converse) tier and
assert the response shape."""

10
tests/e2e/claude_code/count_tokens/test_bedrock_invoke.py
View file

@ -39,6 +39,7 @@ from __future__ import annotations
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy_client
from claude_code.http_probe import (
assert_count_tokens_shape,
@ -54,6 +55,15 @@ BEDROCK_INVOKE_MODELS = [
@pytest.mark.covers("llm.messages.bedrock_invoke.count_tokens.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.COUNT_TOKENS,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_INVOKE_MODELS),
mode=Mode.NONSTREAM,
)
)
def test_count_tokens_bedrock_invoke(compat_result):
"""Probe `/v1/messages/count_tokens` for each Bedrock (Invoke) tier and
assert the response shape."""

10
tests/e2e/claude_code/count_tokens/test_vertex_ai.py
View file

@ -39,6 +39,7 @@ from __future__ import annotations
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy_client
from claude_code.http_probe import (
assert_count_tokens_shape,
@ -55,6 +56,15 @@ VERTEX_AI_MODELS = [
@pytest.mark.skip(reason="stage red: Vertex returns not supported for token counting for Claude aliases")
@pytest.mark.covers("llm.messages.vertex.count_tokens.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.COUNT_TOKENS,
providers=(Provider.VERTEX_AI,),
models=tuple(VERTEX_AI_MODELS),
mode=Mode.NONSTREAM,
)
)
def test_count_tokens_vertex_ai(compat_result):
"""Probe `/v1/messages/count_tokens` for each Vertex AI tier and
assert the response shape."""

7
tests/e2e/claude_code/http_probe.py
View file

@ -42,6 +42,7 @@ from e2e_http import (
UnknownApiError,
ValidationError,
)
from e2e_metadata import step
from models import (
AnthropicAssistantTurn,
AnthropicCustomTool,
@ -109,6 +110,7 @@ def _acquire(model: str, rate_limiter: RateLimiter | None) -> None:
limiter.acquire(infer_provider(model))
@step('Count tokens with /v1/messages/count_tokens for {model} on the message "{message}"')
def probe_count_tokens(
*,
client: ProxyClient,
@ -132,6 +134,7 @@ def probe_count_tokens(
)
@step("Send a /v1/messages request to {model} with the tool_search tool declared")
def probe_tool_search(
*,
client: ProxyClient,
@ -228,6 +231,10 @@ def _replay_history(answer: AnthropicMessagesResponse) -> tuple[AnthropicMessage
)
@step(
"Send a /v1/messages request to {model} with the tool_search tool declared,"
" then send its answer back as history in a second request"
)
def probe_tool_search_multiturn(
*,
client: ProxyClient,

10
tests/e2e/claude_code/long_context_1m/test_anthropic.py
View file

@ -58,6 +58,7 @@ from typing import Sequence
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -155,6 +156,15 @@ def _build_long_prompt(target_tokens: int = TARGET_INPUT_TOKENS) -> str:
@pytest.mark.skip(reason="stage red: 1M long_context not green on stage Anthropic path yet (200k sonnet / model alias)")
@pytest.mark.covers("llm.messages.anthropic.long_context_1m.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.ANTHROPIC,),
models=tuple(ANTHROPIC_MODELS),
mode=Mode.NONSTREAM,
)
)
def test_long_context_1m_anthropic(compat_result):
"""Drive the `claude` CLI with a ~210k-token prompt and the
`context-1m-2025-08-07` beta header; assert no 400 / 413 and a

10
tests/e2e/claude_code/long_context_1m/test_azure.py
View file

@ -58,6 +58,7 @@ from typing import Sequence
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -155,6 +156,15 @@ def _build_long_prompt(target_tokens: int = TARGET_INPUT_TOKENS) -> str:
@pytest.mark.skip(reason="stage red: 1M long_context not green on stage Azure Foundry deployments yet")
@pytest.mark.covers("llm.messages.azure_foundry.long_context_1m.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.AZURE_AI,),
models=tuple(AZURE_MODELS),
mode=Mode.NONSTREAM,
)
)
def test_long_context_1m_azure(compat_result):
"""Drive the `claude` CLI (Azure (Microsoft Foundry)) with a ~210k-token prompt and the
`context-1m-2025-08-07` beta header; assert no 400 / 413 and a

10
tests/e2e/claude_code/long_context_1m/test_bedrock_converse.py
View file

@ -58,6 +58,7 @@ from typing import Sequence
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -155,6 +156,15 @@ def _build_long_prompt(target_tokens: int = TARGET_INPUT_TOKENS) -> str:
@pytest.mark.skip(reason="stage red: 1M long_context not green on stage Bedrock Converse deployments yet")
@pytest.mark.covers("llm.messages.bedrock_converse.long_context_1m.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_CONVERSE_MODELS),
mode=Mode.NONSTREAM,
)
)
def test_long_context_1m_bedrock_converse(compat_result):
"""Drive the `claude` CLI (Bedrock (Converse)) with a ~210k-token prompt and the
`context-1m-2025-08-07` beta header; assert no 400 / 413 and a

10
tests/e2e/claude_code/long_context_1m/test_bedrock_invoke.py
View file

@ -58,6 +58,7 @@ from typing import Sequence
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -155,6 +156,15 @@ def _build_long_prompt(target_tokens: int = TARGET_INPUT_TOKENS) -> str:
@pytest.mark.skip(reason="stage red: 1M long_context not green on stage Bedrock Invoke deployments yet")
@pytest.mark.covers("llm.messages.bedrock_invoke.long_context_1m.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_INVOKE_MODELS),
mode=Mode.NONSTREAM,
)
)
def test_long_context_1m_bedrock_invoke(compat_result):
"""Drive the `claude` CLI (Bedrock (Invoke)) with a ~210k-token prompt and the
`context-1m-2025-08-07` beta header; assert no 400 / 413 and a

10
tests/e2e/claude_code/long_context_1m/test_vertex_ai.py
View file

@ -58,6 +58,7 @@ from typing import Sequence
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -155,6 +156,15 @@ def _build_long_prompt(target_tokens: int = TARGET_INPUT_TOKENS) -> str:
@pytest.mark.skip(reason="stage red: 1M long_context not green on stage Vertex deployments yet")
@pytest.mark.covers("llm.messages.vertex.long_context_1m.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.VERTEX_AI,),
models=tuple(VERTEX_AI_MODELS),
mode=Mode.NONSTREAM,
)
)
def test_long_context_1m_vertex_ai(compat_result):
"""Drive the `claude` CLI (Vertex AI) with a ~210k-token prompt and the
`context-1m-2025-08-07` beta header; assert no 400 / 413 and a

10
tests/e2e/claude_code/passthrough/test_anthropic.py
View file

@ -22,6 +22,7 @@ no per-provider transformation is involved.
from __future__ import annotations
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._passthrough import (
ANTHROPIC_PASSTHROUGH_BASE_PATH,
run_passthrough_cell,
@ -34,6 +35,15 @@ ANTHROPIC_MODELS = [
]
@meta(
Subject(
domain=Domain.PASSTHROUGH,
route=Route.PASSTHROUGH,
providers=(Provider.ANTHROPIC,),
models=tuple(ANTHROPIC_MODELS),
mode=Mode.STREAM,
)
)
def test_passthrough_anthropic(compat_result):
"""Drive the `claude` CLI through `{proxy}/anthropic` and assert a reply."""
run_passthrough_cell(

10
tests/e2e/claude_code/passthrough/test_azure.py
View file

@ -42,6 +42,7 @@ from __future__ import annotations
import pytest
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._passthrough import foundry_extra_env, run_passthrough_cell
AZURE_MODELS = [
@ -52,6 +53,15 @@ AZURE_MODELS = [
@pytest.mark.skip(reason="stage red: /azure passthrough drops client headers (e.g. anthropic-version); product gap")
@meta(
Subject(
domain=Domain.PASSTHROUGH,
route=Route.PASSTHROUGH,
providers=(Provider.AZURE,),
models=tuple(AZURE_MODELS),
mode=Mode.STREAM,
)
)
def test_passthrough_azure(compat_result):
"""Drive the `claude` CLI through `{proxy}/azure` and assert a reply."""
run_passthrough_cell(

3
tests/e2e/claude_code/passthrough/test_bedrock_converse.py
View file

@ -18,7 +18,10 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
from e2e_metadata import Domain, Subject, meta
@meta(Subject(domain=Domain.PASSTHROUGH))
def test_passthrough_bedrock_converse(compat_result):
"""Report not_applicable: Claude Code has no Converse-wire mode."""
compat_result.set(

10
tests/e2e/claude_code/passthrough/test_bedrock_invoke.py
View file

@ -23,6 +23,7 @@ cell.
from __future__ import annotations
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._passthrough import bedrock_extra_env, run_passthrough_cell
BEDROCK_INVOKE_MODELS = [
@ -32,6 +33,15 @@ BEDROCK_INVOKE_MODELS = [
]
@meta(
Subject(
domain=Domain.PASSTHROUGH,
route=Route.PASSTHROUGH,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_INVOKE_MODELS),
mode=Mode.STREAM,
)
)
def test_passthrough_bedrock_invoke(compat_result):
"""Drive the `claude` CLI through `{proxy}/bedrock` and assert a reply."""
run_passthrough_cell(

10
tests/e2e/claude_code/passthrough/test_vertex_ai.py
View file

@ -26,6 +26,7 @@ Google and every tier fails with a 401.
from __future__ import annotations
from e2e_metadata import Domain, Mode, Provider, Route, Subject, meta
from claude_code._passthrough import run_passthrough_cell, vertex_extra_env
VERTEX_MODELS = [
@ -35,6 +36,15 @@ VERTEX_MODELS = [
]
@meta(
Subject(
domain=Domain.PASSTHROUGH,
route=Route.PASSTHROUGH,
providers=(Provider.VERTEX_AI,),
models=tuple(VERTEX_MODELS),
mode=Mode.STREAM,
)
)
def test_passthrough_vertex_ai(compat_result):
"""Drive the `claude` CLI through `{proxy}/vertex_ai` and assert a reply."""
run_passthrough_cell(

11
tests/e2e/claude_code/pdf_input/test_anthropic.py
View file

@ -24,6 +24,7 @@ from __future__ import annotations
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -106,6 +107,16 @@ def _build_minimal_pdf(marker: str) -> bytes:
@pytest.mark.covers("llm.messages.anthropic.pdf_input.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.ANTHROPIC,),
models=tuple(ANTHROPIC_MODELS),
capabilities=(Capability.PDF_INPUT,),
mode=Mode.NONSTREAM,
)
)
def test_pdf_input_anthropic(compat_result, tmp_path):
"""Drive the `claude` CLI against the LiteLLM proxy with a PDF
attached via the Read tool and assert the reply references it."""

11
tests/e2e/claude_code/pdf_input/test_azure.py
View file

@ -17,6 +17,7 @@ from __future__ import annotations
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -83,6 +84,16 @@ def _build_minimal_pdf(marker: str) -> bytes:
@pytest.mark.covers("llm.messages.azure_foundry.pdf_input.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.AZURE_AI,),
models=tuple(AZURE_MODELS),
capabilities=(Capability.PDF_INPUT,),
mode=Mode.NONSTREAM,
)
)
def test_pdf_input_azure(compat_result, tmp_path):
base_url, api_key = require_proxy(compat_result)

11
tests/e2e/claude_code/pdf_input/test_bedrock_converse.py
View file

@ -23,6 +23,7 @@ from __future__ import annotations
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -89,6 +90,16 @@ def _build_minimal_pdf(marker: str) -> bytes:
@pytest.mark.covers("llm.messages.bedrock_converse.pdf_input.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_CONVERSE_MODELS),
capabilities=(Capability.PDF_INPUT,),
mode=Mode.NONSTREAM,
)
)
def test_pdf_input_bedrock_converse(compat_result, tmp_path):
base_url, api_key = require_proxy(compat_result)

11
tests/e2e/claude_code/pdf_input/test_bedrock_invoke.py
View file

@ -22,6 +22,7 @@ from __future__ import annotations
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -88,6 +89,16 @@ def _build_minimal_pdf(marker: str) -> bytes:
@pytest.mark.covers("llm.messages.bedrock_invoke.pdf_input.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_INVOKE_MODELS),
capabilities=(Capability.PDF_INPUT,),
mode=Mode.NONSTREAM,
)
)
def test_pdf_input_bedrock_invoke(compat_result, tmp_path):
base_url, api_key = require_proxy(compat_result)

11
tests/e2e/claude_code/pdf_input/test_vertex_ai.py
View file

@ -17,6 +17,7 @@ from __future__ import annotations
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -83,6 +84,16 @@ def _build_minimal_pdf(marker: str) -> bytes:
@pytest.mark.covers("llm.messages.vertex.pdf_input.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.VERTEX_AI,),
models=tuple(VERTEX_AI_MODELS),
capabilities=(Capability.PDF_INPUT,),
mode=Mode.NONSTREAM,
)
)
def test_pdf_input_vertex_ai(compat_result, tmp_path):
base_url, api_key = require_proxy(compat_result)

11
tests/e2e/claude_code/prompt_caching_1h/test_anthropic.py
View file

@ -28,6 +28,7 @@ from typing import Any, Mapping, Optional
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -64,6 +65,16 @@ def _cache_tokens(usage: Optional[Mapping[str, Any]]) -> int:
@pytest.mark.covers("llm.messages.anthropic.prompt_cache_1h.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.ANTHROPIC,),
models=tuple(ANTHROPIC_MODELS),
capabilities=(Capability.PROMPT_CACHING,),
mode=Mode.NONSTREAM,
)
)
def test_prompt_caching_1h_anthropic(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy with the 1h
TTL opt-in env var set, and assert the upstream usage block

11
tests/e2e/claude_code/prompt_caching_1h/test_azure.py
View file

@ -19,6 +19,7 @@ from typing import Any, Mapping, Optional
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -48,6 +49,16 @@ def _cache_tokens(usage: Optional[Mapping[str, Any]]) -> int:
@pytest.mark.covers("llm.messages.azure_foundry.prompt_cache_1h.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.AZURE_AI,),
models=tuple(AZURE_MODELS),
capabilities=(Capability.PROMPT_CACHING,),
mode=Mode.NONSTREAM,
)
)
def test_prompt_caching_1h_azure(compat_result):
base_url, api_key = require_proxy(compat_result)

11
tests/e2e/claude_code/prompt_caching_1h/test_bedrock_converse.py
View file

@ -24,6 +24,7 @@ from typing import Any, Mapping, Optional
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -56,6 +57,16 @@ def _cache_tokens(usage: Optional[Mapping[str, Any]]) -> int:
@pytest.mark.covers("llm.messages.bedrock_converse.prompt_cache_1h.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_CONVERSE_MODELS),
capabilities=(Capability.PROMPT_CACHING,),
mode=Mode.NONSTREAM,
)
)
def test_prompt_caching_1h_bedrock_converse(compat_result):
base_url, api_key = require_proxy(compat_result)

11
tests/e2e/claude_code/prompt_caching_1h/test_bedrock_invoke.py
View file

@ -26,6 +26,7 @@ from typing import Any, Mapping, Optional
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -60,6 +61,16 @@ def _cache_tokens(usage: Optional[Mapping[str, Any]]) -> int:
@pytest.mark.covers("llm.messages.bedrock_invoke.prompt_cache_1h.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_INVOKE_MODELS),
capabilities=(Capability.PROMPT_CACHING,),
mode=Mode.NONSTREAM,
)
)
def test_prompt_caching_1h_bedrock_invoke(compat_result):
base_url, api_key = require_proxy(compat_result)

11
tests/e2e/claude_code/prompt_caching_1h/test_vertex_ai.py
View file

@ -19,6 +19,7 @@ from typing import Any, Mapping, Optional
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -48,6 +49,16 @@ def _cache_tokens(usage: Optional[Mapping[str, Any]]) -> int:
@pytest.mark.covers("llm.messages.vertex.prompt_cache_1h.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.VERTEX_AI,),
models=tuple(VERTEX_AI_MODELS),
capabilities=(Capability.PROMPT_CACHING,),
mode=Mode.NONSTREAM,
)
)
def test_prompt_caching_1h_vertex_ai(compat_result):
base_url, api_key = require_proxy(compat_result)

11
tests/e2e/claude_code/prompt_caching_5m/test_anthropic.py
View file

@ -26,6 +26,7 @@ from typing import Any, Mapping, Optional
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -55,6 +56,16 @@ def _cache_tokens(usage: Optional[Mapping[str, Any]]) -> int:
@pytest.mark.covers("llm.messages.anthropic.prompt_cache_5m.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.ANTHROPIC,),
models=tuple(ANTHROPIC_MODELS),
capabilities=(Capability.PROMPT_CACHING,),
mode=Mode.NONSTREAM,
)
)
def test_prompt_caching_5m_anthropic(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert the
upstream usage block surfaces a non-zero cache token count."""

11
tests/e2e/claude_code/prompt_caching_5m/test_azure.py
View file

@ -26,6 +26,7 @@ from typing import Any, Mapping, Optional
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -53,6 +54,16 @@ def _cache_tokens(usage: Optional[Mapping[str, Any]]) -> int:
@pytest.mark.covers("llm.messages.azure_foundry.prompt_cache_5m.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.AZURE_AI,),
models=tuple(AZURE_MODELS),
capabilities=(Capability.PROMPT_CACHING,),
mode=Mode.NONSTREAM,
)
)
def test_prompt_caching_5m_azure(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert the
upstream usage block surfaces a non-zero cache token count."""

11
tests/e2e/claude_code/prompt_caching_5m/test_bedrock_converse.py
View file

@ -19,6 +19,7 @@ from typing import Any, Mapping, Optional
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -46,6 +47,16 @@ def _cache_tokens(usage: Optional[Mapping[str, Any]]) -> int:
@pytest.mark.covers("llm.messages.bedrock_converse.prompt_cache_5m.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_CONVERSE_MODELS),
capabilities=(Capability.PROMPT_CACHING,),
mode=Mode.NONSTREAM,
)
)
def test_prompt_caching_5m_bedrock_converse(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert the
upstream usage block surfaces a non-zero cache token count."""

11
tests/e2e/claude_code/prompt_caching_5m/test_bedrock_invoke.py
View file

@ -19,6 +19,7 @@ from typing import Any, Mapping, Optional
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -46,6 +47,16 @@ def _cache_tokens(usage: Optional[Mapping[str, Any]]) -> int:
@pytest.mark.covers("llm.messages.bedrock_invoke.prompt_cache_5m.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_INVOKE_MODELS),
capabilities=(Capability.PROMPT_CACHING,),
mode=Mode.NONSTREAM,
)
)
def test_prompt_caching_5m_bedrock_invoke(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert the
upstream usage block surfaces a non-zero cache token count."""

11
tests/e2e/claude_code/prompt_caching_5m/test_vertex_ai.py
View file

@ -19,6 +19,7 @@ from typing import Any, Mapping, Optional
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -46,6 +47,16 @@ def _cache_tokens(usage: Optional[Mapping[str, Any]]) -> int:
@pytest.mark.covers("llm.messages.vertex.prompt_cache_5m.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.VERTEX_AI,),
models=tuple(VERTEX_AI_MODELS),
capabilities=(Capability.PROMPT_CACHING,),
mode=Mode.NONSTREAM,
)
)
def test_prompt_caching_5m_vertex_ai(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert the
upstream usage block surfaces a non-zero cache token count."""

11
tests/e2e/claude_code/structured_outputs/test_anthropic.py
View file

@ -53,6 +53,7 @@ from typing import Any, Mapping, Optional, Sequence, Tuple
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -151,6 +152,16 @@ def _validate_against_schema(
@pytest.mark.covers("llm.messages.anthropic.structured_output.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.ANTHROPIC,),
models=tuple(ANTHROPIC_MODELS),
capabilities=(Capability.RESPONSE_SCHEMA,),
mode=Mode.NONSTREAM,
)
)
def test_structured_outputs_anthropic(compat_result):
"""Drive `claude --json-schema ...` against the LiteLLM proxy and
assert the trailing `result` event contains a schema-conforming

11
tests/e2e/claude_code/structured_outputs/test_azure.py
View file

@ -53,6 +53,7 @@ from typing import Any, Mapping, Optional, Sequence, Tuple
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -151,6 +152,16 @@ def _validate_against_schema(
@pytest.mark.covers("llm.messages.azure_foundry.structured_output.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.AZURE_AI,),
models=tuple(AZURE_MODELS),
capabilities=(Capability.RESPONSE_SCHEMA,),
mode=Mode.NONSTREAM,
)
)
def test_structured_outputs_azure(compat_result):
"""Drive `claude --json-schema ...` against the LiteLLM proxy and
assert the trailing `result` event contains a schema-conforming

12
tests/e2e/claude_code/structured_outputs/test_bedrock_converse.py
View file

@ -53,6 +53,8 @@ from typing import Any, Mapping, Optional, Sequence, Tuple
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -151,6 +153,16 @@ def _validate_against_schema(
@pytest.mark.covers("llm.messages.bedrock_converse.structured_output.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_CONVERSE_MODELS),
capabilities=(Capability.RESPONSE_SCHEMA,),
mode=Mode.NONSTREAM,
)
)
def test_structured_outputs_bedrock_converse(compat_result):
"""Drive `claude --json-schema ...` against the LiteLLM proxy and
assert the trailing `result` event contains a schema-conforming

12
tests/e2e/claude_code/structured_outputs/test_bedrock_invoke.py
View file

@ -53,6 +53,8 @@ from typing import Any, Mapping, Optional, Sequence, Tuple
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -151,6 +153,16 @@ def _validate_against_schema(
@pytest.mark.covers("llm.messages.bedrock_invoke.structured_output.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_INVOKE_MODELS),
capabilities=(Capability.RESPONSE_SCHEMA,),
mode=Mode.NONSTREAM,
)
)
def test_structured_outputs_bedrock_invoke(compat_result):
"""Drive `claude --json-schema ...` against the LiteLLM proxy and
assert the trailing `result` event contains a schema-conforming

12
tests/e2e/claude_code/structured_outputs/test_vertex_ai.py
View file

@ -53,6 +53,8 @@ from typing import Any, Mapping, Optional, Sequence, Tuple
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -151,6 +153,16 @@ def _validate_against_schema(
@pytest.mark.covers("llm.messages.vertex.structured_output.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.VERTEX_AI,),
models=tuple(VERTEX_AI_MODELS),
capabilities=(Capability.RESPONSE_SCHEMA,),
mode=Mode.NONSTREAM,
)
)
def test_structured_outputs_vertex_ai(compat_result):
"""Drive `claude --json-schema ...` against the LiteLLM proxy and
assert the trailing `result` event contains a schema-conforming

12
tests/e2e/claude_code/thinking/test_anthropic.py
View file

@ -24,6 +24,8 @@ from typing import Any, Mapping, Sequence
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -75,6 +77,16 @@ def _has_thinking_block(events: Sequence[Mapping[str, Any]]) -> bool:
@pytest.mark.covers("llm.messages.anthropic.thinking.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.ANTHROPIC,),
models=tuple(ANTHROPIC_MODELS),
capabilities=(Capability.REASONING,),
mode=Mode.NONSTREAM,
)
)
def test_thinking_anthropic(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy with thinking
enabled and assert a `thinking` content block was emitted."""

12
tests/e2e/claude_code/thinking/test_azure.py
View file

@ -27,6 +27,8 @@ from typing import Any, Mapping, Sequence
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -63,6 +65,16 @@ def _has_thinking_block(events: Sequence[Mapping[str, Any]]) -> bool:
@pytest.mark.covers("llm.messages.azure_foundry.thinking.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.AZURE_AI,),
models=tuple(AZURE_MODELS),
capabilities=(Capability.REASONING,),
mode=Mode.NONSTREAM,
)
)
def test_thinking_azure(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy with thinking
enabled and assert a `thinking` content block was emitted."""

12
tests/e2e/claude_code/thinking/test_bedrock_converse.py
View file

@ -19,6 +19,8 @@ from typing import Any, Mapping, Sequence
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -55,6 +57,16 @@ def _has_thinking_block(events: Sequence[Mapping[str, Any]]) -> bool:
@pytest.mark.covers("llm.messages.bedrock_converse.thinking.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_CONVERSE_MODELS),
capabilities=(Capability.REASONING,),
mode=Mode.NONSTREAM,
)
)
def test_thinking_bedrock_converse(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy with thinking
enabled and assert a `thinking` content block was emitted."""

12
tests/e2e/claude_code/thinking/test_bedrock_invoke.py
View file

@ -19,6 +19,8 @@ from typing import Any, Mapping, Sequence
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -55,6 +57,16 @@ def _has_thinking_block(events: Sequence[Mapping[str, Any]]) -> bool:
@pytest.mark.covers("llm.messages.bedrock_invoke.thinking.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_INVOKE_MODELS),
capabilities=(Capability.REASONING,),
mode=Mode.NONSTREAM,
)
)
def test_thinking_bedrock_invoke(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy with thinking
enabled and assert a `thinking` content block was emitted."""

12
tests/e2e/claude_code/thinking/test_vertex_ai.py
View file

@ -19,6 +19,8 @@ from typing import Any, Mapping, Sequence
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -55,6 +57,16 @@ def _has_thinking_block(events: Sequence[Mapping[str, Any]]) -> bool:
@pytest.mark.covers("llm.messages.vertex.thinking.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.VERTEX_AI,),
models=tuple(VERTEX_AI_MODELS),
capabilities=(Capability.REASONING,),
mode=Mode.NONSTREAM,
)
)
def test_thinking_vertex_ai(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy with thinking
enabled and assert a `thinking` content block was emitted."""

12
tests/e2e/claude_code/thinking_with_tool_use/test_anthropic.py
View file

@ -28,6 +28,8 @@ from typing import Any, Mapping, Sequence
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -89,6 +91,16 @@ def _has_block_type(
@pytest.mark.covers("llm.messages.anthropic.thinking_with_tool_use.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.ANTHROPIC,),
models=tuple(ANTHROPIC_MODELS),
capabilities=(Capability.REASONING, Capability.FUNCTION_CALLING,),
mode=Mode.NONSTREAM,
)
)
def test_thinking_with_tool_use_anthropic(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy with thinking
enabled and tool use, and assert both `thinking` and `tool_use`

12
tests/e2e/claude_code/thinking_with_tool_use/test_azure.py
View file

@ -22,6 +22,8 @@ from typing import Any, Mapping, Sequence
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -70,6 +72,16 @@ def _has_block_type(
@pytest.mark.covers("llm.messages.azure_foundry.thinking_with_tool_use.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.AZURE_AI,),
models=tuple(AZURE_MODELS),
capabilities=(Capability.REASONING, Capability.FUNCTION_CALLING,),
mode=Mode.NONSTREAM,
)
)
def test_thinking_with_tool_use_azure(compat_result):
base_url, api_key = require_proxy(compat_result)

12
tests/e2e/claude_code/thinking_with_tool_use/test_bedrock_converse.py
View file

@ -27,6 +27,8 @@ from typing import Any, Mapping, Sequence
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -75,6 +77,16 @@ def _has_block_type(
@pytest.mark.covers("llm.messages.bedrock_converse.thinking_with_tool_use.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_CONVERSE_MODELS),
capabilities=(Capability.REASONING, Capability.FUNCTION_CALLING,),
mode=Mode.NONSTREAM,
)
)
def test_thinking_with_tool_use_bedrock_converse(compat_result):
base_url, api_key = require_proxy(compat_result)

12
tests/e2e/claude_code/thinking_with_tool_use/test_bedrock_invoke.py
View file

@ -29,6 +29,8 @@ from typing import Any, Mapping, Sequence
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -77,6 +79,16 @@ def _has_block_type(
@pytest.mark.covers("llm.messages.bedrock_invoke.thinking_with_tool_use.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_INVOKE_MODELS),
capabilities=(Capability.REASONING, Capability.FUNCTION_CALLING,),
mode=Mode.NONSTREAM,
)
)
def test_thinking_with_tool_use_bedrock_invoke(compat_result):
base_url, api_key = require_proxy(compat_result)

12
tests/e2e/claude_code/thinking_with_tool_use/test_vertex_ai.py
View file

@ -27,6 +27,8 @@ from typing import Any, Mapping, Sequence
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -75,6 +77,16 @@ def _has_block_type(
@pytest.mark.covers("llm.messages.vertex.thinking_with_tool_use.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.VERTEX_AI,),
models=tuple(VERTEX_AI_MODELS),
capabilities=(Capability.REASONING, Capability.FUNCTION_CALLING,),
mode=Mode.NONSTREAM,
)
)
def test_thinking_with_tool_use_vertex_ai(compat_result):
base_url, api_key = require_proxy(compat_result)

12
tests/e2e/claude_code/tool_search/test_anthropic.py
View file

@ -45,6 +45,8 @@ from __future__ import annotations
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy_client
from claude_code.http_probe import (
assert_tool_search_shape,
@ -60,6 +62,16 @@ ANTHROPIC_MODELS = [
@pytest.mark.covers("llm.messages.anthropic.tool_search.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.ANTHROPIC,),
models=tuple(ANTHROPIC_MODELS),
capabilities=(Capability.TOOL_SEARCH,),
mode=Mode.NONSTREAM,
)
)
def test_tool_search_anthropic(compat_result):
"""Probe `/v1/messages` with a `tool_search_tool_regex_20251119`
tool and assert the proxy + upstream accept it for every Anthropic

12
tests/e2e/claude_code/tool_search/test_azure.py
View file

@ -45,6 +45,8 @@ from __future__ import annotations
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy_client
from claude_code.http_probe import (
assert_tool_search_shape,
@ -61,6 +63,16 @@ AZURE_MODELS = [
@pytest.mark.skip(reason="stage red: Azure Foundry tool_search_server not supported in workspace for probed models")
@pytest.mark.covers("llm.messages.azure_foundry.tool_search.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.AZURE_AI,),
models=tuple(AZURE_MODELS),
capabilities=(Capability.TOOL_SEARCH,),
mode=Mode.NONSTREAM,
)
)
def test_tool_search_azure(compat_result):
"""Probe `/v1/messages` with a `tool_search_tool_regex_20251119`
tool and assert the proxy + upstream accept it for every Azure (Microsoft Foundry)

12
tests/e2e/claude_code/tool_search/test_bedrock_converse.py
View file

@ -45,6 +45,8 @@ from __future__ import annotations
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy_client
from claude_code.http_probe import (
assert_tool_search_shape,
@ -60,6 +62,16 @@ BEDROCK_CONVERSE_MODELS = [
@pytest.mark.covers("llm.messages.bedrock_converse.tool_search.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_CONVERSE_MODELS),
capabilities=(Capability.TOOL_SEARCH,),
mode=Mode.NONSTREAM,
)
)
def test_tool_search_bedrock_converse(compat_result):
"""Probe `/v1/messages` with a `tool_search_tool_regex_20251119`
tool and assert the proxy + upstream accept it for every Bedrock (Converse)

22
tests/e2e/claude_code/tool_search/test_bedrock_invoke.py
View file

@ -50,6 +50,8 @@ from __future__ import annotations
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy_client
from claude_code.http_probe import (
assert_tool_search_replay_shape,
@ -67,6 +69,16 @@ BEDROCK_INVOKE_MODELS = [
@pytest.mark.covers("llm.messages.bedrock_invoke.tool_search.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_INVOKE_MODELS),
capabilities=(Capability.TOOL_SEARCH,),
mode=Mode.NONSTREAM,
)
)
def test_tool_search_bedrock_invoke(compat_result):
"""Probe `/v1/messages` with a `tool_search_tool_regex_20251119`
tool and assert the proxy + upstream accept it for every Bedrock (Invoke)
@ -90,6 +102,16 @@ def test_tool_search_bedrock_invoke(compat_result):
@pytest.mark.covers("llm.messages.bedrock_invoke.tool_search_history.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.BEDROCK,),
models=tuple(BEDROCK_INVOKE_MODELS),
capabilities=(Capability.TOOL_SEARCH,),
mode=Mode.NONSTREAM,
)
)
def test_tool_search_history_bedrock_invoke(compat_result):
"""Send the tool-search request, take the real assistant turn back, and
replay it as history with the tools still declared.

12
tests/e2e/claude_code/tool_search/test_vertex_ai.py
View file

@ -45,6 +45,8 @@ from __future__ import annotations
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy_client
from claude_code.http_probe import (
assert_tool_search_shape,
@ -61,6 +63,16 @@ VERTEX_AI_MODELS = [
@pytest.mark.skip(reason="stage red: Vertex rejects tool_search when deployment extra_headers inject context-1m beta; product/config")
@pytest.mark.covers("llm.messages.vertex.tool_search.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.VERTEX_AI,),
models=tuple(VERTEX_AI_MODELS),
capabilities=(Capability.TOOL_SEARCH,),
mode=Mode.NONSTREAM,
)
)
def test_tool_search_vertex_ai(compat_result):
"""Probe `/v1/messages` with a `tool_search_tool_regex_20251119`
tool and assert the proxy + upstream accept it for every Vertex AI

11
tests/e2e/claude_code/tool_use/test_anthropic.py
View file

@ -19,6 +19,7 @@ from typing import Any, Mapping, Sequence
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -72,6 +73,16 @@ def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool:
@pytest.mark.covers("llm.messages.anthropic.tool_use.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.ANTHROPIC,),
models=tuple(ANTHROPIC_MODELS),
capabilities=(Capability.FUNCTION_CALLING,),
mode=Mode.NONSTREAM,
)
)
def test_tool_use_anthropic(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
tool call was emitted on the wire."""

11
tests/e2e/claude_code/tool_use/test_azure.py
View file

@ -23,6 +23,7 @@ from typing import Any, Mapping, Sequence
import pytest
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
@ -66,6 +67,16 @@ def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool:
@pytest.mark.covers("llm.messages.azure_foundry.tool_use.nonstream.works")
@meta(
Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.MESSAGES,
providers=(Provider.AZURE_AI,),
models=tuple(AZURE_MODELS),
capabilities=(Capability.FUNCTION_CALLING,),
mode=Mode.NONSTREAM,
)
)
def test_tool_use_azure(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
tool call was emitted on the wire."""

Some files were not shown because too many files have changed in this diff Show more