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.
pip install anthropicPara 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]"Se requiere Python 3.9 o posterior.
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.
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())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())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)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).
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) # 10Este 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.
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.
Este SDK proporciona soporte para la API de Message Batches bajo client.messages.batches.
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"}],
},
},
]
)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)Los parámetros de solicitud que corresponden a cargas de archivos se pueden pasar en muchas formas diferentes:
PathLike (por ejemplo, pathlib.Path)(filename, content, content_type)BinaryIOfrom 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.
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 |
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_018EeWyXxfu5pfWkrYcMdjWGA 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.
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",
)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.
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.
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_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.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 del cliente o por solicitud.
Sobrescribir los encabezados predeterminados puede resultar en tipos incorrectos y otros comportamientos inesperados o indefinidos en el SDK.
# Establece encabezados predeterminados para todas las solicitudes en el cliente
client = Anthropic(
default_headers={"anthropic-version": "My-Custom-Value"},
)
# O anúlalos 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"},
)Los parámetros de solicitud anidados son TypedDicts. Las respuestas son modelos 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.
Para convertir un modelo Pydantic a 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()En las respuestas, puedes distinguir entre campos que son explícitamente null y campos que no fueron devueltos (faltantes):
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")Se puede acceder a la Response "sin procesar" devuelta por httpx a través de la propiedad .with_raw_response en el 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.
El enfoque .with_raw_response lee de forma anticipada el cuerpo completo de la respuesta cuando haces la solicitud. Para transmitir el 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 manera confiable.
El SDK usa el módulo logging de la biblioteca estándar.
Puedes habilitar el registro estableciendo la variable de entorno ANTHROPIC_LOG en debug o info:
export ANTHROPIC_LOG=debugEsta 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.
Para hacer 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 hacer estas solicitudes.
import httpx
response = client.post(
"/foo",
cast_to=httpx.Response,
body={"my_param": True},
)
print(response.json())Si deseas enviar explícitamente un parámetro adicional, puedes hacerlo con las opciones de solicitud extra_query, extra_body y extra_headers.
Los parámetros extra_ sobrescriben los parámetros documentados del mismo nombre. Por razones de seguridad, asegúrate de que estos métodos solo se usen con datos de entrada confiables.
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 Pydantic como un dict con response.model_extra.
Puedes sobrescribir directamente el cliente httpx para personalizarlo según tu caso de uso, incluido el soporte para proxies y transportes:
import httpx
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=httpx.HTTPTransport(local_address="0.0.0.0"),
),
)También puedes personalizar el cliente por solicitud usando with_options():
client.with_options(http_client=DefaultHttpxClient(...))Usa DefaultHttpxClient y DefaultAsyncHttpxClient en lugar de httpx.Client y httpx.AsyncClient sin procesar para asegurar que se preserve la configuración predeterminada del SDK (como los tiempos de espera y los límites de conexión).
De forma predeterminada, la biblioteca cierra las conexiones HTTP subyacentes cada vez que el cliente es recolectado por el recolector de basura. Puedes cerrar manualmente el cliente usando el método .close() si lo deseas, o con un administrador de contexto que se cierra al salir.
with Anthropic() as client:
message = client.messages.create(...)
# El cliente HTTP se cierra automáticamenteLas funciones beta están disponibles antes del lanzamiento general para obtener comentarios tempranos 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 funciones beta de la API a través de la propiedad beta del cliente. Para habilitar una función beta en particular, debes agregar el encabezado beta apropiado al campo betas al crear un mensaje.
Por ejemplo, para usar la API de Files:
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"],
)Para guías detalladas de configuración de plataforma con ejemplos de código, consulta:
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 permanece para aplicaciones existentes que usan la API InvokeModel de Bedrock.
Este paquete generalmente sigue las convenciones de SemVer, aunque ciertos cambios incompatibles con versiones anteriores pueden publicarse como versiones menores:
Si has actualizado a la última versión pero no ves las nuevas funciones que esperabas, es probable que tu entorno de Python todavía esté usando una versión anterior. Puedes determinar la versión que se está usando en tiempo de ejecución con:
print(anthropic.__version__)Was this page helpful?