spb/doc-api

1#!/usr/bin/env python32"""Manual SSE parser (no SDK): accumulates text, tool input JSON and thinking; handles ping/error/unknown events.34STATUS: LIVE_VERIFIED 2026-09-18 (stdlib urllib via scripts/live.py). Run: .venv/bin/python examples/anthropic/streaming/manual_sse_parser.py5"""6import json7import sys8from pathlib import Path910sys.path.insert(0, str(Path(__file__).resolve().parents[3]))11from scripts.live import anthropic_request  # noqa: E402121314def parse_sse(lines):15    """Yield (event_name, data_dict). SSE frames are separated by blank lines; we only use event:/data: fields."""16    name, data = None, []17    for line in lines:18        if line == "":19            if data:20                yield name, json.loads("\n".join(data))21            name, data = None, []22        elif line.startswith("event:"):23            name = line[6:].strip()24        elif line.startswith("data:"):25            data.append(line[5:].strip())26    if data:27        yield name, json.loads("\n".join(data))282930body = {"model": "claude-haiku-4-5-20251001", "max_tokens": 64, "stream": True,31        "tools": [{"name": "get_weather", "description": "Weather for a city", "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}}],32        "tool_choice": {"type": "tool", "name": "get_weather"}, "messages": [{"role": "user", "content": "Weather in Paris?"}]}33status, lines, headers = anthropic_request("POST", "/v1/messages", body, stream=True, est_cost_usd=0.0009, note="example manual_sse_parser.py")34assert status == 200, status35blocks: dict[int, dict] = {}36message, order = {}, []37for name, data in parse_sse(lines):38    order.append(name)39    t = data.get("type")40    if t == "message_start":41        message = data["message"]42    elif t == "content_block_start":43        b = dict(data["content_block"]); b["_json"] = ""; blocks[data["index"]] = b44    elif t == "content_block_delta":45        d, b = data["delta"], blocks[data["index"]]46        if d["type"] == "text_delta": b["text"] += d["text"]47        elif d["type"] == "input_json_delta": b["_json"] += d["partial_json"]48        elif d["type"] == "thinking_delta": b["thinking"] += d["thinking"]49        elif d["type"] == "signature_delta": b["signature"] = d["signature"]50        elif d["type"] == "citations_delta": b.setdefault("citations", []).append(d["citation"])51    elif t == "content_block_stop":52        b = blocks[data["index"]]53        if b["type"] in ("tool_use", "server_tool_use"):54            b["input"] = json.loads(b.pop("_json") or "{}")   # guard the parse when using eager_input_streaming55        else:56            b.pop("_json", None)57    elif t == "message_delta":58        message.update({k: v for k, v in data["delta"].items()}); message["usage"].update(data["usage"])  # usage is cumulative59    elif t == "ping":60        pass61    elif t == "error":62        raise RuntimeError(data["error"])        # e.g. overloaded_error mid-stream63    elif t == "message_stop":64        break65    else:66        pass                                      # unknown event types must be ignored (versioning policy)67message["content"] = [blocks[i] for i in sorted(blocks)]68print("order :", " ".join(order))69print("stop  :", message["stop_reason"], "| usage:", message["usage"])70print("blocks:", json.dumps(message["content"])[:200])71