Anthropic Python SDK: установка, настройка base_url, потоковая передача и обработка ошибок

В anthropic — официальный Python-клиент Anthropic для Claude API. Этот пакет обеспечивает аутентификацию, повторные попытки, потоковую передачу и типизированные ошибки, а также работает без изменений с любой конечной точкой, совместимой с Anthropic: задайте base_url (или ANTHROPIC_BASE_URL), и остальной код останется прежним.

Установка

pip install anthropic

Версии 1.x работают на httpx2 и требуют Python 3.10 или новее. Если вы передаёте собственный клиент HTTP, используйте предоставляемый SDK DefaultHttpxClient , чтобы сохранить его стандартные тайм-ауты и ограничения подключений.

Создание клиента

import anthropic

client = anthropic.Anthropic(
    api_key="your_key",   # default: ANTHROPIC_API_KEY
    base_url="https://aiprimetech.io",    # default: ANTHROPIC_BASE_URL, else https://api.anthropic.com
)

base_url — корневой адрес хоста. SDK добавляет /v1/messages самостоятельно, поэтому https://aiprimetech.io/v1 приведёт к /v1/v1/messages и 404 page not found. Если обе переменные заданы в окружении, anthropic.Anthropic() вообще не требует аргументов:

export ANTHROPIC_API_KEY=your_key
export ANTHROPIC_BASE_URL=https://aiprimetech.io

Отправка сообщения

message = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Summarise TCP slow start in two sentences."}],
)

for block in message.content:
    if block.type == "text":
        print(block.text)

print(message.usage.input_tokens, message.usage.output_tokens)
print(message._request_id)   # log this with any failure report

content — список блоков (текст, рассуждения, использование инструментов), поэтому проверяйте block.type вместо того, чтобы читать content[0].text без проверки.

Потоковая передача ответов

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=16000,
    messages=[{"role": "user", "content": "Write a short story about a lighthouse."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()

Используйте потоковую передачу для любых запросов с большим объёмом входных или выходных данных либо большим значением max_tokens: отдельный запрос без потоковой передачи должен завершиться до истечения тайм-аута, а для потока это не требуется. messages.create(..., stream=True) предоставляет итератор исходных событий, если вспомогательный метод вам не нужен.

Асинхронный клиент

import asyncio, anthropic

client = anthropic.AsyncAnthropic(base_url="https://aiprimetech.io")
limit = asyncio.Semaphore(3)   # stay within your key's concurrency

async def ask(question: str) -> str:
    async with limit:
        msg = await client.messages.create(
            model="claude-haiku-4-5",
            max_tokens=512,
            messages=[{"role": "user", "content": question}],
        )
        return next(b.text for b in msg.content if b.type == "text")

async def main():
    answers = await asyncio.gather(*(ask(q) for q in ["What is DNS?", "What is BGP?"]))
    print(answers)

asyncio.run(main())

Семафор важен. Одновременная отправка сотен запросов с помощью asyncio.gather — самая частая причина ошибок Concurrency limit exceeded , с которыми мы сталкиваемся при работе Python-клиентов.

Тайм-ауты и повторные попытки

По умолчанию: тайм-аут 10 мин. на запрос и 2 автоматических повторных попыток с экспоненциальным увеличением задержки. SDK повторяет запросы при ошибках подключения, ответах 408, 409, 429 и 5xx. Поскольку запросы с истёкшим тайм-аутом тоже повторяются, максимальное общее время равно тайм-ауту, умноженному на число попыток.

client = anthropic.Anthropic(base_url="https://aiprimetech.io", timeout=60.0, max_retries=4)

# override for one call only
client.with_options(timeout=300.0).messages.create(
    model="claude-opus-5-5",
    max_tokens=4000,
    messages=[{"role": "user", "content": "Review this design doc..."}],
)

Обработка ошибок

try:
    message = client.messages.create(model="claude-sonnet-5", max_tokens=1024,
                                      messages=[{"role": "user", "content": "Hi"}])
except anthropic.NotFoundError as e:
    print("Bad model ID or URL:", e.message)
except anthropic.RateLimitError as e:
    print("Rate limited, retry after", e.response.headers.get("retry-after"))
except anthropic.APIStatusError as e:
    print(e.status_code, e.type, e.message)
except anthropic.APIConnectionError:
    print("Network problem")

Сначала перехватывайте конкретные классы, а APIStatusError — в последнюю очередь; один общий except скрывает различие между ошибками, при которых стоит повторить запрос, и ошибками, требующими изменения кода. e.type возвращает строку с типом ошибки API, например rate_limit_error или overloaded_error.

ID моделей

client.models.list() возвращает модели, доступные через конечную точку, поэтому вы можете проверить ID перед использованием. Используйте ID точно в указанном виде: без суффиксов с датой.

Устранение неполадок при работе через шлюз

Подробнее об ошибках на стороне модели — в руководстве по ошибкам Claude API.

Часто задаваемые вопросы

AI Prime Tech — независимый API-шлюз. Он не связан с Anthropic, не одобрен этой компанией и не является её реселлером. Claude и названия связанных моделей — товарные знаки их соответствующих владельцев.