import httpx, json, os
with httpx.stream("POST", "https://aiwave.live/v1/chat/completions",
headers={"Authorization": f"Bearer {os.environ['AIWAVE_API_KEY']}"},
json={"model": "deepseek-v4-flash", "stream": True, "messages": [{"role": "user", "content": "Reply with OK"}]},
timeout=60.0) as response:
response.raise_for_status()
if "text/event-stream" not in response.headers.get("content-type", ""):
raise RuntimeError("expected an SSE response")
for line in response.iter_lines():
if not line.startswith("data:"):
continue
data = line[5:].strip()
if data == "[DONE]":
break
chunk = json.loads(data)
# Accumulate content and any route-tested tool-call deltas here.
Wire contract
Signal
Meaning
Handler
HTTP 4xx/5xx
Normal JSON error envelope
Classify before SSE parsing
Content-Type: text/event-stream
Streaming body
Read complete data events
data: {JSON}
One chunk
Accumulate content and tested tool deltas
data: [DONE]
Normal termination
Close the reader
Network disconnect
Completion state unknown
Do not repeat state-changing tool work blindly
Do not replay a request just to estimate tokens.
Save the request ID. Use verified final usage when the route returns it; otherwise reconcile the request ledger. A second generation is a second bill and may produce different output.