SDK Python

Le SDK Python Anthropic offre un accès pratique à la Claude API depuis les applications Python. Il prend en charge les opérations synchrones et asynchrones, le streaming, ainsi que les intégrations avec Amazon Bedrock, Claude Platform sur AWS, Google Cloud et Microsoft Foundry.

Installation

pip install anthropic

Pour les intégrations spécifiques à une plateforme ou de meilleures performances asynchrones, installez avec les extras :

# Pour la prise en charge d'Amazon Bedrock
pip install "anthropic[bedrock]"

# Pour la prise en charge de Google Cloud
pip install "anthropic[vertex]"

# Pour la prise en charge de Claude Platform sur AWS
pip install "anthropic[aws]"

# La prise en charge de Microsoft Foundry est incluse dans le package de base

# Pour de meilleures performances asynchrones avec aiohttp
pip install "anthropic[aiohttp]"

Prérequis

Python 3.10 ou une version ultérieure est requis. Si vous effectuez une mise à niveau depuis une version 0.x du SDK, consultez le guide de migration v1 pour la liste des changements non rétrocompatibles.

Utilisation

import os
from anthropic import Anthropic

client = Anthropic(
    # Valeur par défaut, peut être omise
    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)

Pour les options d'authentification, y compris Workload Identity Federation, consultez Authentification. Si votre clé API est une clé personnelle ou de compte de service ayant accès à plusieurs espaces de travail, définissez l'identifiant de l'espace de travail dans l'en-tête de requête anthropic-workspace-id ; Sélectionner un espace de travail présente l'option par requête pour ce SDK.

Utilisation asynchrone

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

Utiliser aiohttp pour une meilleure concurrence

Pour de meilleures performances asynchrones, vous pouvez utiliser le backend HTTP aiohttp au lieu du backend par défaut 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())

Réponses en streaming

Le SDK prend en charge les réponses en « streaming » (streaming) à l'aide des « Server-Sent Events » (événements envoyés par le serveur), ou 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)

Le client asynchrone utilise exactement la même interface :

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)

Assistants de streaming

Le SDK fournit également des assistants de streaming qui utilisent des gestionnaires de contexte et donnent accès au texte accumulé et au message final :

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

Le streaming avec client.messages.stream(...) expose divers assistants, notamment l'accumulation et des événements spécifiques au SDK.

Vous pouvez également utiliser client.messages.create(..., stream=True), qui renvoie uniquement un itérable des événements du flux et utilise moins de mémoire (il ne construit pas d'objet message final pour vous).

Comptage des tokens

Vous pouvez consulter l'utilisation exacte pour une requête donnée via la propriété de réponse usage :

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

Vous pouvez également compter les tokens avant d'effectuer une requête :

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

Utilisation d'outils

Ce SDK prend en charge le « tool use » (utilisation d'outils), également appelé appel de fonctions. Pour plus de détails, consultez Utilisation d'outils avec Claude.

Assistants d'outils

Le SDK fournit des assistants pour définir et exécuter des outils sous forme de fonctions Python pures. Le décorateur @beta_tool génère le schéma de l'outil à partir de la signature de la fonction et de sa 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",
        }
    )


# Utilisez le tool_runner pour gérer automatiquement les appels d'outils
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)

À chaque itération, une requête API est effectuée. Si la réponse inclut un appel à l'un des outils fournis, l'outil est automatiquement appelé et le résultat est renvoyé directement au modèle lors de l'itération suivante.

Lots de messages

Ce SDK prend en charge le traitement par lots sous client.messages.batches.

Créer un lot

Message Batches accepte un tableau de requêtes, où chaque objet possède un identifiant custom_id et les mêmes params de requête que l'API Messages 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"}],
            },
        },
    ]
)

Obtenir les résultats d'un lot

Une fois qu'un lot de messages a été traité, ce qui est indiqué par .processing_status == 'ended', vous pouvez accéder aux résultats avec .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)

Téléversement de fichiers

Les paramètres de requête correspondant à des téléversements de fichiers peuvent être transmis sous de nombreuses formes différentes :

• Un objet PathLike (par exemple, pathlib.Path)
• Un tuple (filename, content, content_type)
• Un objet de type fichier BinaryIO

from pathlib import Path
from anthropic import Anthropic

client = Anthropic()

# Téléverser à partir d'un chemin de fichier
client.files.upload(
    file=Path("/path/to/file"),
)

# Téléverser à partir d'octets
client.files.upload(
    file=("file.txt", b"my bytes", "text/plain"),
)

Le client asynchrone utilise exactement la même interface. Si vous transmettez une instance PathLike, le contenu du fichier est automatiquement lu de manière asynchrone.

Gestion des erreurs

Lorsque la bibliothèque ne parvient pas à se connecter à l'API, ou si l'API renvoie un code de statut d'échec (c'est-à-dire une réponse 4xx ou 5xx), une sous-classe de APIError est levée :

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)

Les codes d'erreur sont les suivants :

Identifiants de requête

Pour plus d'informations sur le débogage des requêtes, consultez Identifiant de requête.

Toutes les réponses objet du SDK fournissent une propriété _request_id, ajoutée à partir de l'en-tête de réponse request-id, afin que vous puissiez rapidement journaliser les requêtes en échec et les signaler à 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

Nouvelles tentatives

Certaines erreurs font automatiquement l'objet de 2 nouvelles tentatives par défaut, avec un court délai d'attente exponentiel. Les erreurs de connexion (par exemple, en raison d'un problème de connectivité réseau), 408 Request Timeout, 409 Conflict, 429 Rate Limit et les erreurs internes >=500 font toutes l'objet de nouvelles tentatives par défaut.

Vous pouvez utiliser l'option max_retries pour configurer ou désactiver ce comportement :

# Configurez la valeur par défaut pour toutes les requêtes :
client = Anthropic(
    max_retries=0,  # default is 2
)

# Ou configurez-la par requête :
client.with_options(max_retries=5).messages.create(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello, Claude"}],
    model="claude-opus-5",
)

Délais d'expiration

Par défaut, les requêtes expirent après 10 minutes. Vous pouvez configurer cela avec l'option timeout, qui accepte un nombre flottant ou un objet httpx2.Timeout :

import httpx2
from anthropic import Anthropic

# Configurer la valeur par défaut pour toutes les requêtes :
client = Anthropic(
    timeout=20.0,  # 20 seconds (default is 10 minutes)
)

# Contrôle plus granulaire :
client = Anthropic(
    timeout=httpx2.Timeout(60.0, read=5.0, write=10.0, connect=2.0),
)

# Remplacer pour une requête donnée :
client.with_options(timeout=5.0).messages.create(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello, Claude"}],
    model="claude-opus-5",
)

En cas d'expiration du délai, le SDK lève une APITimeoutError.

Notez que les requêtes qui expirent font l'objet de deux nouvelles tentatives par défaut.

Requêtes longues

Évitez de définir une valeur max_tokens élevée sans utiliser le streaming. Certains réseaux peuvent interrompre les connexions inactives après un certain temps, ce qui peut entraîner l'échec de la requête ou son expiration sans recevoir de réponse d'Anthropic.

Le SDK lèvera une ValueError si une requête sans streaming est censée durer plus d'environ 10 minutes. Passer stream=True ou remplacer l'option timeout au niveau du client ou de la requête désactive cette erreur.

Une « latency » (latence) de requête attendue supérieure au délai d'expiration pour une requête sans streaming entraînera la fermeture de la connexion par le client et une nouvelle tentative sans réception de réponse.

Le SDK définit une option TCP socket keep-alive afin de réduire l'impact des expirations de connexions inactives sur certains réseaux. Ce comportement peut être remplacé en passant une option http_client personnalisée au client.

Pagination automatique

Les méthodes de liste de la Claude API sont paginées. Vous pouvez utiliser la syntaxe for pour parcourir les éléments sur toutes les pages :

client = Anthropic()

all_batches = []
# Récupère automatiquement d'autres pages selon les besoins.
for batch in client.messages.batches.list(limit=20):
    all_batches.append(batch)
print(all_batches)

Pour l'itération asynchrone :

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

Vous pouvez également utiliser les méthodes .has_next_page(), .next_page_info() ou .get_next_page() pour un contrôle plus fin du travail avec les pages :

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

# Supprimez `await` pour une utilisation non asynchrone.

Ou travailler directement avec les données renvoyées :

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)

# Supprimez `await` pour une utilisation non asynchrone.

En-têtes par défaut

Le SDK envoie automatiquement l'en-tête anthropic-version défini sur 2023-06-01.

Si nécessaire, vous pouvez le remplacer en définissant des en-têtes par défaut sur l'objet client ou par requête.

# Définir des en-têtes par défaut pour toutes les requêtes du client
client = Anthropic(
    default_headers={"anthropic-version": "My-Custom-Value"},
)

# Ou les remplacer pour une requête donnée
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"},
)

Système de types

Paramètres de requête

Les paramètres de requête imbriqués sont des TypedDicts. Les réponses sont des modèles Pydantic qui disposent également de méthodes d'assistance, par exemple pour la sérialisation en JSON (v1, v2).

Les requêtes et réponses typées offrent l'autocomplétion et la documentation dans votre éditeur. Si vous souhaitez voir les erreurs de type dans VS Code pour détecter les bogues plus tôt, définissez python.analysis.typeCheckingMode sur basic.

Modèles de réponse

Pour convertir un modèle Pydantic en dictionnaire, utilisez les méthodes d'assistance :

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

# Convertir en chaîne JSON
json_str = message.to_json()

# Convertir en dictionnaire
data = message.to_dict()

Gestion des champs null et des champs manquants

Dans les réponses, vous pouvez distinguer les champs explicitement null des champs qui n'ont pas été renvoyés (manquants) :

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

Utilisation avancée

Accéder aux données brutes de la réponse (par exemple, les en-têtes)

La Response « brute » renvoyée par httpx2 est accessible via la propriété .with_raw_response du client. Cela est utile pour accéder aux en-têtes de réponse ou à d'autres métadonnées :

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)

Ces méthodes renvoient un objet APIResponse. Sur le client asynchrone, elles renvoient un AsyncAPIResponse, et .parse(), .read(), .text() et .json() doivent être attendus avec await.

Corps de réponse en streaming

L'approche .with_raw_response lit immédiatement l'intégralité du corps de la réponse lorsque vous effectuez la requête. Pour diffuser le corps de la réponse en streaming à la place, utilisez .with_streaming_response, qui nécessite un gestionnaire de contexte et ne lit le corps de la réponse qu'une fois que vous appelez .read(), .text(), .json(), .iter_bytes(), .iter_text(), .iter_lines() ou .parse(). Dans le client asynchrone, ce sont des méthodes asynchrones.

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)

Le gestionnaire de contexte est requis afin que la réponse soit fermée de manière fiable.

Journalisation

Le SDK utilise le module logging de la bibliothèque standard.

Vous pouvez activer la journalisation en définissant la variable d'environnement ANTHROPIC_LOG sur debug ou info :

export ANTHROPIC_LOG=debug

Effectuer des requêtes personnalisées/non documentées

Cette bibliothèque est typée pour un accès pratique à l'API documentée. Si vous devez accéder à des points de terminaison, paramètres ou propriétés de réponse non documentés, la bibliothèque peut tout de même être utilisée.

Points de terminaison non documentés

Pour effectuer des requêtes vers des points de terminaison non documentés, vous pouvez utiliser client.get, client.post et d'autres verbes HTTP. Les options du client, telles que les nouvelles tentatives, sont respectées lors de ces requêtes.

import httpx2

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

print(response.json())

Paramètres de requête non documentés

Si vous souhaitez envoyer explicitement un paramètre supplémentaire, vous pouvez le faire avec les options de requête extra_query, extra_body et extra_headers.

Propriétés de réponse non documentées

Pour accéder aux propriétés de réponse non documentées, vous pouvez accéder aux champs supplémentaires comme response.unknown_prop. Vous pouvez également obtenir tous les champs supplémentaires du modèle Pydantic sous forme de dict avec response.model_extra.

Configurer le client HTTP

Le SDK envoie les requêtes avec httpx2, un fork de httpx compatible au niveau de l'API. Pour personnaliser le client HTTP, y compris les proxys et les transports, passez votre propre client httpx2 en tant que http_client :

import httpx2
from anthropic import Anthropic, DefaultHttpxClient

client = Anthropic(
    # Ou utilisez la variable d'environnement `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"),
    ),
)

Vous pouvez également personnaliser le client par requête en utilisant with_options() :

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

Les outils de traçage et de simulation qui modifient httpx lui-même, tels que HTTPXClientInstrumentor d'OpenTelemetry, l'intégration httpx de Sentry, respx ou pytest-httpx, ne voient pas les requêtes du SDK par défaut. Pour les utiliser, appelez httpx2.alias_httpx() une fois au démarrage, avant que quoi que ce soit n'importe httpx. Cela fait en sorte que import httpx se résolve en httpx2 pour l'ensemble du processus.

Gestion des ressources HTTP

Par défaut, la bibliothèque ferme les connexions HTTP sous-jacentes chaque fois que le client est collecté par le ramasse-miettes. Vous pouvez fermer manuellement le client à l'aide de la méthode .close() si vous le souhaitez, ou avec un gestionnaire de contexte qui le ferme à la sortie.

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

# Le client HTTP est fermé automatiquement

Fonctionnalités bêta

Les fonctionnalités bêta sont disponibles avant leur publication générale afin de recueillir des retours précoces et de tester de nouvelles fonctionnalités. Vous pouvez vérifier la disponibilité de toutes les capacités et de tous les outils de Claude dans la vue d'ensemble Construire avec Claude.

Vous pouvez accéder à la plupart des fonctionnalités bêta de l'API via la propriété beta du client. Pour activer une fonctionnalité bêta particulière, vous devez ajouter l'en-tête bêta approprié au champ betas lors de la création d'un message.

Par exemple, pour activer l'édition de contexte :

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

Intégrations de plateformes

Les cinq classes de client sont toutes incluses dans le package de base anthropic :

Le client AnthropicAWS est en bêta. Passez workspace_id au constructeur ou définissez la variable d'environnement ANTHROPIC_AWS_WORKSPACE_ID.

Utilisez AnthropicBedrockMantle pour les nouveaux projets ; AnthropicBedrock reste disponible pour les applications existantes utilisant l'API InvokeModel de Bedrock.

Gestion sémantique de version

Ce package suit généralement les conventions SemVer, bien que certains changements non rétrocompatibles puissent être publiés en tant que versions mineures :

1. Les changements qui n'affectent que les types statiques, sans casser le comportement à l'exécution.
2. Les changements apportés aux éléments internes de la bibliothèque qui sont techniquement publics mais ne sont ni destinés ni documentés pour un usage externe.
3. Les changements qui ne devraient pas affecter la grande majorité des utilisateurs en pratique.

Déterminer la version installée

Si vous avez effectué la mise à niveau vers la dernière version mais ne voyez pas les nouvelles fonctionnalités attendues, votre environnement Python utilise probablement encore une version plus ancienne. Vous pouvez déterminer la version utilisée à l'exécution avec :

print(anthropic.__version__)

Ressources supplémentaires

• Dépôt GitHub
• Référence de l'API
• Messages en streaming
• Utilisation d'outils avec Claude