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 un acceso conveniente a la Claude API 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.
Instalación
pip install anthropicPara integraciones específicas de plataforma o un mejor rendimiento asíncrono, instala con extras:
# Para compatibilidad con Amazon Bedrock
pip install "anthropic[bedrock]"
# Para compatibilidad con Google Cloud
pip install "anthropic[vertex]"
# Para compatibilidad con Claude Platform en AWS
pip install "anthropic[aws]"
# La compatibilidad con Microsoft Foundry está incluida en el paquete base
# Para un mejor rendimiento asíncrono con aiohttp
pip install "anthropic[aiohttp]"Requisitos
Se requiere Python 3.10 o posterior. Si estás actualizando desde una versión 0.x del SDK, consulta la guía de migración a v1 para ver la lista de cambios incompatibles.
Uso
import os
from anthropic import Anthropic
client = Anthropic(
# Este es el valor predeterminado y puede omitirse
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)Para conocer las opciones de autenticación, incluida Workload Identity Federation, consulta Autenticación. Si tu clave de API es una clave personal o de cuenta de servicio con acceso a múltiples espacios de trabajo, establece el ID del espacio de trabajo en el encabezado de solicitud anthropic-workspace-id; Seleccionar un espacio de trabajo muestra la opción por solicitud para este SDK.
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 una mejor concurrencia
Para un mejor rendimiento asíncrono, puedes usar el backend HTTP aiohttp en lugar del httpx2 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" (transmisión en tiempo real) usando "Server-Sent Events" (eventos enviados por el servidor), o 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, incluyendo 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 del 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 realizar una solicitud:
count = client.messages.count_tokens(
model="claude-opus-5", messages=[{"role": "user", "content": "Hello, world"}]
)
print(count.input_tokens) # 10Uso de herramientas
Este SDK proporciona soporte para "tool use" (uso de herramientas), también conocido como llamada de 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 su 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 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 el procesamiento por lotes en client.messages.batches.
Creación de un lote
Message Batches recibe un arreglo de solicitudes, donde cada objeto tiene un identificador custom_id y los mismos params de solicitud que la Messages API 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 lote de mensajes ha sido procesado, lo cual se indica con .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 de muchas formas diferentes:
- Un objeto
PathLike(por ejemplo,pathlib.Path) - Una tupla de
(filename, content, content_type) - Un objeto tipo archivo
BinaryIO
from pathlib import Path
from anthropic import Anthropic
client = Anthropic()
# Subir usando una ruta de archivo
client.files.upload(
file=Path("/path/to/file"),
)
# Subir usando bytes
client.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 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)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 objeto en el SDK proporcionan una propiedad _request_id que se agrega a partir del 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_018EeWyXxfu5pfWkrYcMdjWGReintentos
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 bien, configúralo 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 agotan su tiempo de espera después de 10 minutos. Puedes configurar esto con la opción timeout, que acepta un float o un objeto httpx2.Timeout:
import httpx2
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=httpx2.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 agotarse el tiempo de espera, el SDK lanza un APITimeoutError.
Ten en cuenta que las solicitudes que agotan su tiempo de espera se reintentan dos veces de forma predeterminada.
Solicitudes largas
Evita establecer un valor grande de max_tokens sin usar streaming. Algunas redes pueden descartar conexiones inactivas después de cierto período de tiempo, lo que puede hacer que la solicitud falle o agote su tiempo de espera 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 de cliente o de solicitud deshabilita este error.
Una latencia de solicitud esperada mayor que el tiempo de espera para una solicitud sin streaming hará 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 por 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 Claude API están paginados. Puedes usar la sintaxis for para iterar a través de los elementos de 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_next_page() para un control más granular al trabajar con páginas:
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)}")
# Elimina `await` para uso no asíncrono.O trabajar directamente con los datos devueltos:
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)
# Elimina `await` para uso no asíncrono.Encabezados predeterminados
El SDK envía automáticamente el encabezado anthropic-version establecido en 2023-06-01.
Si lo necesitas, puedes sobrescribirlo estableciendo encabezados predeterminados en el objeto cliente o por solicitud.
# Establece encabezados predeterminados para todas las solicitudes del cliente
client = Anthropic(
default_headers={"anthropic-version": "My-Custom-Value"},
)
# O sobrescríbelos por solicitud
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 de tipos
Parámetros de solicitud
Los parámetros de solicitud anidados son TypedDicts. Las respuestas son modelos de Pydantic que también tienen métodos auxiliares para cosas como serializar de vuelta a JSON (v1, v2).
Las solicitudes y respuestas tipadas proporcionan autocompletado y documentación dentro de tu editor. Si deseas ver errores de tipo en VS Code para ayudar a detectar errores antes, establece python.analysis.typeCheckingMode en basic.
Modelos de respuesta
Para convertir un modelo de Pydantic en un diccionario, usa los métodos auxiliares:
message = client.messages.create(...)
# Convertir a cadena JSON
json_str = message.to_json()
# Convertir a diccionario
data = message.to_dict()Manejo de campos null frente a campos ausentes
En las respuestas, puedes distinguir entre campos que son explícitamente null y campos que no fueron devueltos (ausentes):
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")Uso avanzado
Acceso a los datos de respuesta sin procesar (por ejemplo, encabezados)
Se puede acceder a la Response "sin procesar" devuelta por httpx2 a través de la propiedad .with_raw_response del cliente. Esto es útil para acceder a los encabezados de respuesta u otros metadatos:
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)Estos métodos devuelven un objeto APIResponse. En el cliente asíncrono devuelven un AsyncAPIResponse, y .parse(), .read(), .text() y .json() deben esperarse con await.
Streaming del cuerpo de la respuesta
El enfoque .with_raw_response lee de forma anticipada el cuerpo completo de la respuesta cuando realizas la solicitud. Para hacer streaming del cuerpo de la respuesta en su lugar, usa .with_streaming_response, que requiere un administrador de contexto y solo lee el cuerpo de la respuesta una vez que llamas a .read(), .text(), .json(), .iter_bytes(), .iter_text(), .iter_lines() o .parse(). En el cliente asíncrono, estos son métodos asíncronos.
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)El administrador de contexto es necesario para que la respuesta se cierre de forma confiable.
Registro de logs
El SDK usa el módulo logging de la biblioteca estándar.
Puedes habilitar el registro de logs estableciendo la variable de entorno ANTHROPIC_LOG en debug o info:
export ANTHROPIC_LOG=debugRealización de solicitudes personalizadas/no documentadas
Esta biblioteca está tipada para un acceso conveniente a la API documentada. Si necesitas acceder a endpoints, parámetros o propiedades de respuesta no documentados, la biblioteca aún se puede usar.
Endpoints no documentados
Para realizar solicitudes a endpoints no documentados, puedes usar client.get, client.post y otros verbos HTTP. Las opciones del cliente, como los reintentos, se respetan al realizar estas solicitudes.
import httpx2
response = client.post(
"/foo",
cast_to=httpx2.Response,
body={"my_param": True},
)
print(response.json())Parámetros de solicitud no documentados
Si deseas enviar explícitamente un parámetro adicional, puedes hacerlo con las opciones de solicitud extra_query, extra_body y extra_headers.
Propiedades de respuesta no documentadas
Para acceder a propiedades de respuesta no documentadas, puedes acceder a los campos adicionales como response.unknown_prop. También puedes obtener todos los campos adicionales del modelo de Pydantic como un dict con response.model_extra.
Configuración del cliente HTTP
El SDK envía solicitudes con httpx2, un fork de httpx compatible a nivel de API. Para personalizar el cliente HTTP, incluyendo proxies y transportes, pasa tu propio cliente httpx2 como http_client:
import httpx2
from anthropic import Anthropic, DefaultHttpxClient
client = Anthropic(
# O usa la variable de entorno `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"),
),
)También puedes personalizar el cliente por solicitud usando with_options():
client.with_options(http_client=DefaultHttpxClient(...))Las herramientas de trazado y simulación que parchean httpx directamente, como HTTPXClientInstrumentor de OpenTelemetry, la integración httpx de Sentry, respx o pytest-httpx, no ven las solicitudes del SDK de forma predeterminada. Para usarlas, llama a httpx2.alias_httpx() una vez al inicio, antes de que cualquier cosa importe httpx. Esto hace que import httpx se resuelva a httpx2 para todo el proceso.
Administración de recursos HTTP
De forma predeterminada, la biblioteca cierra las conexiones HTTP subyacentes cuando el cliente es recolectado por el recolector de basura. Puedes cerrar el cliente manualmente usando el método .close() si lo deseas, o con un administrador de contexto que lo cierra al salir.
with Anthropic() as client:
message = client.messages.create(...)
# El cliente HTTP se cierra automáticamenteFuncionalidades beta
Las funcionalidades beta están disponibles antes del lanzamiento general para obtener retroalimentación temprana y probar nuevas funcionalidades. Puedes verificar la disponibilidad de todas las capacidades y herramientas de Claude en la descripción general de construir con Claude.
Puedes acceder a la mayoría de las funcionalidades beta de la API a través de la propiedad beta del cliente. Para habilitar una funcionalidad beta en particular, debes agregar el encabezado beta correspondiente al campo betas al crear un mensaje.
Por ejemplo, para habilitar la edición de contexto:
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"],
)Integraciones de plataforma
Las cinco clases de cliente están incluidas en el paquete base anthropic:
| Proveedor | Cliente | Dependencias adicionales |
|---|---|---|
| Agent Platform | from anthropic import AnthropicVertex | pip install "anthropic[vertex]" |
| Bedrock | from anthropic import AnthropicBedrockMantle | pip install "anthropic[bedrock]" |
Bedrock (ruta bedrock-runtime) | from anthropic import AnthropicBedrock | pip install "anthropic[bedrock]" |
| Claude Platform en AWS | from anthropic import AnthropicAWS | pip install "anthropic[aws]" |
| Foundry | from anthropic import AnthropicFoundry | Ninguna |
El cliente AnthropicAWS está en beta. Pasa workspace_id al constructor o establece la variable de entorno ANTHROPIC_AWS_WORKSPACE_ID.
Usa AnthropicBedrockMantle para proyectos nuevos; AnthropicBedrock se mantiene para aplicaciones existentes que usan la API InvokeModel de Bedrock.
Versionado semántico
Este paquete generalmente sigue las convenciones de SemVer, aunque ciertos cambios incompatibles con versiones anteriores pueden publicarse como versiones menores:
- Cambios que solo afectan a los tipos estáticos, sin romper el comportamiento en tiempo de ejecución.
- Cambios en los componentes internos de la biblioteca que son técnicamente públicos pero no están destinados ni documentados para uso externo.
- Cambios que no se espera que afecten a la gran mayoría de los usuarios en la práctica.
Determinación de la versión instalada
Si has actualizado a la última versión pero no ves las nuevas funcionalidades que esperabas, es probable que tu entorno de Python aún esté usando una versión anterior. Puedes determinar la versión que se está usando en tiempo de ejecución con:
print(anthropic.__version__)Recursos adicionales
Was this page helpful?