anthropic_bug.py

#!/usr/bin/env python3

import json
import os
import sys
import time
import urllib.error
import urllib.request


API_URL = "https://api.anthropic.com/v1/messages"
API_VERSION = "2023-06-01"
MODEL = "claude-opus-4-6"
PROMPT = "2 + 2 is:"
MAX_TOKENS = 64
EFFORT = "low"


def load_api_key() -> str:
    env_key = os.environ.get("ANTHROPIC_API_KEY", "").strip()
    if env_key:
        return env_key

    token_path = os.path.expanduser("~/.config/anthropic.token")
    try:
        with open(token_path, "r", encoding="utf-8") as f:
            return f.read().strip()
    except FileNotFoundError:
        pass

    print(
        "Missing API key. Set ANTHROPIC_API_KEY or create ~/.config/anthropic.token",
        file=sys.stderr,
    )
    sys.exit(1)


def summarize_event(event: dict) -> str:
    event_type = event.get("type", "unknown")

    if event_type == "message_start":
        message = event.get("message") or {}
        return f"{event_type}:model={message.get('model', '')}"

    if event_type == "content_block_start":
        block = event.get("content_block") or {}
        return f"{event_type}:{block.get('type', 'unknown')}"

    if event_type == "content_block_delta":
        delta = event.get("delta") or {}
        delta_type = delta.get("type", "unknown")
        if delta_type == "text_delta":
            return f"{event_type}:{delta_type}:{delta.get('text', '')!r}"
        return f"{event_type}:{delta_type}"

    if event_type == "message_delta":
        delta = event.get("delta") or {}
        return f"{event_type}:stop_reason={delta.get('stop_reason')}"

    return event_type


def main() -> int:
    api_key = load_api_key()

    payload = {
        "model": MODEL,
        "max_tokens": MAX_TOKENS,
        "messages": [{"role": "user", "content": PROMPT}],
        "output_config": {"effort": EFFORT},
        "stream": True,
    }

    request = urllib.request.Request(
        API_URL,
        data=json.dumps(payload).encode("utf-8"),
        method="POST",
        headers={
            "x-api-key": api_key,
            "anthropic-version": API_VERSION,
            "content-type": "application/json",
        },
    )

    text_parts: list[str] = []
    last_text_time = None
    start_time = time.perf_counter()

    try:
        with urllib.request.urlopen(request) as response:
            print(f"HTTP_STATUS={response.status}")

            for raw_line in response:
                now = time.perf_counter()
                line = raw_line.decode("utf-8", errors="replace").strip()
                if not line or line.startswith("event: "):
                    continue

                if not line.startswith("data: "):
                    print(f"EVENT {now - start_time:.6f} RAW {line!r}")
                    continue

                data = line[6:]
                if data == "[DONE]":
                    print(f"EVENT {now - start_time:.6f} [DONE]")
                    continue

                try:
                    event = json.loads(data)
                except json.JSONDecodeError:
                    print(f"EVENT {now - start_time:.6f} JSON_DECODE_ERROR {data!r}")
                    continue

                print(f"EVENT {now - start_time:.6f} {summarize_event(event)}")

                if event.get("type") == "content_block_delta":
                    delta = event.get("delta") or {}
                    if delta.get("type") == "text_delta":
                        text = delta.get("text", "")
                        if text:
                            text_parts.append(text)
                            last_text_time = now
    except urllib.error.HTTPError as exc:
        body = exc.read().decode("utf-8", errors="replace")
        print(f"HTTP_ERROR={exc.code}", file=sys.stderr)
        print(body, file=sys.stderr)
        return 1

    end_time = time.perf_counter()
    answer = "".join(text_parts)

    print(f"ANSWER={answer}")
    print(
        "LAST_TEXT_SECONDS="
        + ("NA" if last_text_time is None else f"{last_text_time - start_time:.6f}")
    )
    print(f"FULL_FINISH_SECONDS={end_time - start_time:.6f}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())