Python SDK

Das Anthropic Python SDK bietet bequemen Zugriff auf die Anthropic REST API aus Python-Anwendungen heraus. Es unterstützt sowohl synchrone als auch asynchrone Operationen, Streaming und Integrationen mit Amazon Bedrock, Vertex AI, Microsoft Foundry und Claude Platform on AWS.

Installation

pip install anthropic

Für plattformspezifische Integrationen oder verbesserte asynchrone Performance installiere das Paket mit Extras:

# Für Amazon Bedrock-Unterstützung
pip install "anthropic[bedrock]"

# Für Vertex AI-Unterstützung
pip install "anthropic[vertex]"

# Für Claude Platform auf AWS-Unterstützung
pip install "anthropic[aws]"

# Microsoft Foundry-Unterstützung ist im Basispaket enthalten

# Für verbesserte Async-Performance mit aiohttp
pip install "anthropic[aiohttp]"

Voraussetzungen

Python 3.9 oder höher ist erforderlich.

Verwendung

import os
from anthropic import Anthropic

client = Anthropic(
    # Dies ist der Standardwert und kann weggelassen werden
    api_key=os.environ.get("ANTHROPIC_API_KEY"),
)

message = client.messages.create(
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": "Hello, Claude",
        }
    ],
    model="claude-opus-4-8",
)
print(message.content)

Für Authentifizierungsoptionen einschließlich Workload Identity Federation siehe Authentifizierung.

Asynchrone Verwendung

import os
import asyncio
from anthropic import AsyncAnthropic

client = AsyncAnthropic(
    api_key=os.environ.get("ANTHROPIC_API_KEY"),
)


async def main() -> None:
    message = await client.messages.create(
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": "Hello, Claude",
            }
        ],
        model="claude-opus-4-8",
    )
    print(message.content)


asyncio.run(main())

Verwendung von aiohttp für bessere Nebenläufigkeit

Für verbesserte asynchrone Performance kannst du das aiohttp-HTTP-Backend anstelle des standardmäßigen httpx verwenden:

import os
import asyncio
from anthropic import AsyncAnthropic, DefaultAioHttpClient


async def main() -> None:
    async with AsyncAnthropic(
        api_key=os.environ.get("ANTHROPIC_API_KEY"),
        http_client=DefaultAioHttpClient(),
    ) as client:
        message = await client.messages.create(
            max_tokens=1024,
            messages=[
                {
                    "role": "user",
                    "content": "Hello, Claude",
                }
            ],
            model="claude-opus-4-8",
        )
        print(message.content)


asyncio.run(main())

Streaming von Antworten

Das SDK bietet Unterstützung für das Streaming von Antworten mittels „Server-Sent Events" (servergesendete Ereignisse), oder SSE.

client = Anthropic()

stream = client.messages.create(
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": "Hello, Claude",
        }
    ],
    model="claude-opus-4-8",
    stream=True,
)
for event in stream:
    print(event.type)

Der asynchrone Client verwendet genau dieselbe Schnittstelle:

client = AsyncAnthropic()

stream = await client.messages.create(
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": "Hello, Claude",
        }
    ],
    model="claude-opus-4-8",
    stream=True,
)
async for event in stream:
    print(event.type)

Streaming-Helfer

Das SDK bietet auch Streaming-Helfer, die Kontextmanager verwenden und Zugriff auf den akkumulierten Text und die finale Nachricht bieten:

async def main() -> None:
    async with client.messages.stream(
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": "Say hello there!",
            }
        ],
        model="claude-opus-4-8",
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)
        print()

        message = await stream.get_final_message()
        print(message.to_json())


asyncio.run(main())

Streaming mit client.messages.stream(...) stellt verschiedene Helfer bereit, einschließlich Akkumulation und SDK-spezifischer Events.

Alternativ kannst du client.messages.create(..., stream=True) verwenden, was nur ein Iterable der Events im Stream zurückgibt und weniger Speicher verbraucht (es baut kein finales Nachrichtenobjekt für dich auf).

Token-Zählung

Du kannst die genaue Nutzung für eine bestimmte Anfrage über die usage-Antworteigenschaft einsehen:

message = client.messages.create(...)
print(message.usage)
# Usage(input_tokens=25, output_tokens=13)

Du kannst Token auch zählen, bevor du eine Anfrage stellst:

count = client.messages.count_tokens(
    model="claude-opus-4-8", messages=[{"role": "user", "content": "Hello, world"}]
)
print(count.input_tokens)  # 10

Tool-Nutzung

Dieses SDK bietet Unterstützung für „tool use" (Tool-Nutzung), auch bekannt als Function Calling. Weitere Details findest du unter Tool-Nutzung mit Claude.

Tool-Helfer

Das SDK bietet Helfer zum Definieren und Ausführen von Tools als reine Python-Funktionen. Der @beta_tool-Decorator generiert das Tool-Schema aus der Funktionssignatur und dem Docstring:

import json
from anthropic import Anthropic, beta_tool

client = Anthropic()


@beta_tool
def get_weather(location: str) -> str:
    """Get the weather for a given location.

    Args:
        location: The city and state, for example, San Francisco, CA
    Returns:
        A JSON-encoded string with the location, temperature, and weather condition.
    """
    return json.dumps(
        {
            "location": location,
            "temperature": "68°F",
            "condition": "Sunny",
        }
    )


# Verwende den tool_runner, um Tool-Aufrufe automatisch zu verarbeiten
runner = client.beta.messages.tool_runner(
    max_tokens=1024,
    model="claude-opus-4-8",
    tools=[get_weather],
    messages=[
        {"role": "user", "content": "What is the weather in SF?"},
    ],
)
for message in runner:
    print(message)

Bei jeder Iteration wird eine API-Anfrage gestellt. Wenn Claude eines der angegebenen Tools aufrufen möchte, wird es automatisch aufgerufen, und das Ergebnis wird in der nächsten Iteration direkt an das Modell zurückgegeben.

Message Batches

Dieses SDK bietet Unterstützung für die Message Batches API unter client.messages.batches.

Erstellen eines Batches

Message Batches nimmt ein Array von Anfragen entgegen, wobei jedes Objekt einen custom_id-Bezeichner und dieselben Anfrage-params wie die Standard-Messages-API hat:

client.messages.batches.create(
    requests=[
        {
            "custom_id": "my-first-request",
            "params": {
                "model": "claude-opus-4-8",
                "max_tokens": 1024,
                "messages": [{"role": "user", "content": "Hello, world"}],
            },
        },
        {
            "custom_id": "my-second-request",
            "params": {
                "model": "claude-opus-4-8",
                "max_tokens": 1024,
                "messages": [{"role": "user", "content": "Hi again, friend"}],
            },
        },
    ]
)

Ergebnisse aus einem Batch abrufen

Sobald ein Message Batch verarbeitet wurde, angezeigt durch .processing_status == 'ended', kannst du mit .batches.results() auf die Ergebnisse zugreifen:

client = anthropic.Anthropic()
batch_id = "batch_abc123"
result_stream = client.messages.batches.results(batch_id)
for entry in result_stream:
    if entry.result.type == "succeeded":
        print(entry.result.message.content)

Datei-Uploads

Anfrageparameter, die Datei-Uploads entsprechen, können in vielen verschiedenen Formen übergeben werden:

• Ein PathLike-Objekt (zum Beispiel pathlib.Path)
• Ein Tupel aus (filename, content, content_type)
• Ein BinaryIO-dateiähnliches Objekt

from pathlib import Path
from anthropic import Anthropic

client = Anthropic()

# Hochladen über einen Dateipfad
client.beta.files.upload(
    file=Path("/path/to/file"),
)

# Hochladen über Bytes
client.beta.files.upload(
    file=("file.txt", b"my bytes", "text/plain"),
)

Der asynchrone Client verwendet genau dieselbe Schnittstelle. Wenn du eine PathLike-Instanz übergibst, werden die Dateiinhalte automatisch asynchron gelesen.

Fehlerbehandlung

Wenn die Bibliothek keine Verbindung zur API herstellen kann oder wenn die API einen Nicht-Erfolgs-Statuscode zurückgibt (d. h. eine 4xx- oder 5xx-Antwort), wird eine Unterklasse von APIError ausgelöst:

import anthropic
# ...
try:
    message = client.messages.create(
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": "Hello, Claude",
            }
        ],
        model="claude-opus-4-8",
    )
except anthropic.APIConnectionError as e:
    print("The server could not be reached")
    print(e.__cause__)  # an underlying Exception, likely raised within httpx
except anthropic.RateLimitError as e:
    print("A 429 status code was received; we should back off a bit.")
except anthropic.APIStatusError as e:
    print("Another non-200-range status code was received")
    print(e.status_code)
    print(e.response)

Die Fehlercodes sind wie folgt:

Request-IDs

Weitere Informationen zum Debuggen von Anfragen findest du unter Request-ID.

Alle Objektantworten im SDK stellen eine _request_id-Eigenschaft bereit, die aus dem request-id-Antwort-Header hinzugefügt wird, damit du fehlgeschlagene Anfragen schnell protokollieren und an Anthropic melden kannst.

message = client.messages.create(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello, Claude"}],
    model="claude-opus-4-8",
)
print(message._request_id)  # e.g., req_018EeWyXxfu5pfWkrYcMdjWG

Wiederholungsversuche

Bestimmte Fehler werden standardmäßig automatisch 2-mal wiederholt, mit einem kurzen exponentiellen Backoff. Verbindungsfehler (zum Beispiel aufgrund eines Netzwerkverbindungsproblems), 408 Request Timeout, 409 Conflict, 429 Rate Limit und >=500 Internal-Fehler werden alle standardmäßig wiederholt.

Du kannst die Option max_retries verwenden, um dies zu konfigurieren oder zu deaktivieren:

# Konfiguriere den Standard für alle Anfragen:
client = Anthropic(
    max_retries=0,  # default is 2
)

# Oder konfiguriere pro Anfrage:
client.with_options(max_retries=5).messages.create(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello, Claude"}],
    model="claude-opus-4-8",
)

Timeouts

Standardmäßig laufen Anfragen nach 10 Minuten ab. Du kannst dies mit einer timeout-Option konfigurieren, die einen Float oder ein httpx.Timeout-Objekt akzeptiert:

import httpx
from anthropic import Anthropic

# Konfiguriere den Standard für alle Anfragen:
client = Anthropic(
    timeout=20.0,  # 20 seconds (default is 10 minutes)
)

# Feinere Steuerung:
client = Anthropic(
    timeout=httpx.Timeout(60.0, read=5.0, write=10.0, connect=2.0),
)

# Überschreibe pro Anfrage:
client.with_options(timeout=5.0).messages.create(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello, Claude"}],
    model="claude-opus-4-8",
)

Bei einem Timeout wird ein APITimeoutError ausgelöst.

Beachte, dass Anfragen, die ein Timeout erreichen, standardmäßig zweimal wiederholt werden.

Lange Anfragen

Vermeide es, einen großen max_tokens-Wert ohne Streaming zu setzen. Einige Netzwerke können inaktive Verbindungen nach einer bestimmten Zeit trennen, was dazu führen kann, dass die Anfrage fehlschlägt oder ein Timeout erreicht, ohne eine Antwort von Anthropic zu erhalten.

Das SDK löst einen ValueError aus, wenn erwartet wird, dass eine Nicht-Streaming-Anfrage länger als ungefähr 10 Minuten dauert. Das Übergeben von stream=True oder das Überschreiben der timeout-Option auf Client- oder Anfrageebene deaktiviert diesen Fehler.

Eine erwartete Anfrage-Latency, die länger als das Timeout für eine Nicht-Streaming-Anfrage ist, führt dazu, dass der Client die Verbindung beendet und einen Wiederholungsversuch startet, ohne eine Antwort zu erhalten.

Das SDK setzt eine TCP-Socket-Keep-Alive-Option, um die Auswirkungen von Timeouts bei inaktiven Verbindungen in einigen Netzwerken zu reduzieren. Dies kann durch Übergeben einer benutzerdefinierten http_client-Option an den Client überschrieben werden.

Automatische Paginierung

Listenmethoden in der Claude API sind paginiert. Du kannst die for-Syntax verwenden, um über Elemente auf allen Seiten zu iterieren:

client = Anthropic()

all_batches = []
# Ruft bei Bedarf automatisch weitere Seiten ab.
for batch in client.messages.batches.list(limit=20):
    all_batches.append(batch)
print(all_batches)

Für asynchrone Iteration:

async def main() -> None:
    all_batches = []
    async for batch in client.messages.batches.list(limit=20):
        all_batches.append(batch)
    print(all_batches)


asyncio.run(main())

Alternativ kannst du die Methoden .has_next_page(), .next_page_info() oder .get_next_page() für eine feinere Kontrolle bei der Arbeit mit Seiten verwenden:

first_page = await client.messages.batches.list(limit=20)

if first_page.has_next_page():
    print(f"will fetch next page using these details: {first_page.next_page_info()}")
    next_page = await first_page.get_next_page()
    print(f"number of items we just fetched: {len(next_page.data)}")

# Entferne `await` für nicht-asynchrone Nutzung.

Oder arbeite direkt mit den zurückgegebenen Daten:

first_page = await client.messages.batches.list(limit=20)

print(f"next page cursor: {first_page.last_id}")
for batch in first_page.data:
    print(batch.id)

# Entferne `await` für nicht-asynchrone Nutzung.

Standard-Header

Das SDK sendet automatisch den anthropic-version-Header mit dem Wert 2023-06-01.

Falls nötig, kannst du ihn überschreiben, indem du Standard-Header auf dem Client-Objekt oder pro Anfrage setzt.

# Setze Standard-Header für alle Anfragen auf dem Client
client = Anthropic(
    default_headers={"anthropic-version": "My-Custom-Value"},
)

# Oder überschreibe sie pro Anfrage
client.messages.with_raw_response.create(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello, Claude"}],
    model="claude-opus-4-8",
    extra_headers={"anthropic-version": "My-Custom-Value"},
)

Typsystem

Anfrageparameter

Verschachtelte Anfrageparameter sind TypedDicts. Antworten sind Pydantic-Modelle, die auch Hilfsmethoden für Dinge wie die Serialisierung zurück in JSON haben (v1, v2).

Typisierte Anfragen und Antworten bieten Autovervollständigung und Dokumentation in deinem Editor. Wenn du Typfehler in VS Code sehen möchtest, um Bugs früher zu erkennen, setze python.analysis.typeCheckingMode auf basic.

Antwortmodelle

Um ein Pydantic-Modell in ein Dictionary zu konvertieren, verwende die Hilfsmethoden:

message = client.messages.create(...)

# In JSON-String konvertieren
json_str = message.to_json()

# In Dictionary konvertieren
data = message.to_dict()

Umgang mit null vs. fehlenden Feldern

In Antworten kannst du zwischen Feldern unterscheiden, die explizit null sind, und Feldern, die nicht zurückgegeben wurden (fehlend):

response = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}],
)
if response.my_field is None:
    if "my_field" not in response.model_fields_set:
        print("field was not in the response")
    else:
        print("field was null")

Erweiterte Verwendung

Zugriff auf rohe Antwortdaten (zum Beispiel Header)

Auf die „rohe" Response, die von httpx zurückgegeben wird, kann über die .with_raw_response-Eigenschaft des Clients zugegriffen werden. Dies ist nützlich für den Zugriff auf Antwort-Header oder andere Metadaten:

client = Anthropic()

response = client.messages.with_raw_response.create(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello, Claude"}],
    model="claude-opus-4-8",
)

print(response.headers.get("request-id"))
message = (
    response.parse()
)  # get the object that `messages.create()` would have returned
print(message.content)

Diese Methoden geben ein APIResponse-Objekt zurück.

Streaming des Antwort-Bodys

Der .with_raw_response-Ansatz liest den vollständigen Antwort-Body eifrig ein, wenn du die Anfrage stellst. Um stattdessen den Antwort-Body zu streamen, verwende .with_streaming_response, was einen Kontextmanager erfordert und den Antwort-Body erst liest, wenn du .read(), .text(), .json(), .iter_bytes(), .iter_text(), .iter_lines() oder .parse() aufrufst. Im asynchronen Client sind dies asynchrone Methoden.

with client.messages.with_streaming_response.create(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello, Claude"}],
    model="claude-opus-4-8",
) as response:
    print(response.headers.get("request-id"))

    for line in response.iter_lines():
        print(line)

Der Kontextmanager ist erforderlich, damit die Antwort zuverlässig geschlossen wird.

Logging

Das SDK verwendet das logging-Modul der Standardbibliothek.

Du kannst Logging aktivieren, indem du die Umgebungsvariable ANTHROPIC_LOG auf debug oder info setzt:

export ANTHROPIC_LOG=debug

Benutzerdefinierte/undokumentierte Anfragen stellen

Diese Bibliothek ist für bequemen Zugriff auf die dokumentierte API typisiert. Wenn du auf undokumentierte Endpunkte, Parameter oder Antworteigenschaften zugreifen musst, kann die Bibliothek trotzdem verwendet werden.

Undokumentierte Endpunkte

Um Anfragen an undokumentierte Endpunkte zu stellen, kannst du client.get, client.post und andere HTTP-Verben verwenden. Optionen auf dem Client, wie Wiederholungsversuche, werden bei diesen Anfragen berücksichtigt.

import httpx

response = client.post(
    "/foo",
    cast_to=httpx.Response,
    body={"my_param": True},
)

print(response.json())

Undokumentierte Anfrageparameter

Wenn du explizit einen zusätzlichen Parameter senden möchtest, kannst du dies mit den Anfrageoptionen extra_query, extra_body und extra_headers tun.

Undokumentierte Antworteigenschaften

Um auf undokumentierte Antworteigenschaften zuzugreifen, kannst du auf die zusätzlichen Felder wie response.unknown_prop zugreifen. Du kannst auch alle zusätzlichen Felder des Pydantic-Modells als Dict mit response.model_extra erhalten.

Konfigurieren des HTTP-Clients

Du kannst den httpx-Client direkt überschreiben, um ihn für deinen Anwendungsfall anzupassen, einschließlich Unterstützung für Proxies und Transports:

import httpx
from anthropic import Anthropic, DefaultHttpxClient

client = Anthropic(
    # Oder verwende die Umgebungsvariable `ANTHROPIC_BASE_URL`
    base_url="http://my.test.server.example.com:8083",
    http_client=DefaultHttpxClient(
        proxy="http://my.test.proxy.example.com",
        transport=httpx.HTTPTransport(local_address="0.0.0.0"),
    ),
)

Du kannst den Client auch pro Anfrage anpassen, indem du with_options() verwendest:

client.with_options(http_client=DefaultHttpxClient(...))

Verwalten von HTTP-Ressourcen

Standardmäßig schließt die Bibliothek zugrunde liegende HTTP-Verbindungen, wenn der Client vom Garbage Collector bereinigt wird. Du kannst den Client bei Bedarf manuell mit der .close()-Methode schließen oder mit einem Kontextmanager, der beim Verlassen schließt.

with Anthropic() as client:
    message = client.messages.create(...)

# HTTP-Client wird automatisch geschlossen

Beta-Funktionen

Beta-Funktionen sind vor der allgemeinen Veröffentlichung verfügbar, um frühes Feedback zu erhalten und neue Funktionalität zu testen. Du kannst die Verfügbarkeit aller Fähigkeiten und Tools von Claude in der Übersicht zum Entwickeln mit Claude überprüfen.

Du kannst auf die meisten Beta-API-Funktionen über die beta-Eigenschaft des Clients zugreifen. Um eine bestimmte Beta-Funktion zu aktivieren, musst du den entsprechenden Beta-Header zum betas-Feld hinzufügen, wenn du eine Nachricht erstellst.

Zum Beispiel, um die Files API zu verwenden:

client = Anthropic()

response = client.beta.messages.create(
    model="claude-opus-4-8",
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "Please summarize this document for me."},
                {
                    "type": "document",
                    "source": {
                        "type": "file",
                        "file_id": "file_abc123",
                    },
                },
            ],
        },
    ],
    betas=["files-api-2025-04-14"],
)

Plattform-Integrationen

Alle fünf Client-Klassen sind im Basis-Paket anthropic enthalten:

Der AnthropicAWS-Client befindet sich in der Beta-Phase. Übergib workspace_id an den Konstruktor oder setze die Umgebungsvariable ANTHROPIC_AWS_WORKSPACE_ID.

Verwende AnthropicBedrockMantle für neue Projekte; AnthropicBedrock bleibt für bestehende Anwendungen erhalten, die die Bedrock-InvokeModel-API verwenden.

Semantische Versionierung

Dieses Paket folgt im Allgemeinen den SemVer-Konventionen, obwohl bestimmte rückwärtsinkompatible Änderungen als Minor-Versionen veröffentlicht werden können:

1. Änderungen, die nur statische Typen betreffen, ohne das Laufzeitverhalten zu beeinträchtigen.
2. Änderungen an Bibliotheksinterna, die technisch öffentlich sind, aber nicht für die externe Verwendung vorgesehen oder dokumentiert sind.
3. Änderungen, von denen nicht erwartet wird, dass sie die überwiegende Mehrheit der Nutzer in der Praxis betreffen.

Ermitteln der installierten Version

Wenn du auf die neueste Version aktualisiert hast, aber neue Funktionen, die du erwartet hast, nicht siehst, verwendet deine Python-Umgebung wahrscheinlich noch eine ältere Version. Du kannst die zur Laufzeit verwendete Version ermitteln mit:

print(anthropic.__version__)

Zusätzliche Ressourcen

• GitHub-Repository
• API-Referenz
• Streaming von Nachrichten
• Tool-Nutzung mit Claude