diff --git a/tests/e2e/claude_code/_tool_use.py b/tests/e2e/claude_code/_tool_use.py new file mode 100644 index 000000000000..f8637c60b2b3 --- /dev/null +++ b/tests/e2e/claude_code/_tool_use.py @@ -0,0 +1,168 @@ +"""Shared body for the `tool_use` / `tool_use_streaming` x 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//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 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) diff --git a/tests/e2e/claude_code/tool_use/test_anthropic.py b/tests/e2e/claude_code/tool_use/test_anthropic.py index 9ff4c58907ff..598ba0619cdc 100644 --- a/tests/e2e/claude_code/tool_use/test_anthropic.py +++ b/tests/e2e/claude_code/tool_use/test_anthropic.py @@ -5,6 +5,10 @@ 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 @@ 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 @@ "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//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) diff --git a/tests/e2e/claude_code/tool_use/test_azure.py b/tests/e2e/claude_code/tool_use/test_azure.py index 9e7398267c45..46938e44b22e 100644 --- a/tests/e2e/claude_code/tool_use/test_azure.py +++ b/tests/e2e/claude_code/tool_use/test_azure.py @@ -9,6 +9,10 @@ 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 @@ 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 @@ "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) diff --git a/tests/e2e/claude_code/tool_use/test_azure_openai.py b/tests/e2e/claude_code/tool_use/test_azure_openai.py index 7e1eecdbc03a..66e2cbf2a799 100644 --- a/tests/e2e/claude_code/tool_use/test_azure_openai.py +++ b/tests/e2e/claude_code/tool_use/test_azure_openai.py @@ -12,7 +12,7 @@ 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 @@ 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 @@ "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) diff --git a/tests/e2e/claude_code/tool_use/test_bedrock_converse.py b/tests/e2e/claude_code/tool_use/test_bedrock_converse.py index 33d4d3820d25..175ce1a7fb8f 100644 --- a/tests/e2e/claude_code/tool_use/test_bedrock_converse.py +++ b/tests/e2e/claude_code/tool_use/test_bedrock_converse.py @@ -5,6 +5,10 @@ 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 @@ 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 @@ "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) diff --git a/tests/e2e/claude_code/tool_use/test_bedrock_invoke.py b/tests/e2e/claude_code/tool_use/test_bedrock_invoke.py index 47ae3aef1dac..a347cdeea929 100644 --- a/tests/e2e/claude_code/tool_use/test_bedrock_invoke.py +++ b/tests/e2e/claude_code/tool_use/test_bedrock_invoke.py @@ -5,6 +5,10 @@ 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 @@ 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 @@ "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) diff --git a/tests/e2e/claude_code/tool_use/test_bedrock_mantle.py b/tests/e2e/claude_code/tool_use/test_bedrock_mantle.py index e9cb70e74e97..fe7f0a1f07c9 100644 --- a/tests/e2e/claude_code/tool_use/test_bedrock_mantle.py +++ b/tests/e2e/claude_code/tool_use/test_bedrock_mantle.py @@ -13,7 +13,7 @@ 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 @@ 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 @@ "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) diff --git a/tests/e2e/claude_code/tool_use/test_openai.py b/tests/e2e/claude_code/tool_use/test_openai.py index dbe60a65281c..c326639a18e6 100644 --- a/tests/e2e/claude_code/tool_use/test_openai.py +++ b/tests/e2e/claude_code/tool_use/test_openai.py @@ -11,7 +11,7 @@ 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 @@ 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 @@ "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) diff --git a/tests/e2e/claude_code/tool_use/test_vertex_ai.py b/tests/e2e/claude_code/tool_use/test_vertex_ai.py index 79a3016345cd..c4c481bcfe95 100644 --- a/tests/e2e/claude_code/tool_use/test_vertex_ai.py +++ b/tests/e2e/claude_code/tool_use/test_vertex_ai.py @@ -5,6 +5,10 @@ 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 @@ 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 @@ "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) diff --git a/tests/e2e/claude_code/tool_use_streaming/test_anthropic.py b/tests/e2e/claude_code/tool_use_streaming/test_anthropic.py index 152652dcf3c8..aab96583114a 100644 --- a/tests/e2e/claude_code/tool_use_streaming/test_anthropic.py +++ b/tests/e2e/claude_code/tool_use_streaming/test_anthropic.py @@ -4,7 +4,7 @@ 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 @@ 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 @@ 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 @@ "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) diff --git a/tests/e2e/claude_code/tool_use_streaming/test_azure.py b/tests/e2e/claude_code/tool_use_streaming/test_azure.py index 8a1cc1852dd0..da810cd55971 100644 --- a/tests/e2e/claude_code/tool_use_streaming/test_azure.py +++ b/tests/e2e/claude_code/tool_use_streaming/test_azure.py @@ -7,6 +7,10 @@ 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 @@ 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 @@ "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) diff --git a/tests/e2e/claude_code/tool_use_streaming/test_azure_openai.py b/tests/e2e/claude_code/tool_use_streaming/test_azure_openai.py index ad5d4e0f613f..02137e32f18b 100644 --- a/tests/e2e/claude_code/tool_use_streaming/test_azure_openai.py +++ b/tests/e2e/claude_code/tool_use_streaming/test_azure_openai.py @@ -14,7 +14,7 @@ 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 @@ 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 @@ "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) diff --git a/tests/e2e/claude_code/tool_use_streaming/test_bedrock_converse.py b/tests/e2e/claude_code/tool_use_streaming/test_bedrock_converse.py index 3b04ed5962fb..29b72a3522bf 100644 --- a/tests/e2e/claude_code/tool_use_streaming/test_bedrock_converse.py +++ b/tests/e2e/claude_code/tool_use_streaming/test_bedrock_converse.py @@ -13,6 +13,10 @@ 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 @@ 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 @@ "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) diff --git a/tests/e2e/claude_code/tool_use_streaming/test_bedrock_invoke.py b/tests/e2e/claude_code/tool_use_streaming/test_bedrock_invoke.py index c7b61129782a..4a8bc8222a78 100644 --- a/tests/e2e/claude_code/tool_use_streaming/test_bedrock_invoke.py +++ b/tests/e2e/claude_code/tool_use_streaming/test_bedrock_invoke.py @@ -11,6 +11,10 @@ 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 @@ 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 @@ "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) diff --git a/tests/e2e/claude_code/tool_use_streaming/test_bedrock_mantle.py b/tests/e2e/claude_code/tool_use_streaming/test_bedrock_mantle.py index 20fae5d48dba..175e5356abc6 100644 --- a/tests/e2e/claude_code/tool_use_streaming/test_bedrock_mantle.py +++ b/tests/e2e/claude_code/tool_use_streaming/test_bedrock_mantle.py @@ -14,7 +14,7 @@ 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 @@ 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 @@ "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) diff --git a/tests/e2e/claude_code/tool_use_streaming/test_openai.py b/tests/e2e/claude_code/tool_use_streaming/test_openai.py index 895f88d994b8..0a21d99c831c 100644 --- a/tests/e2e/claude_code/tool_use_streaming/test_openai.py +++ b/tests/e2e/claude_code/tool_use_streaming/test_openai.py @@ -12,7 +12,7 @@ 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 @@ 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 @@ "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) diff --git a/tests/e2e/claude_code/tool_use_streaming/test_vertex_ai.py b/tests/e2e/claude_code/tool_use_streaming/test_vertex_ai.py index 2912e3aae3d2..80c96205781d 100644 --- a/tests/e2e/claude_code/tool_use_streaming/test_vertex_ai.py +++ b/tests/e2e/claude_code/tool_use_streaming/test_vertex_ai.py @@ -10,6 +10,10 @@ 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 @@ 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 @@ "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)