Function _parse_anthropic_events

tests/test_anthropic_compat.py:561–578

Parse Anthropic SSE events into (event_type, data) tuples.

(raw_events: list[bytes])

def _parse_anthropic_events(raw_events: list[bytes]) -> list[dict]:
    """Parse Anthropic SSE events into (event_type, data) tuples."""
    results: list[dict] = []
    for raw in raw_events:
        text = raw.decode()
        lines = text.strip().split("\n")
        event_type = ""
        data_str = ""
        for line in lines:
            if line.startswith("event: "):
                event_type = line[7:]
            elif line.startswith("data: "):
                data_str = line[6:]
        if data_str:
            parsed = json.loads(data_str)
            parsed["_event"] = event_type
            results.append(parsed)
    return results


def _parse_openai_events(raw_events: list[bytes]) -> tuple[list[dict], bool]: