or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

client.mdcollections-training.mddeployments-webhooks.mdexceptions.mdfiles.mdindex.mdmodels-predictions.md
tile.json

tessl/pypi-replicate

Python client for Replicate

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/replicate@1.0.x

To install, run

npx @tessl/cli install tessl/pypi-replicate@1.0.0

index.mddocs/

Replicate

A comprehensive Python client library for the Replicate platform, enabling developers to run AI models, manage predictions, handle file outputs, and interact with the Replicate API programmatically. It offers both synchronous and asynchronous APIs for running models locally or in the cloud, supports streaming outputs for real-time model results, provides file handling capabilities with efficient FileOutput objects for managing model-generated files, and includes comprehensive prediction management with background execution, webhooks, and cancellation support.

Package Information

  • Package Name: replicate
  • Language: Python
  • Installation: pip install replicate
  • Minimum Python Version: 3.8+

Core Imports

import replicate

For client-based usage:

from replicate import Client
from replicate.exceptions import ModelError, ReplicateError

Basic Usage

Quick Start - Run a Model

import replicate

# Set your API token as environment variable: REPLICATE_API_TOKEN=<your token>

# Run a model and get output
outputs = replicate.run(
    "black-forest-labs/flux-schnell",
    input={"prompt": "astronaut riding a rocket like a horse"}
)

# Handle file outputs
for index, output in enumerate(outputs):
    with open(f"output_{index}.webp", "wb") as file:
        file.write(output.read())

Stream Model Output

import replicate

# Stream tokens as they are generated
for event in replicate.stream(
    "meta/meta-llama-3-70b-instruct",
    input={"prompt": "Please write a haiku about llamas."}
):
    print(str(event), end="")

Background Predictions

import replicate

# Start a prediction in the background
model = replicate.models.get("kvfrans/clipdraw")
version = model.versions.get("5797a99edc939ea0e9242d5e8c9cb3bc7d125b1eac21bda852e5cb79ede2cd9b")
prediction = replicate.predictions.create(
    version=version,
    input={"prompt": "Watercolor painting of an underwater submarine"}
)

# Check status and wait for completion
print(prediction.status)  # 'starting'
prediction.wait()
print(prediction.status)  # 'succeeded'

# Access output
with open("output.png", "wb") as file:
    file.write(prediction.output.read())

Architecture

The Replicate client is built around several key components:

  • Client: Main interface for API interaction with configurable authentication and HTTP settings
  • Resources: Model, Prediction, Training, Collection, Deployment, File objects representing API entities
  • Namespaces: Organized API endpoints (models, predictions, trainings, etc.) accessed via client properties
  • FileOutput: Efficient file handling for model outputs with streaming and URL access
  • Streaming: Server-sent events for real-time model output consumption
  • Pagination: Automated handling of paginated API responses

Capabilities

Core Model Execution

Primary functions for running AI models synchronously and asynchronously, with support for streaming outputs and file handling.

def run(
    ref: str,
    input: Optional[Dict[str, Any]] = None,
    *,
    use_file_output: Optional[bool] = True,
    **params
) -> Union[Any, Iterator[Any]]: ...

async def async_run(
    ref: str,
    input: Optional[Dict[str, Any]] = None,
    *,
    use_file_output: Optional[bool] = True,
    **params
) -> Union[Any, AsyncIterator[Any]]: ...

def stream(
    ref: str,
    *,
    input: Optional[Dict[str, Any]] = None,
    use_file_output: Optional[bool] = True,
    **params
) -> Iterator[ServerSentEvent]: ...

async def async_stream(
    ref: str,
    input: Optional[Dict[str, Any]] = None,
    *,
    use_file_output: Optional[bool] = True,
    **params
) -> AsyncIterator[ServerSentEvent]: ...

Client & Authentication

Model and Prediction Management

Comprehensive management of models, versions, and prediction lifecycle including creation, monitoring, cancellation, and output retrieval.

# Namespace access
accounts: Accounts
models: Models
predictions: Predictions

# Core prediction operations
def predictions.create(
    model: Optional[str] = None,
    version: Optional[str] = None,
    input: Optional[Dict[str, Any]] = None,
    **params
) -> Prediction: ...

def predictions.get(id: str) -> Prediction: ...
def predictions.list(**params) -> Page[Prediction]: ...
def predictions.cancel(id: str) -> Prediction: ...

Models & Predictions

File Handling and Outputs

Efficient handling of file inputs and outputs with FileOutput objects that provide streaming, URL access, and metadata.

class FileOutput:
    def read() -> bytes: ...
    def __iter__() -> Iterator[bytes]: ...
    async def aread() -> bytes: ...
    async def __aiter__() -> AsyncIterator[bytes]: ...
    @property
    def url() -> str: ...

# File management namespace
files: Files

def files.create(file: BinaryIO, **params) -> File: ...
def files.get(id: str) -> File: ...
def files.list(**params) -> Page[File]: ...

File Handling

Collections and Training

Management of model collections and training operations for custom model development.

# Collection operations
collections: Collections
def collections.get(name: str) -> Collection: ...
def collections.list(**params) -> Page[Collection]: ...

# Training operations  
trainings: Trainings
def trainings.create(
    model: str,
    version: str,
    input: Dict[str, Any],
    **params
) -> Training: ...
def trainings.get(id: str) -> Training: ...
def trainings.list(**params) -> Page[Training]: ...

Collections & Training

Deployments and Webhooks

Management of model deployments and webhook configuration for production use cases.

# Deployment operations
deployments: Deployments
def deployments.get(owner: str, name: str) -> Deployment: ...
def deployments.list(**params) -> Page[Deployment]: ...

# Webhook operations
webhooks: Webhooks
def webhooks.default.secret() -> WebhookSigningSecret: ...
@staticmethod
def webhooks.validate(
    request: Optional[httpx.Request] = None,
    headers: Optional[Dict[str, str]] = None,
    body: Optional[str] = None,
    secret: Optional[WebhookSigningSecret] = None,
    tolerance: Optional[int] = None
) -> None: ...

Deployments & Webhooks

Pagination Utilities

Helper functions for handling paginated API responses.

def paginate(pages, **kwargs): ...
async def async_paginate(pages, **kwargs): ...

class Page[T]:
    results: List[T]
    next: Optional[str]
    previous: Optional[str]

Error Handling

Comprehensive exception classes for handling API errors and model failures.

class ReplicateException(Exception): ...
class ModelError(ReplicateException):
    prediction: Prediction
class ReplicateError(ReplicateException):
    type: Optional[str]
    title: Optional[str]
    status: Optional[int]
    detail: Optional[str]
    instance: Optional[str]

Error Handling

Types

from typing import Any, Dict, List, Optional, Union, Iterator, AsyncIterator
from typing_extensions import Literal

# Server-sent event for streaming
class ServerSentEvent:
    event: Optional[str]
    data: Optional[str]
    id: Optional[str]
    retry: Optional[int]

# File encoding strategies
FileEncodingStrategy = Literal["base64", "url"]

# Prediction status types
PredictionStatus = Literal["starting", "processing", "succeeded", "failed", "canceled"]

# Model visibility types  
ModelVisibility = Literal["public", "private"]