CLI, SDKs y bibliotecasSDKs de cliente

SDK de Python

Instala y configura el SDK de Python de Anthropic con soporte para clientes síncronos y asíncronos

El SDK de Python de Anthropic proporciona acceso conveniente a la API REST de Anthropic desde aplicaciones Python. Admite operaciones tanto síncronas como asíncronas, streaming e integraciones con Amazon Bedrock, Claude Platform en AWS, Google Cloud y Microsoft Foundry.

Para la documentación de las funciones de la API con ejemplos de código, consulta la referencia de la API. Esta página cubre las funciones y la configuración del SDK específicas de Python.

Instalación

pip install anthropic

Para integraciones específicas de plataforma o un mejor rendimiento asíncrono, instala con extras:

# Para soporte de Amazon Bedrock
pip install "anthropic[bedrock]"

# Para soporte de Google Cloud
pip install "anthropic[vertex]"

# Para soporte de Claude Platform en AWS
pip install "anthropic[aws]"

# El soporte de Microsoft Foundry está incluido en el paquete base

# Para mejor rendimiento asíncrono con aiohttp
pip install "anthropic[aiohttp]"

Requisitos

Se requiere Python 3.9 o posterior.

Uso

import os
from anthropic import Anthropic

client = Anthropic(
    # Este es el valor predeterminado y se puede omitir
    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)

Considera usar python-dotenv para agregar ANTHROPIC_API_KEY="my-anthropic-api-key" a tu archivo .env de modo que tu clave de API no se almacene en el control de versiones.

Para conocer las opciones de autenticación, incluida Workload Identity Federation, consulta Autenticación.

Uso asíncrono

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

Uso de aiohttp para mejor concurrencia

Para un mejor rendimiento asíncrono, puedes usar el backend HTTP aiohttp en lugar del httpx predeterminado:

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

Respuestas en streaming

El SDK proporciona soporte para respuestas en streaming usando 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)

El cliente asíncrono usa exactamente la misma interfaz:

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)

Ayudantes de streaming

El SDK también proporciona ayudantes de streaming que usan administradores de contexto y brindan acceso al texto acumulado y al mensaje 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())

El streaming con client.messages.stream(...) expone varios ayudantes, incluida la acumulación y eventos específicos del SDK.

Alternativamente, puedes usar client.messages.create(..., stream=True), que solo devuelve un iterable de los eventos en el stream y usa menos memoria (no construye un objeto de mensaje final por ti).

Conteo de tokens

Puedes ver el uso exacto de una solicitud determinada a través de la propiedad de respuesta usage:

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

También puedes contar tokens antes de hacer una solicitud:

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

Uso de herramientas

Este SDK proporciona soporte para el uso de herramientas, también conocido como llamada a funciones. Para más detalles, consulta Uso de herramientas con Claude.

Ayudantes de herramientas

El SDK proporciona ayudantes para definir y ejecutar herramientas como funciones puras de Python. El decorador @beta_tool genera el esquema de la herramienta a partir de la firma de la función y el 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 el tool_runner para manejar automáticamente las llamadas a herramientas
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)

En cada iteración, se realiza una solicitud a la API. Si la respuesta incluye una llamada a una de las herramientas proporcionadas, la herramienta se llama automáticamente y el resultado se devuelve directamente al modelo en la siguiente iteración.

Lotes de mensajes

Este SDK proporciona soporte para la API de Message Batches bajo client.messages.batches.

Creación de un lote

Message Batches toma un arreglo de solicitudes, donde cada objeto tiene un identificador custom_id y los mismos params de solicitud que la API de Messages estándar:

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

Obtención de resultados de un lote

Una vez que un Message Batch ha sido procesado, indicado por .processing_status == 'ended', puedes acceder a los resultados 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)

Carga de archivos

Los parámetros de solicitud que corresponden a cargas de archivos se pueden pasar en muchas formas diferentes:

Un objeto PathLike (por ejemplo, pathlib.Path)
Una tupla de (filename, content, content_type)
Un objeto similar a archivo BinaryIO

from pathlib import Path
from anthropic import Anthropic

client = Anthropic()

# Subir usando una ruta de archivo
client.beta.files.upload(
    file=Path("/path/to/file"),
)

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

El cliente asíncrono usa exactamente la misma interfaz. Si pasas una instancia de PathLike, el contenido del archivo se lee de forma asíncrona automáticamente.

Manejo de errores

Cuando la biblioteca no puede conectarse a la API, o si la API devuelve un código de estado no exitoso (es decir, una respuesta 4xx o 5xx), se lanza una subclase de 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 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)

Los códigos de error son los siguientes:

Código de estado	Tipo de error
400	`BadRequestError`
401	`AuthenticationError`
403	`PermissionDeniedError`
404	`NotFoundError`
409	`ConflictError`
422	`UnprocessableEntityError`
429	`RateLimitError`
>=500	`InternalServerError`
N/A	`APIConnectionError`

IDs de solicitud

Para más información sobre la depuración de solicitudes, consulta ID de solicitud.

Todas las respuestas de objetos en el SDK proporcionan una propiedad _request_id que se agrega desde el encabezado de respuesta request-id para que puedas registrar rápidamente las solicitudes fallidas y reportarlas a 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

A diferencia de otras propiedades que usan un prefijo _, la propiedad _request_id es pública. A menos que se documente lo contrario, todas las demás propiedades, métodos y módulos con prefijo _ son privados.

Reintentos

Ciertos errores se reintentan automáticamente 2 veces de forma predeterminada, con un breve retroceso exponencial. Los errores de conexión (por ejemplo, debido a un problema de conectividad de red), 408 Request Timeout, 409 Conflict, 429 Rate Limit y los errores internos >=500 se reintentan de forma predeterminada.

Puedes usar la opción max_retries para configurar o deshabilitar esto:

# Configura el valor predeterminado para todas las solicitudes:
client = Anthropic(
    max_retries=0,  # default is 2
)

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

Tiempos de espera

De forma predeterminada, las solicitudes expiran después de 10 minutos. Puedes configurar esto con una opción timeout, que acepta un float o un objeto httpx.Timeout:

import httpx
from anthropic import Anthropic

# Configura el valor predeterminado para todas las solicitudes:
client = Anthropic(
    timeout=20.0,  # 20 seconds (default is 10 minutes)
)

# Control más granular:
client = Anthropic(
    timeout=httpx.Timeout(60.0, read=5.0, write=10.0, connect=2.0),
)

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

Al expirar el tiempo de espera, el SDK lanza un APITimeoutError.

Ten en cuenta que las solicitudes que expiran se reintentan dos veces de forma predeterminada.

Solicitudes largas

Considera usar la API de Messages con streaming para solicitudes de mayor duración.

Evita establecer un valor grande de max_tokens sin usar streaming. Algunas redes pueden descartar conexiones inactivas después de un cierto período de tiempo, lo que puede hacer que la solicitud falle o expire sin recibir una respuesta de Anthropic.

El SDK lanzará un ValueError si se espera que una solicitud sin streaming tarde más de aproximadamente 10 minutos. Pasar stream=True o sobrescribir la opción timeout a nivel del cliente o de la solicitud deshabilita este error.

Una latencia de solicitud esperada mayor que el tiempo de espera para una solicitud sin streaming resultará en que el cliente termine la conexión y reintente sin recibir una respuesta.

El SDK establece una opción de TCP socket keep-alive para reducir el impacto de los tiempos de espera de conexiones inactivas en algunas redes. Esto se puede sobrescribir pasando una opción http_client personalizada al cliente.

Paginación automática

Los métodos de listado en la API de Claude están paginados. Puedes usar la sintaxis for para iterar a través de los elementos en todas las páginas:

client = Anthropic()

all_batches = []
# Obtiene automáticamente más páginas según sea necesario.
for batch in client.messages.batches.list(limit=20):
    all_batches.append(batch)
print(all_batches)

Para iteración asíncrona:

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

Alternativamente, puedes usar los métodos .has_next_page(), .next_page_info() o .get

Instalación

Requisitos

Uso

Uso asíncrono

Uso de aiohttp para mejor concurrencia

Respuestas en streaming

Ayudantes de streaming

Conteo de tokens

Uso de herramientas

Ayudantes de herramientas

Lotes de mensajes

Creación de un lote

Obtención de resultados de un lote

Carga de archivos

Manejo de errores

IDs de solicitud

Reintentos

Tiempos de espera

Solicitudes largas

Paginación automática

Instalación

Requisitos

Uso

Uso asíncrono

Uso de aiohttp para mejor concurrencia

Respuestas en streaming

Ayudantes de streaming

Conteo de tokens

Uso de herramientas

Ayudantes de herramientas

Lotes de mensajes

Creación de un lote

Obtención de resultados de un lote

Carga de archivos

Manejo de errores

IDs de solicitud

Reintentos

Tiempos de espera

Solicitudes largas

Paginación automática