Claude API для Python: от первого запроса до продакшна

pip install anthropic

export ANTHROPIC_API_KEY="sk-ant-..."

import os
from anthropic import Anthropic

client = Anthropic()  # автоматически читает ANTHROPIC_API_KEY из окружения

message = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Объясни разницу между list и tuple в Python за три предложения."}
    ]
)

print(message.content[0].text)

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=2048,
    messages=[{"role": "user", "content": "Напиши функцию быстрой сортировки на Python"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

import asyncio
from anthropic import AsyncAnthropic

async_client = AsyncAnthropic()

async def ask(question: str) -> str:
    message = await async_client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=512,
        messages=[{"role": "user", "content": question}]
    )
    return message.content[0].text

async def main():
    results = await asyncio.gather(
        ask("Что такое GIL?"),
        ask("Что такое event loop?"),
        ask("Чем asyncio отличается от threading?"),
    )
    for r in results:
        print(r, "\n---")

asyncio.run(main())

history = []

def chat(user_input: str) -> str:
    history.append({"role": "user", "content": user_input})
    
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=history
    )
    
    assistant_reply = response.content[0].text
    history.append({"role": "assistant", "content": assistant_reply})
    return assistant_reply

print(chat("Меня зовут Алексей, я пишу на Python"))
print(chat("Как меня зовут?"))  # Claude вспомнит имя из контекста

SYSTEM_PROMPT = """
Ты — опытный Python-разработчик и технический ревьюер.
[... 2000 токенов документации и правил ...]
"""

def review_code(code: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": SYSTEM_PROMPT,
                "cache_control": {"type": "ephemeral"}  # пометить для кэширования
            }
        ],
        messages=[{"role": "user", "content": f"Проверь этот код:\n```python\n{code}\n```"}]
    )
    
    usage = response.usage
    print(f"Cache write: {usage.cache_creation_input_tokens}, "
          f"Cache read: {usage.cache_read_input_tokens}, "
          f"Input: {usage.input_tokens}")
    
    return response.content[0].text

import time
import anthropic
from anthropic import APIError, APIConnectionError, RateLimitError, APIStatusError

def robust_request(messages: list, max_retries: int = 5) -> str:
    delay = 2.0
    
    for attempt in range(max_retries):
        try:
            response = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=1024,
                messages=messages
            )
            return response.content[0].text
        
        except RateLimitError as e:
            # 429 — превышен лимит. Ждём retry-after из заголовка
            retry_after = float(e.response.headers.get("retry-after", delay))
            print(f"Rate limit. Ждём {retry_after:.1f}s...")
            time.sleep(retry_after)
            delay = min(delay * 2, 120)  # exponential backoff, cap 2 min
        
        except APIStatusError as e:
            if e.status_code == 529:
                # 529 — API перегружен (отличается от 429!)
                print(f"API overloaded. Попытка {attempt + 1}/{max_retries}")
                time.sleep(delay)
                delay = min(delay * 2, 120)
            else:
                raise  # прочие статус-ошибки не ретраим
        
        except APIConnectionError:
            print(f"Сетевая ошибка. Попытка {attempt + 1}/{max_retries}")
            time.sleep(delay)
    
    raise RuntimeError(f"Не удалось выполнить запрос после {max_retries} попыток")

# Создаём батч из нескольких запросов
batch = client.messages.batches.create(
    requests=[
        {
            "custom_id": f"req-{i}",
            "params": {
                "model": "claude-haiku-4-5",
                "max_tokens": 256,
                "messages": [{"role": "user", "content": text}]
            }
        }
        for i, text in enumerate(texts_to_process)
    ]
)

print(f"Батч создан: {batch.id}")

# Через некоторое время — получаем результаты
results = client.messages.batches.results(batch.id)
for result in results:
    print(result.custom_id, result.result.message.content[0].text)