Python SDK

Das Anthropic Python SDK bietet bequemen Zugriff auf die Anthropic REST 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 Async-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 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]"



Anforderungen

Python 3.9 oder höher ist erforderlich.



Verwendung

import os
from anthropic import Anthropic

client = Anthropic(
    # Dies ist der Standard 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)

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


asyncio.run(main())



Verwendung von aiohttp für bessere Nebenläufigkeit

Für verbesserte Async-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-5",
        )
        print(message.content)


asyncio.run(main())



Streaming von Antworten

Das SDK bietet Unterstützung für das Streaming von Antworten mittels Server-Sent Events (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 Context Manager 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-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, einschließlich Akkumulation und SDK-spezifischer Events.

Alternativ kannst du client.messages.create(..., stream=True) verwenden, das 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-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 @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-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 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-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, erkennbar an .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()

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

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

Der asynchrone Client verwendet exakt 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 erfolgreichen Statuscode zurückgibt (also 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 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 bieten eine _request_id-Eigenschaft, 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-5",
)
print(message._request_id)  # e.g., req_018EeWyXxfu5pfWkrYcMdjWG



Wiederholungsversuche

Bestimmte Fehler werden standardmäßig 2-mal automatisch wiederholt, mit einem kurzen exponentiellen Backoff. Verbindungsfehler (zum Beispiel aufgrund eines Netzwerkverbindungsproblems), 408 Request Timeout, 409 Conflict, 429 Rate Limit und >=500 Internal Errors 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 es 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 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 Kontrolle:
client = Anthropic(
    timeout=httpx.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 einen Timeout erreichen, standardmäßig zweimal wiederholt werden.



Lange Anfragen

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

Das SDK wirft einen ValueError, wenn erwartet wird, dass eine Nicht-Streaming-Anfrage länger als etwa 10 Minuten dauert. Das Übergeben von stream=True oder das Überschreiben der timeout-Option 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 es erneut versucht, 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 granularere Kontrolle beim Arbeiten 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 Verwendung.

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 Verwendung.



Standard-Header

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

Bei Bedarf 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-5",
    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 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 zu konvertieren, 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 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-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.



Streaming des Antwort-Bodys

Der .with_raw_response-Ansatz 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 Context Manager 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 Context Manager 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 den bequemen Zugriff auf die dokumentierte API typisiert. Wenn du auf undokumentierte Endpunkte, Parameter oder Antworteigenschaften zugreifen musst, kann die Bibliothek dennoch 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 abrufen.



Konfigurieren des HTTP-Clients

Du kannst den httpx-Client direkt überschreiben, um ihn für deinen Anwendungsfall anzupassen, einschließlich Unterstützung für Proxys 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 mit with_options() anpassen:

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



Verwalten von HTTP-Ressourcen

Standardmäßig schließt die Bibliothek zugrunde liegende HTTP-Verbindungen, sobald der Client der Garbage Collection unterliegt. Du kannst den Client bei Bedarf manuell mit der .close()-Methode schließen oder mit einem Context Manager, 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" überprüfen.

Auf die meisten Beta-API-Funktionen kannst du ü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-5",
    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, 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 beeinträchtigen.



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 wie folgt ermitteln:

print(anthropic.__version__)



Zusätzliche Ressourcen

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