diff --git a/pyrightconfig.json b/pyrightconfig.json index 97f099d5b2c..eabfbf515c4 100644 --- a/pyrightconfig.json +++ b/pyrightconfig.json @@ -1,7 +1,7 @@ { "include": ["litellm"], "ignore": [], - "exclude": ["**/node_modules", "**/__pycache__", "litellm/types/utils.py", "litellm/proxy/_types.py"], + "exclude": ["**/node_modules", "**/__pycache__", "tests/e2e/claude_code", "litellm/types/utils.py", "litellm/proxy/_types.py"], "pythonVersion": "3.12", "typeCheckingMode": "strict", "enableTypeIgnoreComments": false, 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 d81a15d5ca0..9aa94c89241 100644 --- a/tests/e2e/claude_code/tool_use_streaming/test_anthropic.py +++ b/tests/e2e/claude_code/tool_use_streaming/test_anthropic.py @@ -1,18 +1,19 @@ """tool_use_streaming x Anthropic. Drive the real `claude` CLI in headless `--output-format stream-json` -mode 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 -events incrementally — i.e. more than one stream-json record was -observed before the final `result`. +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. +`input_json_delta` stream events were observed for the block. This is the "fine-grained tool streaming" path. Historically gateways -break it in two ways: they either buffer the entire response before -flushing (in which case `len(events)` collapses to ~1 final record) 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. +break it in two ways: they either buffer/collapse the streamed tool +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 (feature, provider) for this cell is inferred from the file path by `tests/e2e/claude_code/conftest.py`: @@ -45,11 +46,10 @@ ANTHROPIC_MODELS = [ ] # Same shape as the non-streaming `tool_use` cell: ask Claude to call -# the built-in `Bash` tool. The CLI is already in stream-json mode by -# default in `run_claude`, so we don't need to toggle anything to -# exercise the streaming wire — what we want to assert is that the -# stream-json transport actually carried more than one record, which -# is the wire-level signal that the proxy didn't buffer the upstream. +# 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." ) @@ -61,21 +61,9 @@ TOOL_USE_ARGS = [ "Bash(echo pong)", "--permission-mode", "dontAsk", + "--include-partial-messages", ] -# Floor on the number of stream-json records we expect to see for a -# tool-use turn. A buffered (non-streamed) wire for this multi-turn -# flow collapses to roughly: one `system` init + one `assistant` with -# the `tool_use` block + a `user` tool_result + one `assistant` final -# text + one `result`, i.e. ~5 records (the CLI executes the tool -# locally and sends the result back, producing a second model turn -# even on a fully buffered proxy). Real fine-grained streaming -# produces many more (incremental input_json_delta events, -# intermediate assistant deltas, etc., typically 15+). We pick a -# floor comfortably above the buffered case so the assertion catches -# the regression without being flaky on short responses. -MIN_STREAM_EVENTS = 8 - def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool: """Walk the stream-json events and return True if any assistant @@ -93,6 +81,24 @@ def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool: 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_anthropic(compat_result): """Drive the `claude` CLI against the LiteLLM proxy and assert the proxy preserves fine-grained tool streaming end-to-end.""" @@ -143,10 +149,10 @@ def test_tool_use_streaming_anthropic(compat_result): failures.append(error) continue - if len(outcome.events) < MIN_STREAM_EVENTS: + if _count_input_json_deltas(outcome.events) == 0: error = ( - f"[{model}] only {len(outcome.events)} stream-json events observed " - f"(< {MIN_STREAM_EVENTS}); proxy likely buffered the response or " + 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}) 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 5446d541175..c73062b72cd 100644 --- a/tests/e2e/claude_code/tool_use_streaming/test_azure.py +++ b/tests/e2e/claude_code/tool_use_streaming/test_azure.py @@ -48,21 +48,9 @@ TOOL_USE_ARGS = [ "Bash(echo pong)", "--permission-mode", "dontAsk", + "--include-partial-messages", ] -# Floor on the number of stream-json records we expect to see for a -# tool-use turn. A buffered (non-streamed) wire for this multi-turn -# flow collapses to roughly: one `system` init + one `assistant` with -# the `tool_use` block + a `user` tool_result + one `assistant` final -# text + one `result`, i.e. ~5 records (the CLI executes the tool -# locally and sends the result back, producing a second model turn -# even on a fully buffered proxy). Real fine-grained streaming -# produces many more (incremental input_json_delta events, -# intermediate assistant deltas, etc., typically 15+). We pick a -# floor comfortably above the buffered case so the assertion catches -# the regression without being flaky on short responses. -MIN_STREAM_EVENTS = 8 - def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool: for event in events: @@ -78,6 +66,24 @@ def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool: 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(compat_result): base_url = os.environ.get(PROXY_BASE_URL_ENV) api_key = os.environ.get(PROXY_API_KEY_ENV) @@ -126,10 +132,11 @@ def test_tool_use_streaming_azure(compat_result): failures.append(error) continue - if len(outcome.events) < MIN_STREAM_EVENTS: + if _count_input_json_deltas(outcome.events) == 0: error = ( - f"[{model}] only {len(outcome.events)} stream-json events observed " - f"(< {MIN_STREAM_EVENTS}); proxy likely buffered the response" + 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) 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 1ad3766436d..3642551c7c3 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 @@ -54,21 +54,9 @@ TOOL_USE_ARGS = [ "Bash(echo pong)", "--permission-mode", "dontAsk", + "--include-partial-messages", ] -# Floor on the number of stream-json records we expect to see for a -# tool-use turn. A buffered (non-streamed) wire for this multi-turn -# flow collapses to roughly: one `system` init + one `assistant` with -# the `tool_use` block + a `user` tool_result + one `assistant` final -# text + one `result`, i.e. ~5 records (the CLI executes the tool -# locally and sends the result back, producing a second model turn -# even on a fully buffered proxy). Real fine-grained streaming -# produces many more (incremental input_json_delta events, -# intermediate assistant deltas, etc., typically 15+). We pick a -# floor comfortably above the buffered case so the assertion catches -# the regression without being flaky on short responses. -MIN_STREAM_EVENTS = 8 - def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool: for event in events: @@ -84,6 +72,24 @@ def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool: 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_converse(compat_result): base_url = os.environ.get(PROXY_BASE_URL_ENV) api_key = os.environ.get(PROXY_API_KEY_ENV) @@ -132,10 +138,11 @@ def test_tool_use_streaming_bedrock_converse(compat_result): failures.append(error) continue - if len(outcome.events) < MIN_STREAM_EVENTS: + if _count_input_json_deltas(outcome.events) == 0: error = ( - f"[{model}] only {len(outcome.events)} stream-json events observed " - f"(< {MIN_STREAM_EVENTS}); proxy likely buffered the response" + 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) 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 e1e46b92887..af4689b2847 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 @@ -52,21 +52,9 @@ TOOL_USE_ARGS = [ "Bash(echo pong)", "--permission-mode", "dontAsk", + "--include-partial-messages", ] -# Floor on the number of stream-json records we expect to see for a -# tool-use turn. A buffered (non-streamed) wire for this multi-turn -# flow collapses to roughly: one `system` init + one `assistant` with -# the `tool_use` block + a `user` tool_result + one `assistant` final -# text + one `result`, i.e. ~5 records (the CLI executes the tool -# locally and sends the result back, producing a second model turn -# even on a fully buffered proxy). Real fine-grained streaming -# produces many more (incremental input_json_delta events, -# intermediate assistant deltas, etc., typically 15+). We pick a -# floor comfortably above the buffered case so the assertion catches -# the regression without being flaky on short responses. -MIN_STREAM_EVENTS = 8 - def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool: for event in events: @@ -82,6 +70,24 @@ def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool: 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_invoke(compat_result): base_url = os.environ.get(PROXY_BASE_URL_ENV) api_key = os.environ.get(PROXY_API_KEY_ENV) @@ -130,10 +136,11 @@ def test_tool_use_streaming_bedrock_invoke(compat_result): failures.append(error) continue - if len(outcome.events) < MIN_STREAM_EVENTS: + if _count_input_json_deltas(outcome.events) == 0: error = ( - f"[{model}] only {len(outcome.events)} stream-json events observed " - f"(< {MIN_STREAM_EVENTS}); proxy likely buffered the response" + 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) 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 4e98af7e7e4..19ef9a4e90e 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 @@ -51,21 +51,9 @@ TOOL_USE_ARGS = [ "Bash(echo pong)", "--permission-mode", "dontAsk", + "--include-partial-messages", ] -# Floor on the number of stream-json records we expect to see for a -# tool-use turn. A buffered (non-streamed) wire for this multi-turn -# flow collapses to roughly: one `system` init + one `assistant` with -# the `tool_use` block + a `user` tool_result + one `assistant` final -# text + one `result`, i.e. ~5 records (the CLI executes the tool -# locally and sends the result back, producing a second model turn -# even on a fully buffered proxy). Real fine-grained streaming -# produces many more (incremental input_json_delta events, -# intermediate assistant deltas, etc., typically 15+). We pick a -# floor comfortably above the buffered case so the assertion catches -# the regression without being flaky on short responses. -MIN_STREAM_EVENTS = 8 - def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool: for event in events: @@ -81,6 +69,24 @@ def _has_tool_use_event(events: Sequence[Mapping[str, Any]]) -> bool: 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_vertex_ai(compat_result): base_url = os.environ.get(PROXY_BASE_URL_ENV) api_key = os.environ.get(PROXY_API_KEY_ENV) @@ -129,10 +135,11 @@ def test_tool_use_streaming_vertex_ai(compat_result): failures.append(error) continue - if len(outcome.events) < MIN_STREAM_EVENTS: + if _count_input_json_deltas(outcome.events) == 0: error = ( - f"[{model}] only {len(outcome.events)} stream-json events observed " - f"(< {MIN_STREAM_EVENTS}); proxy likely buffered the response" + 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)