Sonnet 5 strict tool use returns a string for an array-typed property

"""Reproduce an array-type violation in a completed strict tool call.

Requires Python 3.10+ and httpx. Set ANTHROPIC_API_KEY, then run:
    python translations_repro.py --attempts 10 --output translations_evidence.json

Calls the HTTP API directly, without the Anthropic SDK or an agent framework.
Runs up to ten requests and records the raw tool arguments and request IDs.
"""

import argparse
import hashlib
import json
import os
import re
from datetime import datetime, timezone
from pathlib import Path

import httpx


REQUEST = {
    "model": "claude-sonnet-5",
    "system": "Copy the supplied example sentences exactly into save_translations. Preserve every character, including quotation marks and accents.",
    "messages": [{
        "role": "user",
        "content": 'Save these two example sentences:\nGerman: Das Café "Möwe" öffnet um 8 Uhr.\nFrench: Le café "Étoile" coûte 8 €.\n\nThe tool arguments must use ASCII-only JSON. Encode non-ASCII characters using JSON Unicode escapes. After JSON parsing, the texts must equal the original sentences exactly.',
    }],
    "tools": [{
        "name": "save_translations",
        "description": "Save example sentences with their language labels.",
        "strict": True,
        "eager_input_streaming": False,
        "input_schema": {
            "type": "object",
            "properties": {
                "translations": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "language": {"type": "string", "enum": ["German", "French"]},
                            "text": {"type": "string"},
                        },
                        "required": ["language", "text"],
                        "additionalProperties": False,
                    },
                },
            },
            "required": ["translations"],
            "additionalProperties": False,
        },
    }],
    "tool_choice": {"type": "auto"},
    "thinking": {"type": "adaptive", "display": "summarized"},
    "max_tokens": 8192,
    "stream": True,
}

def run_once(client, number):
    record = {
        "attempt": number,
        "started_at": datetime.now(timezone.utc).isoformat(),
        "complete_stream": False,
        "stop_reason": None,
        "usage": {},
        "tool_calls": [],
    }
    blocks = {}
    with client.stream("POST", "https://api.anthropic.com/v1/messages", json=REQUEST) as response:
        record["http_status"] = response.status_code
        record["request_id"] = response.headers.get("request-id")
        if response.status_code != 200:
            # Do not print request headers, credentials, or arbitrary error bodies.
            raise RuntimeError(f"HTTP {response.status_code}; request-id={record['request_id']}")
        for line in response.iter_lines():
            if not line.startswith("data: "):
                continue
            event = json.loads(line[6:])
            kind = event["type"]
            if kind == "message_start":
                record["message_id"] = event["message"]["id"]
                record["usage"].update(event["message"].get("usage", {}))
            elif kind == "content_block_start" and event["content_block"]["type"] == "tool_use":
                block = event["content_block"]
                blocks[event["index"]] = {
                    "name": block["name"], "id": block["id"], "raw_arguments": "",
                }
            elif kind == "content_block_delta" and event["delta"]["type"] == "input_json_delta":
                blocks[event["index"]]["raw_arguments"] += event["delta"]["partial_json"]
            elif kind == "message_delta":
                record["stop_reason"] = event["delta"].get("stop_reason")
                record["usage"].update({k: v for k, v in event.get("usage", {}).items() if v is not None})
            elif kind == "message_stop":
                record["complete_stream"] = True
            elif kind == "error":
                raise RuntimeError(f"SSE error; request-id={record['request_id']}")
    violation = False
    for block in blocks.values():
        try:
            arguments = json.loads(block["raw_arguments"])
            block["parsed_arguments"] = arguments
            if block["name"] == "save_translations" and isinstance(arguments, dict):
                block["translations_python_type"] = type(arguments.get("translations")).__name__
                violation |= not isinstance(arguments.get("translations"), list)
        except json.JSONDecodeError as error:
            block["parse_error"] = str(error)
        record["tool_calls"].append(block)
    record["schema_violation"] = (
        record["complete_stream"] and record["stop_reason"] == "tool_use" and violation
    )
    values = {}
    for block in record["tool_calls"]:
        payload = block.get("parsed_arguments", {})
        items = payload.get("translations") if isinstance(payload, dict) else None
        if isinstance(items, list):
            for item in items:
                if isinstance(item, dict) and isinstance(item.get("text"), str):
                    values[item.get("language")] = item["text"]
    record["exact_text"] = values == {
        "German": 'Das Café "Möwe" öffnet um 8 Uhr.',
        "French": 'Le café "Étoile" coûte 8 €.',
    }
    record["literal_unicode"] = [
        match for value in values.values()
        for match in re.findall(r"\\u[0-9a-fA-F]{4}", value)
    ]
    record["control_characters"] = sorted({
        f"U+{ord(char):04X}" for value in values.values() for char in value
        if ord(char) < 32 or 127 <= ord(char) <= 159
    })
    return record


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--attempts", type=int, default=10, choices=range(1, 11), metavar="1..10")
    parser.add_argument("--output", type=Path, default=Path("translations_evidence.json"))
    parser.add_argument("--no-strict", action="store_true", help="Run the same request with strict mode disabled")
    args = parser.parse_args()
    REQUEST["tools"][0]["strict"] = not args.no_strict
    key = os.environ["ANTHROPIC_API_KEY"]
    evidence = {
        "endpoint": "https://api.anthropic.com/v1/messages",
        "anthropic_version": "2023-06-01",
        "httpx_version": httpx.__version__,
        "request": REQUEST,
        "request_sha256": hashlib.sha256(json.dumps(REQUEST, sort_keys=True, ensure_ascii=False).encode()).hexdigest(),
        "attempts": [],
        "estimated_cost_usd": 0,
    }
    # Conservative reservation using request bytes and the full output cap.
    # Direct Sonnet 5 prices at verification: $2/M input, $10/M output.
    reserve = ((len(json.dumps(REQUEST, ensure_ascii=False).encode()) + 4000) * 2 + 8192 * 10) / 1e6 * 1.25
    with httpx.Client(headers={"x-api-key": key, "anthropic-version": "2023-06-01"}, timeout=180) as client:
        for number in range(1, args.attempts + 1):
            if evidence["estimated_cost_usd"] + reserve > 1:
                print("Stopped before exceeding the $1 reservation cap.")
                break
            record = run_once(client, number)
            evidence["attempts"].append(record)
            usage = record["usage"]
            cost = (usage.get("input_tokens", 0) * 2 + usage.get("output_tokens", 0) * 10) / 1e6
            evidence["estimated_cost_usd"] += cost if record["complete_stream"] else reserve
            args.output.write_text(json.dumps(evidence, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
            print(json.dumps({k: record.get(k) for k in ("attempt", "request_id", "message_id", "stop_reason", "schema_violation", "exact_text", "literal_unicode", "control_characters")}), flush=True)
            if record["schema_violation"]:
                print("REPRODUCED: completed tool call violated the translations array constraint.")
    print(f"Estimated cost: ${evidence['estimated_cost_usd']:.5f}")


if __name__ == "__main__":
    main()

{
  "attempt": 7,
  "started_at": "2026-09-10T19:22:59.944020+00:00",
  "complete_stream": true,
  "stop_reason": "tool_use",
  "usage": {
    "input_tokens": 679,
    "cache_creation_input_tokens": 0,
    "cache_read_input_tokens": 0,
    "cache_creation": {
      "ephemeral_5m_input_tokens": 0,
      "ephemeral_1h_input_tokens": 0
    },
    "output_tokens": 503,
    "service_tier": "standard",
    "inference_geo": "global",
    "output_tokens_details": {
      "thinking_tokens": 178
    }
  },
  "tool_calls": [
    {
      "name": "save_translations",
      "id": "toolu_01Cjfi3VJ9g6mbCLZqCVCceA",
      "raw_arguments": "{\"translations\": \"[{\\\"language\\\": \\\"German\\\", \\\"text\\\": \\\"Das Caf\\\\ffnnet um 8 Uhr.\\\"}, {\\\"language\\\": \\\"French\\\", \\\"text\\\": \\\"Le caf\\\\tooile\\\\\\\" co\\\\te }]\"}",
      "parsed_arguments": {
        "translations": "[{\"language\": \"German\", \"text\": \"Das Caf\\ffnnet um 8 Uhr.\"}, {\"language\": \"French\", \"text\": \"Le caf\\tooile\\\" co\\te }]"
      },
      "translations_python_type": "str",
      "schema_errors": [
        "'[{\"language\": \"German\", \"text\": \"Das Caf\\\\ffnnet um 8 Uhr.\"}, {\"language\": \"French\", \"text\": \"Le caf\\\\tooile\\\\\" co\\\\te }]' is not of type 'array'"
      ]
    }
  ],
  "http_status": 200,
  "request_id": "req_011CevJagiwTs4PSNi4UNmYy",
  "message_id": "msg_011CevJahSc5jTtQtetFN4nd",
  "schema_violation": true,
  "exact_text": false,
  "controls": [],
  "literal_unicode": []
}