SDK Python

L'SDK Python di Anthropic fornisce un accesso comodo alla Claude API dalle applicazioni Python. Supporta operazioni sia sincrone che asincrone, lo streaming e le integrazioni con Amazon Bedrock, Claude Platform su AWS, Google Cloud e Microsoft Foundry.

Installazione

pip install anthropic

Per integrazioni specifiche della piattaforma o per prestazioni asincrone migliorate, installa con gli extra:

# Per il supporto di Amazon Bedrock
pip install "anthropic[bedrock]"

# Per il supporto di Google Cloud
pip install "anthropic[vertex]"

# Per il supporto di Claude Platform su AWS
pip install "anthropic[aws]"

# Il supporto di Microsoft Foundry è incluso nel pacchetto base

# Per prestazioni asincrone migliorate con aiohttp
pip install "anthropic[aiohttp]"

Requisiti

È richiesto Python 3.10 o successivo. Se stai eseguendo l'aggiornamento da una versione 0.x dell'SDK, consulta la guida alla migrazione v1 per l'elenco delle modifiche incompatibili.

Utilizzo

import os
from anthropic import Anthropic

client = Anthropic(
    # Questo è il valore predefinito e può essere omesso
    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)

Per le opzioni di autenticazione, inclusa la Workload Identity Federation, consulta Autenticazione. Se la tua chiave API è una chiave personale o di account di servizio con accesso a più workspace, imposta l'ID del workspace nell'header di richiesta anthropic-workspace-id; Seleziona un workspace mostra l'opzione per singola richiesta per questo SDK.

Utilizzo asincrono

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())

Usare aiohttp per una migliore concorrenza

Per prestazioni asincrone migliorate, puoi usare il backend HTTP aiohttp invece del predefinito httpx2:

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())

Risposte in streaming

L'SDK fornisce supporto per le risposte in "streaming" (streaming) utilizzando 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)

Il client asincrono usa esattamente la stessa interfaccia:

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)

Helper per lo streaming

L'SDK fornisce anche helper per lo streaming che usano context manager e forniscono accesso al testo accumulato e al messaggio finale:

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())

Lo streaming con client.messages.stream(...) espone vari helper, tra cui l'accumulo e gli eventi specifici dell'SDK.

In alternativa, puoi usare client.messages.create(..., stream=True), che restituisce solo un iterabile degli eventi nello stream e usa meno memoria (non costruisce per te un oggetto messaggio finale).

Conteggio dei token

Puoi vedere l'utilizzo esatto per una determinata richiesta tramite la proprietà di risposta usage:

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

Puoi anche contare i token prima di effettuare una richiesta:

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

Uso degli strumenti

Questo SDK fornisce supporto per il "tool use" (uso degli strumenti), noto anche come function calling. Per maggiori dettagli, consulta Uso degli strumenti con Claude.

Helper per gli strumenti

L'SDK fornisce helper per definire ed eseguire strumenti come pure funzioni Python. Il decoratore @beta_tool genera lo schema dello strumento a partire dalla firma della funzione e dalla 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",
        }
    )


# Usa tool_runner per gestire automaticamente le chiamate agli strumenti
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)

A ogni iterazione viene effettuata una richiesta API. Se la risposta include una chiamata a uno degli strumenti forniti, lo strumento viene chiamato automaticamente e il risultato viene restituito direttamente al modello nell'iterazione successiva.

Batch di messaggi

Questo SDK fornisce supporto per l'elaborazione batch sotto client.messages.batches.

Creare un batch

Message Batches accetta un array di richieste, in cui ogni oggetto ha un identificatore custom_id e gli stessi params di richiesta della Messages API standard:

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"}],
            },
        },
    ]
)

Ottenere i risultati da un batch

Una volta che un Message Batch è stato elaborato, come indicato da .processing_status == 'ended', puoi accedere ai risultati con .batches.results():

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)

Caricamento di file

I parametri di richiesta che corrispondono a caricamenti di file possono essere passati in molte forme diverse:

• Un oggetto PathLike (ad esempio, pathlib.Path)
• Una tupla di (filename, content, content_type)
• Un oggetto file-like BinaryIO

from pathlib import Path
from anthropic import Anthropic

client = Anthropic()

# Carica usando un percorso di file
client.files.upload(
    file=Path("/path/to/file"),
)

# Carica usando byte
client.files.upload(
    file=("file.txt", b"my bytes", "text/plain"),
)

Il client asincrono usa esattamente la stessa interfaccia. Se passi un'istanza PathLike, il contenuto del file viene letto automaticamente in modo asincrono.

Gestione degli errori

Quando la libreria non riesce a connettersi all'API, o se l'API restituisce un codice di stato di non successo (ovvero una risposta 4xx o 5xx), viene sollevata una sottoclasse di APIError:

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)

I codici di errore sono i seguenti:

ID delle richieste

Per maggiori informazioni sul debug delle richieste, consulta ID richiesta.

Tutte le risposte oggetto nell'SDK forniscono una proprietà _request_id, aggiunta a partire dall'header di risposta request-id, in modo che tu possa registrare rapidamente le richieste non riuscite e segnalarle ad Anthropic.

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

Tentativi ripetuti

Alcuni errori vengono ritentati automaticamente 2 volte per impostazione predefinita, con un breve backoff esponenziale. Gli errori di connessione (ad esempio, a causa di un problema di connettività di rete), 408 Request Timeout, 409 Conflict, 429 Rate Limit e gli errori interni >=500 vengono tutti ritentati per impostazione predefinita.

Puoi usare l'opzione max_retries per configurare o disabilitare questo comportamento:

# Configura il valore predefinito per tutte le richieste:
client = Anthropic(
    max_retries=0,  # default is 2
)

# Oppure, configura per singola richiesta:
client.with_options(max_retries=5).messages.create(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello, Claude"}],
    model="claude-opus-5",
)

Timeout

Per impostazione predefinita le richieste vanno in timeout dopo 10 minuti. Puoi configurarlo con un'opzione timeout, che accetta un float o un oggetto httpx2.Timeout:

import httpx2
from anthropic import Anthropic

# Configura il valore predefinito per tutte le richieste:
client = Anthropic(
    timeout=20.0,  # 20 seconds (default is 10 minutes)
)

# Controllo più granulare:
client = Anthropic(
    timeout=httpx2.Timeout(60.0, read=5.0, write=10.0, connect=2.0),
)

# Sovrascrivi per singola richiesta:
client.with_options(timeout=5.0).messages.create(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello, Claude"}],
    model="claude-opus-5",
)

In caso di timeout, l'SDK solleva un APITimeoutError.

Nota che le richieste che vanno in timeout vengono ritentate due volte per impostazione predefinita.

Richieste lunghe

Evita di impostare un valore elevato di max_tokens senza usare lo streaming. Alcune reti potrebbero interrompere le connessioni inattive dopo un certo periodo di tempo, il che può causare il fallimento della richiesta o un timeout senza ricevere una risposta da Anthropic.

L'SDK solleverà un ValueError se si prevede che una richiesta non in streaming richieda più di circa 10 minuti. Passare stream=True o sovrascrivere l'opzione timeout a livello di client o di richiesta disabilita questo errore.

Una "latency" (latenza) prevista della richiesta superiore al timeout per una richiesta non in streaming farà sì che il client termini la connessione e ritenti senza ricevere una risposta.

L'SDK imposta un'opzione TCP socket keep-alive per ridurre l'impatto dei timeout delle connessioni inattive su alcune reti. Questo può essere sovrascritto passando un'opzione http_client personalizzata al client.

Paginazione automatica

I metodi di elenco nella Claude API sono paginati. Puoi usare la sintassi for per iterare sugli elementi attraverso tutte le pagine:

client = Anthropic()

all_batches = []
# Recupera automaticamente altre pagine secondo necessità.
for batch in client.messages.batches.list(limit=20):
    all_batches.append(batch)
print(all_batches)

Per l'iterazione asincrona:

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())

In alternativa, puoi usare i metodi .has_next_page(), .next_page_info() o .get_next_page() per un controllo più granulare nel lavorare con le pagine:

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)}")

# Rimuovi `await` per l'uso non asincrono.

Oppure lavora direttamente con i dati restituiti:

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)

# Rimuovi `await` per l'uso non asincrono.

Header predefiniti

L'SDK invia automaticamente l'header anthropic-version impostato su 2023-06-01.

Se necessario, puoi sovrascriverlo impostando header predefiniti sull'oggetto client o per singola richiesta.

# Imposta le intestazioni predefinite per tutte le richieste sul client
client = Anthropic(
    default_headers={"anthropic-version": "My-Custom-Value"},
)

# Oppure sovrascrivile per singola richiesta
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"},
)

Sistema di tipi

Parametri di richiesta

I parametri di richiesta annidati sono TypedDict. Le risposte sono modelli Pydantic che dispongono anche di metodi helper per operazioni come la serializzazione in JSON (v1, v2).

Le richieste e le risposte tipizzate forniscono completamento automatico e documentazione all'interno del tuo editor. Se desideri vedere gli errori di tipo in VS Code per individuare i bug prima, imposta python.analysis.typeCheckingMode su basic.

Modelli di risposta

Per convertire un modello Pydantic in un dizionario, usa i metodi helper:

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

# Converti in stringa JSON
json_str = message.to_json()

# Converti in dizionario
data = message.to_dict()

Gestione dei campi null rispetto ai campi mancanti

Nelle risposte, puoi distinguere tra i campi che sono esplicitamente null e i campi che non sono stati restituiti (mancanti):

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

Utilizzo avanzato

Accesso ai dati grezzi della risposta (ad esempio, gli header)

La Response "grezza" restituita da httpx2 è accessibile tramite la proprietà .with_raw_response sul client. Questo è utile per accedere agli header della risposta o ad altri metadati:

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)

Questi metodi restituiscono un oggetto APIResponse. Sul client asincrono restituiscono un AsyncAPIResponse, e .parse(), .read(), .text() e .json() devono essere attesi con await.

Streaming del corpo della risposta

L'approccio .with_raw_response legge immediatamente l'intero corpo della risposta quando effettui la richiesta. Per eseguire invece lo streaming del corpo della risposta, usa .with_streaming_response, che richiede un context manager e legge il corpo della risposta solo quando chiami .read(), .text(), .json(), .iter_bytes(), .iter_text(), .iter_lines() o .parse(). Nel client asincrono, questi sono metodi asincroni.

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)

Il context manager è necessario affinché la risposta venga chiusa in modo affidabile.

Logging

L'SDK usa il modulo logging della libreria standard.

Puoi abilitare il logging impostando la variabile d'ambiente ANTHROPIC_LOG su debug o info:

export ANTHROPIC_LOG=debug

Effettuare richieste personalizzate/non documentate

Questa libreria è tipizzata per un accesso comodo all'API documentata. Se hai bisogno di accedere a endpoint, parametri o proprietà di risposta non documentati, la libreria può comunque essere usata.

Endpoint non documentati

Per effettuare richieste a endpoint non documentati, puoi usare client.get, client.post e altri verbi HTTP. Le opzioni sul client, come i tentativi ripetuti, vengono rispettate quando si effettuano queste richieste.

import httpx2

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

print(response.json())

Parametri di richiesta non documentati

Se vuoi inviare esplicitamente un parametro aggiuntivo, puoi farlo con le opzioni di richiesta extra_query, extra_body ed extra_headers.

Proprietà di risposta non documentate

Per accedere a proprietà di risposta non documentate, puoi accedere ai campi extra come response.unknown_prop. Puoi anche ottenere tutti i campi extra sul modello Pydantic come dict con response.model_extra.

Configurazione del client HTTP

L'SDK invia le richieste con httpx2, un fork di httpx compatibile a livello di API. Per personalizzare il client HTTP, inclusi proxy e transport, passa il tuo client httpx2 come http_client:

import httpx2
from anthropic import Anthropic, DefaultHttpxClient

client = Anthropic(
    # Oppure usa la variabile d'ambiente `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"),
    ),
)

Puoi anche personalizzare il client per singola richiesta usando with_options():

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

Gli strumenti di tracing e mocking che applicano patch a httpx stesso, come HTTPXClientInstrumentor di OpenTelemetry, l'integrazione httpx di Sentry, respx o pytest-httpx, non vedono le richieste dell'SDK per impostazione predefinita. Per usarli, chiama httpx2.alias_httpx() una volta all'avvio, prima che qualsiasi cosa importi httpx. Questo fa sì che import httpx si risolva in httpx2 per l'intero processo.

Gestione delle risorse HTTP

Per impostazione predefinita la libreria chiude le connessioni HTTP sottostanti ogni volta che il client viene sottoposto a garbage collection. Puoi chiudere manualmente il client usando il metodo .close() se lo desideri, oppure con un context manager che chiude all'uscita.

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

# Il client HTTP viene chiuso automaticamente

Funzionalità beta

Le funzionalità beta sono disponibili prima del rilascio generale per raccogliere feedback anticipati e testare nuove funzionalità. Puoi verificare la disponibilità di tutte le capacità e gli strumenti di Claude nella panoramica su come costruire con Claude.

Puoi accedere alla maggior parte delle funzionalità beta dell'API tramite la proprietà beta del client. Per abilitare una particolare funzionalità beta, devi aggiungere l'header beta appropriato al campo betas quando crei un messaggio.

Ad esempio, per abilitare la modifica del contesto:

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

Integrazioni con le piattaforme

Tutte e cinque le classi client sono incluse nel pacchetto base anthropic:

Il client AnthropicAWS è in beta. Passa workspace_id al costruttore oppure imposta la variabile d'ambiente ANTHROPIC_AWS_WORKSPACE_ID.

Usa AnthropicBedrockMantle per i nuovi progetti; AnthropicBedrock rimane disponibile per le applicazioni esistenti che usano l'API InvokeModel di Bedrock.

Versionamento semantico

Questo pacchetto segue generalmente le convenzioni SemVer, anche se alcune modifiche non retrocompatibili potrebbero essere rilasciate come versioni minori:

1. Modifiche che riguardano solo i tipi statici, senza alterare il comportamento a runtime.
2. Modifiche agli interni della libreria che sono tecnicamente pubblici ma non destinati o documentati per l'uso esterno.
3. Modifiche che non si prevede abbiano un impatto sulla grande maggioranza degli utenti nella pratica.

Determinare la versione installata

Se hai eseguito l'aggiornamento all'ultima versione ma non vedi le nuove funzionalità che ti aspettavi, è probabile che il tuo ambiente Python stia ancora usando una versione precedente. Puoi determinare la versione in uso a runtime con:

print(anthropic.__version__)

Risorse aggiuntive

• Repository GitHub
• Riferimento API
• Messaggi in streaming
• Uso degli strumenti con Claude