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

refactor(e2e/claude_code): dedupe tool_use cell body into shared _tool_use module

Browse source
This commit is contained in:
mateo-berri 2026-07-16 18:13:23 -07:00
parent a017b95e2f
commit 62c00fbf83
17 changed files with 264 additions and 1387 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

168
tests/e2e/claude_code/_tool_use.py Normal file
View file

@ -0,0 +1,168 @@
"""Shared body for the `tool_use` / `tool_use_streaming` x <provider> compat cells.
Every tool_use cell follows the same skeleton:
1. Read the proxy base URL + API key from env, fail-early if missing.
2. Ask each model tier (fanned out via `run_claude_models_parallel`)
to invoke the built-in `Bash` tool.
3. Inspect each model's outcome and report one `compat_result` row per
model; `ClaudeCLIError`, non-zero exit, and a missing `tool_use`
content block are all per-model fails, everything else is a
per-model pass.
4. Surface a joined failure message via `pytest.fail(...)` so the
pytest run also goes red.
The streaming variant (`verify_streaming=True`) adds the
`--include-partial-messages` CLI flag and additionally requires that
`input_json_delta` stream events were observed; zero deltas means the
proxy collapsed the streamed tool input into a single complete block or
stripped fine-grained tool streaming.
Security rationale for `TOOL_USE_ARGS`: the prompt asks for the exact
command `echo pong`, the `--allowed-tools` rule restricts the Bash tool
to that command, and `--permission-mode dontAsk` auto-denies anything
else the model returns instead of executing it. `dontAsk` mode in
headless `--print` mode only runs tools matching an explicit `allow`
rule (plus the built-in read-only set), so a compromised provider
response cannot turn the `Bash` allowlist into arbitrary host execution
(which would expose `docker inspect compat-proxy` /
`/proc/<proxy_pid>/environ` and thereby provider credentials living in
the proxy container).
The conftest infers `(feature_id, provider)` purely from the test file
path, so each per-provider file just declares its model list and calls
`run_tool_use_cell(...)`. This keeps all cell logic in one place; a
future tweak to the outcome checks or the failure-loop shape now
propagates to every cell automatically.
The leading underscore in the filename is what keeps pytest from
collecting this module as a test file.
"""
from __future__ import annotations
from typing import Any, Callable, Mapping, Sequence
import pytest
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
DriverResult,
failure_diagnostic,
run_claude_models_parallel,
)
ClaudeRunner = Callable[..., Mapping[str, DriverResult | ClaudeCLIError]]
TOOL_USE_PROMPT = "Use the Bash tool to run the command `echo pong` and report what it printed."
TOOL_USE_ARGS = (
"--allowed-tools",
"Bash(echo pong)",
"--permission-mode",
"dontAsk",
)
TOOL_USE_STREAMING_ARGS = TOOL_USE_ARGS + ("--include-partial-messages",)
def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool:
"""Walk the stream-json events and return True if any assistant
message included a `tool_use` content block."""
for event in events:
if event.get("type") != "assistant":
continue
message = event.get("message") or {}
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and block.get("type") == "tool_use":
return True
return False
def _count_input_json_deltas(events: Sequence[Mapping[str, Any]]) -> int:
"""Count `input_json_delta` records among the `stream_event`
entries. Zero means the proxy collapsed the streamed tool input
into a single complete block instead of forwarding the incremental
deltas the upstream emitted."""
inner_events = (event.get("event") for event in events if event.get("type") == "stream_event")
return sum(
1
for inner in inner_events
if isinstance(inner, Mapping)
and inner.get("type") == "content_block_delta"
and isinstance(inner.get("delta"), Mapping)
and inner["delta"].get("type") == "input_json_delta"
)
def run_tool_use_cell(
*,
compat_result,
models: Sequence[str],
verify_streaming: bool = False,
env: Mapping[str, str] | None = None,
runner: ClaudeRunner = run_claude_models_parallel,
) -> None:
"""Run the shared `tool_use` / `tool_use_streaming` x <provider> cell body.
With ``verify_streaming=False`` the cell asserts that a `tool_use`
content block came back over the wire. With ``verify_streaming=True``
it additionally passes ``--include-partial-messages`` and asserts
that `input_json_delta` stream events were observed for the block,
which catches both known gateway regressions: buffering the streamed
tool input into one complete block and stripping the
fine-grained-tool-streaming beta so the upstream falls back to
non-streaming tool_use.
"""
base_url, api_key = require_proxy(compat_result, env=env)
extra_args = TOOL_USE_STREAMING_ARGS if verify_streaming else TOOL_USE_ARGS
outcomes = runner(
models=models,
prompt=TOOL_USE_PROMPT,
base_url=base_url,
api_key=api_key,
extra_args=extra_args,
)
failures = []
for model in models:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not _has_tool_use_event(outcome.events):
error = f"[{model}] no tool_use content block observed in stream-json events"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if verify_streaming and _count_input_json_deltas(outcome.events) == 0:
error = (
f"[{model}] no input_json_delta stream events observed; proxy "
f"likely buffered the tool input into a complete block or "
f"stripped fine-grained tool streaming"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)

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

@ -5,6 +5,10 @@ to Anthropic, ask Claude to invoke a built-in tool (`Bash`), and assert
that the upstream returned a `tool_use` content block. This proves the
proxy preserves Claude Code's tool-call wire shape end-to-end.
Bash is restricted to the exact command `echo pong` plus
`--permission-mode dontAsk`; the security rationale lives in
`claude_code/_tool_use.py` alongside the shared cell body.
The (feature, provider) for this cell is inferred from the file path by
`tests/e2e/claude_code/conftest.py`:
@ -15,16 +19,9 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
from typing import Any, Mapping, Sequence
import pytest
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
from claude_code._tool_use import run_tool_use_cell
ANTHROPIC_MODELS = [
@ -33,82 +30,9 @@ ANTHROPIC_MODELS = [
"claude-opus-4-7",
]
# Built-in tool-use prompt: ask Claude to use the `Bash` tool. The CLI
# allow-lists the tool via `--allowed-tools` so the run completes without
# an interactive permission prompt.
TOOL_USE_PROMPT = (
"Use the Bash tool to run the command `echo pong` and report what it printed."
)
# Restrict the Bash tool to the exact command `echo pong` and put the
# CLI in `dontAsk` mode so anything else the model returns is auto-
# denied instead of executed. `dontAsk` mode in headless `--print` mode
# only runs tools matching an explicit `allow` rule (plus the built-in
# read-only set), so a compromised provider response cannot turn the
# `Bash` allowlist into arbitrary host execution (which would expose
# `docker inspect compat-proxy` / `/proc/<proxy_pid>/environ` and
# thereby provider credentials living in the proxy container).
TOOL_USE_ARGS = [
"--allowed-tools",
"Bash(echo pong)",
"--permission-mode",
"dontAsk",
]
def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool:
"""Walk the stream-json events and return True if any assistant
message included a `tool_use` content block."""
for event in events:
if event.get("type") != "assistant":
continue
message = event.get("message") or {}
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and block.get("type") == "tool_use":
return True
return False
@pytest.mark.covers("llm.messages.anthropic.tool_use.nonstream.works")
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."""
base_url, api_key = require_proxy(compat_result)
outcomes = run_claude_models_parallel(
models=ANTHROPIC_MODELS,
prompt=TOOL_USE_PROMPT,
base_url=base_url,
api_key=api_key,
extra_args=TOOL_USE_ARGS,
)
failures = []
for model in ANTHROPIC_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not _has_tool_use_event(outcome.events):
error = (
f"[{model}] no tool_use content block observed in stream-json events"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)
run_tool_use_cell(compat_result=compat_result, models=ANTHROPIC_MODELS)

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

@ -9,6 +9,10 @@ Foundry's Anthropic deployments support function/tool calling
identically to anthropic.com; LiteLLM's `azure_ai/claude-*` route
inherits the full Anthropic tool-use transformation.
Bash is restricted to the exact command `echo pong` plus
`--permission-mode dontAsk`; see `claude_code/_tool_use.py` for the
security rationale.
The (feature, provider) for this cell is inferred from the file path by
`tests/e2e/claude_code/conftest.py`:
@ -19,16 +23,9 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
from typing import Any, Mapping, Sequence
import pytest
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
from claude_code._tool_use import run_tool_use_cell
AZURE_MODELS = [
@ -37,72 +34,9 @@ AZURE_MODELS = [
"claude-opus-4-7-azure",
]
TOOL_USE_PROMPT = (
"Use the Bash tool to run the command `echo pong` and report what it printed."
)
# Bash is restricted to the exact command `echo pong` + `dontAsk`
# permission mode; see `tool_use/test_anthropic.py` for the security
# rationale.
TOOL_USE_ARGS = [
"--allowed-tools",
"Bash(echo pong)",
"--permission-mode",
"dontAsk",
]
def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool:
for event in events:
if event.get("type") != "assistant":
continue
message = event.get("message") or {}
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and block.get("type") == "tool_use":
return True
return False
@pytest.mark.covers("llm.messages.azure_foundry.tool_use.nonstream.works")
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."""
base_url, api_key = require_proxy(compat_result)
outcomes = run_claude_models_parallel(
models=AZURE_MODELS,
prompt=TOOL_USE_PROMPT,
base_url=base_url,
api_key=api_key,
extra_args=TOOL_USE_ARGS,
)
failures = []
for model in AZURE_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not _has_tool_use_event(outcome.events):
error = (
f"[{model}] no tool_use content block observed in stream-json events"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)
run_tool_use_cell(compat_result=compat_result, models=AZURE_MODELS)

75
tests/e2e/claude_code/tool_use/test_azure_openai.py
View file

@ -12,7 +12,7 @@ route reuses the OpenAI tool translation on top of Azure's deployment
addressing and auth.
Bash is restricted to the exact command `echo pong` plus
`--permission-mode dontAsk`; see `tool_use/test_anthropic.py` for the
`--permission-mode dontAsk`; see `claude_code/_tool_use.py` for the
security rationale.
The (feature, provider) for this cell is inferred from the file path by
@ -25,16 +25,7 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
from typing import Any, Mapping, Sequence
import pytest
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
from claude_code._tool_use import run_tool_use_cell
AZURE_OPENAI_MODELS = [
"gpt-5-6-sol-azure-openai",
@ -42,68 +33,8 @@ AZURE_OPENAI_MODELS = [
"gpt-5-6-luna-azure-openai",
]
TOOL_USE_PROMPT = (
"Use the Bash tool to run the command `echo pong` and report what it printed."
)
TOOL_USE_ARGS = [
"--allowed-tools",
"Bash(echo pong)",
"--permission-mode",
"dontAsk",
]
def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool:
for event in events:
if event.get("type") != "assistant":
continue
message = event.get("message") or {}
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and block.get("type") == "tool_use":
return True
return False
def test_tool_use_azure_openai(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
tool call was emitted on the wire by each GPT-5.6 tier."""
proxy = require_proxy(compat_result)
outcomes = run_claude_models_parallel(
models=AZURE_OPENAI_MODELS,
prompt=TOOL_USE_PROMPT,
base_url=proxy.base_url,
api_key=proxy.api_key,
extra_args=TOOL_USE_ARGS,
)
failures = []
for model in AZURE_OPENAI_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not _has_tool_use_event(outcome.events):
error = (
f"[{model}] no tool_use content block observed in stream-json events"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)
run_tool_use_cell(compat_result=compat_result, models=AZURE_OPENAI_MODELS)

78
tests/e2e/claude_code/tool_use/test_bedrock_converse.py
View file

@ -5,6 +5,10 @@ Claude requests to AWS Bedrock via the unified `Converse` API path, ask
Claude to invoke a built-in tool (`Bash`), and assert that the upstream
returned a `tool_use` content block.
Bash is restricted to the exact command `echo pong` plus
`--permission-mode dontAsk`; see `claude_code/_tool_use.py` for the
security rationale.
The (feature, provider) for this cell is inferred from the file path by
`tests/e2e/claude_code/conftest.py`:
@ -15,16 +19,9 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
from typing import Any, Mapping, Sequence
import pytest
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
from claude_code._tool_use import run_tool_use_cell
BEDROCK_CONVERSE_MODELS = [
@ -33,72 +30,9 @@ BEDROCK_CONVERSE_MODELS = [
"claude-opus-4-7-bedrock-converse",
]
TOOL_USE_PROMPT = (
"Use the Bash tool to run the command `echo pong` and report what it printed."
)
# Bash is restricted to the exact command `echo pong` + `dontAsk`
# permission mode; see `tool_use/test_anthropic.py` for the security
# rationale.
TOOL_USE_ARGS = [
"--allowed-tools",
"Bash(echo pong)",
"--permission-mode",
"dontAsk",
]
def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool:
for event in events:
if event.get("type") != "assistant":
continue
message = event.get("message") or {}
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and block.get("type") == "tool_use":
return True
return False
@pytest.mark.covers("llm.messages.bedrock_converse.tool_use.nonstream.works")
def test_tool_use_bedrock_converse(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
tool call was emitted on the wire."""
base_url, api_key = require_proxy(compat_result)
outcomes = run_claude_models_parallel(
models=BEDROCK_CONVERSE_MODELS,
prompt=TOOL_USE_PROMPT,
base_url=base_url,
api_key=api_key,
extra_args=TOOL_USE_ARGS,
)
failures = []
for model in BEDROCK_CONVERSE_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not _has_tool_use_event(outcome.events):
error = (
f"[{model}] no tool_use content block observed in stream-json events"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)
run_tool_use_cell(compat_result=compat_result, models=BEDROCK_CONVERSE_MODELS)

78
tests/e2e/claude_code/tool_use/test_bedrock_invoke.py
View file

@ -5,6 +5,10 @@ Claude requests to AWS Bedrock via the legacy `InvokeModel` API path,
ask Claude to invoke a built-in tool (`Bash`), and assert that the
upstream returned a `tool_use` content block.
Bash is restricted to the exact command `echo pong` plus
`--permission-mode dontAsk`; see `claude_code/_tool_use.py` for the
security rationale.
The (feature, provider) for this cell is inferred from the file path by
`tests/e2e/claude_code/conftest.py`:
@ -15,16 +19,9 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
from typing import Any, Mapping, Sequence
import pytest
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
from claude_code._tool_use import run_tool_use_cell
BEDROCK_INVOKE_MODELS = [
@ -33,72 +30,9 @@ BEDROCK_INVOKE_MODELS = [
"claude-opus-4-7-bedrock-invoke",
]
TOOL_USE_PROMPT = (
"Use the Bash tool to run the command `echo pong` and report what it printed."
)
# Bash is restricted to the exact command `echo pong` + `dontAsk`
# permission mode; see `tool_use/test_anthropic.py` for the security
# rationale.
TOOL_USE_ARGS = [
"--allowed-tools",
"Bash(echo pong)",
"--permission-mode",
"dontAsk",
]
def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool:
for event in events:
if event.get("type") != "assistant":
continue
message = event.get("message") or {}
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and block.get("type") == "tool_use":
return True
return False
@pytest.mark.covers("llm.messages.bedrock_invoke.tool_use.nonstream.works")
def test_tool_use_bedrock_invoke(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
tool call was emitted on the wire."""
base_url, api_key = require_proxy(compat_result)
outcomes = run_claude_models_parallel(
models=BEDROCK_INVOKE_MODELS,
prompt=TOOL_USE_PROMPT,
base_url=base_url,
api_key=api_key,
extra_args=TOOL_USE_ARGS,
)
failures = []
for model in BEDROCK_INVOKE_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not _has_tool_use_event(outcome.events):
error = (
f"[{model}] no tool_use content block observed in stream-json events"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)
run_tool_use_cell(compat_result=compat_result, models=BEDROCK_INVOKE_MODELS)

75
tests/e2e/claude_code/tool_use/test_bedrock_mantle.py
View file

@ -13,7 +13,7 @@ translates Anthropic `tools` into Responses tool declarations and maps
the emitted function calls back to `tool_use` blocks.
Bash is restricted to the exact command `echo pong` plus
`--permission-mode dontAsk`; see `tool_use/test_anthropic.py` for the
`--permission-mode dontAsk`; see `claude_code/_tool_use.py` for the
security rationale.
Mantle cells are opt-in via COMPAT_MANTLE_CELLS=1 (see
@ -29,17 +29,8 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
from typing import Any, Mapping, Sequence
import pytest
from claude_code._env import require_proxy
from claude_code._gpt_cells import skip_unless_mantle_cells_enabled
from claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
from claude_code._tool_use import run_tool_use_cell
BEDROCK_MANTLE_MODELS = [
"gpt-5-6-sol-bedrock-mantle",
@ -47,69 +38,9 @@ BEDROCK_MANTLE_MODELS = [
"gpt-5-6-luna-bedrock-mantle",
]
TOOL_USE_PROMPT = (
"Use the Bash tool to run the command `echo pong` and report what it printed."
)
TOOL_USE_ARGS = [
"--allowed-tools",
"Bash(echo pong)",
"--permission-mode",
"dontAsk",
]
def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool:
for event in events:
if event.get("type") != "assistant":
continue
message = event.get("message") or {}
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and block.get("type") == "tool_use":
return True
return False
def test_tool_use_bedrock_mantle(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
tool call was emitted on the wire by each GPT-5.6 tier."""
skip_unless_mantle_cells_enabled()
proxy = require_proxy(compat_result)
outcomes = run_claude_models_parallel(
models=BEDROCK_MANTLE_MODELS,
prompt=TOOL_USE_PROMPT,
base_url=proxy.base_url,
api_key=proxy.api_key,
extra_args=TOOL_USE_ARGS,
)
failures = []
for model in BEDROCK_MANTLE_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not _has_tool_use_event(outcome.events):
error = (
f"[{model}] no tool_use content block observed in stream-json events"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)
run_tool_use_cell(compat_result=compat_result, models=BEDROCK_MANTLE_MODELS)

75
tests/e2e/claude_code/tool_use/test_openai.py
View file

@ -11,7 +11,7 @@ maps the returned `tool_calls` back to Anthropic `tool_use` blocks, so
this cell exercises the tool-schema translation in both directions.
Bash is restricted to the exact command `echo pong` plus
`--permission-mode dontAsk`; see `tool_use/test_anthropic.py` for the
`--permission-mode dontAsk`; see `claude_code/_tool_use.py` for the
security rationale.
The (feature, provider) for this cell is inferred from the file path by
@ -24,16 +24,7 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
from typing import Any, Mapping, Sequence
import pytest
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
from claude_code._tool_use import run_tool_use_cell
OPENAI_MODELS = [
"gpt-5-6-sol-openai",
@ -41,68 +32,8 @@ OPENAI_MODELS = [
"gpt-5-6-luna-openai",
]
TOOL_USE_PROMPT = (
"Use the Bash tool to run the command `echo pong` and report what it printed."
)
TOOL_USE_ARGS = [
"--allowed-tools",
"Bash(echo pong)",
"--permission-mode",
"dontAsk",
]
def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool:
for event in events:
if event.get("type") != "assistant":
continue
message = event.get("message") or {}
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and block.get("type") == "tool_use":
return True
return False
def test_tool_use_openai(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
tool call was emitted on the wire by each GPT-5.6 tier."""
proxy = require_proxy(compat_result)
outcomes = run_claude_models_parallel(
models=OPENAI_MODELS,
prompt=TOOL_USE_PROMPT,
base_url=proxy.base_url,
api_key=proxy.api_key,
extra_args=TOOL_USE_ARGS,
)
failures = []
for model in OPENAI_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not _has_tool_use_event(outcome.events):
error = (
f"[{model}] no tool_use content block observed in stream-json events"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)
run_tool_use_cell(compat_result=compat_result, models=OPENAI_MODELS)

78
tests/e2e/claude_code/tool_use/test_vertex_ai.py
View file

@ -5,6 +5,10 @@ Claude requests to Anthropic's models on Google Cloud Vertex AI, ask
Claude to invoke a built-in tool (`Bash`), and assert that the upstream
returned a `tool_use` content block.
Bash is restricted to the exact command `echo pong` plus
`--permission-mode dontAsk`; see `claude_code/_tool_use.py` for the
security rationale.
The (feature, provider) for this cell is inferred from the file path by
`tests/e2e/claude_code/conftest.py`:
@ -15,16 +19,9 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
from typing import Any, Mapping, Sequence
import pytest
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
from claude_code._tool_use import run_tool_use_cell
VERTEX_AI_MODELS = [
@ -33,72 +30,9 @@ VERTEX_AI_MODELS = [
"claude-opus-4-7-vertex",
]
TOOL_USE_PROMPT = (
"Use the Bash tool to run the command `echo pong` and report what it printed."
)
# Bash is restricted to the exact command `echo pong` + `dontAsk`
# permission mode; see `tool_use/test_anthropic.py` for the security
# rationale.
TOOL_USE_ARGS = [
"--allowed-tools",
"Bash(echo pong)",
"--permission-mode",
"dontAsk",
]
def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool:
for event in events:
if event.get("type") != "assistant":
continue
message = event.get("message") or {}
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and block.get("type") == "tool_use":
return True
return False
@pytest.mark.covers("llm.messages.vertex.tool_use.nonstream.works")
def test_tool_use_vertex_ai(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert a
tool call was emitted on the wire."""
base_url, api_key = require_proxy(compat_result)
outcomes = run_claude_models_parallel(
models=VERTEX_AI_MODELS,
prompt=TOOL_USE_PROMPT,
base_url=base_url,
api_key=api_key,
extra_args=TOOL_USE_ARGS,
)
failures = []
for model in VERTEX_AI_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not _has_tool_use_event(outcome.events):
error = (
f"[{model}] no tool_use content block observed in stream-json events"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)
run_tool_use_cell(compat_result=compat_result, models=VERTEX_AI_MODELS)

118
tests/e2e/claude_code/tool_use_streaming/test_anthropic.py
View file

@ -4,7 +4,7 @@ Drive the real `claude` CLI in headless `--output-format stream-json`
mode (with `--include-partial-messages`) against a running LiteLLM
proxy that routes to Anthropic, ask Claude to invoke a built-in tool
(`Bash`), and assert that the upstream (a) emitted a `tool_use` content
block and (b) actually streamed the tool input incrementally — i.e.
block and (b) actually streamed the tool input incrementally, i.e.
`input_json_delta` stream events were observed for the block.
This is the "fine-grained tool streaming" path. Historically gateways
@ -13,7 +13,11 @@ input into a single complete block (no `input_json_delta` records
reach the client) or they strip the
`fine-grained-tool-streaming-2025-05-14` beta header and the upstream
falls back to non-streaming tool_use. Both regressions are caught by
the assertions below.
the shared cell body in `claude_code/_tool_use.py`.
Bash is restricted to the exact command `echo pong` plus
`--permission-mode dontAsk`; see `claude_code/_tool_use.py` for the
security rationale.
The (feature, provider) for this cell is inferred from the file path by
`tests/e2e/claude_code/conftest.py`:
@ -25,16 +29,9 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
from typing import Any, Mapping, Sequence
import pytest
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
from claude_code._tool_use import run_tool_use_cell
ANTHROPIC_MODELS = [
@ -43,108 +40,13 @@ ANTHROPIC_MODELS = [
"claude-opus-4-7",
]
# Same shape as the non-streaming `tool_use` cell: ask Claude to call
# the built-in `Bash` tool. `--include-partial-messages` surfaces the
# raw SSE records as `stream_event` entries in the stream-json output,
# which is the wire-level signal for whether the proxy preserved
# incremental `input_json_delta` events for the tool_use block.
TOOL_USE_PROMPT = (
"Use the Bash tool to run the command `echo pong` and report what it printed."
)
# Bash is restricted to the exact command `echo pong` + `dontAsk`
# permission mode; see `tool_use/test_anthropic.py` for the security
# rationale.
TOOL_USE_ARGS = [
"--allowed-tools",
"Bash(echo pong)",
"--permission-mode",
"dontAsk",
"--include-partial-messages",
]
def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool:
"""Walk the stream-json events and return True if any assistant
message included a `tool_use` content block."""
for event in events:
if event.get("type") != "assistant":
continue
message = event.get("message") or {}
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and block.get("type") == "tool_use":
return True
return False
def _count_input_json_deltas(events: Sequence[Mapping[str, Any]]) -> int:
"""Count `input_json_delta` records among the `stream_event`
entries. Zero means the proxy collapsed the streamed tool input
into a single complete block instead of forwarding the incremental
deltas the upstream emitted."""
inner_events = (
event.get("event") for event in events if event.get("type") == "stream_event"
)
return sum(
1
for inner in inner_events
if isinstance(inner, Mapping)
and inner.get("type") == "content_block_delta"
and isinstance(inner.get("delta"), Mapping)
and inner["delta"].get("type") == "input_json_delta"
)
@pytest.mark.covers("llm.messages.anthropic.tool_use.stream.works")
def test_tool_use_streaming_anthropic(compat_result):
"""Drive the `claude` CLI against the LiteLLM proxy and assert the
proxy preserves fine-grained tool streaming end-to-end."""
base_url, api_key = require_proxy(compat_result)
outcomes = run_claude_models_parallel(
run_tool_use_cell(
compat_result=compat_result,
models=ANTHROPIC_MODELS,
prompt=TOOL_USE_PROMPT,
base_url=base_url,
api_key=api_key,
extra_args=TOOL_USE_ARGS,
verify_streaming=True,
)
failures = []
for model in ANTHROPIC_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not _has_tool_use_event(outcome.events):
error = (
f"[{model}] no tool_use content block observed in stream-json events"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if _count_input_json_deltas(outcome.events) == 0:
error = (
f"[{model}] no input_json_delta stream events observed; proxy "
f"likely buffered the tool input into a complete block or "
f"stripped fine-grained tool streaming"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)

107
tests/e2e/claude_code/tool_use_streaming/test_azure.py
View file

@ -7,6 +7,10 @@ invoke a built-in tool (`Bash`), and assert that the upstream (a)
emitted a `tool_use` content block and (b) actually streamed events
incrementally.
Bash is restricted to the exact command `echo pong` plus
`--permission-mode dontAsk`; see `claude_code/_tool_use.py` for the
security rationale.
The (feature, provider) for this cell is inferred from the file path by
`tests/e2e/claude_code/conftest.py`:
@ -17,16 +21,9 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
from typing import Any, Mapping, Sequence
import pytest
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
from claude_code._tool_use import run_tool_use_cell
AZURE_MODELS = [
@ -35,99 +32,11 @@ AZURE_MODELS = [
"claude-opus-4-7-azure",
]
TOOL_USE_PROMPT = (
"Use the Bash tool to run the command `echo pong` and report what it printed."
)
# Bash is restricted to the exact command `echo pong` + `dontAsk`
# permission mode; see `tool_use/test_anthropic.py` for the security
# rationale.
TOOL_USE_ARGS = [
"--allowed-tools",
"Bash(echo pong)",
"--permission-mode",
"dontAsk",
"--include-partial-messages",
]
def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool:
for event in events:
if event.get("type") != "assistant":
continue
message = event.get("message") or {}
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and block.get("type") == "tool_use":
return True
return False
def _count_input_json_deltas(events: Sequence[Mapping[str, Any]]) -> int:
"""Count `input_json_delta` records among the `stream_event`
entries. Zero means the proxy collapsed the streamed tool input
into a single complete block instead of forwarding the incremental
deltas the upstream emitted."""
inner_events = (
event.get("event") for event in events if event.get("type") == "stream_event"
)
return sum(
1
for inner in inner_events
if isinstance(inner, Mapping)
and inner.get("type") == "content_block_delta"
and isinstance(inner.get("delta"), Mapping)
and inner["delta"].get("type") == "input_json_delta"
)
@pytest.mark.covers("llm.messages.azure_foundry.tool_use.stream.works")
def test_tool_use_streaming_azure(compat_result):
base_url, api_key = require_proxy(compat_result)
outcomes = run_claude_models_parallel(
run_tool_use_cell(
compat_result=compat_result,
models=AZURE_MODELS,
prompt=TOOL_USE_PROMPT,
base_url=base_url,
api_key=api_key,
extra_args=TOOL_USE_ARGS,
verify_streaming=True,
)
failures = []
for model in AZURE_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not _has_tool_use_event(outcome.events):
error = (
f"[{model}] no tool_use content block observed in stream-json events"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if _count_input_json_deltas(outcome.events) == 0:
error = (
f"[{model}] no input_json_delta stream events observed; proxy "
f"likely buffered the tool input into a complete block or "
f"stripped fine-grained tool streaming"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)

104
tests/e2e/claude_code/tool_use_streaming/test_azure_openai.py
View file

@ -14,7 +14,7 @@ fragment shape as openai.com; LiteLLM must re-emit them as Anthropic
one complete block.
Bash is restricted to the exact command `echo pong` plus
`--permission-mode dontAsk`; see `tool_use/test_anthropic.py` for the
`--permission-mode dontAsk`; see `claude_code/_tool_use.py` for the
security rationale.
The (feature, provider) for this cell is inferred from the file path by
@ -27,16 +27,7 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
from typing import Any, Mapping, Sequence
import pytest
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
from claude_code._tool_use import run_tool_use_cell
AZURE_OPENAI_MODELS = [
"gpt-5-6-sol-azure-openai",
@ -44,95 +35,10 @@ AZURE_OPENAI_MODELS = [
"gpt-5-6-luna-azure-openai",
]
TOOL_USE_PROMPT = (
"Use the Bash tool to run the command `echo pong` and report what it printed."
)
TOOL_USE_ARGS = [
"--allowed-tools",
"Bash(echo pong)",
"--permission-mode",
"dontAsk",
"--include-partial-messages",
]
def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool:
for event in events:
if event.get("type") != "assistant":
continue
message = event.get("message") or {}
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and block.get("type") == "tool_use":
return True
return False
def _count_input_json_deltas(events: Sequence[Mapping[str, Any]]) -> int:
"""Count `input_json_delta` records among the `stream_event`
entries. Zero means the proxy collapsed the streamed tool input
into a single complete block instead of forwarding the incremental
deltas the upstream emitted."""
inner_events = (
event.get("event") for event in events if event.get("type") == "stream_event"
)
return sum(
1
for inner in inner_events
if isinstance(inner, Mapping)
and inner.get("type") == "content_block_delta"
and isinstance(inner.get("delta"), Mapping)
and inner["delta"].get("type") == "input_json_delta"
)
def test_tool_use_streaming_azure_openai(compat_result):
proxy = require_proxy(compat_result)
outcomes = run_claude_models_parallel(
run_tool_use_cell(
compat_result=compat_result,
models=AZURE_OPENAI_MODELS,
prompt=TOOL_USE_PROMPT,
base_url=proxy.base_url,
api_key=proxy.api_key,
extra_args=TOOL_USE_ARGS,
verify_streaming=True,
)
failures = []
for model in AZURE_OPENAI_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not _has_tool_use_event(outcome.events):
error = (
f"[{model}] no tool_use content block observed in stream-json events"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if _count_input_json_deltas(outcome.events) == 0:
error = (
f"[{model}] no input_json_delta stream events observed; proxy "
f"likely buffered the tool input into a complete block or "
f"stripped fine-grained tool streaming"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)

107
tests/e2e/claude_code/tool_use_streaming/test_bedrock_converse.py
View file

@ -13,6 +13,10 @@ proxy buffers the response or fails to translate Converse's streaming
envelope back to the Anthropic `message_*` event shape Claude Code
expects.
Bash is restricted to the exact command `echo pong` plus
`--permission-mode dontAsk`; see `claude_code/_tool_use.py` for the
security rationale.
The (feature, provider) for this cell is inferred from the file path by
`tests/e2e/claude_code/conftest.py`:
@ -23,16 +27,9 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
from typing import Any, Mapping, Sequence
import pytest
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
from claude_code._tool_use import run_tool_use_cell
BEDROCK_CONVERSE_MODELS = [
@ -41,99 +38,11 @@ BEDROCK_CONVERSE_MODELS = [
"claude-opus-4-7-bedrock-converse",
]
TOOL_USE_PROMPT = (
"Use the Bash tool to run the command `echo pong` and report what it printed."
)
# Bash is restricted to the exact command `echo pong` + `dontAsk`
# permission mode; see `tool_use/test_anthropic.py` for the security
# rationale.
TOOL_USE_ARGS = [
"--allowed-tools",
"Bash(echo pong)",
"--permission-mode",
"dontAsk",
"--include-partial-messages",
]
def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool:
for event in events:
if event.get("type") != "assistant":
continue
message = event.get("message") or {}
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and block.get("type") == "tool_use":
return True
return False
def _count_input_json_deltas(events: Sequence[Mapping[str, Any]]) -> int:
"""Count `input_json_delta` records among the `stream_event`
entries. Zero means the proxy collapsed the streamed tool input
into a single complete block instead of forwarding the incremental
deltas the upstream emitted."""
inner_events = (
event.get("event") for event in events if event.get("type") == "stream_event"
)
return sum(
1
for inner in inner_events
if isinstance(inner, Mapping)
and inner.get("type") == "content_block_delta"
and isinstance(inner.get("delta"), Mapping)
and inner["delta"].get("type") == "input_json_delta"
)
@pytest.mark.covers("llm.messages.bedrock_converse.tool_use.stream.works")
def test_tool_use_streaming_bedrock_converse(compat_result):
base_url, api_key = require_proxy(compat_result)
outcomes = run_claude_models_parallel(
run_tool_use_cell(
compat_result=compat_result,
models=BEDROCK_CONVERSE_MODELS,
prompt=TOOL_USE_PROMPT,
base_url=base_url,
api_key=api_key,
extra_args=TOOL_USE_ARGS,
verify_streaming=True,
)
failures = []
for model in BEDROCK_CONVERSE_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not _has_tool_use_event(outcome.events):
error = (
f"[{model}] no tool_use content block observed in stream-json events"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if _count_input_json_deltas(outcome.events) == 0:
error = (
f"[{model}] no input_json_delta stream events observed; proxy "
f"likely buffered the tool input into a complete block or "
f"stripped fine-grained tool streaming"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)

107
tests/e2e/claude_code/tool_use_streaming/test_bedrock_invoke.py
View file

@ -11,6 +11,10 @@ this cell catches gateway regressions where the proxy buffers the
response or fails to translate the streaming envelope to Anthropic
`message_*` event shape.
Bash is restricted to the exact command `echo pong` plus
`--permission-mode dontAsk`; see `claude_code/_tool_use.py` for the
security rationale.
The (feature, provider) for this cell is inferred from the file path by
`tests/e2e/claude_code/conftest.py`:
@ -21,16 +25,9 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
from typing import Any, Mapping, Sequence
import pytest
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
from claude_code._tool_use import run_tool_use_cell
BEDROCK_INVOKE_MODELS = [
@ -39,99 +36,11 @@ BEDROCK_INVOKE_MODELS = [
"claude-opus-4-7-bedrock-invoke",
]
TOOL_USE_PROMPT = (
"Use the Bash tool to run the command `echo pong` and report what it printed."
)
# Bash is restricted to the exact command `echo pong` + `dontAsk`
# permission mode; see `tool_use/test_anthropic.py` for the security
# rationale.
TOOL_USE_ARGS = [
"--allowed-tools",
"Bash(echo pong)",
"--permission-mode",
"dontAsk",
"--include-partial-messages",
]
def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool:
for event in events:
if event.get("type") != "assistant":
continue
message = event.get("message") or {}
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and block.get("type") == "tool_use":
return True
return False
def _count_input_json_deltas(events: Sequence[Mapping[str, Any]]) -> int:
"""Count `input_json_delta` records among the `stream_event`
entries. Zero means the proxy collapsed the streamed tool input
into a single complete block instead of forwarding the incremental
deltas the upstream emitted."""
inner_events = (
event.get("event") for event in events if event.get("type") == "stream_event"
)
return sum(
1
for inner in inner_events
if isinstance(inner, Mapping)
and inner.get("type") == "content_block_delta"
and isinstance(inner.get("delta"), Mapping)
and inner["delta"].get("type") == "input_json_delta"
)
@pytest.mark.covers("llm.messages.bedrock_invoke.tool_use.stream.works")
def test_tool_use_streaming_bedrock_invoke(compat_result):
base_url, api_key = require_proxy(compat_result)
outcomes = run_claude_models_parallel(
run_tool_use_cell(
compat_result=compat_result,
models=BEDROCK_INVOKE_MODELS,
prompt=TOOL_USE_PROMPT,
base_url=base_url,
api_key=api_key,
extra_args=TOOL_USE_ARGS,
verify_streaming=True,
)
failures = []
for model in BEDROCK_INVOKE_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not _has_tool_use_event(outcome.events):
error = (
f"[{model}] no tool_use content block observed in stream-json events"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if _count_input_json_deltas(outcome.events) == 0:
error = (
f"[{model}] no input_json_delta stream events observed; proxy "
f"likely buffered the tool input into a complete block or "
f"stripped fine-grained tool streaming"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)

104
tests/e2e/claude_code/tool_use_streaming/test_bedrock_mantle.py
View file

@ -14,7 +14,7 @@ events over SigV4-signed SSE; LiteLLM must re-emit them as Anthropic
one complete block.
Bash is restricted to the exact command `echo pong` plus
`--permission-mode dontAsk`; see `tool_use/test_anthropic.py` for the
`--permission-mode dontAsk`; see `claude_code/_tool_use.py` for the
security rationale.
Mantle cells are opt-in via COMPAT_MANTLE_CELLS=1 (see
@ -30,17 +30,8 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
from typing import Any, Mapping, Sequence
import pytest
from claude_code._env import require_proxy
from claude_code._gpt_cells import skip_unless_mantle_cells_enabled
from claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
from claude_code._tool_use import run_tool_use_cell
BEDROCK_MANTLE_MODELS = [
"gpt-5-6-sol-bedrock-mantle",
@ -48,96 +39,11 @@ BEDROCK_MANTLE_MODELS = [
"gpt-5-6-luna-bedrock-mantle",
]
TOOL_USE_PROMPT = (
"Use the Bash tool to run the command `echo pong` and report what it printed."
)
TOOL_USE_ARGS = [
"--allowed-tools",
"Bash(echo pong)",
"--permission-mode",
"dontAsk",
"--include-partial-messages",
]
def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool:
for event in events:
if event.get("type") != "assistant":
continue
message = event.get("message") or {}
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and block.get("type") == "tool_use":
return True
return False
def _count_input_json_deltas(events: Sequence[Mapping[str, Any]]) -> int:
"""Count `input_json_delta` records among the `stream_event`
entries. Zero means the proxy collapsed the streamed tool input
into a single complete block instead of forwarding the incremental
deltas the upstream emitted."""
inner_events = (
event.get("event") for event in events if event.get("type") == "stream_event"
)
return sum(
1
for inner in inner_events
if isinstance(inner, Mapping)
and inner.get("type") == "content_block_delta"
and isinstance(inner.get("delta"), Mapping)
and inner["delta"].get("type") == "input_json_delta"
)
def test_tool_use_streaming_bedrock_mantle(compat_result):
skip_unless_mantle_cells_enabled()
proxy = require_proxy(compat_result)
outcomes = run_claude_models_parallel(
run_tool_use_cell(
compat_result=compat_result,
models=BEDROCK_MANTLE_MODELS,
prompt=TOOL_USE_PROMPT,
base_url=proxy.base_url,
api_key=proxy.api_key,
extra_args=TOOL_USE_ARGS,
verify_streaming=True,
)
failures = []
for model in BEDROCK_MANTLE_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not _has_tool_use_event(outcome.events):
error = (
f"[{model}] no tool_use content block observed in stream-json events"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if _count_input_json_deltas(outcome.events) == 0:
error = (
f"[{model}] no input_json_delta stream events observed; proxy "
f"likely buffered the tool input into a complete block or "
f"stripped fine-grained tool streaming"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)

104
tests/e2e/claude_code/tool_use_streaming/test_openai.py
View file

@ -12,7 +12,7 @@ fragments; LiteLLM must re-emit them as Anthropic `input_json_delta`
deltas rather than buffering the full input into one complete block.
Bash is restricted to the exact command `echo pong` plus
`--permission-mode dontAsk`; see `tool_use/test_anthropic.py` for the
`--permission-mode dontAsk`; see `claude_code/_tool_use.py` for the
security rationale.
The (feature, provider) for this cell is inferred from the file path by
@ -25,16 +25,7 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
from typing import Any, Mapping, Sequence
import pytest
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
from claude_code._tool_use import run_tool_use_cell
OPENAI_MODELS = [
"gpt-5-6-sol-openai",
@ -42,95 +33,10 @@ OPENAI_MODELS = [
"gpt-5-6-luna-openai",
]
TOOL_USE_PROMPT = (
"Use the Bash tool to run the command `echo pong` and report what it printed."
)
TOOL_USE_ARGS = [
"--allowed-tools",
"Bash(echo pong)",
"--permission-mode",
"dontAsk",
"--include-partial-messages",
]
def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool:
for event in events:
if event.get("type") != "assistant":
continue
message = event.get("message") or {}
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and block.get("type") == "tool_use":
return True
return False
def _count_input_json_deltas(events: Sequence[Mapping[str, Any]]) -> int:
"""Count `input_json_delta` records among the `stream_event`
entries. Zero means the proxy collapsed the streamed tool input
into a single complete block instead of forwarding the incremental
deltas the upstream emitted."""
inner_events = (
event.get("event") for event in events if event.get("type") == "stream_event"
)
return sum(
1
for inner in inner_events
if isinstance(inner, Mapping)
and inner.get("type") == "content_block_delta"
and isinstance(inner.get("delta"), Mapping)
and inner["delta"].get("type") == "input_json_delta"
)
def test_tool_use_streaming_openai(compat_result):
proxy = require_proxy(compat_result)
outcomes = run_claude_models_parallel(
run_tool_use_cell(
compat_result=compat_result,
models=OPENAI_MODELS,
prompt=TOOL_USE_PROMPT,
base_url=proxy.base_url,
api_key=proxy.api_key,
extra_args=TOOL_USE_ARGS,
verify_streaming=True,
)
failures = []
for model in OPENAI_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not _has_tool_use_event(outcome.events):
error = (
f"[{model}] no tool_use content block observed in stream-json events"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if _count_input_json_deltas(outcome.events) == 0:
error = (
f"[{model}] no input_json_delta stream events observed; proxy "
f"likely buffered the tool input into a complete block or "
f"stripped fine-grained tool streaming"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)

107
tests/e2e/claude_code/tool_use_streaming/test_vertex_ai.py
View file

@ -10,6 +10,10 @@ Vertex AI exposes Anthropic models via `:streamRawPredict`; this cell
catches gateway regressions where the proxy buffers the response or
strips the streaming beta header on the way to Vertex.
Bash is restricted to the exact command `echo pong` plus
`--permission-mode dontAsk`; see `claude_code/_tool_use.py` for the
security rationale.
The (feature, provider) for this cell is inferred from the file path by
`tests/e2e/claude_code/conftest.py`:
@ -20,16 +24,9 @@ The (feature, provider) for this cell is inferred from the file path by
from __future__ import annotations
from typing import Any, Mapping, Sequence
import pytest
from claude_code._env import require_proxy
from claude_code.cli_driver import (
ClaudeCLIError,
failure_diagnostic,
run_claude_models_parallel,
)
from claude_code._tool_use import run_tool_use_cell
VERTEX_AI_MODELS = [
@ -38,99 +35,11 @@ VERTEX_AI_MODELS = [
"claude-opus-4-7-vertex",
]
TOOL_USE_PROMPT = (
"Use the Bash tool to run the command `echo pong` and report what it printed."
)
# Bash is restricted to the exact command `echo pong` + `dontAsk`
# permission mode; see `tool_use/test_anthropic.py` for the security
# rationale.
TOOL_USE_ARGS = [
"--allowed-tools",
"Bash(echo pong)",
"--permission-mode",
"dontAsk",
"--include-partial-messages",
]
def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool:
for event in events:
if event.get("type") != "assistant":
continue
message = event.get("message") or {}
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and block.get("type") == "tool_use":
return True
return False
def _count_input_json_deltas(events: Sequence[Mapping[str, Any]]) -> int:
"""Count `input_json_delta` records among the `stream_event`
entries. Zero means the proxy collapsed the streamed tool input
into a single complete block instead of forwarding the incremental
deltas the upstream emitted."""
inner_events = (
event.get("event") for event in events if event.get("type") == "stream_event"
)
return sum(
1
for inner in inner_events
if isinstance(inner, Mapping)
and inner.get("type") == "content_block_delta"
and isinstance(inner.get("delta"), Mapping)
and inner["delta"].get("type") == "input_json_delta"
)
@pytest.mark.covers("llm.messages.vertex.tool_use.stream.works")
def test_tool_use_streaming_vertex_ai(compat_result):
base_url, api_key = require_proxy(compat_result)
outcomes = run_claude_models_parallel(
run_tool_use_cell(
compat_result=compat_result,
models=VERTEX_AI_MODELS,
prompt=TOOL_USE_PROMPT,
base_url=base_url,
api_key=api_key,
extra_args=TOOL_USE_ARGS,
verify_streaming=True,
)
failures = []
for model in VERTEX_AI_MODELS:
outcome = outcomes[model]
if isinstance(outcome, ClaudeCLIError):
error = f"[{model}] {outcome}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if outcome.exit_code != 0:
error = f"[{model}] claude CLI failed: {failure_diagnostic(outcome)}"
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if not _has_tool_use_event(outcome.events):
error = (
f"[{model}] no tool_use content block observed in stream-json events"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
if _count_input_json_deltas(outcome.events) == 0:
error = (
f"[{model}] no input_json_delta stream events observed; proxy "
f"likely buffered the tool input into a complete block or "
f"stripped fine-grained tool streaming"
)
compat_result.add({"status": "fail", "error": error})
failures.append(error)
continue
compat_result.add({"status": "pass"})
if failures:
pytest.fail("; ".join(failures), pytrace=False)