scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-14 23:21:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 102

test(claude_code): assert fine-grained tool streaming via input_json_delta instead of an event-count floor
Some checks failed
Terraform Provider / gofmt, vet, build, test (push) Has been cancelled
Details
Terraform Provider / Provider endpoints vs proxy OpenAPI schema (push) Has been cancelled
Details

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
mateo 2026-07-15 02:02:20 +00:00
parent 61b5d93794
commit a61aa82eea
6 changed files with 130 additions and 96 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
pyrightconfig.json
View file

@ -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,

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

@ -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})

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

@ -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)

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

@ -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)

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

@ -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)

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

@ -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)