Python 88.3%
TypeScript 7.6%
Shell 4.1%
1"""Smoke tests for POST /v1/messages with stream=true (SSE). Verified live 2026-09-18, < $0.0002 total."""2from __future__ import annotations34import json56MSG = "/v1/messages"789def _events(lines):10 name = None11 for line in lines:12 if line.startswith("event:"):13 name = line[6:].strip()14 elif line.startswith("data:"):15 yield name, json.loads(line[5:].strip())161718def test_sse_event_order_and_shapes(anthropic, models):19 st, gen, hdrs = anthropic("POST", MSG, {"model": models["anthropic"], "max_tokens": 16, "stream": True,20 "messages": [{"role": "user", "content": "Reply with OK."}]}, stream=True, note="test_streaming basic")21 assert st == 20022 assert "text/event-stream" in {k.lower(): v for k, v in hdrs.items()}["content-type"]23 evs = list(_events(gen))24 names = [n for n, _ in evs]25 for n, d in evs:26 assert d["type"] == n # SSE event name == data.type27 assert names[0] == "message_start" and names[-1] == "message_stop"28 assert names.count("message_start") == 1 and names.count("message_stop") == 129 assert names.index("content_block_start") < names.index("content_block_delta") < names.index("content_block_stop") < names.index("message_delta")30 start = evs[0][1]["message"]31 assert start["content"] == [] and start["stop_reason"] is None and "usage" in start32 delta = next(d for n, d in evs if n == "message_delta")33 assert delta["delta"]["stop_reason"] == "end_turn"34 assert set(delta["usage"]) >= {"input_tokens", "output_tokens", "cache_creation_input_tokens", "cache_read_input_tokens"}35 text = "".join(d["delta"]["text"] for n, d in evs if n == "content_block_delta" and d["delta"]["type"] == "text_delta")36 assert "OK" in text37 assert all(n in {"message_start", "content_block_start", "content_block_delta", "content_block_stop", "message_delta", "message_stop", "ping"} for n in names)383940def test_sse_tool_use_input_json_delta(anthropic, models):41 tool = {"name": "get_weather", "description": "Weather for a city", "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}}42 st, gen, _ = anthropic("POST", MSG, {"model": models["anthropic"], "max_tokens": 48, "stream": True, "tools": [tool],43 "tool_choice": {"type": "tool", "name": "get_weather"},44 "messages": [{"role": "user", "content": "Weather in Paris?"}]}, stream=True, note="test_streaming tool_use")45 assert st == 20046 evs = list(_events(gen))47 start = next(d for n, d in evs if n == "content_block_start")48 assert start["content_block"]["type"] == "tool_use" and start["content_block"]["input"] == {}49 partial = "".join(d["delta"]["partial_json"] for n, d in evs if n == "content_block_delta" and d["delta"]["type"] == "input_json_delta")50 assert json.loads(partial).get("city")51 assert next(d for n, d in evs if n == "message_delta")["delta"]["stop_reason"] == "tool_use"52