tessl/pypi-replicate

Python client for Replicate

—

Pending

Overview

Eval results

Files

Client & Authentication

Name: tessl/pypi-replicate
Author: tessl

Client configuration, authentication management, and HTTP transport customization for the Replicate API.

Capabilities

Client Initialization

Create and configure Replicate API clients with custom authentication, base URLs, timeouts, and HTTP settings.

class Client:
    def __init__(
        self,
        api_token: Optional[str] = None,
        *,
        base_url: Optional[str] = None,
        timeout: Optional[httpx.Timeout] = None,
        **kwargs
    ) -> None:
        """
        Initialize a Replicate API client.

        Parameters:
        - api_token: API token for authentication (defaults to REPLICATE_API_TOKEN env var)
        - base_url: Base URL for API requests (defaults to https://api.replicate.com)
        - timeout: HTTP timeout configuration
        - **kwargs: Additional HTTP client arguments
        """

Default Client

The package provides a default client instance used by top-level functions.

default_client: Client

Usage examples:

import replicate

# Using default client (uses REPLICATE_API_TOKEN env var)
output = replicate.run("model-name", input={"prompt": "hello"})

# Using custom client
client = replicate.Client(api_token="your-token-here")
output = client.run("model-name", input={"prompt": "hello"})

# Custom base URL and timeout
import httpx
client = replicate.Client(
    api_token="your-token",
    base_url="https://custom-api.example.com",
    timeout=httpx.Timeout(30.0)
)

Authentication Methods

Multiple authentication approaches depending on your use case.

Environment Variable (Recommended)

import os
os.environ['REPLICATE_API_TOKEN'] = 'your-token-here'

import replicate
output = replicate.run("model-name", input={})

Client Constructor

from replicate import Client

client = Client(api_token="your-token-here")
output = client.run("model-name", input={})

Cog Context Integration

For models running within Cog, the client automatically detects context-provided tokens:

# In a Cog prediction, token is automatically available
import replicate
output = replicate.run("model-name", input={})

Client Methods

All core functionality available on Client instances.

class Client:
    def run(
        self,
        ref: str,
        input: Optional[Dict[str, Any]] = None,
        *,
        use_file_output: Optional[bool] = True,
        **params
    ) -> Union[Any, Iterator[Any]]:
        """Run a model and wait for its output."""

    async def async_run(
        self,
        ref: str,
        input: Optional[Dict[str, Any]] = None,
        *,
        use_file_output: Optional[bool] = True,
        **params
    ) -> Union[Any, AsyncIterator[Any]]:
        """Run a model and wait for its output asynchronously."""

    def stream(
        self,
        ref: str,
        *,
        input: Optional[Dict[str, Any]] = None,
        use_file_output: Optional[bool] = True,
        **params
    ) -> Iterator[ServerSentEvent]:
        """Stream a model's output."""

    async def async_stream(
        self,
        ref: str,
        input: Optional[Dict[str, Any]] = None,
        *,
        use_file_output: Optional[bool] = True,
        **params
    ) -> AsyncIterator[ServerSentEvent]:
        """Stream a model's output asynchronously."""

Client Properties (Namespaces)

Access to all API functionality through organized namespaces.

class Client:
    @property
    def accounts(self) -> Accounts:
        """Namespace for operations related to accounts."""

    @property  
    def collections(self) -> Collections:
        """Namespace for operations related to collections of models."""

    @property
    def deployments(self) -> Deployments:
        """Namespace for operations related to deployments."""

    @property
    def files(self) -> Files:
        """Namespace for operations related to files."""

    @property
    def hardware(self) -> Hardware:
        """Namespace for operations related to hardware."""

    @property
    def models(self) -> Models:
        """Namespace for operations related to models."""

    @property
    def predictions(self) -> Predictions:
        """Namespace for operations related to predictions."""

    @property
    def trainings(self) -> Trainings:
        """Namespace for operations related to trainings."""

    @property
    def webhooks(self) -> Webhooks:
        """Namespace for operations related to webhooks."""

HTTP Transport Customization

Advanced HTTP configuration with automatic retry logic and custom transport options.

class RetryTransport:
    def __init__(
        self,
        wrapped_transport: Union[httpx.BaseTransport, httpx.AsyncBaseTransport],
        *,
        max_attempts: int = 10,
        max_backoff_wait: float = 60,
        backoff_factor: float = 0.1,
        jitter_ratio: float = 0.1,
        retryable_methods: Optional[Iterable[str]] = None,
        retry_status_codes: Optional[Iterable[int]] = None
    ) -> None:
        """
        Custom HTTP transport with automatic retry using exponential backoff.

        Parameters:
        - wrapped_transport: Underlying HTTP transport
        - max_attempts: Maximum retry attempts
        - max_backoff_wait: Maximum wait time between retries
        - backoff_factor: Exponential backoff factor
        - jitter_ratio: Jitter ratio for randomization (0-0.5)
        - retryable_methods: HTTP methods to retry (defaults to GET, PUT, DELETE, etc.)
        - retry_status_codes: Status codes to retry (defaults to 429, 503, 504)
        """

Account Management

Access account information and settings.

class Accounts:
    def current(self) -> Account:
        """
        Get the current account information.

        Returns:
        Account object with username, name, and type information
        """

    async def async_current(self) -> Account:
        """
        Get the current account information asynchronously.

        Returns:
        Account object with username, name, and type information
        """

class Account:
    type: Literal["user", "organization"]
    """The type of account."""

    username: str
    """The username of the account."""

    name: str
    """The name of the account."""

    github_url: Optional[str]
    """The GitHub URL of the account."""

Usage examples:

import replicate

# Get current account info
account = replicate.accounts.current()
print(f"Account: {account.username} ({account.type})")
print(f"Name: {account.name}")

# Using custom client
client = replicate.Client(api_token="your-token")
account = client.accounts.current()

Configuration Options

Environment variables for default configuration:

REPLICATE_API_TOKEN: API authentication token
REPLICATE_BASE_URL: Custom API base URL (defaults to https://api.replicate.com)
REPLICATE_POLL_INTERVAL: Polling interval for prediction status (defaults to 0.5 seconds)

Usage examples:

import os
import httpx
from replicate import Client

# Basic configuration
os.environ['REPLICATE_API_TOKEN'] = 'your-token'
os.environ['REPLICATE_POLL_INTERVAL'] = '1.0'

# Advanced client configuration
client = Client(
    api_token="your-token",
    base_url="https://api.replicate.com",
    timeout=httpx.Timeout(connect=5.0, read=30.0),
    headers={"User-Agent": "MyApp/1.0"},
    transport=custom_transport
)

Install with Tessl CLI