mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-06 02:48:13 +00:00
test(integration): group anthropic messages tests into feature subfolders
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
kerry
parent
d95448382f
commit
533fab43ab
21 changed files with 2360 additions and 1 deletions
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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]
|
||||
)
|
||||
|
|
@ -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)
|
||||
|
|
@ -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": "<summary>"}
|
||||
|
||||
|
||||
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
|
||||
|
|
@ -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() == ()
|
||||
|
|
@ -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
|
||||
|
|
@ -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)
|
||||
|
|
@ -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
|
||||
|
|
@ -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<</Type/Catalog/Pages 2 0 R>>endobj\n"
|
||||
b"2 0 obj<</Type/Pages/Kids[3 0 R]/Count 1>>endobj\n"
|
||||
b"3 0 obj<</Type/Page/Parent 2 0 R/MediaBox[0 0 72 72]>>endobj\n"
|
||||
b"trailer<</Root 1 0 R>>\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
|
||||
|
|
@ -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
|
||||
|
|
@ -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)
|
||||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -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}"
|
||||
|
|
@ -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"]
|
||||
|
|
@ -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"]
|
||||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -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])
|
||||
|
|
@ -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
|
||||
)
|
||||
|
|
@ -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])
|
||||
Loading…
Add table
Reference in a new issue