Replicate Python API SDK (beta)

Docs

Installation

pip install --pre replicate

Usage

export REPLICATE_API_TOKEN="r8_..."

import replicate
claude = replicate.use("anthropic/claude-4.5-sonnet")
seedream = replicate.use("bytedance/seedream-4")
veo = replicate.use("google/veo-3-fast")
# Enhance a simple prompt
image_prompt = claude(prompt="bananas wearing cowboy hats", system_prompt="turn prompts into image prompts")
# Generate an image from the enhanced prompt
images = seedream(prompt=image_prompt)
# Generate a video from the image
video = veo(prompt="dancing bananas", image_input=images[0])
open(video)

Initialization and authentication

import os
from replicate import Replicate
client = Replicate(
 bearer_token=os.environ.get("REPLICATE_API_TOKEN")
)

Using replicate.use()

import replicate
# Create a model function
flux_dev = replicate.use("black-forest-labs/flux-dev")
# Call it like a regular Python function
outputs = flux_dev(
 prompt="a cat wearing a wizard hat, digital art",
 num_outputs=1,
 aspect_ratio="1:1",
 output_format="webp",
)
# outputs is a list of URLPath objects that auto-download when accessed
for output in outputs:
 print(output) # e.g., Path(/tmp/a1b2c3/output.webp)

Language models with streaming

import replicate
# Create a streaming language model function
claude = replicate.use("anthropic/claude-4.5-sonnet", streaming=True)
# Stream the output
output = claude(prompt="Write a haiku about Python programming")
for chunk in output:
 print(chunk, end="", flush=True)

Chaining models

import replicate
# Create two model functions
flux_dev = replicate.use("black-forest-labs/flux-dev")
claude = replicate.use("anthropic/claude-4.5-sonnet")
# Generate an image
images = flux_dev(prompt="a mysterious ancient artifact")
# Describe the image
description = claude(
 prompt="Describe this image in detail",
 image=images[0], # Pass the first image directly
)
print(description)

Async support

import asyncio
import replicate
async def main():
 # Create an async model function
 flux_dev = replicate.use("black-forest-labs/flux-dev", use_async=True)
 # Await the result
 outputs = await flux_dev(prompt="futuristic city at sunset")
 for output in outputs:
 print(output)
asyncio.run(main())

Accessing URLs without downloading

import replicate
from replicate.lib._predictions_use import get_path_url
flux_dev = replicate.use("black-forest-labs/flux-dev")
outputs = flux_dev(prompt="a serene landscape")
for output in outputs:
 url = get_path_url(output)
 print(f"URL: {url}") # https://replicate.delivery/...

Creating predictions without waiting

import replicate
claude = replicate.use("anthropic/claude-4.5-sonnet")
# Start the prediction
run = claude.create(prompt="Explain quantum computing")
# Check logs while it's running
print(run.logs())
# Get the output when ready
result = run.output()
print(result)

Current limitations

Run a model

import replicate
output = replicate.run(
 "black-forest-labs/flux-schnell", input={"prompt": "astronaut riding a rocket like a horse"}
)
print(output)

Handling errors

import replicate
from replicate.exceptions import ModelError
try:
 output = replicate.run(
 "stability-ai/stable-diffusion-3", input={"prompt": "An astronaut riding a rainbow unicorn"}
 )
except ModelError as e:
 print(f"Prediction failed: {e}")
 # The prediction object is available as e.prediction
 print(f"Prediction ID: {e.prediction.id}")
 print(f"Status: {e.prediction.status}")

File inputs

# Using a URL
output = replicate.run(
 "andreasjansson/blip-2:f677695e5e89f8b236e52ecd1d3f01beb44c34606419bcc19345e046d8f786f9",
 input={"image": "https://example.com/image.jpg"},
)
# Using a local file
with open("path/to/image.jpg", "rb") as f:
 output = replicate.run(
 "andreasjansson/blip-2:f677695e5e89f8b236e52ecd1d3f01beb44c34606419bcc19345e046d8f786f9",
 input={"image": f},
 )

Wait parameter

# Wait up to 30 seconds
output = replicate.run("...", input={...}, wait=30)
# Don't wait at all - returns immediately
output = replicate.run("...", input={...}, wait=False)

Streaming output

import replicate
claude = replicate.use("anthropic/claude-4.5-sonnet", streaming=True)
for event in claude(input={"prompt": "Please write a haiku about streaming Python."}):
 print(str(event), end="")

Async usage

import os
import asyncio
from replicate import AsyncReplicate
replicate = AsyncReplicate(
 bearer_token=os.environ.get("REPLICATE_API_TOKEN"), # This is the default and can be omitted
)
async def main() -> None:
 prediction = await replicate.predictions.get(
 prediction_id="gm3qorzdhgbfurvjtvhg6dckhu",
 )
 print(prediction.id)
asyncio.run(main())

With aiohttp

# install from PyPI
pip install --pre replicate[aiohttp]

import asyncio
from replicate import DefaultAioHttpClient
from replicate import AsyncReplicate
async def main() -> None:
 async with AsyncReplicate(
 bearer_token="My Bearer Token",
 http_client=DefaultAioHttpClient(),
 ) as replicate:
 prediction = await replicate.predictions.get(
 prediction_id="gm3qorzdhgbfurvjtvhg6dckhu",
 )
 print(prediction.id)
asyncio.run(main())

Using types

Pagination

from replicate import Replicate
replicate = Replicate()
all_models = []
# Automatically fetches more pages as needed.
for model in replicate.models.list():
 # Do something with model here
 all_models.append(model)
print(all_models)

import asyncio
from replicate import AsyncReplicate
replicate = AsyncReplicate()
async def main() -> None:
 all_models = []
 # Iterate through items across all pages, issuing requests as needed.
 async for model in replicate.models.list():
 all_models.append(model)
 print(all_models)
asyncio.run(main())

first_page = await replicate.models.list()
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.results)}")
# Remove `await` for non-async usage.

first_page = await replicate.models.list()
print(f"next URL: {first_page.next}") # => "next URL: ..."
for model in first_page.results:
 print(model.cover_image_url)
# Remove `await` for non-async usage.

File uploads

from pathlib import Path
from replicate import Replicate
replicate = Replicate()
replicate.files.create(
 content=Path("/path/to/file"),
)

Handling errors

import replicate
from replicate import Replicate
replicate = Replicate()
try:
 replicate.predictions.create(
 input={"text": "Alice"},
 version="replicate/hello-world:5c7d5dc6dd8bf75c1acaa8565735e7986bc5b66206b55cca93cb72c9bf15ccaa",
 )
except replicate.APIConnectionError as e:
 print("The server could not be reached")
 print(e.__cause__) # an underlying Exception, likely raised within httpx.
except replicate.RateLimitError as e:
 print("A 429 status code was received; we should back off a bit.")
except replicate.APIStatusError as e:
 print("Another non-200-range status code was received")
 print(e.status_code)
 print(e.response)

Retries

from replicate import Replicate
# Configure the default for all requests:
replicate = Replicate(
 # default is 2
 max_retries=0,
)
# Or, configure per-request:
replicate.with_options(max_retries=5).predictions.create(
 input={"text": "Alice"},
 version="replicate/hello-world:5c7d5dc6dd8bf75c1acaa8565735e7986bc5b66206b55cca93cb72c9bf15ccaa",
)

Timeouts

from replicate import Replicate
# Configure the default for all requests:
replicate = Replicate(
 # 20 seconds (default is 1 minute)
 timeout=20.0,
)
# More granular control:
replicate = Replicate(
 timeout=httpx.Timeout(60.0, read=5.0, write=10.0, connect=2.0),
)
# Override per-request:
replicate.with_options(timeout=5.0).predictions.create(
 input={"text": "Alice"},
 version="replicate/hello-world:5c7d5dc6dd8bf75c1acaa8565735e7986bc5b66206b55cca93cb72c9bf15ccaa",
)

Advanced

Logging

$ export REPLICATE_LOG=info

How to tell whether None means null or missing

if response.my_field is None:
 if 'my_field' not in response.model_fields_set:
 print('Got json like {}, without a "my_field" key present at all.')
 else:
 print('Got json like {"my_field": null}.')

Accessing raw response data (e.g. headers)

from replicate import Replicate
replicate = Replicate()
response = replicate.predictions.with_raw_response.create(
 input={
 "text": "Alice"
 },
 version="replicate/hello-world:5c7d5dc6dd8bf75c1acaa8565735e7986bc5b66206b55cca93cb72c9bf15ccaa",
)
print(response.headers.get('X-My-Header'))
prediction = response.parse() # get the object that `predictions.create()` would have returned
print(prediction.id)

.with_streaming_response

with replicate.predictions.with_streaming_response.create(
 input={"text": "Alice"},
 version="replicate/hello-world:5c7d5dc6dd8bf75c1acaa8565735e7986bc5b66206b55cca93cb72c9bf15ccaa",
) as response:
 print(response.headers.get("X-My-Header"))
 for line in response.iter_lines():
 print(line)

Making custom/undocumented requests

Undocumented endpoints

import httpx
response = replicate.post(
 "/foo",
 cast_to=httpx.Response,
 body={"my_param": True},
)
print(response.headers.get("x-foo"))

Undocumented request params

Undocumented response properties

Configuring the HTTP client

import httpx
from replicate import Replicate, DefaultHttpxClient
replicate = Replicate(
 # Or use the `REPLICATE_BASE_URL` env var
 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"),
 ),
)

replicate.with_options(http_client=DefaultHttpxClient(...))

Managing HTTP resources

from replicate import Replicate
with Replicate() as replicate:
 # make requests here
 ...
# HTTP client is now closed

Versioning

Determining the installed version

import replicate
print(replicate.__version__)

Requirements

Contributing