diff --git a/tests/integration/README.md b/tests/integration/README.md index 7c5297cc5e1..1af3004b40e 100644 --- a/tests/integration/README.md +++ b/tests/integration/README.md @@ -30,7 +30,7 @@ Streaming checks send real HTTP transfer chunks, including one-byte partitions, The `messages_endpoint/` directory holds `/v1/messages` endpoint contracts: native-provider backends under `providers/` (`anthropic`, `bedrock`, `gemini`) and the translation bridges (`responses_bridge`, `chat_bridge`) at the top level. It runs in the providers shard; `run.py` selects test files recursively under each scheduled directory -A provider folder holds only what depends on that provider's wire format: request and response fidelity, multi-turn history, request parameters, content types, and provider-specific pricing. Behavior every provider shares, such as fallback or billing after a client disconnect, lives in the feature directory it exercises (`routing/`, `streaming/`, `spend/`). `_support/claude_code.py` holds a captured Claude Code request and stream builders that any directory can use as a realistic agent payload +A provider folder holds only what depends on that provider's wire format, and each subfolder is one feature that provider implements its own way: `headers/`, `streaming/`, `reasoning/`, `tools/`, `caching/`, `usage/` (reading the provider's token counts and pricing them), `multimodal/`, `context/` and `errors/`. A test goes in the folder of the feature it varies; one that fits no single folder tests two things and gets split. Behavior every provider shares, such as fallback or billing after a client disconnect, lives in the feature directory it exercises (`routing/`, `streaming/`, `spend/`). `_support/claude_code.py` holds a captured Claude Code request and stream builders that any directory can use as a realistic agent payload The sdk shard exercises the SDK's own HTTP clients against local protocol peers with no gateway in the path, so a case here fails only when the client library or its wire behavior changes. The HTTP/2 case runs a hypercorn TLS peer offering h2 and http/1.1 over ALPN, drives the sync and async httpx handlers at it with `LITELLM_HTTP2` off and on, and asserts the version both the client and the peer observed on the wire. Put a test here only when it needs no proxy, database or Redis; a case that reaches the gateway belongs in one of the other shards diff --git a/tests/integration/messages_endpoint/providers/anthropic/caching/test_anthropic_prompt_cache_pricing_wire.py b/tests/integration/messages_endpoint/providers/anthropic/caching/test_anthropic_prompt_cache_pricing_wire.py new file mode 100644 index 00000000000..06318b52aee --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/caching/test_anthropic_prompt_cache_pricing_wire.py @@ -0,0 +1,85 @@ +import uuid +from typing import Final + +import pytest +from integration._support import claude_code as cc +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows +from integration._support.wire import Reply, Request, wire_server + +_USAGE: Final = { + "input_tokens": 10, + "cache_read_input_tokens": 3000, + "cache_creation_input_tokens": 200, + "output_tokens": 5, +} + + +def test_cached_turn_charges_cache_read_and_creation_rates(gateway: Gateway) -> None: + identity: Final = f"msg_pc_{uuid.uuid4().hex}" + turn1: Final = cc.frontier_request( + f"cache-bust-{uuid.uuid4().hex}", + "high", + 64000, + prompt_text="Read /tmp/cc_probe/hello.txt and reply with its single word", + ) + turn2: Final = cc.tool_loop_turn2( + turn1, + ( + {"type": "thinking", "thinking": "need to read the file", "signature": "sig_anthropic_1"}, + { + "type": "tool_use", + "id": "toolu_read_1", + "name": "Read", + "input": {"file_path": "/tmp/cc_probe/hello.txt"}, + }, + ), + (("toolu_read_1", "1\tPROBE\n2\t"),), + ) + + def respond(request: Request) -> Reply: + assert request.method == "POST" + assert request.target == "/v1/messages", request.target + body: Final = cc.JSON_OBJECT.validate_json(request.body) + expected: Final = {**turn2, "model": cc.FABLE} + assert body == expected, { + key: (expected.get(key), body.get(key)) + for key in expected.keys() | body.keys() + if expected.get(key) != body.get(key) + } + return Reply(content_type="text/event-stream", chunks=cc.text_stream(identity, cc.FABLE, "PROBE", _USAGE)) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model( + model=f"anthropic/{cc.FABLE}", + api_base=wire.url, + api_key=cc.ANTHROPIC_API_KEY, + input_cost_per_token=1e-6, + output_cost_per_token=5e-6, + cache_read_input_token_cost=1e-7, + cache_creation_input_token_cost=1.25e-6, + ) + response: Final = gateway.request( + "POST", + "/v1/messages", + {**turn2, "model": model}, + params={"beta": "true"}, + headers=cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA), + ) + assert response.status_code == 200, response.text + events: Final = cc.sse_events(response.text) + usage: Final = events[0][1]["message"]["usage"] + assert usage["cache_read_input_tokens"] == 3000, usage + assert usage["cache_creation_input_tokens"] == 200, usage + assert len(wire.drain()) == 1 + rows: Final = eventually( + lambda: read_rows( + 'SELECT spend, prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" WHERE request_id=%s', + (identity,), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert float(rows[0]["spend"]) == pytest.approx(10 * 1e-6 + 3000 * 1e-7 + 200 * 1.25e-6 + 5 * 5e-6), dict( + rows[0] + ) diff --git a/tests/integration/messages_endpoint/providers/anthropic/caching/test_anthropic_tool_history_cache_tokens_wire.py b/tests/integration/messages_endpoint/providers/anthropic/caching/test_anthropic_tool_history_cache_tokens_wire.py new file mode 100644 index 00000000000..eff5fd565b5 --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/caching/test_anthropic_tool_history_cache_tokens_wire.py @@ -0,0 +1,128 @@ +import json +import time +import uuid +from typing import Final + +import pytest +from integration._support.client import Gateway, eventually, object_value +from integration._support.database import read_rows +from integration._support.wire import Reply, Request, wire_server + + +@pytest.mark.covers( + "other.provider_wire.anthropic.tool_history_system_cache_and_internal_fields", + "quota_management.spend_tracking.cache_tokens.disjoint_classes_use_explicit_rates", +) +def test_anthropic_tool_history_and_cache_tokens_keep_wire_and_accounting_contracts(gateway: Gateway) -> None: + identity: Final = "anthropic-wire-" + uuid.uuid4().hex + tool_schema: Final = { + "type": "object", + "properties": {"x": {"type": "integer"}, "y": {"type": "integer"}}, + "required": ["x", "y"], + } + + def respond(request: Request) -> Reply: + assert request.method == "POST" and request.target == "/v1/messages" + assert request.headers["x-api-key"] == "synthetic-anthropic-key" + body: Final = json.loads(request.body) + assert body["model"] == "claude-sonnet-4-5-20250929" + assert body["system"] == [{"type": "text", "text": "synthetic policy", "cache_control": {"type": "ephemeral"}}] + assert body["tools"][0]["name"] == "add" and body["tools"][0]["input_schema"] == tool_schema + assert body["max_tokens"] == 16 + assert not {"timeout", "stream_chunk_size", "litellm_params", "litellm_metadata", "rpm", "tpm"}.intersection( + body + ) + messages: Final = body["messages"] + assert [message["role"] for message in messages] == ["user", "assistant", "user"] + assert messages[0]["content"] == [{"type": "text", "text": "first"}] + assert messages[1]["content"] == [ + {"type": "tool_use", "id": "history-call", "name": "add", "input": {"x": 1, "y": 2}} + ] + assert messages[2]["content"] == [ + {"type": "tool_result", "tool_use_id": "history-call", "content": "3"}, + {"type": "text", "text": "next"}, + ] + return Reply( + body=json.dumps( + { + "id": identity, + "type": "message", + "role": "assistant", + "model": "claude-sonnet-4-5-20250929", + "content": [{"type": "tool_use", "id": "next-call", "name": "add", "input": {"x": 3, "y": 4}}], + "stop_reason": "tool_use", + "stop_sequence": None, + "usage": { + "input_tokens": 10, + "output_tokens": 4, + "cache_read_input_tokens": 5, + "cache_creation_input_tokens": 7, + }, + } + ).encode() + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model( + model="anthropic/claude-sonnet-4-5-20250929", + api_base=wire.url, + api_key="synthetic-anthropic-key", + input_cost_per_token=0.001, + output_cost_per_token=0.002, + cache_read_input_token_cost=0.0001, + cache_creation_input_token_cost=0.002, + ) + response: Final = gateway.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "max_tokens": 16, + "timeout": 5, + "messages": [ + { + "role": "system", + "content": [ + {"type": "text", "text": "synthetic policy", "cache_control": {"type": "ephemeral"}} + ], + }, + {"role": "user", "content": "first"}, + { + "role": "assistant", + "tool_calls": [ + { + "id": "history-call", + "type": "function", + "function": {"name": "add", "arguments": '{"x":1,"y":2}'}, + } + ], + }, + {"role": "tool", "tool_call_id": "history-call", "content": "3"}, + {"role": "user", "content": "next"}, + ], + "tools": [{"type": "function", "function": {"name": "add", "parameters": tool_schema}}], + }, + ) + assert response.status_code == 200, response.text + body: Final = response.json() + assert body["id"].startswith("chatcmpl-") + assert body["choices"][0]["finish_reason"] == "tool_calls" + tool: Final = body["choices"][0]["message"]["tool_calls"][0] + assert tool["id"] == "next-call" and tool["function"]["name"] == "add" + assert json.loads(tool["function"]["arguments"]) == {"x": 3, "y": 4} + assert body["usage"]["prompt_tokens"] == 22 and body["usage"]["completion_tokens"] == 4 + assert len(wire.drain()) == 1 + rows: Final = eventually( + lambda: read_rows( + 'SELECT spend, prompt_tokens, completion_tokens, metadata FROM "LiteLLM_SpendLogs" WHERE request_id=%s', + (body["id"],), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert float(rows[0]["spend"]) == pytest.approx(10 * 0.001 + 5 * 0.0001 + 7 * 0.002 + 4 * 0.002) + assert rows[0]["prompt_tokens"] == 22 and rows[0]["completion_tokens"] == 4 + metadata: Final = rows[0]["metadata"] + parsed: Final = json.loads(metadata) if isinstance(metadata, str) else object_value(metadata) + assert parsed["cost_breakdown"]["input_cost"] == pytest.approx(0.0245) + assert parsed["cost_breakdown"]["output_cost"] == pytest.approx(0.008) diff --git a/tests/integration/messages_endpoint/providers/anthropic/context/test_anthropic_compaction_wire.py b/tests/integration/messages_endpoint/providers/anthropic/context/test_anthropic_compaction_wire.py new file mode 100644 index 00000000000..c6ac00c20e8 --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/context/test_anthropic_compaction_wire.py @@ -0,0 +1,108 @@ +import uuid +from typing import Final + +from integration._support import claude_code as cc +from integration._support.client import Gateway +from integration._support.wire import Reply, Request, wire_server + +_CONTEXT_MANAGEMENT: Final = { + "edits": [ + {"type": "clear_thinking_20251015", "keep": "all"}, + {"type": "compact_20260112", "trigger": {"type": "input_tokens", "value": 150000}}, + ] +} +_COMPACTION_BLOCK: Final = {"type": "compaction", "content": ""} + + +def _compaction_stream(identity: str) -> tuple[bytes, ...]: + return ( + cc.sse_frame( + "message_start", + { + "type": "message_start", + "message": { + "id": identity, + "type": "message", + "role": "assistant", + "model": cc.FABLE, + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 20, "output_tokens": 1}, + }, + }, + ), + cc.sse_frame( + "content_block_start", + {"type": "content_block_start", "index": 0, "content_block": dict(_COMPACTION_BLOCK)}, + ), + cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 0}), + cc.sse_frame( + "message_delta", + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": {"output_tokens": 5}, + "context_management": { + "applied_edits": [{"type": "compact_20260112", "compacted_at": "2026-09-26T00:00:00Z"}] + }, + }, + ), + cc.sse_frame("message_stop", {"type": "message_stop"}), + ) + + +def test_compaction_edit_and_applied_edit_block_round_trip_through_anthropic(gateway: Gateway) -> None: + identity: Final = f"msg_cm_{uuid.uuid4().hex}" + request_body: Final = { + **cc.frontier_request(f"cache-bust-{uuid.uuid4().hex}", "high", 64000), + "context_management": _CONTEXT_MANAGEMENT, + } + turn3: Final = cc.tool_loop_turn2( + request_body, + ( + dict(_COMPACTION_BLOCK), + {"type": "text", "text": "continuing after compaction"}, + ), + (), + ) + first_expected: Final = {**request_body, "model": cc.FABLE} + second_expected: Final = {**turn3, "model": cc.FABLE} + + def respond(request: Request) -> Reply: + assert request.method == "POST" + assert request.target == "/v1/messages", request.target + body: Final = cc.JSON_OBJECT.validate_json(request.body) + if body == first_expected: + assert body["context_management"] == _CONTEXT_MANAGEMENT + return Reply(content_type="text/event-stream", chunks=_compaction_stream(identity)) + assert body == second_expected, { + key: (second_expected.get(key), body.get(key)) + for key in second_expected.keys() | body.keys() + if second_expected.get(key) != body.get(key) + } + assert body["context_management"] == _CONTEXT_MANAGEMENT + return Reply( + content_type="text/event-stream", + chunks=cc.text_stream("msg_cm_next", cc.FABLE, "OK", {"input_tokens": 20, "output_tokens": 2}), + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) + headers: Final = cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA) + response1: Final = gateway.request( + "POST", "/v1/messages", {**request_body, "model": model}, params={"beta": "true"}, headers=headers + ) + assert response1.status_code == 200, response1.text + events: Final = cc.sse_events(response1.text) + assert events[1][1]["content_block"] == _COMPACTION_BLOCK, events[1] + deltas: Final = [data for event, data in events if event == "message_delta"] + assert len(deltas) == 1 and deltas[0].get("context_management") == { + "applied_edits": [{"type": "compact_20260112", "compacted_at": "2026-09-26T00:00:00Z"}] + }, events + response2: Final = gateway.request( + "POST", "/v1/messages", {**turn3, "model": model}, params={"beta": "true"}, headers=headers + ) + assert response2.status_code == 200, response2.text + bodies: Final = tuple(cc.JSON_OBJECT.validate_json(request.body) for request in wire.drain()) + assert bodies == (first_expected, second_expected), bodies diff --git a/tests/integration/messages_endpoint/providers/anthropic/errors/test_anthropic_bare_string_content_rejected_wire.py b/tests/integration/messages_endpoint/providers/anthropic/errors/test_anthropic_bare_string_content_rejected_wire.py new file mode 100644 index 00000000000..ffea1830e54 --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/errors/test_anthropic_bare_string_content_rejected_wire.py @@ -0,0 +1,32 @@ +import json +import time +import uuid +from typing import Final + +import pytest +from integration._support.client import Gateway, eventually, object_value +from integration._support.database import read_rows +from integration._support.wire import Reply, Request, wire_server + + +@pytest.mark.covers("other.provider_wire.anthropic.bare_string_content_item_is_client_error") +@pytest.mark.parametrize( + "text", [pytest.param("what type of file is this?", id="type_word"), pytest.param("hello", id="plain")] +) +def test_anthropic_bare_string_content_item_is_rejected_as_client_error_before_the_wire( + gateway: Gateway, text: str +) -> None: + def respond(request: Request) -> Reply: + raise AssertionError(f"upstream must not be reached: {request.target}") + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model( + model="anthropic/claude-sonnet-4-5-20250929", api_base=wire.url, api_key="synthetic-anthropic-key" + ) + response: Final = gateway.request( + "POST", + "/v1/chat/completions", + {"model": model, "max_tokens": 16, "timeout": 5, "messages": [{"role": "system", "content": [text]}]}, + ) + assert response.status_code == 400, response.text + assert wire.drain() == () diff --git a/tests/integration/messages_endpoint/providers/anthropic/errors/test_anthropic_messages_timeout_wire.py b/tests/integration/messages_endpoint/providers/anthropic/errors/test_anthropic_messages_timeout_wire.py new file mode 100644 index 00000000000..76c29ad7763 --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/errors/test_anthropic_messages_timeout_wire.py @@ -0,0 +1,58 @@ +import json +import time +import uuid +from typing import Final + +import pytest +from integration._support.client import JSON_OBJECT, Gateway +from integration._support.wire import Reply, Request, wire_server + +_UPSTREAM_STALL_SECONDS: Final = 4.0 +_CONFIGURED_TIMEOUT_SECONDS: Final = 1.0 + + +@pytest.mark.covers("providers.anthropic_messages.configured_timeout_aborts_stalled_upstream") +def test_messages_endpoint_honors_configured_timeout_against_stalled_upstream(gateway: Gateway) -> None: + prompt: Final = "stall-" + uuid.uuid4().hex + + def respond(request: Request) -> Reply: + assert request.method == "POST" and request.target == "/v1/messages" + assert request.headers["x-api-key"] == "synthetic-anthropic-key" + body: Final = JSON_OBJECT.validate_json(request.body) + assert body["model"] == "claude-sonnet-4-5-20250929" + assert body["messages"] == [{"role": "user", "content": prompt}] + assert body["max_tokens"] == 16 + assert "timeout" not in body + time.sleep(_UPSTREAM_STALL_SECONDS) + return Reply( + body=json.dumps( + { + "id": "msg_stalled", + "type": "message", + "role": "assistant", + "model": "claude-sonnet-4-5-20250929", + "content": [{"type": "text", "text": "too late"}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 1, "output_tokens": 2}, + } + ).encode() + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model( + model="anthropic/claude-sonnet-4-5-20250929", + api_base=wire.url, + api_key="synthetic-anthropic-key", + timeout=_CONFIGURED_TIMEOUT_SECONDS, + ) + started: Final = time.monotonic() + response: Final = gateway.request( + "POST", + "/v1/messages", + {"model": model, "max_tokens": 16, "messages": [{"role": "user", "content": prompt}]}, + ) + elapsed: Final = time.monotonic() - started + assert response.status_code == 408, response.text + assert elapsed < _UPSTREAM_STALL_SECONDS, f"timed out only after {elapsed:.2f}s: {response.text}" + assert len(wire.drain()) == 1 diff --git a/tests/integration/messages_endpoint/providers/anthropic/errors/test_anthropic_slow_upstream_cutoff_wire.py b/tests/integration/messages_endpoint/providers/anthropic/errors/test_anthropic_slow_upstream_cutoff_wire.py new file mode 100644 index 00000000000..7532afe718d --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/errors/test_anthropic_slow_upstream_cutoff_wire.py @@ -0,0 +1,64 @@ +import json +import time +import uuid +from typing import Final + +import pytest +from integration._support.client import Gateway, eventually, object_value +from integration._support.database import read_rows +from integration._support.wire import Reply, Request, wire_server + + +@pytest.mark.covers("other.provider_wire.anthropic.messages_request_timeout_reaches_transport") +def test_anthropic_messages_slow_upstream_is_cut_off_at_the_deployment_request_timeout(gateway: Gateway) -> None: + identity: Final = "anthropic-timeout-" + uuid.uuid4().hex + prompt: Final = f"slow answer {identity}" + + def respond(request: Request) -> Reply: + assert request.method == "POST" and request.target == "/v1/messages" + assert request.headers["x-api-key"] == "synthetic-anthropic-key" + body: Final = json.loads(request.body) + assert body["model"] == "claude-sonnet-4-5-20250929" + assert body["max_tokens"] == 16 + assert body["messages"] == [{"role": "user", "content": prompt}] + assert not { + "timeout", + "request_timeout", + "stream_chunk_size", + "litellm_params", + "litellm_metadata", + "rpm", + "tpm", + }.intersection(body) + time.sleep(1.5) + return Reply( + body=json.dumps( + { + "id": identity, + "type": "message", + "role": "assistant", + "model": "claude-sonnet-4-5-20250929", + "content": [{"type": "text", "text": "late"}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 3, "output_tokens": 1}, + } + ).encode() + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model( + model="anthropic/claude-sonnet-4-5-20250929", + api_base=wire.url, + api_key="synthetic-anthropic-key", + request_timeout=0.3, + ) + response: Final = gateway.request( + "POST", + "/v1/messages", + {"model": model, "max_tokens": 16, "messages": [{"role": "user", "content": prompt}]}, + headers={"anthropic-version": "2023-06-01"}, + ) + assert response.status_code == 408, response.text + assert "Timeout" in response.json()["error"]["message"], response.text + assert eventually(wire.drain, lambda requests: len(requests) == 1, seconds=5, return_last_on_timeout=True) diff --git a/tests/integration/messages_endpoint/providers/anthropic/headers/test_anthropic_request_fidelity_wire.py b/tests/integration/messages_endpoint/providers/anthropic/headers/test_anthropic_request_fidelity_wire.py new file mode 100644 index 00000000000..a5a57737fc1 --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/headers/test_anthropic_request_fidelity_wire.py @@ -0,0 +1,71 @@ +import uuid +from typing import Final + +from integration._support import claude_code as cc +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows +from integration._support.wire import Reply, Request, wire_server + +_MODEL: Final = cc.SONNET + + +def test_streaming_request_reaches_anthropic_intact_and_streams_back(gateway: Gateway) -> None: + identity: Final = f"msg_cc_{uuid.uuid4().hex}" + request_body: Final = cc.claude_code_request(f"cache-bust-{uuid.uuid4().hex}") + cli_beta: Final = frozenset(cc.CLI_BETA.split(",")) + + def respond(request: Request) -> Reply: + assert request.method == "POST" + assert request.target == "/v1/messages", request.target + assert request.headers["x-api-key"] == cc.ANTHROPIC_API_KEY + assert request.headers["anthropic-version"] == "2023-06-01" + assert frozenset(request.headers.get("anthropic-beta", "").split(",")) == cli_beta, request.headers.get( + "anthropic-beta" + ) + assert "authorization" not in request.headers, dict(request.headers) + assert all(gateway.key not in value for value in request.headers.values()), dict(request.headers) + body: Final = cc.JSON_OBJECT.validate_json(request.body) + expected: Final = {**request_body, "model": _MODEL} + assert body == expected, { + key: (expected.get(key), body.get(key)) + for key in expected.keys() | body.keys() + if expected.get(key) != body.get(key) + } + return Reply( + content_type="text/event-stream", + chunks=cc.text_stream(identity, _MODEL, "PONG", {"input_tokens": 12, "output_tokens": 4}), + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{_MODEL}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) + response: Final = gateway.request( + "POST", + "/v1/messages", + {**request_body, "model": model}, + params={"beta": "true"}, + headers=cc.cli_headers(gateway.key), + ) + assert response.status_code == 200, response.text + assert response.headers["content-type"].startswith("text/event-stream"), dict(response.headers) + events: Final = cc.sse_events(response.text) + assert [event for event, _ in events] == [ + "message_start", + "content_block_start", + "content_block_delta", + "content_block_stop", + "message_delta", + "message_stop", + ] + assert events[2][1]["delta"] == {"type": "text_delta", "text": "PONG"} + assert events[4][1]["delta"]["stop_reason"] == "end_turn" + assert events[4][1]["usage"]["output_tokens"] == 4 + assert len(wire.drain()) == 1 + rows: Final = eventually( + lambda: read_rows( + 'SELECT prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" WHERE request_id=%s', + (identity,), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert rows[0]["prompt_tokens"] == 12 and rows[0]["completion_tokens"] == 4 diff --git a/tests/integration/messages_endpoint/providers/anthropic/multimodal/test_anthropic_document_input_wire.py b/tests/integration/messages_endpoint/providers/anthropic/multimodal/test_anthropic_document_input_wire.py new file mode 100644 index 00000000000..48b914a7374 --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/multimodal/test_anthropic_document_input_wire.py @@ -0,0 +1,133 @@ +import base64 +import uuid +from typing import Final + +from integration._support import claude_code as cc +from integration._support.client import Gateway +from integration._support.wire import Reply, Request, wire_server + +_PDF_BYTES: Final = ( + b"%PDF-1.1\n" + b"1 0 obj<>endobj\n" + b"2 0 obj<>endobj\n" + b"3 0 obj<>endobj\n" + b"trailer<>\n%%EOF" +) +_DOC_BLOCK: Final = { + "type": "document", + "source": {"type": "base64", "data": base64.b64encode(_PDF_BYTES).decode(), "media_type": "application/pdf"}, + "citations": {"enabled": True}, +} + + +def _cited_stream(identity: str) -> tuple[bytes, ...]: + return ( + cc.sse_frame( + "message_start", + { + "type": "message_start", + "message": { + "id": identity, + "type": "message", + "role": "assistant", + "model": cc.FABLE, + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 20, "output_tokens": 1}, + }, + }, + ), + cc.sse_frame( + "content_block_start", + {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}, + ), + cc.sse_frame( + "content_block_delta", + {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "A page."}}, + ), + cc.sse_frame( + "content_block_delta", + { + "type": "content_block_delta", + "index": 0, + "delta": { + "type": "citations_delta", + "citation": { + "type": "page_location", + "document_index": 0, + "document_title": "dot.pdf", + "start_page_number": 1, + "end_page_number": 1, + "cited_text": "Page", + }, + }, + }, + ), + cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 0}), + cc.sse_frame( + "message_delta", + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": {"output_tokens": 6}, + }, + ), + cc.sse_frame("message_stop", {"type": "message_stop"}), + ) + + +def test_base64_pdf_document_with_citations_reaches_anthropic_identical(gateway: Gateway) -> None: + request_body: Final = { + **cc.frontier_request(f"cache-bust-{uuid.uuid4().hex}", "high", 64000), + "messages": [ + { + "role": "user", + "content": [ + dict(_DOC_BLOCK), + {"type": "text", "text": f"What is on page one? {uuid.uuid4().hex}"}, + ], + } + ], + } + + def respond(request: Request) -> Reply: + assert request.method == "POST" + assert request.target == "/v1/messages", request.target + body: Final = cc.JSON_OBJECT.validate_json(request.body) + expected: Final = {**request_body, "model": cc.FABLE} + assert body == expected, { + key: (expected.get(key), body.get(key)) + for key in expected.keys() | body.keys() + if expected.get(key) != body.get(key) + } + return Reply(content_type="text/event-stream", chunks=_cited_stream(f"msg_doc_{uuid.uuid4().hex}")) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) + response: Final = gateway.request( + "POST", + "/v1/messages", + {**request_body, "model": model}, + params={"beta": "true"}, + headers=cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA), + ) + assert response.status_code == 200, response.text + events: Final = cc.sse_events(response.text) + citations: Final = [ + data["delta"] for event, data in events if data.get("delta", {}).get("type") == "citations_delta" + ] + assert citations == [ + { + "type": "citations_delta", + "citation": { + "type": "page_location", + "document_index": 0, + "document_title": "dot.pdf", + "start_page_number": 1, + "end_page_number": 1, + "cited_text": "Page", + }, + } + ], citations + assert len(wire.drain()) == 1 diff --git a/tests/integration/messages_endpoint/providers/anthropic/multimodal/test_anthropic_image_input_wire.py b/tests/integration/messages_endpoint/providers/anthropic/multimodal/test_anthropic_image_input_wire.py new file mode 100644 index 00000000000..8a929ee2a06 --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/multimodal/test_anthropic_image_input_wire.py @@ -0,0 +1,67 @@ +import uuid +from typing import Final + +from integration._support import claude_code as cc +from integration._support.client import Gateway +from integration._support.wire import Reply, Request, wire_server + +_PNG_B64: Final = "iVBORw0KGgoAAAANSUhEUgAAAAQAAAAECAIAAAAmkwkpAAAAEElEQVR4nGP4z8AARwzEcQCukw/x0F8jngAAAABJRU5ErkJggg==" +_IMAGE_BLOCK: Final = { + "type": "image", + "source": {"type": "base64", "data": _PNG_B64, "media_type": "image/png"}, +} + + +def test_tool_result_image_block_and_pasted_image_reach_anthropic_identical(gateway: Gateway) -> None: + turn1: Final = cc.frontier_request( + f"cache-bust-{uuid.uuid4().hex}", + "high", + 64000, + prompt_text="Read /tmp/cc_probe/dot.png and say what colour it is", + ) + with_image_result: Final = cc.tool_loop_turn2( + turn1, + ({"type": "tool_use", "id": "toolu_img", "name": "Read", "input": {"file_path": "/tmp/cc_probe/dot.png"}},), + (("toolu_img", [dict(_IMAGE_BLOCK)]),), + ) + pasted: Final = { + **cc.frontier_request(f"cache-bust-{uuid.uuid4().hex}", "high", 64000), + "messages": [ + { + "role": "user", + "content": [ + dict(_IMAGE_BLOCK), + {"type": "text", "text": f"What colour is this? {uuid.uuid4().hex}"}, + ], + } + ], + } + first_expected: Final = {**with_image_result, "model": cc.FABLE} + second_expected: Final = {**pasted, "model": cc.FABLE} + + def respond(request: Request) -> Reply: + assert request.method == "POST" + assert request.target == "/v1/messages", request.target + body: Final = cc.JSON_OBJECT.validate_json(request.body) + if body != first_expected: + assert body == second_expected, body + return Reply( + content_type="text/event-stream", + chunks=cc.text_stream( + f"msg_img_{uuid.uuid4().hex}", cc.FABLE, "RED", {"input_tokens": 20, "output_tokens": 2} + ), + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) + headers: Final = cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA) + response1: Final = gateway.request( + "POST", "/v1/messages", {**with_image_result, "model": model}, params={"beta": "true"}, headers=headers + ) + assert response1.status_code == 200, response1.text + response2: Final = gateway.request( + "POST", "/v1/messages", {**pasted, "model": model}, params={"beta": "true"}, headers=headers + ) + assert response2.status_code == 200, response2.text + bodies: Final = tuple(cc.JSON_OBJECT.validate_json(request.body) for request in wire.drain()) + assert bodies == (first_expected, second_expected), bodies diff --git a/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_adaptive_thinking_effort_wire.py b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_adaptive_thinking_effort_wire.py new file mode 100644 index 00000000000..9f929d69a05 --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_adaptive_thinking_effort_wire.py @@ -0,0 +1,111 @@ +import uuid +from typing import Final + +import pytest +from integration._support import claude_code as cc +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows +from integration._support.wire import Reply, Request, wire_server +from pydantic import JsonValue + + +def _diff(expected: dict[str, JsonValue], body: dict[str, JsonValue]) -> dict[str, JsonValue]: + return { + key: {"expected": expected.get(key), "upstream": body.get(key)} + for key in expected.keys() | body.keys() + if expected.get(key) != body.get(key) + } + + +def test_adaptive_thinking_and_effort_reach_anthropic_intact(gateway: Gateway) -> None: + identity: Final = f"msg_fable_{uuid.uuid4().hex}" + request_body: Final = cc.frontier_request(f"cache-bust-{uuid.uuid4().hex}", "high", 64000) + cli_beta: Final = frozenset(cc.FRONTIER_CLI_BETA.split(",")) + + def respond(request: Request) -> Reply: + assert request.method == "POST" + assert request.target == "/v1/messages", request.target + assert request.headers["x-api-key"] == cc.ANTHROPIC_API_KEY + assert request.headers["anthropic-version"] == "2023-06-01" + upstream_beta: Final = request.headers.get("anthropic-beta", "") + assert cli_beta <= frozenset(upstream_beta.split(",")), upstream_beta + assert upstream_beta.split(",").count("effort-2025-11-24") == 1, upstream_beta + body: Final = cc.JSON_OBJECT.validate_json(request.body) + expected: Final = {**request_body, "model": cc.FABLE} + assert body == expected, _diff(expected, body) + return Reply( + content_type="text/event-stream", + chunks=cc.text_stream(identity, cc.FABLE, "PONG", {"input_tokens": 12, "output_tokens": 4}), + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) + response: Final = gateway.request( + "POST", + "/v1/messages", + {**request_body, "model": model}, + params={"beta": "true"}, + headers=cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA), + ) + assert response.status_code == 200, response.text + events: Final = cc.sse_events(response.text) + assert [event for event, _ in events][-1] == "message_stop" + assert events[4][1]["delta"]["stop_reason"] == "end_turn" + assert len(wire.drain()) == 1 + rows: Final = eventually( + lambda: read_rows( + 'SELECT prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" WHERE request_id=%s', + (identity,), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert rows[0]["prompt_tokens"] == 12 and rows[0]["completion_tokens"] == 4 + + +def test_xhigh_effort_reaches_anthropic_and_charges_by_usage(gateway: Gateway) -> None: + identity: Final = f"msg_opus_{uuid.uuid4().hex}" + request_body: Final = cc.frontier_request(f"cache-bust-{uuid.uuid4().hex}", "xhigh", 128000) + cli_beta: Final = frozenset(cc.FRONTIER_CLI_BETA.split(",")) + + def respond(request: Request) -> Reply: + assert request.method == "POST" + assert request.target == "/v1/messages", request.target + assert request.headers["x-api-key"] == cc.ANTHROPIC_API_KEY + upstream_beta: Final = request.headers.get("anthropic-beta", "") + assert cli_beta <= frozenset(upstream_beta.split(",")), upstream_beta + assert upstream_beta.split(",").count("effort-2025-11-24") == 1, upstream_beta + body: Final = cc.JSON_OBJECT.validate_json(request.body) + expected: Final = {**request_body, "model": cc.OPUS} + assert body == expected, _diff(expected, body) + return Reply( + content_type="text/event-stream", + chunks=cc.text_stream(identity, cc.OPUS, "PONG", {"input_tokens": 10, "output_tokens": 5}), + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model( + model=f"anthropic/{cc.OPUS}", + api_base=wire.url, + api_key=cc.ANTHROPIC_API_KEY, + input_cost_per_token=1e-6, + output_cost_per_token=5e-6, + ) + response: Final = gateway.request( + "POST", + "/v1/messages", + {**request_body, "model": model}, + params={"beta": "true"}, + headers=cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA), + ) + assert response.status_code == 200, response.text + assert len(wire.drain()) == 1 + rows: Final = eventually( + lambda: read_rows( + 'SELECT spend, prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" WHERE request_id=%s', + (identity,), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert float(rows[0]["spend"]) == pytest.approx(10 * 1e-6 + 5 * 5e-6) diff --git a/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_interleaved_thinking_history_wire.py b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_interleaved_thinking_history_wire.py new file mode 100644 index 00000000000..6990f33e51c --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_interleaved_thinking_history_wire.py @@ -0,0 +1,167 @@ +import uuid +from typing import Final + +from integration._support import claude_code as cc +from integration._support.client import Gateway +from integration._support.wire import Reply, Request, wire_server +from pydantic import JsonValue + +_TURN1_TOOL: Final = ("toolu_a", "Read", {"file_path": "/tmp/cc_probe/a.txt"}) +_TURN2_TOOL: Final = ("toolu_b", "Read", {"file_path": "/tmp/cc_probe/b.txt"}) + + +def _diff(expected: dict[str, JsonValue], body: dict[str, JsonValue]) -> dict[str, JsonValue]: + return { + key: {"expected": expected.get(key), "upstream": body.get(key)} + for key in expected.keys() | body.keys() + if expected.get(key) != body.get(key) + } + + +def _turn2(base: dict[str, JsonValue]) -> dict[str, JsonValue]: + return cc.tool_loop_turn2( + base, + ( + {"type": "thinking", "thinking": "plan", "signature": "sig1"}, + {"type": "tool_use", "id": _TURN1_TOOL[0], "name": _TURN1_TOOL[1], "input": _TURN1_TOOL[2]}, + ), + ((_TURN1_TOOL[0], "ALPHA"),), + ) + + +def _turn3(turn2: dict[str, JsonValue]) -> dict[str, JsonValue]: + return cc.tool_loop_turn2( + turn2, + ( + {"type": "thinking", "thinking": "got A", "signature": "sig2"}, + {"type": "text", "text": "got A"}, + {"type": "tool_use", "id": _TURN2_TOOL[0], "name": _TURN2_TOOL[1], "input": _TURN2_TOOL[2]}, + ), + ((_TURN2_TOOL[0], "BRAVO"),), + ) + + +def _interleaved_stream(identity: str) -> tuple[bytes, ...]: + usage: Final = {"input_tokens": 20, "output_tokens": 12} + return ( + cc.sse_frame( + "message_start", + { + "type": "message_start", + "message": { + "id": identity, + "type": "message", + "role": "assistant", + "model": cc.FABLE, + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": usage["input_tokens"], "output_tokens": 1}, + }, + }, + ), + cc.sse_frame( + "content_block_start", + {"type": "content_block_start", "index": 0, "content_block": {"type": "thinking", "thinking": ""}}, + ), + cc.sse_frame( + "content_block_delta", + {"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": "got A"}}, + ), + cc.sse_frame( + "content_block_delta", + {"type": "content_block_delta", "index": 0, "delta": {"type": "signature_delta", "signature": "sig2"}}, + ), + cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 0}), + cc.sse_frame( + "content_block_start", + {"type": "content_block_start", "index": 1, "content_block": {"type": "text", "text": ""}}, + ), + cc.sse_frame( + "content_block_delta", + {"type": "content_block_delta", "index": 1, "delta": {"type": "text_delta", "text": "got A"}}, + ), + cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 1}), + cc.sse_frame( + "content_block_start", + { + "type": "content_block_start", + "index": 2, + "content_block": {"type": "tool_use", "id": _TURN2_TOOL[0], "name": _TURN2_TOOL[1], "input": {}}, + }, + ), + cc.sse_frame( + "content_block_delta", + { + "type": "content_block_delta", + "index": 2, + "delta": {"type": "input_json_delta", "partial_json": '{"file_path": "/tmp/cc_pr'}, + }, + ), + cc.sse_frame( + "content_block_delta", + { + "type": "content_block_delta", + "index": 2, + "delta": {"type": "input_json_delta", "partial_json": 'obe/b.txt"}'}, + }, + ), + cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 2}), + cc.sse_frame( + "message_delta", + { + "type": "message_delta", + "delta": {"stop_reason": "tool_use", "stop_sequence": None}, + "usage": {"output_tokens": usage["output_tokens"]}, + }, + ), + cc.sse_frame("message_stop", {"type": "message_stop"}), + ) + + +def test_interleaved_thinking_text_and_tool_use_history_reaches_anthropic_identical(gateway: Gateway) -> None: + identity: Final = f"msg_il_{uuid.uuid4().hex}" + turn1: Final = cc.frontier_request( + f"cache-bust-{uuid.uuid4().hex}", + "high", + 64000, + prompt_text="Read /tmp/cc_probe/a.txt then /tmp/cc_probe/b.txt one at a time and reply with both words", + ) + turn2: Final = _turn2(turn1) + turn3: Final = _turn3(turn2) + turn2_expected: Final = {**turn2, "model": cc.FABLE} + turn3_expected: Final = {**turn3, "model": cc.FABLE} + + def respond(request: Request) -> Reply: + assert request.method == "POST" + assert request.target == "/v1/messages", request.target + upstream_beta: Final = request.headers.get("anthropic-beta", "") + assert upstream_beta.split(",").count("interleaved-thinking-2025-05-14") == 1, upstream_beta + body: Final = cc.JSON_OBJECT.validate_json(request.body) + if body == turn2_expected: + return Reply( + content_type="text/event-stream", + chunks=cc.text_stream("msg_il_turn2", cc.FABLE, "got A", {"input_tokens": 20, "output_tokens": 4}), + ) + assert body == turn3_expected, _diff(turn3_expected, body) + return Reply(content_type="text/event-stream", chunks=_interleaved_stream(identity)) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) + headers: Final = cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA) + response2: Final = gateway.request( + "POST", "/v1/messages", {**turn2, "model": model}, params={"beta": "true"}, headers=headers + ) + assert response2.status_code == 200, response2.text + response3: Final = gateway.request( + "POST", "/v1/messages", {**turn3, "model": model}, params={"beta": "true"}, headers=headers + ) + assert response3.status_code == 200, response3.text + events: Final = cc.sse_events(response3.text) + started: Final = [ + (data["index"], data["content_block"]["type"]) for event, data in events if event == "content_block_start" + ] + assert started == [(0, "thinking"), (1, "text"), (2, "tool_use")], started + assert events[-1][0] == "message_stop" + bodies: Final = tuple(cc.JSON_OBJECT.validate_json(request.body) for request in wire.drain()) + assert bodies == (turn2_expected, turn3_expected), bodies diff --git a/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_legacy_thinking_budget_wire.py b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_legacy_thinking_budget_wire.py new file mode 100644 index 00000000000..242e5c7ec5a --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_legacy_thinking_budget_wire.py @@ -0,0 +1,77 @@ +import json +import uuid +from typing import Final + +import pytest +from integration._support.client import Gateway +from integration._support.wire import Reply, Request, wire_server + +_MODEL: Final = "claude-sonnet-4-6" +_KEY: Final = "synthetic-anthropic-key" +_THINKING: Final = {"type": "enabled", "budget_tokens": 8000} +_TOOL: Final = { + "name": "read_file", + "description": "read a file", + "input_schema": {"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]}, +} +_NEXT_CALL: Final = {"type": "tool_use", "id": "call-2", "name": "read_file", "input": {"path": "schema.prisma"}} + + +def _tool_loop_history(identity: str) -> tuple[dict[str, object], ...]: + return ( + {"role": "user", "content": f"open the config for {identity}"}, + { + "role": "assistant", + "content": [{"type": "tool_use", "id": "call-1", "name": "read_file", "input": {"path": "config.yaml"}}], + }, + {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "call-1", "content": "model_list: []"}]}, + ) + + +def _tool_use_reply(identity: str) -> Reply: + return Reply( + body=json.dumps( + { + "id": f"msg-{identity}", + "type": "message", + "role": "assistant", + "model": _MODEL, + "content": [_NEXT_CALL], + "stop_reason": "tool_use", + "stop_sequence": None, + "usage": {"input_tokens": 40, "output_tokens": 12}, + } + ).encode() + ) + + +@pytest.mark.covers("providers.anthropic_messages.claude_4_6_legacy_thinking_budget_reaches_the_wire_unchanged") +def test_claude_4_6_thinking_budget_tokens_on_messages_is_forwarded_instead_of_rewritten_to_adaptive( + gateway: Gateway, +) -> None: + identity: Final = "legacy-thinking-" + uuid.uuid4().hex + history: Final = _tool_loop_history(identity) + + def respond(request: Request) -> Reply: + assert request.method == "POST" and request.target == "/v1/messages", request.target + assert request.headers["x-api-key"] == _KEY + body: Final = json.loads(request.body) + assert body["thinking"] == _THINKING, body + assert "output_config" not in body, body + assert body["max_tokens"] == 32768, body + assert body["messages"] == list(history), body + assert body["tools"] == [_TOOL], body + return _tool_use_reply(identity) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{_MODEL}", api_base=wire.url, api_key=_KEY) + response: Final = gateway.request( + "POST", + "/v1/messages", + {"model": model, "max_tokens": 32768, "thinking": _THINKING, "messages": history, "tools": [_TOOL]}, + ) + assert response.status_code == 200, response.text + body: Final = response.json() + assert body["content"] == [_NEXT_CALL], response.text + assert body["stop_reason"] == "tool_use", response.text + assert len(wire.drain()) == 1 diff --git a/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_model_switch_history_wire.py b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_model_switch_history_wire.py new file mode 100644 index 00000000000..5753d395bc4 --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_model_switch_history_wire.py @@ -0,0 +1,83 @@ +import uuid +from typing import Final + +from integration._support import claude_code as cc +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows +from integration._support.wire import Reply, Request, wire_server + + +def test_mid_loop_model_switch_replays_history_byte_identical(gateway: Gateway) -> None: + identity1: Final = f"msg_sw1_{uuid.uuid4().hex}" + identity2: Final = f"msg_sw2_{uuid.uuid4().hex}" + turn1: Final = cc.frontier_request( + f"cache-bust-{uuid.uuid4().hex}", + "high", + 64000, + prompt_text="Read /tmp/cc_probe/hello.txt and reply with its single word", + ) + turn2: Final = cc.tool_loop_turn2( + turn1, + ( + {"type": "thinking", "thinking": "need to read the file", "signature": "sig_anthropic_1"}, + { + "type": "tool_use", + "id": "toolu_read_1", + "name": "Read", + "input": {"file_path": "/tmp/cc_probe/hello.txt"}, + }, + ), + (("toolu_read_1", "1\tPROBE\n2\t"),), + ) + first_expected: Final = {**turn1, "model": cc.FABLE} + second_expected: Final = {**turn2, "model": cc.OPUS} + + def respond(request: Request) -> Reply: + assert request.method == "POST" + assert request.target == "/v1/messages", request.target + body: Final = cc.JSON_OBJECT.validate_json(request.body) + if body == first_expected: + return Reply( + content_type="text/event-stream", + chunks=cc.tool_use_stream( + identity1, + cc.FABLE, + "need to read the file", + "sig_anthropic_1", + (("toolu_read_1", "Read", {"file_path": "/tmp/cc_probe/hello.txt"}),), + {"input_tokens": 20, "output_tokens": 10}, + ), + ) + assert body == second_expected, { + key: (second_expected.get(key), body.get(key)) + for key in second_expected.keys() | body.keys() + if second_expected.get(key) != body.get(key) + } + return Reply( + content_type="text/event-stream", + chunks=cc.text_stream(identity2, cc.OPUS, "PROBE", {"input_tokens": 30, "output_tokens": 3}), + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + fable: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) + opus: Final = scenario.model(model=f"anthropic/{cc.OPUS}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) + headers: Final = cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA) + response1: Final = gateway.request( + "POST", "/v1/messages", {**turn1, "model": fable}, params={"beta": "true"}, headers=headers + ) + assert response1.status_code == 200, response1.text + response2: Final = gateway.request( + "POST", "/v1/messages", {**turn2, "model": opus}, params={"beta": "true"}, headers=headers + ) + assert response2.status_code == 200, response2.text + bodies: Final = tuple(cc.JSON_OBJECT.validate_json(request.body) for request in wire.drain()) + assert bodies == (first_expected, second_expected), bodies + rows: Final = eventually( + lambda: read_rows( + 'SELECT model FROM "LiteLLM_SpendLogs" WHERE request_id=%s', + (identity2,), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert rows[0]["model"] == f"anthropic/{cc.OPUS}" diff --git a/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_thinking_signature_retry_wire.py b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_thinking_signature_retry_wire.py new file mode 100644 index 00000000000..e414d8f0d11 --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_thinking_signature_retry_wire.py @@ -0,0 +1,95 @@ +import json +import uuid +from typing import Final + +import pytest +from integration._support.client import Gateway +from integration._support.wire import Reply, Request, wire_server + +MODEL: Final = "claude-sonnet-4-5-20250929" +KEY: Final = "synthetic-anthropic-key" +SIGNATURE_ERROR: Final = json.dumps( + { + "type": "error", + "error": { + "type": "invalid_request_error", + "message": "messages.2.content.0.thinking.signature.str: Input should be a valid string", + }, + } +).encode() +TOOLS: Final = ({"name": "lookup", "input_schema": {"type": "object", "properties": {"key": {"type": "string"}}}},) + + +def _history_with_unsigned_thinking(identity: str) -> tuple[dict[str, object], ...]: + return ( + {"role": "user", "content": [{"type": "text", "text": f"first question {identity}"}]}, + {"role": "assistant", "content": [{"type": "text", "text": "first answer"}]}, + {"role": "user", "content": [{"type": "text", "text": "second question"}]}, + { + "role": "assistant", + "content": [ + {"type": "thinking", "thinking": "replayed from another provider", "signature": None}, + {"type": "tool_use", "id": "call-1", "name": "lookup", "input": {"key": "value"}}, + ], + }, + {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "call-1", "content": "found"}]}, + ) + + +@pytest.mark.covers("providers.anthropic_messages.missing_thinking_signature_400_retries_without_thinking_blocks") +def test_missing_thinking_signature_400_retries_once_without_thinking_blocks_and_returns_200( + gateway: Gateway, +) -> None: + identity: Final = "thinking-signature-" + uuid.uuid4().hex + history: Final = _history_with_unsigned_thinking(identity) + tool_use_only_turn: Final = { + "role": "assistant", + "content": [{"type": "tool_use", "id": "call-1", "name": "lookup", "input": {"key": "value"}}], + } + + def respond(request: Request) -> Reply: + assert request.method == "POST" and request.target == "/v1/messages" + assert request.headers["x-api-key"] == KEY + body: Final = json.loads(request.body) + assert body["model"] == MODEL + assert body["tools"] == list(TOOLS), body + if body["messages"][3]["content"][0]["type"] == "thinking": + assert body["messages"] == list(history), body + assert body["thinking"] == {"type": "enabled", "budget_tokens": 1024}, body + return Reply(status=400, body=SIGNATURE_ERROR) + assert body["messages"] == [*history[:3], tool_use_only_turn, history[4]], body + assert "thinking" not in body, body + return Reply( + body=json.dumps( + { + "id": identity, + "type": "message", + "role": "assistant", + "model": MODEL, + "content": [{"type": "text", "text": "recovered without thinking history"}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 30, "output_tokens": 6}, + } + ).encode() + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{MODEL}", api_base=wire.url, api_key=KEY) + response: Final = gateway.request( + "POST", + "/v1/messages", + { + "model": model, + "max_tokens": 64, + "thinking": {"type": "enabled", "budget_tokens": 1024}, + "tools": list(TOOLS), + "messages": list(history), + }, + ) + assert response.status_code == 200, response.text + body: Final = response.json() + assert body["id"] == identity, response.text + assert body["content"] == [{"type": "text", "text": "recovered without thinking history"}], response.text + assert body["stop_reason"] == "end_turn", response.text + assert [request.target for request in wire.drain()] == ["/v1/messages", "/v1/messages"] diff --git a/tests/integration/messages_endpoint/providers/anthropic/streaming/test_anthropic_messages_live_lifecycle_wire.py b/tests/integration/messages_endpoint/providers/anthropic/streaming/test_anthropic_messages_live_lifecycle_wire.py new file mode 100644 index 00000000000..cb7043c0362 --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/streaming/test_anthropic_messages_live_lifecycle_wire.py @@ -0,0 +1,82 @@ +import json +import threading +import uuid +from typing import Final + +from integration._support.client import Gateway +from integration._support.wire import Reply, Request, wire_server + +_MODEL: Final = "claude-sonnet-4-5-20250929" +_API_KEY: Final = "synthetic-anthropic-key" + + +def _sse(event: str, payload: dict[str, object]) -> bytes: + return f"event: {event}\ndata: {json.dumps(payload)}\n\n".encode() + + +def test_messages_stream_message_start_reaches_client_before_content_without_fallback( + gateway: Gateway, +) -> None: + """With no fallback able to take over, the proxy must not hold lifecycle + frames back for a retry that cannot happen: message_start reaches the + client while the upstream is still thinking.""" + gate: Final = threading.Event() + head: Final = _sse("message_start", {"type": "message_start", "message": {"id": "msg_live_1"}}) + _sse( + "content_block_start", + {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}, + ) + tail: Final = ( + _sse( + "content_block_delta", + {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "Hello"}}, + ) + + _sse("content_block_stop", {"type": "content_block_stop", "index": 0}) + + _sse( + "message_delta", + {"type": "message_delta", "delta": {"stop_reason": "end_turn"}, "usage": {"output_tokens": 3}}, + ) + + _sse("message_stop", {"type": "message_stop"}) + ) + prompt: Final = "live-lifecycle-" + uuid.uuid4().hex + + def respond(request: Request) -> Reply: + assert request.method == "POST" and request.target == "/v1/messages" + assert request.headers["x-api-key"] == _API_KEY + body: Final = json.loads(request.body) + assert body["model"] == _MODEL + assert body["stream"] is True + assert body["messages"] == [{"role": "user", "content": prompt}] + return Reply(content_type="text/event-stream", chunks=(head, tail), gate_after_first=gate) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{_MODEL}", api_base=wire.url, api_key=_API_KEY) + with gateway.client.stream( + "POST", + "/v1/messages", + json={ + "model": model, + "max_tokens": 16, + "stream": True, + "messages": [{"role": "user", "content": prompt}], + }, + headers={"Authorization": f"Bearer {gateway.key}"}, + ) as response: + assert response.status_code == 200, response.read().decode() + lines = response.iter_lines() + first_event: Final = next( + json.loads(line.removeprefix("data: ")) for line in lines if line.startswith("data: ") + ) + assert first_event["type"] == "message_start" + gate.set() + events: Final = (first_event,) + tuple( + json.loads(line.removeprefix("data: ")) for line in lines if line.startswith("data: ") + ) + assert tuple(event["type"] for event in events) == ( + "message_start", + "content_block_start", + "content_block_delta", + "content_block_stop", + "message_delta", + "message_stop", + ), f"observed events: {events!r}" + assert [request.target for request in wire.drain()] == ["/v1/messages"] diff --git a/tests/integration/messages_endpoint/providers/anthropic/tools/test_anthropic_advisor_wire.py b/tests/integration/messages_endpoint/providers/anthropic/tools/test_anthropic_advisor_wire.py new file mode 100644 index 00000000000..b2f44d8f155 --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/tools/test_anthropic_advisor_wire.py @@ -0,0 +1,201 @@ +import json +import uuid +from pathlib import Path +from typing import Final + +import pytest +import yaml +from integration._support.client import Gateway +from integration._support.process import owned_proxy +from integration._support.wire import Reply, Request, wire_server + +_ADVISOR_KEY: Final = "synthetic-advisor-key" +_PROXY_ANTHROPIC_KEY: Final = "sk-proxy-owned-anthropic-secret" +_QUESTION: Final = "which index should this query use" +_ADVICE: Final = "use the composite index on (tenant_id, created_at)" +_FINAL_ANSWER: Final = "done, the composite index is the right one" + + +def _advisor_call_message(question: str) -> dict[str, object]: + return { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "advisor-call", + "type": "function", + "function": {"name": "advisor", "arguments": json.dumps({"question": question})}, + } + ], + } + + +_FINAL_MESSAGE: Final = {"role": "assistant", "content": _FINAL_ANSWER} + + +def _chat_completion(identity: str, message: dict[str, object], finish_reason: str) -> Reply: + return Reply( + body=json.dumps( + { + "id": f"chatcmpl-{identity}", + "object": "chat.completion", + "created": 1, + "model": "llama-3.3-70b-versatile", + "choices": [{"index": 0, "message": message, "finish_reason": finish_reason}], + "usage": {"prompt_tokens": 10, "completion_tokens": 4, "total_tokens": 14}, + } + ).encode() + ) + + +def _executor_reply(body: dict[str, object], identity: str, question: str) -> Reply: + messages: Final = body["messages"] + assert isinstance(messages, list) + if any(message.get("role") == "tool" for message in messages): + assert messages[-1]["content"] == _ADVICE + return _chat_completion(identity, _FINAL_MESSAGE, "stop") + tools: Final = body["tools"] + assert isinstance(tools, list) + assert tools[0]["function"]["name"] == "advisor" + return _chat_completion(identity, _advisor_call_message(question), "tool_calls") + + +@pytest.mark.covers("providers.anthropic_messages_advisor.sub_call_uses_the_configured_advisor_deployment") +def test_advisor_sub_call_reaches_the_router_deployment_with_its_key_instead_of_anthropic_unauthenticated( + gateway: Gateway, +) -> None: + identity: Final = "advisor-wire-" + uuid.uuid4().hex + migration: Final = "please plan the migration " + identity + question: Final = _QUESTION + " " + identity + + def respond(request: Request) -> Reply: + body: Final = json.loads(request.body) + if request.target == "/v1/chat/completions": + assert request.headers["authorization"] == "Bearer integration-provider-key" + return _executor_reply(body, identity, question) + assert request.target == "/v1/messages" + assert request.headers["x-api-key"] == _ADVISOR_KEY + assert body["model"] == "claude-opus-4-1-20250805" + assert body["messages"] == [ + {"role": "user", "content": migration}, + {"role": "user", "content": question}, + ] + assert "tools" not in body + return Reply( + body=json.dumps( + { + "id": f"msg-{identity}", + "type": "message", + "role": "assistant", + "model": "claude-opus-4-1-20250805", + "content": [{"type": "text", "text": _ADVICE}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 12, "output_tokens": 6}, + } + ).encode() + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + executor: Final = scenario.model(model="hosted_vllm/gpt-4o-mini", api_base=wire.url + "/v1") + advisor: Final = scenario.model( + model="anthropic/claude-opus-4-1-20250805", api_base=wire.url, api_key=_ADVISOR_KEY + ) + response: Final = gateway.request( + "POST", + "/v1/messages", + { + "model": executor, + "max_tokens": 64, + "messages": [{"role": "user", "content": migration}], + "tools": [{"type": "advisor_20260301", "name": "advisor", "model": advisor}], + }, + ) + assert response.status_code == 200, response.text + body: Final = response.json() + assert body["content"] == [{"type": "text", "text": _FINAL_ANSWER}], response.text + assert body["stop_reason"] == "end_turn", response.text + assert [request.target for request in wire.drain()] == [ + "/v1/chat/completions", + "/v1/messages", + "/v1/chat/completions", + ] + + +def _advice_reply(identity: str) -> Reply: + return Reply( + body=json.dumps( + { + "id": f"msg-{identity}", + "type": "message", + "role": "assistant", + "model": "claude-opus-4-1-20250805", + "content": [{"type": "text", "text": _ADVICE}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 12, "output_tokens": 6}, + } + ).encode() + ) + + +@pytest.mark.covers("providers.anthropic_messages_advisor.caller_api_base_without_api_key_never_receives_the_proxy_key") +def test_advisor_api_base_without_api_key_is_rejected_before_the_proxy_anthropic_key_reaches_the_caller_host( + gateway: Gateway, tmp_path: Path +) -> None: + identity: Final = "advisor-leak-" + uuid.uuid4().hex + question: Final = _QUESTION + " " + identity + + def executor(request: Request) -> Reply: + assert request.target == "/v1/chat/completions", request.target + return _executor_reply(json.loads(request.body), identity, question) + + def caller_host(request: Request) -> Reply: + return _advice_reply(identity) + + config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + config["general_settings"]["allow_client_side_credentials"] = True + path: Final = tmp_path / "client-side-credentials.yaml" + path.write_text(yaml.safe_dump(config)) + with ( + wire_server(executor) as executor_wire, + wire_server(caller_host) as caller_wire, + owned_proxy(gateway, tmp_path, {"ANTHROPIC_API_KEY": _PROXY_ANTHROPIC_KEY}, config=path) as candidate, + candidate.scenario() as scenario, + ): + model: Final = scenario.model(model="hosted_vllm/gpt-4o-mini", api_base=executor_wire.url + "/v1") + response: Final = candidate.request( + "POST", + "/v1/messages", + { + "model": model, + "max_tokens": 64, + "messages": [{"role": "user", "content": "please plan the migration"}], + "tools": [ + { + "type": "advisor_20260301", + "name": "advisor", + "model": "anthropic/claude-opus-4-1-20250805", + "api_base": caller_wire.url, + } + ], + }, + ) + received: Final = caller_wire.drain() + assert [ + (request.target, request.headers.get("x-api-key"), json.loads(request.body)["messages"]) + for request in received + ] == [], response.text + assert response.is_error, response.text + assert response.json() == { + "type": "error", + "error": { + "type": "api_error", + "message": ( + "advisor tool definition sets 'api_base' without 'api_key'. A caller-supplied api_base is only " + "honored alongside a caller-supplied api_key, so the proxy's own credentials are never sent to a " + "caller-chosen destination." + ), + }, + }, response.text + assert executor_wire.drain() == (), response.text diff --git a/tests/integration/messages_endpoint/providers/anthropic/tools/test_anthropic_tool_loop_wire.py b/tests/integration/messages_endpoint/providers/anthropic/tools/test_anthropic_tool_loop_wire.py new file mode 100644 index 00000000000..f3fb9786e02 --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/tools/test_anthropic_tool_loop_wire.py @@ -0,0 +1,165 @@ +import uuid +from typing import Final + +from integration._support import claude_code as cc +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows +from integration._support.wire import Reply, Request, wire_server +from pydantic import JsonValue + +_THINKING: Final = "need to read the file" +_SIGNATURE: Final = "sig_probe_1" +_USAGE: Final = {"input_tokens": 20, "output_tokens": 10} + + +def _diff(expected: dict[str, JsonValue], body: dict[str, JsonValue]) -> dict[str, JsonValue]: + return { + key: {"expected": expected.get(key), "upstream": body.get(key)} + for key in expected.keys() | body.keys() + if expected.get(key) != body.get(key) + } + + +def test_tool_loop_round_trips_thinking_tool_use_and_tool_result(gateway: Gateway) -> None: + identity1: Final = f"msg_tl1_{uuid.uuid4().hex}" + identity2: Final = f"msg_tl2_{uuid.uuid4().hex}" + turn1: Final = cc.frontier_request( + f"cache-bust-{uuid.uuid4().hex}", + "high", + 64000, + prompt_text="Read /tmp/cc_probe/hello.txt and reply with its single word", + ) + calls: Final = (("toolu_read_1", "Read", {"file_path": "/tmp/cc_probe/hello.txt"}),) + turn2: Final = cc.tool_loop_turn2( + turn1, + ( + {"type": "thinking", "thinking": _THINKING, "signature": _SIGNATURE}, + { + "type": "tool_use", + "id": "toolu_read_1", + "name": "Read", + "input": {"file_path": "/tmp/cc_probe/hello.txt"}, + }, + ), + (("toolu_read_1", "1\tPROBE\n2\t"),), + ) + first_expected: Final = {**turn1, "model": cc.FABLE} + second_expected: Final = {**turn2, "model": cc.FABLE} + + def respond(request: Request) -> Reply: + assert request.method == "POST" + assert request.target == "/v1/messages", request.target + body: Final = cc.JSON_OBJECT.validate_json(request.body) + if body == first_expected: + return Reply( + content_type="text/event-stream", + chunks=cc.tool_use_stream(identity1, cc.FABLE, _THINKING, _SIGNATURE, calls, _USAGE), + ) + assert body == second_expected, _diff(second_expected, body) + return Reply( + content_type="text/event-stream", + chunks=cc.text_stream(identity2, cc.FABLE, "PROBE", {"input_tokens": 30, "output_tokens": 3}), + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) + headers: Final = cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA) + response1: Final = gateway.request( + "POST", "/v1/messages", {**turn1, "model": model}, params={"beta": "true"}, headers=headers + ) + assert response1.status_code == 200, response1.text + events: Final = cc.sse_events(response1.text) + assert [ + ( + event, + data.get("delta", {}).get( + "type", data.get("content_block", {}).get("type", data.get("delta", {}).get("stop_reason")) + ), + ) + for event, data in events + ] == [ + ("message_start", None), + ("content_block_start", "thinking"), + ("content_block_delta", "thinking_delta"), + ("content_block_delta", "signature_delta"), + ("content_block_stop", None), + ("content_block_start", "tool_use"), + ("content_block_delta", "input_json_delta"), + ("content_block_delta", "input_json_delta"), + ("content_block_stop", None), + ("message_delta", "tool_use"), + ("message_stop", None), + ] + assert events[5][1]["content_block"]["id"] == "toolu_read_1" + assert events[5][1]["content_block"]["name"] == "Read" + partial: Final = events[6][1]["delta"]["partial_json"] + events[7][1]["delta"]["partial_json"] + assert partial == '{"file_path": "/tmp/cc_probe/hello.txt"}' + response2: Final = gateway.request( + "POST", "/v1/messages", {**turn2, "model": model}, params={"beta": "true"}, headers=headers + ) + assert response2.status_code == 200, response2.text + events2: Final = cc.sse_events(response2.text) + assert events2[2][1]["delta"] == {"type": "text_delta", "text": "PROBE"} + assert events2[4][1]["delta"]["stop_reason"] == "end_turn" + bodies: Final = tuple(cc.JSON_OBJECT.validate_json(request.body) for request in wire.drain()) + assert bodies == (first_expected, second_expected), bodies + rows: Final = eventually( + lambda: read_rows( + 'SELECT prompt_tokens FROM "LiteLLM_SpendLogs" WHERE request_id=%s', + (identity2,), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert rows[0]["prompt_tokens"] == 30 + + +def test_parallel_tool_results_reach_anthropic_in_client_order(gateway: Gateway) -> None: + turn1: Final = cc.frontier_request( + f"cache-bust-{uuid.uuid4().hex}", + "high", + 64000, + prompt_text="Read /tmp/cc_probe/hello.txt and /tmp/cc_probe/world.txt and reply with both words", + ) + turn2: Final = cc.tool_loop_turn2( + turn1, + ( + {"type": "thinking", "thinking": _THINKING, "signature": _SIGNATURE}, + { + "type": "tool_use", + "id": "toolu_read_1", + "name": "Read", + "input": {"file_path": "/tmp/cc_probe/hello.txt"}, + }, + { + "type": "tool_use", + "id": "toolu_read_2", + "name": "Read", + "input": {"file_path": "/tmp/cc_probe/world.txt"}, + }, + ), + (("toolu_read_2", "1\tPROBE2\n2\t"), ("toolu_read_1", "1\tPROBE\n2\t")), + ) + + def respond(request: Request) -> Reply: + body: Final = cc.JSON_OBJECT.validate_json(request.body) + expected: Final = {**turn2, "model": cc.FABLE} + assert body == expected, _diff(expected, body) + results: Final = [block for block in body["messages"][3]["content"] if block["type"] == "tool_result"] + assert [block["tool_use_id"] for block in results] == ["toolu_read_2", "toolu_read_1"] + return Reply( + content_type="text/event-stream", + chunks=cc.text_stream(f"msg_mt_{uuid.uuid4().hex}", cc.FABLE, "PROBE PROBE2", _USAGE), + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) + response: Final = gateway.request( + "POST", + "/v1/messages", + {**turn2, "model": model}, + params={"beta": "true"}, + headers=cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA), + ) + assert response.status_code == 200, response.text + assert len(wire.drain()) == 1 diff --git a/tests/integration/messages_endpoint/providers/anthropic/tools/test_anthropic_web_search_citations_wire.py b/tests/integration/messages_endpoint/providers/anthropic/tools/test_anthropic_web_search_citations_wire.py new file mode 100644 index 00000000000..66852e6e21b --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/tools/test_anthropic_web_search_citations_wire.py @@ -0,0 +1,170 @@ +import uuid +from typing import Final + +from integration._support import claude_code as cc +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows +from integration._support.wire import Reply, Request, wire_server + +WEB_SEARCH_TOOL: Final = {"type": "web_search_20250305", "name": "web_search", "max_uses": 8} + + +def _web_search_stream(identity: str) -> tuple[bytes, ...]: + return ( + cc.sse_frame( + "message_start", + { + "type": "message_start", + "message": { + "id": identity, + "type": "message", + "role": "assistant", + "model": cc.FABLE, + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 20, "output_tokens": 1, "server_tool_use": {"web_search_requests": 1}}, + }, + }, + ), + cc.sse_frame( + "content_block_start", + { + "type": "content_block_start", + "index": 0, + "content_block": {"type": "server_tool_use", "id": "srvtoolu_1", "name": "web_search", "input": {}}, + }, + ), + cc.sse_frame( + "content_block_delta", + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "input_json_delta", "partial_json": '{"query": "current LiteLLM version"}'}, + }, + ), + cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 0}), + cc.sse_frame( + "content_block_start", + { + "type": "content_block_start", + "index": 1, + "content_block": { + "type": "web_search_tool_result", + "tool_use_id": "srvtoolu_1", + "content": [ + { + "type": "web_search_result", + "title": "litellm releases", + "url": "https://example.com/litellm", + "page_age": None, + "encrypted_content": "enc_ws_1", + } + ], + }, + }, + ), + cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 1}), + cc.sse_frame( + "content_block_start", + {"type": "content_block_start", "index": 2, "content_block": {"type": "text", "text": ""}}, + ), + cc.sse_frame( + "content_block_delta", + {"type": "content_block_delta", "index": 2, "delta": {"type": "text_delta", "text": "1.104.0"}}, + ), + cc.sse_frame( + "content_block_delta", + { + "type": "content_block_delta", + "index": 2, + "delta": { + "type": "citations_delta", + "citation": { + "type": "web_search_result_location", + "url": "https://example.com/litellm", + "title": "litellm releases", + "cited_text": "version 1.104.0", + "encrypted_index": "eidx_1", + }, + }, + }, + ), + cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 2}), + cc.sse_frame( + "message_delta", + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": {"output_tokens": 15, "server_tool_use": {"web_search_requests": 1}}, + }, + ), + cc.sse_frame("message_stop", {"type": "message_stop"}), + ) + + +def test_web_search_tool_passthrough_and_cited_response(gateway: Gateway) -> None: + identity: Final = f"msg_ws_{uuid.uuid4().hex}" + base: Final = cc.frontier_request( + f"cache-bust-{uuid.uuid4().hex}", + "high", + 64000, + prompt_text="Use web search to find the current LiteLLM version and answer in one word", + ) + request_body: Final = { + **base, + "tools": [*base["tools"], WEB_SEARCH_TOOL], + } + + def respond(request: Request) -> Reply: + assert request.method == "POST" + assert request.target == "/v1/messages", request.target + body: Final = cc.JSON_OBJECT.validate_json(request.body) + expected: Final = {**request_body, "model": cc.FABLE} + assert body == expected, { + key: (expected.get(key), body.get(key)) + for key in expected.keys() | body.keys() + if expected.get(key) != body.get(key) + } + assert body["tools"][-1] == WEB_SEARCH_TOOL + assert len({tool["name"] for tool in body["tools"]}) == len(body["tools"]), body["tools"] + return Reply(content_type="text/event-stream", chunks=_web_search_stream(identity)) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model( + model=f"anthropic/{cc.FABLE}", + api_base=wire.url, + api_key=cc.ANTHROPIC_API_KEY, + input_cost_per_token=1e-6, + output_cost_per_token=5e-6, + ) + response: Final = gateway.request( + "POST", + "/v1/messages", + {**request_body, "model": model}, + params={"beta": "true"}, + headers=cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA), + ) + assert response.status_code == 200, response.text + events: Final = cc.sse_events(response.text) + started: Final = [ + (data["index"], data["content_block"]["type"]) for event, data in events if event == "content_block_start" + ] + assert started == [(0, "server_tool_use"), (1, "web_search_tool_result"), (2, "text")], started + citations: Final = [ + data["delta"] for event, data in events if data.get("delta", {}).get("type") == "citations_delta" + ] + assert len(citations) == 1 and citations[0]["citation"]["url"] == "https://example.com/litellm", citations + start_usage: Final = events[0][1]["message"]["usage"] + assert start_usage["server_tool_use"]["web_search_requests"] == 1, start_usage + assert len(wire.drain()) == 1 + rows: Final = eventually( + lambda: read_rows( + 'SELECT spend, prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" WHERE request_id=%s', + (identity,), + ), + lambda values: len(values) == 1, + seconds=70, + ) + token_cost: Final = 20 * 1e-6 + 15 * 5e-6 + assert float(rows[0]["spend"]) >= token_cost, dict(rows[0]) diff --git a/tests/integration/messages_endpoint/providers/anthropic/tools/test_websearch_interception_wire.py b/tests/integration/messages_endpoint/providers/anthropic/tools/test_websearch_interception_wire.py new file mode 100644 index 00000000000..a6f098cf64f --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/tools/test_websearch_interception_wire.py @@ -0,0 +1,409 @@ +import json +from pathlib import Path +from typing import Final + +import pytest +import yaml +from integration._support.client import Gateway +from integration._support.process import owned_proxy +from integration._support.wire import Reply, Request, wire_server + +BEDROCK_MODEL: Final = "us.anthropic.claude-haiku-4-5-20251001-v1:0" +INVOKE_TARGET: Final = f"/model/{BEDROCK_MODEL}/invoke" +SEARCH_TARGET: Final = "/tavily/search" +SEARCH_RESULT: Final = { + "title": "Synthetic result", + "url": "https://example.test/result", + "content": "the snippet text", +} + + +def sse_events(text: str) -> tuple[tuple[str, dict[str, object]], ...]: + frames: Final = tuple(frame for frame in text.split("\n\n") if frame.strip()) + return tuple( + ( + next(line.removeprefix("event: ") for line in frame.splitlines() if line.startswith("event: ")), + json.loads(next(line.removeprefix("data: ") for line in frame.splitlines() if line.startswith("data: "))), + ) + for frame in frames + ) + + +@pytest.mark.covers("other.provider_wire.bedrock.websearch_interception_streamed_capped_turn_ends_with_native_results") +def test_streamed_web_search_turn_capped_by_max_agentic_loops_ends_turn_with_snippets_and_ordered_blocks( + gateway: Gateway, tmp_path: Path +) -> None: + def respond(request: Request) -> Reply: + assert request.method == "POST", request.target + body: Final = json.loads(request.body) + if request.target == SEARCH_TARGET: + assert request.headers["authorization"] == "Bearer synthetic-tavily-key" + assert body["query"] == "query-0", body + return Reply(body=json.dumps({"query": "query-0", "results": [SEARCH_RESULT]}).encode()) + assert request.target == INVOKE_TARGET + assert request.headers["authorization"] == "Bearer synthetic-bedrock-token" + assert [tool["name"] for tool in body["tools"]] == ["litellm_web_search"], body["tools"] + assert "stream" not in body, body + depth: Final = sum( + 1 + for message in body["messages"] + if isinstance(message["content"], list) + for block in message["content"] + if block["type"] == "tool_result" + ) + if depth == 1: + assert body["messages"][2]["content"] == [ + { + "type": "tool_result", + "tool_use_id": "toolu_0", + "content": "Title: Synthetic result\nURL: https://example.test/result\nSnippet: the snippet text", + } + ], body["messages"] + return Reply( + body=json.dumps( + { + "id": f"msg_{depth}", + "type": "message", + "role": "assistant", + "model": BEDROCK_MODEL, + "content": [ + {"type": "text", "text": f"turn-{depth}"}, + { + "type": "tool_use", + "id": f"toolu_{depth}", + "name": "litellm_web_search", + "input": {"query": f"query-{depth}"}, + }, + ], + "stop_reason": "tool_use", + "stop_sequence": None, + "usage": {"input_tokens": 10, "output_tokens": 4}, + } + ).encode() + ) + + with wire_server(respond) as wire: + config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + config["search_tools"] = [ + { + "search_tool_name": "integration-search", + "litellm_params": { + "search_provider": "tavily", + "api_key": "synthetic-tavily-key", + "api_base": wire.url + "/tavily", + }, + } + ] + config["litellm_settings"].update( + { + "callbacks": ["websearch_interception"], + "websearch_interception_params": { + "enabled_providers": ["bedrock"], + "search_tool_name": "integration-search", + "max_agentic_loops": 1, + }, + } + ) + path: Final = tmp_path / "websearch.yaml" + path.write_text(yaml.safe_dump(config)) + with owned_proxy(gateway, tmp_path, {}, config=path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model( + model=f"bedrock/{BEDROCK_MODEL}", + api_key="synthetic-bedrock-token", + api_base=wire.url, + aws_region_name="us-east-1", + aws_bedrock_runtime_endpoint=wire.url, + ) + response: Final = candidate.request( + "POST", + "/v1/messages", + { + "model": model, + "max_tokens": 64, + "stream": True, + "messages": [{"role": "user", "content": "search control"}], + "tools": [{"type": "web_search_20250305", "name": "web_search"}], + }, + ) + assert response.status_code == 200, response.text + events: Final = sse_events(response.text) + assert [name for name, _ in events][:1] == ["message_start"], response.text + assert [name for name, _ in events][-2:] == ["message_delta", "message_stop"], response.text + for position, (name, event) in enumerate(events): + if name == "content_block_stop": + assert event["index"] in { + earlier_event["index"] + for earlier, earlier_event in events[:position] + if earlier == "content_block_start" + }, response.text + started: Final = tuple(event["content_block"] for name, event in events if name == "content_block_start") + search_ids: Final = tuple(block["id"] for block in started if block["type"] == "server_tool_use") + assert search_ids and all(search_id.startswith("srvtoolu_") for search_id in search_ids), response.text + assert started[-1] == {"type": "text", "text": ""}, response.text + assert started[:-1] == tuple( + block + for search_id in search_ids + for block in ( + {"type": "server_tool_use", "id": search_id, "name": "web_search", "input": {"query": "query-0"}}, + { + "type": "web_search_tool_result", + "tool_use_id": search_id, + "content": [ + { + "type": "web_search_result", + "url": "https://example.test/result", + "title": "Synthetic result", + "page_age": None, + "encrypted_content": "", + "snippet": "the snippet text", + } + ], + }, + ) + ), response.text + assert ( + "".join(event["delta"]["text"] for name, event in events if name == "content_block_delta") == "turn-1" + ), response.text + assert [event["delta"]["stop_reason"] for name, event in events if name == "message_delta"] == [ + "end_turn" + ], response.text + assert "litellm_web_search" not in response.text, response.text + assert [request.target for request in wire.drain()] == [INVOKE_TARGET, SEARCH_TARGET, INVOKE_TARGET] + + +import threading +import uuid +from typing import Final +from urllib.parse import parse_qs, urlsplit + +import httpx +import pytest +from integration._support.client import Gateway, eventually + +_QUERY: Final = "integration capped search" +_TEXT_BLOCK: Final = {"type": "text", "text": "searching once more"} +_NOT_INTERCEPTED: Final = "native tool reached the provider" +_FINAL_BLOCK: Final = {"type": "text", "text": "answered from the stored backend"} +_OWNED_RESULT_TEXT: Final = "Title: Owned result\nURL: https://owned.invalid/a\nSnippet: owned snippet" +_SEARCH_RESULT_BLOCK: Final = { + "type": "web_search_result", + "url": "https://owned.invalid/a", + "title": "Owned result", + "page_age": None, + "encrypted_content": "", + "snippet": "owned snippet", +} + + +def _search_tool_use(identity: str) -> dict[str, object]: + return {"type": "tool_use", "id": identity, "name": "litellm_web_search", "input": {"query": _QUERY}} + + +def _anthropic_reply(identity: str, content: list[dict[str, object]], stop_reason: str) -> Reply: + return Reply( + body=json.dumps( + { + "id": identity, + "type": "message", + "role": "assistant", + "model": "claude-sonnet-4-5-20250929", + "content": content, + "stop_reason": stop_reason, + "stop_sequence": None, + "usage": {"input_tokens": 10, "output_tokens": 4}, + } + ).encode() + ) + + +@pytest.mark.covers( + "other.provider_wire.anthropic.websearch_interception_capped_loop_ends_turn_without_internal_tool_use" +) +def test_capped_websearch_interception_loop_ends_turn_instead_of_exposing_internal_tool_use( + gateway: Gateway, tmp_path: Path +) -> None: + identity: Final = "websearch-wire-" + uuid.uuid4().hex + searched: Final = threading.Event() + + def respond(request: Request) -> Reply: + parts: Final = urlsplit(request.target) + if request.method == "GET" and parts.path == "/search": + assert parse_qs(parts.query)["q"] == [_QUERY], request.target + searched.set() + return Reply( + body=json.dumps( + { + "results": [ + {"title": "Owned result", "url": "https://owned.invalid/a", "content": "owned snippet"} + ] + } + ).encode() + ) + assert request.method == "POST" and parts.path == "/v1/messages", request.target + body: Final = json.loads(request.body) + if any(tool.get("type") == "web_search_20250305" for tool in body["tools"]): + return _anthropic_reply(identity, [{"type": "text", "text": _NOT_INTERCEPTED}], "end_turn") + assert [tool["name"] for tool in body["tools"]] == ["litellm_web_search"], body["tools"] + return _anthropic_reply(identity, [_TEXT_BLOCK, _search_tool_use(identity)], "tool_use") + + def send(candidate: Gateway, model: str) -> httpx.Response: + return candidate.request( + "POST", + "/v1/messages", + { + "model": model, + "max_tokens": 64, + "messages": [{"role": "user", "content": identity + " attempt " + uuid.uuid4().hex}], + "tools": [{"type": "web_search_20250305", "name": "web_search", "max_uses": 3}], + }, + ) + + def searched_through_proxy(response: httpx.Response) -> bool: + return searched.is_set() and _NOT_INTERCEPTED not in response.text + + with wire_server(respond) as wire: + config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + config["search_tools"] = [ + { + "search_tool_name": "integration-searxng", + "litellm_params": {"search_provider": "searxng", "api_base": wire.url}, + } + ] + config["litellm_settings"].update( + { + "callbacks": ["websearch_interception"], + "websearch_interception_params": { + "enabled": True, + "enabled_providers": ["anthropic"], + "search_tool_name": "integration-searxng", + }, + } + ) + path: Final = tmp_path / "websearch.yaml" + path.write_text(yaml.safe_dump(config)) + with owned_proxy(gateway, tmp_path, {}, config=path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model( + model="anthropic/claude-sonnet-4-5-20250929", api_base=wire.url, api_key="synthetic-anthropic-key" + ) + response: Final = eventually(lambda: send(candidate, model), searched_through_proxy, seconds=40) + assert response.status_code == 200, response.text + body: Final = response.json() + assert body["stop_reason"] == "end_turn", response.text + content: Final = body["content"] + assert [block["type"] for block in content] == ["server_tool_use", "web_search_tool_result", "text"], ( + response.text + ) + assert content[0]["name"] == "web_search" and content[0]["input"] == {"query": _QUERY}, response.text + assert content[1]["tool_use_id"] == content[0]["id"], response.text + assert content[1]["content"] == [_SEARCH_RESULT_BLOCK], response.text + assert content[2] == _TEXT_BLOCK, response.text + targets: Final = tuple((request.method, urlsplit(request.target).path) for request in wire.drain()) + assert targets[-3:] == (("POST", "/v1/messages"), ("GET", "/search"), ("POST", "/v1/messages")), targets + + +@pytest.mark.covers("other.provider_wire.anthropic.websearch_interception_uses_database_search_tool_backend") +def test_database_created_search_tool_backend_receives_the_intercepted_query_over_a_same_named_config_tool( + gateway: Gateway, tmp_path: Path +) -> None: + identity: Final = "websearch-db-" + uuid.uuid4().hex + tool_name: Final = "integration-db-searxng-" + uuid.uuid4().hex + searched: Final = threading.Event() + + def respond(request: Request) -> Reply: + parts: Final = urlsplit(request.target) + if request.method == "GET" and parts.path == "/database/search": + assert parse_qs(parts.query)["q"] == [_QUERY], request.target + searched.set() + return Reply( + body=json.dumps( + { + "results": [ + {"title": "Owned result", "url": "https://owned.invalid/a", "content": "owned snippet"} + ] + } + ).encode() + ) + assert request.method == "POST" and parts.path == "/v1/messages", request.target + body: Final = json.loads(request.body) + if any(tool.get("type") == "web_search_20250305" for tool in body["tools"]): + return _anthropic_reply(identity, [{"type": "text", "text": _NOT_INTERCEPTED}], "end_turn") + assert [tool["name"] for tool in body["tools"]] == ["litellm_web_search"], body["tools"] + results: Final = [ + block + for message in body["messages"] + if isinstance(message["content"], list) + for block in message["content"] + if block["type"] == "tool_result" + ] + if not results: + return _anthropic_reply(identity, [_TEXT_BLOCK, _search_tool_use(identity)], "tool_use") + assert results == [{"type": "tool_result", "tool_use_id": identity, "content": _OWNED_RESULT_TEXT}], results + return _anthropic_reply(identity, [_FINAL_BLOCK], "end_turn") + + def send(candidate: Gateway, model: str) -> httpx.Response: + return candidate.request( + "POST", + "/v1/messages", + { + "model": model, + "max_tokens": 64, + "messages": [{"role": "user", "content": identity + " attempt " + uuid.uuid4().hex}], + "tools": [{"type": "web_search_20250305", "name": "web_search", "max_uses": 3}], + }, + ) + + def searched_through_proxy(response: httpx.Response) -> bool: + return searched.is_set() and _NOT_INTERCEPTED not in response.text + + with wire_server(respond) as wire, gateway.scenario() as scenario: + created: Final = gateway.post( + "/search_tools", + { + "search_tool": { + "search_tool_name": tool_name, + "litellm_params": {"search_provider": "searxng", "api_base": wire.url + "/database"}, + } + }, + ) + scenario.cleanups.callback(gateway.request, "DELETE", f"/search_tools/{created['search_tool_id']}") + config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + config["search_tools"] = [ + { + "search_tool_name": tool_name, + "litellm_params": {"search_provider": "searxng", "api_base": wire.url + "/config"}, + } + ] + config["litellm_settings"].update( + { + "callbacks": ["websearch_interception"], + "websearch_interception_params": { + "enabled": True, + "enabled_providers": ["anthropic"], + "search_tool_name": tool_name, + }, + } + ) + path: Final = tmp_path / "websearch-db.yaml" + path.write_text(yaml.safe_dump(config)) + environment: Final = {"ANTHROPIC_API_BASE": wire.url} + with owned_proxy(gateway, tmp_path, environment, config=path) as candidate, candidate.scenario() as models: + model: Final = models.model( + model="anthropic/claude-sonnet-4-5-20250929", api_base=wire.url, api_key="synthetic-anthropic-key" + ) + response: Final = eventually(lambda: send(candidate, model), searched_through_proxy, seconds=40) + assert response.status_code == 200, response.text + body: Final = response.json() + assert body["stop_reason"] == "end_turn", response.text + assert body["content"][-1] == _FINAL_BLOCK, response.text + found: Final = [ + (result["url"], result["title"]) + for block in body["content"] + if block["type"] == "web_search_tool_result" + for result in block["content"] + ] + assert found == [("https://owned.invalid/a", "Owned result")], response.text + assert "litellm_web_search" not in response.text, response.text + targets: Final = tuple((request.method, urlsplit(request.target).path) for request in wire.drain()) + assert targets[-3:] == (("POST", "/v1/messages"), ("GET", "/database/search"), ("POST", "/v1/messages")), ( + targets + ) diff --git a/tests/integration/messages_endpoint/providers/anthropic/usage/test_anthropic_long_context_beta_wire.py b/tests/integration/messages_endpoint/providers/anthropic/usage/test_anthropic_long_context_beta_wire.py new file mode 100644 index 00000000000..7fe405ca439 --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/usage/test_anthropic_long_context_beta_wire.py @@ -0,0 +1,53 @@ +import uuid +from typing import Final + +import pytest +from integration._support import claude_code as cc +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows +from integration._support.wire import Reply, Request, wire_server + +_BETA_1M: Final = f"{cc.FRONTIER_CLI_BETA.replace(',effort-2025-11-24', ',context-1m-2025-08-07,effort-2025-11-24')}" + + +def test_1m_context_beta_forwarded_and_tiered_prompt_priced_above_200k(gateway: Gateway) -> None: + identity: Final = f"msg_1m_{uuid.uuid4().hex}" + request_body: Final = cc.frontier_request(f"cache-bust-{uuid.uuid4().hex}", "high", 64000) + + def respond(request: Request) -> Reply: + assert request.method == "POST" + assert request.target == "/v1/messages", request.target + upstream_beta: Final = request.headers.get("anthropic-beta", "") + assert upstream_beta.split(",").count("context-1m-2025-08-07") == 1, upstream_beta + return Reply( + content_type="text/event-stream", + chunks=cc.text_stream(identity, cc.FABLE, "PONG", {"input_tokens": 250000, "output_tokens": 100}), + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model( + model=f"anthropic/{cc.FABLE}", + api_base=wire.url, + api_key=cc.ANTHROPIC_API_KEY, + input_cost_per_token=1e-6, + input_cost_per_token_above_200k_tokens=2e-6, + output_cost_per_token=5e-6, + ) + response: Final = gateway.request( + "POST", + "/v1/messages", + {**request_body, "model": model}, + params={"beta": "true"}, + headers=cc.cli_headers(gateway.key, _BETA_1M), + ) + assert response.status_code == 200, response.text + assert len(wire.drain()) == 1 + rows: Final = eventually( + lambda: read_rows( + 'SELECT spend, prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" WHERE request_id=%s', + (identity,), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert float(rows[0]["spend"]) == pytest.approx(250000 * 2e-6 + 100 * 5e-6), dict(rows[0])