Python SDK

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

Installation

pip install anthropic

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

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

# Für Google Cloud-Unterstützung
pip install "anthropic[vertex]"

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

# Microsoft Foundry-Unterstützung ist im Basispaket enthalten

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

Anforderungen

Python 3.10 oder höher ist erforderlich. Wenn du von einer 0.x-Version des SDK aktualisierst, findest du im v1-Migrationsleitfaden die Liste der Breaking Changes.

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-5",
)

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

Authentifizierungsoptionen einschließlich Workload Identity Federation findest du unter Authentifizierung. Wenn dein API-Key ein persönlicher Key oder ein Service-Account-Key mit Zugriff auf mehrere Workspaces ist, setze die Workspace-ID im Request-Header anthropic-workspace-id; Einen Workspace auswählen zeigt die Option pro Anfrage für dieses SDK.

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-5",
    )
    print(message.content)


asyncio.run(main())

aiohttp für bessere Nebenläufigkeit verwenden

Für verbesserte asynchrone Performance kannst du das HTTP-Backend aiohttp anstelle des standardmäßigen httpx2 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-5",
        )
        print(message.content)


asyncio.run(main())

Streaming-Antworten

Das SDK bietet Unterstützung für „streaming“ (Streaming) von Antworten mithilfe von „Server-Sent Events“ (vom Server gesendete Ereignisse), oder SSE.

client = Anthropic()

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

Der asynchrone Client verwendet exakt dieselbe Schnittstelle:

client = AsyncAnthropic()

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

Streaming-Helfer

Das SDK bietet außerdem Streaming-Helfer, die Kontextmanager verwenden und Zugriff auf den akkumulierten Text und die finale Nachricht ermöglichen:

async def main() -> None:
    async with client.messages.stream(
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": "Say hello there!",
            }
        ],
        model="claude-opus-5",
    ) 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, darunter Akkumulation und SDK-spezifische Ereignisse.

Alternativ kannst du client.messages.create(..., stream=True) verwenden, das lediglich ein Iterable der Ereignisse 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 Antwort-Eigenschaft usage 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-5", 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 Decorator @beta_tool 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-5",
    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 die Antwort einen Aufruf eines der angegebenen Tools enthält, wird das Tool automatisch aufgerufen und das Ergebnis in der nächsten Iteration direkt an das Modell zurückgegeben.

Message Batches

Dieses SDK bietet Unterstützung für Batch-Verarbeitung unter client.messages.batches.

Einen Batch erstellen

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-5",
                "max_tokens": 1024,
                "messages": [{"role": "user", "content": "Hello, world"}],
            },
        },
        {
            "custom_id": "my-second-request",
            "params": {
                "model": "claude-opus-5",
                "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 dateiähnliches BinaryIO-Objekt

from pathlib import Path
from anthropic import Anthropic

client = Anthropic()

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

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

Der asynchrone Client verwendet exakt dieselbe Schnittstelle. Wenn du eine PathLike-Instanz übergibst, wird der Dateiinhalt automatisch asynchron gelesen.

Fehlerbehandlung

Wenn die Bibliothek keine Verbindung zur API herstellen kann oder die API einen nicht erfolgreichen Statuscode zurückgibt (das heißt 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-5",
    )
except anthropic.APIConnectionError as e:
    print("The server could not be reached")
    print(e.__cause__)  # an underlying Exception, likely raised within httpx2
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 lauten wie folgt:

Request-IDs

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

Alle Objektantworten im SDK stellen eine Eigenschaft _request_id bereit, die aus dem Antwort-Header request-id übernommen wird, sodass 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-5",
)
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 Ratenlimit und interne Fehler >=500 werden standardmäßig alle wiederholt.

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

# Konfiguriere den Standardwert 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-5",
)

Timeouts

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

import httpx2
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=httpx2.Timeout(60.0, read=5.0, write=10.0, connect=2.0),
)

# Pro Anfrage überschreiben:
client.with_options(timeout=5.0).messages.create(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello, Claude"}],
    model="claude-opus-5",
)

Bei einem Timeout wirft das SDK einen APITimeoutError.

Beachte, dass Anfragen, die in einen Timeout laufen, standardmäßig zweimal wiederholt werden.

Lange Anfragen

Vermeide es, einen großen max_tokens-Wert zu setzen, ohne Streaming zu verwenden. Manche Netzwerke trennen inaktive Verbindungen nach einer bestimmten Zeit, was dazu führen kann, dass die Anfrage fehlschlägt oder in einen Timeout läuft, ohne eine Antwort von Anthropic zu erhalten.

Das SDK wirft einen ValueError, wenn eine Nicht-Streaming-Anfrage voraussichtlich länger als etwa 10 Minuten dauert. Das Übergeben von stream=True oder das Überschreiben der Option timeout auf Client- oder Anfrageebene deaktiviert diesen Fehler.

Eine erwartete Anfragelatenz, die länger als der 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 inaktiver Verbindungen in manchen Netzwerken zu verringern. Dies kann überschrieben werden, indem eine benutzerdefinierte Option http_client an den Client übergeben wird.

Automatische Paginierung

List-Methoden in der Claude API sind paginiert. Du kannst die for-Syntax verwenden, um über Elemente aller Seiten hinweg 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 Header anthropic-version 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.

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

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

Typsystem

Anfrageparameter

Verschachtelte Anfrageparameter sind TypedDicts. Antworten sind Pydantic-Modelle, die außerdem Hilfsmethoden für Dinge wie die Rückserialisierung in JSON bieten (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 umzuwandeln, verwende die Hilfsmethoden:

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

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

# In Dictionary umwandeln
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-5",
    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 httpx2 zurückgegeben wird, kann über die Eigenschaft .with_raw_response auf dem Client zugegriffen werden. Dies ist nützlich, um auf Antwort-Header oder andere Metadaten zuzugreifen:

client = Anthropic()

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

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. Auf dem asynchronen Client geben sie ein AsyncAPIResponse zurück, und .parse(), .read(), .text() und .json() müssen mit await aufgerufen werden.

Streaming des Antwort-Bodys

Der Ansatz mit .with_raw_response liest den vollständigen Antwort-Body sofort, wenn du die Anfrage stellst. Um den Antwort-Body stattdessen zu streamen, verwende .with_streaming_response, das 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-5",
) 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 Modul logging 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 den 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 etwa Wiederholungsversuche, werden bei diesen Anfragen berücksichtigt.

import httpx2

response = client.post(
    "/foo",
    cast_to=httpx2.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 außerdem alle zusätzlichen Felder des Pydantic-Modells als Dict mit response.model_extra abrufen.

Den HTTP-Client konfigurieren

Das SDK sendet Anfragen mit httpx2, einem API-kompatiblen Fork von httpx. Um den HTTP-Client anzupassen, einschließlich Proxys und Transports, übergib deinen eigenen httpx2-Client als http_client:

import httpx2
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=httpx2.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(...))

Tracing- und Mocking-Tools, die httpx selbst patchen, wie etwa der HTTPXClientInstrumentor von OpenTelemetry, die httpx-Integration von Sentry, respx oder pytest-httpx, sehen die Anfragen des SDK standardmäßig nicht. Um sie zu verwenden, rufe httpx2.alias_httpx() einmal beim Start auf, bevor irgendetwas httpx importiert. Dadurch wird import httpx für den gesamten Prozess zu httpx2 aufgelöst.

HTTP-Ressourcen verwalten

Standardmäßig schließt die Bibliothek die zugrunde liegenden HTTP-Verbindungen, sobald der Client vom Garbage Collector eingesammelt wird. Du kannst den Client bei Bedarf manuell mit der Methode .close() 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 „Mit Claude entwickeln“ prüfen.

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

Zum Beispiel, um Kontextbearbeitung zu aktivieren:

client = Anthropic()

response = client.beta.messages.create(
    model="claude-opus-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello, Claude"}],
    betas=["context-management-2025-06-27"],
)

Plattformintegrationen

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

Der Client AnthropicAWS 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-API InvokeModel verwenden.

Semantische Versionierung

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

1. Änderungen, die nur statische Typen betreffen, ohne das Laufzeitverhalten zu brechen.
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.

Die installierte Version ermitteln

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

print(anthropic.__version__)

Zusätzliche Ressourcen

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