test(anthropic): replay a real Claude Code /v1/messages request through the native wire

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
kerry 2026-09-26 22:34:19 +00:00
parent 9413b82477
commit bac7119572
2 changed files with 1010 additions and 0 deletions

View file

@ -0,0 +1,859 @@
{
"model": "claude-sonnet-4-5",
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": "<system-reminder>\nSynthetic environment reminder.\n</system-reminder>"
},
{
"type": "text",
"text": "<system-reminder>\nSynthetic model identity reminder.\n</system-reminder>"
},
{
"type": "text",
"text": "<system-reminder>\nSynthetic agent types reminder.\n</system-reminder>"
},
{
"type": "text",
"text": "<system-reminder>\nSynthetic skills reminder.\n</system-reminder>"
},
{
"type": "text",
"text": "<system-reminder>\n<total_tokens>15000000 tokens left</total_tokens>\n</system-reminder>"
},
{
"type": "text",
"text": "<system-reminder>\nSynthetic date reminder.\n</system-reminder>"
},
{
"type": "text",
"text": "<system-reminder>\nSynthetic attribution reminder.\n</system-reminder>"
},
{
"type": "text",
"text": "Reply with exactly the word PONG",
"cache_control": {
"type": "ephemeral"
}
}
]
}
],
"system": [
{
"type": "text",
"text": "Synthetic billing header block from a Claude Code request."
},
{
"type": "text",
"text": "Synthetic agent identity system prompt.",
"cache_control": {
"type": "ephemeral"
}
},
{
"type": "text",
"text": "Synthetic interactive agent instructions.",
"cache_control": {
"type": "ephemeral"
}
}
],
"tools": [
{
"name": "Agent",
"description": "Launch a new agent to handle complex, multi-step tasks.",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"description": {
"description": "A short (3-5 word) description of the task",
"type": "string"
},
"prompt": {
"description": "The task for the agent to perform",
"type": "string"
},
"subagent_type": {
"description": "The type of specialized agent to use for this task",
"type": "string"
},
"model": {
"description": "Optional model override for this agent.",
"type": "string",
"enum": [
"sonnet",
"opus",
"haiku",
"fable"
]
},
"run_in_background": {
"description": "Agents run in the background by default; you will be notified when one completes.",
"type": "boolean"
},
"isolation": {
"description": "Isolation mode.",
"type": "string",
"enum": [
"worktree",
"remote"
]
}
},
"required": [
"description",
"prompt"
],
"additionalProperties": false
}
},
{
"name": "Bash",
"description": "Executes a given bash command and returns its output.",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"command": {
"description": "The command to execute",
"type": "string"
},
"timeout": {
"description": "Optional timeout in milliseconds (max 600000)",
"type": "number"
},
"description": {
"description": "Clear, concise description of what this command does in active voice.",
"type": "string"
},
"run_in_background": {
"description": "Set to true to run this command in the background.",
"type": "boolean"
},
"dangerouslyDisableSandbox": {
"description": "Set this to true to dangerously override sandbox mode and run commands without sandboxing.",
"type": "boolean"
}
},
"required": [
"command"
],
"additionalProperties": false
}
},
{
"name": "CronCreate",
"description": "Schedule a prompt to be enqueued at a future time.",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"cron": {
"description": "Standard 5-field cron expression in local time: \"M H DoM Mon DoW\" (e.g.",
"type": "string"
},
"prompt": {
"description": "The prompt to enqueue at each fire time.",
"type": "string"
},
"recurring": {
"description": "true (default) = fire on every cron match until deleted or auto-expired after 7 days.",
"type": "boolean"
},
"durable": {
"description": "true = persist to .claude/scheduled_tasks.json and survive restarts.",
"type": "boolean"
}
},
"required": [
"cron",
"prompt"
],
"additionalProperties": false
}
},
{
"name": "CronDelete",
"description": "Cancel a cron job previously scheduled with CronCreate.",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"id": {
"description": "Job ID returned by CronCreate.",
"type": "string"
}
},
"required": [
"id"
],
"additionalProperties": false
}
},
{
"name": "CronList",
"description": "List all cron jobs scheduled via CronCreate, both durable (.claude/scheduled_tasks.json) and session-only.",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {},
"additionalProperties": false
}
},
{
"name": "Edit",
"description": "Performs exact string replacements in files.",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"file_path": {
"description": "The absolute path to the file to modify",
"type": "string"
},
"old_string": {
"description": "The text to replace",
"type": "string"
},
"new_string": {
"description": "The text to replace it with (must be different from old_string)",
"type": "string"
},
"replace_all": {
"description": "Replace all occurrences of old_string (default false)",
"default": false,
"type": "boolean"
}
},
"required": [
"file_path",
"old_string",
"new_string"
],
"additionalProperties": false
}
},
{
"name": "EnterWorktree",
"description": "Use this tool ONLY when explicitly instructed to work in a worktree \u2014 either by the user directly, or by project instruc",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"name": {
"description": "Optional name for a new worktree.",
"type": "string"
},
"path": {
"description": "Path to an existing worktree to switch into instead of creating a new one.",
"type": "string"
}
},
"additionalProperties": false
}
},
{
"name": "ExitWorktree",
"description": "Exit a worktree session created by EnterWorktree and return the session to the original working directory.",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"action": {
"description": "\"keep\" leaves the worktree and branch on disk; \"remove\" deletes both.",
"type": "string",
"enum": [
"keep",
"remove"
]
},
"discard_changes": {
"description": "Required true when action is \"remove\" and the worktree has uncommitted files or unmerged commits.",
"type": "boolean"
}
},
"required": [
"action"
],
"additionalProperties": false
}
},
{
"name": "ListAgents",
"description": "Lists agents you can SendMessage to \u2014 in-process subagents you spawned, the teammates on your team, other local Claude s",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"channel": {
"description": "Not available in this build; leave unset.",
"type": "string",
"maxLength": 256
},
"q": {
"description": "Not available in this build; leave unset.",
"type": "string",
"maxLength": 256
}
},
"additionalProperties": false
}
},
{
"name": "NotebookEdit",
"description": "Replaces, inserts, or deletes a single cell in a Jupyter notebook (.ipynb file).",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"notebook_path": {
"description": "The absolute path to the Jupyter notebook file to edit (must be absolute, not relative)",
"type": "string"
},
"cell_id": {
"description": "The ID of the cell to edit.",
"type": "string"
},
"new_source": {
"description": "The new source for the cell",
"type": "string"
},
"cell_type": {
"description": "The type of the cell (code or markdown).",
"type": "string",
"enum": [
"code",
"markdown"
]
},
"edit_mode": {
"description": "The type of edit to make (replace, insert, delete).",
"type": "string",
"enum": [
"replace",
"insert",
"delete"
]
}
},
"required": [
"notebook_path",
"new_source"
],
"additionalProperties": false
}
},
{
"name": "Read",
"description": "Reads a file from the local filesystem.",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"file_path": {
"description": "The absolute path to the file to read",
"type": "string"
},
"offset": {
"description": "The line number to start reading from.",
"type": "integer",
"minimum": 0,
"maximum": 9007199254740991
},
"limit": {
"description": "The number of lines to read.",
"type": "integer",
"exclusiveMinimum": 0,
"maximum": 9007199254740991
},
"pages": {
"description": "Page range for PDF files (e.g., \"1-5\", \"3\", \"10-20\").",
"type": "string"
}
},
"required": [
"file_path"
],
"additionalProperties": false
}
},
{
"name": "ReportFindings",
"description": "Report code-review findings as a typed list so the host UI can render them.",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"level": {
"description": "Effort level the review ran at",
"type": "string",
"enum": [
"low",
"medium",
"high",
"xhigh",
"max"
]
},
"findings": {
"description": "Verified findings, most-severe first; empty if none survived",
"maxItems": 32,
"type": "array",
"items": {
"type": "object",
"properties": {
"file": {
"description": "Repo-relative path of the file the finding is in",
"type": "string"
},
"line": {
"description": "1-indexed line the finding anchors to",
"type": "integer",
"minimum": -9007199254740991,
"maximum": 9007199254740991
},
"summary": {
"description": "One-sentence statement of the defect",
"type": "string"
},
"short_summary": {
"description": "Compressed label for compact UI (\u226460 chars): the claim alone, no rationale or consequence clause",
"type": "string",
"maxLength": 60
},
"failure_scenario": {
"description": "Concrete inputs/state \u2192 wrong output/crash",
"type": "string"
},
"category": {
"description": "Short kebab-case slug of the finding type, e.g.",
"type": "string",
"maxLength": 40
},
"verdict": {
"description": "Set when a verify pass ran; absent on inline-only reviews",
"type": "string",
"enum": [
"CONFIRMED",
"PLAUSIBLE"
]
},
"outcome": {
"description": "Set ONLY when re-reporting after applying fixes: what happened to this finding",
"type": "string",
"enum": [
"fixed",
"skipped",
"no_change_needed"
]
}
},
"required": [
"file",
"summary",
"failure_scenario"
],
"additionalProperties": false
}
}
},
"required": [
"findings"
],
"additionalProperties": false
}
},
{
"name": "ScheduleWakeup",
"description": "Schedule when to resume work in /loop dynamic mode \u2014 the user invoked /loop without an interval, asking you to self-pace",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"delaySeconds": {
"description": "Seconds from now to wake up.",
"type": "number"
},
"reason": {
"description": "One short sentence explaining the chosen delay.",
"type": "string"
},
"prompt": {
"description": "The /loop input to fire on wake-up.",
"type": "string"
},
"stop": {
"description": "Set to true to end the dynamic loop immediately instead of scheduling another wakeup.",
"type": "boolean"
},
"noop": {
"description": "true = nothing changed (you checked and there is nothing to report).",
"type": "boolean"
}
},
"additionalProperties": false
}
},
{
"name": "SendMessage",
"description": "# SendMessage\n\nSend a message to another agent.",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"to": {
"description": "Recipient: a name from ListAgents (append its \" [ref]\" only when a listing or an error shows one), a teammate name, \"mai",
"type": "string",
"allOf": [
{
"pattern": "^[^\\n\\r]*$"
},
{
"pattern": "^[\\s\\S]{0,300}$"
}
]
},
"summary": {
"description": "A 5-10 word label for your own transcript row (not transmitted \u2014 the recipient previews the first line of `message`).",
"type": "string",
"maxLength": 200
},
"message": {
"default": "",
"description": "Plain text message content.",
"type": "string"
},
"notify_when_idle": {
"description": "Ask a session ON THIS MACHINE to send you ONE notice when it next goes idle (finishes its turn with nothing queued) or e",
"type": "boolean"
}
},
"required": [
"to",
"message"
],
"additionalProperties": false
}
},
{
"name": "Skill",
"description": "Invoke a skill.",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"skill": {
"description": "The name of a skill from the available-skills list.",
"type": "string"
},
"args": {
"description": "Optional arguments for the skill",
"type": "string"
}
},
"required": [
"skill"
],
"additionalProperties": false
}
},
{
"name": "TaskCreate",
"description": "Use this tool to create a structured task list for your current coding session.",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"subject": {
"description": "A brief title for the task",
"type": "string"
},
"description": {
"description": "What needs to be done",
"type": "string"
},
"activeForm": {
"description": "Present continuous form shown in spinner when in_progress (e.g., \"Running tests\")",
"type": "string"
},
"metadata": {
"description": "Arbitrary metadata to attach to the task",
"type": "object",
"propertyNames": {
"type": "string"
},
"additionalProperties": {}
}
},
"required": [
"subject",
"description"
],
"additionalProperties": false
}
},
{
"name": "TaskGet",
"description": "Use this tool to retrieve a task by its ID from the task list.",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"taskId": {
"description": "The ID of the task to retrieve",
"type": "string"
}
},
"required": [
"taskId"
],
"additionalProperties": false
}
},
{
"name": "TaskList",
"description": "Use this tool to list all tasks in the task list.",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {},
"additionalProperties": false
}
},
{
"name": "TaskStop",
"description": "- Stops a running background task by its ID\n- Takes a task_id parameter identifying the task to stop\n- To stop an agent-",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"task_id": {
"description": "The ID of the background task to stop.",
"type": "string"
},
"shell_id": {
"description": "Deprecated: use task_id instead",
"type": "string"
}
},
"additionalProperties": false
}
},
{
"name": "TaskUpdate",
"description": "Use this tool to update a task in the task list.",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"taskId": {
"description": "The ID of the task to update",
"type": "string"
},
"subject": {
"description": "New subject for the task",
"type": "string"
},
"description": {
"description": "New description for the task",
"type": "string"
},
"activeForm": {
"description": "Present continuous form shown in spinner when in_progress (e.g., \"Running tests\")",
"type": "string"
},
"status": {
"description": "New status for the task",
"anyOf": [
{
"type": "string",
"enum": [
"pending",
"in_progress",
"completed"
]
},
{
"type": "string",
"const": "deleted"
}
]
},
"addBlocks": {
"description": "Task IDs that this task blocks",
"type": "array",
"items": {
"type": "string"
}
},
"addBlockedBy": {
"description": "Task IDs that block this task",
"type": "array",
"items": {
"type": "string"
}
},
"owner": {
"description": "New owner for the task",
"type": "string"
},
"metadata": {
"description": "Metadata keys to merge into the task.",
"type": "object",
"propertyNames": {
"type": "string"
},
"additionalProperties": {}
}
},
"required": [
"taskId"
],
"additionalProperties": false
}
},
{
"name": "WebFetch",
"description": "IMPORTANT: WebFetch WILL FAIL for authenticated or private URLs.",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"url": {
"description": "The URL to fetch content from",
"type": "string",
"format": "uri"
},
"prompt": {
"description": "The prompt to run on the fetched content",
"type": "string"
}
},
"required": [
"url",
"prompt"
],
"additionalProperties": false
}
},
{
"name": "WebSearch",
"description": "- Allows Claude to search the web and use the results to inform responses\n- Provides up-to-date information for current ",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"query": {
"description": "The search query to use",
"type": "string",
"minLength": 2
},
"allowed_domains": {
"description": "Only include search results from these domains",
"type": "array",
"items": {
"type": "string"
}
},
"blocked_domains": {
"description": "Never include search results from these domains",
"type": "array",
"items": {
"type": "string"
}
}
},
"required": [
"query"
],
"additionalProperties": false
}
},
{
"name": "Workflow",
"description": "Execute a workflow script that orchestrates multiple subagents deterministically.",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"script": {
"description": "Self-contained workflow script.",
"type": "string",
"maxLength": 524288
},
"name": {
"description": "Name of a predefined workflow (built-in or from .claude/workflows/).",
"type": "string"
},
"description": {
"description": "Ignored \u2014 set the workflow description in the script's `meta` block.",
"type": "string"
},
"title": {
"description": "Ignored \u2014 set the workflow title in the script's `meta` block.",
"type": "string"
},
"args": {
"description": "Optional input value exposed to the script as the global `args`, verbatim."
},
"scriptPath": {
"description": "Path to a workflow script file on disk.",
"type": "string"
},
"resumeFromRunId": {
"description": "Run ID of a prior Workflow invocation to resume from.",
"type": "string",
"pattern": "^wf_[a-z0-9-]{6,}$"
}
},
"additionalProperties": false
}
},
{
"name": "Write",
"description": "Writes a file to the local filesystem.",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"file_path": {
"description": "The absolute path to the file to write (must be absolute, not relative)",
"type": "string"
},
"content": {
"description": "The content to write to the file",
"type": "string"
}
},
"required": [
"file_path",
"content"
],
"additionalProperties": false
}
}
],
"metadata": {
"user_id": "{\"device_id\": \"0000000000000000000000000000000000000000000000000000000000000000\", \"account_uuid\": \"\", \"session_id\": \"00000000-0000-4000-8000-000000000000\"}"
},
"max_tokens": 32000,
"thinking": {
"budget_tokens": 31999,
"type": "enabled",
"display": "omitted"
},
"context_management": {
"edits": [
{
"type": "clear_thinking_20251015",
"keep": "all"
}
]
},
"stream": true
}

View file

@ -0,0 +1,151 @@
import json
import uuid
from pathlib import Path
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
from pydantic import JsonValue, TypeAdapter
_API_KEY: Final = "synthetic-anthropic-key"
_MODEL: Final = "claude-sonnet-4-5"
_FIXTURE: Final = Path(__file__).with_name("claude_code_request.json")
_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
_CLI_BETA: Final = (
"claude-code-20250219,interleaved-thinking-2025-05-14,thinking-token-count-2026-05-13,"
"context-management-2025-06-27,prompt-caching-scope-2026-01-05"
)
def _sse_frame(event: str, data: JsonValue) -> bytes:
return f"event: {event}\ndata: {json.dumps(data)}\n\n".encode()
def _message_stream(identity: str) -> tuple[bytes, ...]:
return (
_sse_frame(
"message_start",
{
"type": "message_start",
"message": {
"id": identity,
"type": "message",
"role": "assistant",
"model": _MODEL,
"content": [],
"stop_reason": None,
"stop_sequence": None,
"usage": {"input_tokens": 12, "output_tokens": 1},
},
},
),
_sse_frame(
"content_block_start",
{"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}},
),
_sse_frame(
"content_block_delta",
{"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "PONG"}},
),
_sse_frame("content_block_stop", {"type": "content_block_stop", "index": 0}),
_sse_frame(
"message_delta",
{
"type": "message_delta",
"delta": {"stop_reason": "end_turn", "stop_sequence": None},
"usage": {"output_tokens": 4},
},
),
_sse_frame("message_stop", {"type": "message_stop"}),
)
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.anthropic.claude_code_native_request_survives_and_streams_back")
def test_claude_code_streaming_request_reaches_anthropic_intact_and_streams_back(gateway: Gateway) -> None:
identity: Final = f"msg_cc_{uuid.uuid4().hex}"
fixture: Final = _JSON_OBJECT.validate_json(_FIXTURE.read_bytes())
cli_beta: Final = frozenset(_CLI_BETA.split(","))
content: Final = list(object_value(fixture["messages"][0])["content"])
content[0] = {**content[0], "text": f"cache-bust-{uuid.uuid4().hex}"}
request_body: Final = {**fixture, "messages": [{"role": "user", "content": content}]}
def respond(request: Request) -> Reply:
assert request.method == "POST"
assert request.target == "/v1/messages", request.target
assert request.headers["x-api-key"] == _API_KEY
assert request.headers["anthropic-version"] == "2023-06-01"
upstream_beta: Final = frozenset(request.headers.get("anthropic-beta", "").split(","))
assert cli_beta <= upstream_beta, request.headers.get("anthropic-beta")
body: Final = _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=_message_stream(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=_API_KEY)
response: Final = gateway.request(
"POST",
"/v1/messages",
{**request_body, "model": model},
params={"beta": "true"},
headers={
"accept": "application/json",
"content-type": "application/json",
"user-agent": "claude-cli/2.1.283 (external, sdk-cli)",
"x-claude-code-session-id": "00000000-0000-4000-8000-000000000000",
"x-stainless-arch": "x64",
"x-stainless-lang": "js",
"x-stainless-os": "Linux",
"x-stainless-package-version": "0.112.1",
"x-stainless-retry-count": "0",
"x-stainless-runtime": "node",
"x-stainless-runtime-version": "v26.3.0",
"x-stainless-timeout": "600",
"anthropic-beta": _CLI_BETA,
"anthropic-dangerous-direct-browser-access": "true",
"anthropic-version": "2023-06-01",
"x-app": "cli",
"x-api-key": gateway.key,
},
)
assert response.status_code == 200, response.text
assert response.headers["content-type"].startswith("text/event-stream"), dict(response.headers)
events: Final = 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