### Set up replicate-python Development Environment

Source: https://github.com/replicate/replicate-python/blob/main/CONTRIBUTING.md

The Python project is managed using `uv`. Run the `./script/setup` command to install uv and the project's dependencies. This script is essential for setting up a local development environment.

```console
./script/setup
```

--------------------------------

### Create a Model - Python

Source: https://github.com/replicate/replicate-python/blob/main/README.md

Provides a Python example for creating a new model on Replicate. It specifies the owner, model name, visibility (public/private), and the required hardware SKU for running the model.

```python
import replicate

model = replicate.models.create(
    owner="your-username",
    name="my-model",
    visibility="public",
    hardware="gpu-a40-large"
)
```

--------------------------------

### Get Model Details and Versions (Python)

Source: https://context7.com/replicate/replicate-python/llms.txt

Provides examples for retrieving detailed information about a specific Replicate model and its versions. It covers getting a model by its owner and name, accessing its latest version, and inspecting the model's input/output schema using OpenAPI specifications. The snippet also shows how to list all versions of a model.

```python
import replicate

# Get model by owner/name
model = replicate.models.get("stability-ai/sdxl")
print(f"Model: {model.id}")
print(f"Description: {model.description}")
print(f"Visibility: {model.visibility}")
print(f"Run count: {model.run_count}")

# Get alternate form
model = replicate.models.get("stability-ai", "sdxl")

# Access latest version
if model.latest_version:
    version = model.latest_version
    print(f"Latest version: {version.id}")
    print(f"Created: {version.created_at}")
    print(f"Cog version: {version.cog_version}")

    # Inspect input/output schema
    schema = version.openapi_schema
    print("Input parameters:")
    for param, details in schema["components"]["schemas"]["Input"]["properties"].items():
        print(f"  {param}: {details.get('type')} - {details.get('description')}")

# List all versions
for version in model.versions.list():
    print(f"Version {version.id} from {version.created_at}")
```

--------------------------------

### Start and Monitor Model Training (Python)

Source: https://context7.com/replicate/replicate-python/llms.txt

Shows how to initiate a model training job on Replicate, specifying the base model, version, input data, and destination for the trained model. It includes monitoring the training status, retrieving logs, and handling success or failure.

```python
import replicate
import time

# Start a training job
training = replicate.trainings.create(
    model="stability-ai/sdxl",
    version="39ed52f2a78e934b3ba6e2a89f5b1c712de7dfea535525255b1aa35c5565e08b",
    input={
        "input_images": "https://example.com/training-images.zip",
        "token_string": "TOK",
        "caption_prefix": "a photo of TOK",
        "max_train_steps": 1000,
        "learning_rate": 1e-6,
        "train_batch_size": 4
    },
    destination="your-username/my-fine-tuned-model",  # Where to save
    webhook="https://example.com/webhook",
    webhook_completed="https://example.com/training-complete"
)

print(f"Training started: {training.id}")
print(f"Status: {training.status}")

# Monitor training progress
while training.status not in ["succeeded", "failed", "canceled"]:
    time.sleep(10)
    training.reload()
    print(f"Status: {training.status}")

    # Show training logs
    if training.logs:
        lines = training.logs.split('\n')
        print(lines[-5:])  # Last 5 lines

# Check results
if training.status == "succeeded":
    print(f"Training completed!")
    print(f"New version available at: {training.output['version']}")

    # Use the trained model
    output = replicate.run(
        f"your-username/my-fine-tuned-model",
        input={"prompt": "TOK in space"}
    )
elif training.status == "failed":
    print(f"Training failed: {training.error}")

# Cancel training
if training.status == "processing":
    training.cancel()
    print("Training canceled")

# List all trainings
page = replicate.trainings.list()
for t in page:
    print(f"{t.id}: {t.status}")
```

--------------------------------

### Install Replicate Python Package via pip

Source: https://github.com/replicate/replicate-python/blob/main/README.md

Installation command to add the Replicate Python client to your project. Requires Python 3.8 or higher.

```shell
pip install replicate
```

--------------------------------

### List All Models - Python

Source: https://github.com/replicate/replicate-python/blob/main/README.md

Demonstrates how to list all models available on Replicate using `replicate.models.list()`. It includes examples of both automatic pagination with `replicate.paginate` and manual pagination using the `next` cursor for fetching models in pages.

```python
replicate.models.list()
```

```python
# Automatic pagination using `replicate.paginate` (recommended)
models = []
for page in replicate.paginate(replicate.models.list):
    models.extend(page.results)
    if len(models) > 100:
        break
```

```python
# Manual pagination using `next` cursors
page = replicate.models.list()
while page:
    models.extend(page.results)
    if len(models) > 100:
          break
    page = replicate.models.list(page.next) if page.next else None
```

--------------------------------

### Stream Response with FastAPI

Source: https://github.com/replicate/replicate-python/blob/main/README.md

Demonstrates how to use Replicate file output with FastAPI's StreamingResponse for asynchronous streaming. This example defines a FastAPI GET endpoint that runs a Replicate model and returns the output as a stream.

```python
@app.get("/")
async def main():
    output = replicate.run("black-forest-labs/flux-schnell", input={...}, use_file_output =True)
    return StreamingResponse(output)
```

--------------------------------

### List Available Hardware with Replicate Python SDK

Source: https://context7.com/replicate/replicate-python/llms.txt

Demonstrates how to query the available hardware options for running Replicate models and deployments. It shows how to list all hardware options, access their SKUs and names, and provides common hardware SKU examples. It also illustrates how to specify hardware when creating a deployment and an asynchronous version of the list function.

```python
import replicate

# List all hardware options
hardware_list = replicate.hardware.list()

for hw in hardware_list:
    print(f"SKU: {hw.sku}")
    print(f"Name: {hw.name}")
    print()

# Common hardware SKUs:
# - "cpu" - CPU only
# - "gpu-t4" - NVIDIA T4 (entry level)
# - "gpu-a40-small" - NVIDIA A40 (small)
# - "gpu-a40-large" - NVIDIA A40 (large)
# - "gpu-a100" - NVIDIA A100

# Use hardware when creating deployments
# deployment = replicate.deployments.create(
#     name="my-deployment",
#     model="stability-ai/sdxl",
#     version="39ed52f2...",
#     hardware="gpu-a40-large",  # Use specific hardware
#     min_instances=1,
#     max_instances=3
# )

# Async version
async def list_hardware():
    hardware = await replicate.hardware.async_list()
    for hw in hardware:
        print(f"{hw.sku}: {hw.name}")
```

--------------------------------

### Get Account Information with Replicate Python SDK

Source: https://context7.com/replicate/replicate-python/llms.txt

Provides code examples for retrieving information about the authenticated Replicate account. It shows how to get the current user or organization account details, including its type, username, name, and GitHub URL. An asynchronous version of the function is also included.

```python
import replicate

# Get current account
account = replicate.accounts.current()

print(f"Type: {account.type}")  # "user" or "organization"
print(f"Username: {account.username}")
print(f"Name: {account.name}")

if account.github_url:
    print(f"GitHub: {account.github_url}")

# Async version
async def get_account():
    account = await replicate.accounts.async_current()
    print(f"Logged in as: {account.username}")
```

--------------------------------

### List Predictions with Replicate Python

Source: https://github.com/replicate/replicate-python/blob/main/README.md

This Python example demonstrates how to list all predictions associated with your Replicate account. It shows the basic usage of `replicate.predictions.list()` and how to access subsequent pages of results using the `next` property. This is useful for monitoring past and ongoing predictions.

```python
replicate.predictions.list()
# [<Prediction: 8b0ba5ab4d85>, <Prediction: 494900564e8c>]

page1 = replicate.predictions.list()

if page1.next:
    page2 = replicate.predictions.list(page1.next)
```

--------------------------------

### List and Search Models (Python)

Source: https://context7.com/replicate/replicate-python/llms.txt

Shows how to discover and explore models available in Replicate's catalog. It demonstrates paginated listing of models, manual pagination control using cursors, and an automatic pagination helper. The snippet also includes searching for models by query and an example of asynchronous pagination.

```python
import replicate

# List all models (paginated)
page = replicate.models.list()
for model in page.results:
    print(f"{model.id}: {model.description}")
    print(f"  Runs: {model.run_count}")
    if model.latest_version:
        print(f"  Latest version: {model.latest_version.id}")

# Navigate pagination manually
if page.next:
    next_page = replicate.models.list(cursor=page.next)

# Automatic pagination helper
for page in replicate.paginate(replicate.models.list):
    for model in page:
        print(model.id)
    # Can break early if you find what you need
    if len(collected) > 100:
        break

# Search for models
results = replicate.models.search(query="stable diffusion")
for model in results:
    print(f"Found: {model.id}")

# Async pagination
async def list_all_models():
    async for page in replicate.async_paginate(replicate.models.async_list):
        for model in page:
            print(model.id)
```

--------------------------------

### List and Get Replicate Collections with Python

Source: https://context7.com/replicate/replicate-python/llms.txt

Shows how to list all available Replicate collections and retrieve details for a specific collection. It demonstrates iterating through collections, accessing collection properties like name and description, and listing models within a collection. Also shows how to run a model directly from a collection.

```python
import replicate

# List all collections
page = replicate.collections.list()
for collection in page:
    print(f"{collection.name} ({collection.slug})")
    print(f"  {collection.description}")
    print(f"  {len(collection.models)} models")

# Get specific collection
collection = replicate.collections.get("text-to-image")
print(f"Collection: {collection.name}")

# Iterate through models in collection
for model in collection:
    print(f"  {model.id}: {model.description[:100]}")

# Access models by index
first_model = collection[0]
print(f"First model: {first_model.id}")

# Slice collections
top_three = collection[0:3]

# Check collection size
print(f"Total models: {len(collection)}")

# Run model from collection
output = replicate.run(
    collection[0].id,
    input={"prompt": "test"}
)
```

--------------------------------

### Get and Reload Model Versions (Python)

Source: https://context7.com/replicate/replicate-python/llms.txt

Demonstrates how to retrieve a specific model version by its ID and how to reload a model object to fetch the latest data. This is useful for ensuring you are working with the most up-to-date model information.

```python
import replicate

# Assuming 'model' is already initialized, e.g., model = replicate.models.get("stability-ai/sdxl")

# Get specific version
version = model.versions.get("39ed52f2a78e934b3ba6e2a89f5b1c712de7dfea535525255b1aa35c5565e08b")

# Reload model to get latest data
model.reload()
```

--------------------------------

### Handle Progressive Iterator Outputs in Replicate Python

Source: https://context7.com/replicate/replicate-python/llms.txt

Demonstrates how to handle models that return results progressively using iterators. It emphasizes the importance of not using `wait=True` when creating predictions for such models, as this would block until all outputs are ready. The example shows how to get the prediction object immediately and process results as they become available.

```python
import replicate

# Models with x-cog-array-type: iterator return progressive results
# Don't use wait=True, as it waits for ALL outputs before returning

# Get prediction object instead of blocking
prediction = replicate.predictions.create(
    model="pixray/text2image",
    input={"prompts": "cyberpunk city"}
)

```

--------------------------------

### Monitor Replicate Prediction Progress with Python

Source: https://context7.com/replicate/replicate-python/llms.txt

Shows how to monitor the progress of a Replicate prediction in real-time. It includes creating a prediction, polling its status, displaying parsed progress percentages and current/total steps from tqdm-style logs, and printing recent log entries. The example also covers accessing final status, metrics, and calculating the prediction duration.

```python
import replicate
import time

# Create prediction
prediction = replicate.predictions.create(
    model="stability-ai/sdxl",
    input={"prompt": "detailed illustration"}
)

# Monitor with progress information
while prediction.status not in ["succeeded", "failed", "canceled"]:
    time.sleep(2)
    prediction.reload()

    # Get parsed progress (from tqdm-style logs)
    if prediction.progress:
        print(f"Progress: {prediction.progress.percentage}%")
        print(f"Current: {prediction.progress.current}/{prediction.progress.total}")

    # Show recent logs
    if prediction.logs:
        recent_logs = prediction.logs.split('\n')[-3:]
        for line in recent_logs:
            print(line)

# Final status
print(f"Final status: {prediction.status}")
if prediction.status == "succeeded":
    print(f"Completed in {prediction.metrics}")

# Access timing information
if prediction.started_at and prediction.completed_at:
    from datetime import datetime
    started = datetime.fromisoformat(prediction.started_at.replace('Z', '+00:00'))
    completed = datetime.fromisoformat(prediction.completed_at.replace('Z', '+00:00'))
    duration = (completed - started).total_seconds()
    print(f"Duration: {duration:.2f} seconds")

```

--------------------------------

### Create Prediction and Handle Completion (Python)

Source: https://context7.com/replicate/replicate-python/llms.txt

Demonstrates how to create a prediction asynchronously without waiting for its completion. It includes optional webhook notifications for status updates and manual polling using `prediction.reload()`. The example also shows how to retrieve, cancel, and wait for prediction results, including handling success and failure outcomes.

```python
import replicate
import time

# Create prediction without waiting for completion
prediction = replicate.predictions.create(
    model="stability-ai/sdxl",
    input={"prompt": "futuristic vehicle design"},
    webhook="https://example.com/webhook",  # Optional: get notified when done
    webhook_events_filter=["completed"]  # Only notify on completion
)

print(f"Created prediction: {prediction.id}")
print(f"Status: {prediction.status}")

# Manual polling
while prediction.status not in ["succeeded", "failed", "canceled"]:
    time.sleep(1)
    prediction.reload()
    print(f"Status: {prediction.status}")
    if prediction.logs:
        print(f"Logs: {prediction.logs[-100:]}")  # Last 100 chars

# Or use wait() method
prediction.wait()

# Handle results
if prediction.status == "succeeded":
    print(f"Output: {prediction.output}")
elif prediction.status == "failed":
    print(f"Error: {prediction.error}")

# Get prediction by ID
retrieved = replicate.predictions.get(prediction.id)

# Cancel running prediction
if prediction.status == "processing":
    prediction.cancel()

# Use wait parameter to keep connection open (preferred for fast models)
output = replicate.run(
    "stability-ai/sdxl",
    input={"prompt": "sunset"},
    wait=True  # Blocks up to 60 seconds, default in run()
)
```

--------------------------------

### Stream Response with Django

Source: https://github.com/replicate/replicate-python/blob/main/README.md

Provides an example of how to integrate Replicate file output with Django's HttpResponse for streaming responses. It shows a Django view function that runs a Replicate model and returns its output as an image/webp stream. The `@condition` decorator is used for caching control.

```python
@condition(etag_func=None)
def stream_response(request):
    output = replicate.run("black-forest-labs/flux-schnell", input={...}, use_file_output =True)
    return HttpResponse(output, content_type='image/webp')
```

--------------------------------

### Get Output from Running Model with Replicate Python

Source: https://github.com/replicate/replicate-python/blob/main/README.md

This Python code demonstrates how to retrieve output from a model while it is still running using `replicate.run`. It iterates over the output stream, saving each generated image to a file. This allows for processing intermediate results or displaying progress during generation.

```python
iterator = replicate.run(
    "pixray/text2image:5c347a4bfa1d4523a58ae614c2194e15f2ae682b57e3797a5bb468920aa70ebf",
    input={"prompts": "san francisco sunset"}
)

for index, image in enumerate(iterator):
    with open(f"file_{index}.png", "wb") as file:
        file.write(image.read())
```

--------------------------------

### Stream Prediction Output with Replicate Python

Source: https://github.com/replicate/replicate-python/blob/main/README.md

This example shows how to stream the output of a specific prediction initiated via the Replicate API. It first creates a prediction with streaming enabled and then iterates over the prediction's stream to print output as it becomes available. This approach is useful when you need the prediction ID before processing the output.

```python
prediction = replicate.predictions.create(
    model="meta/meta-llama-3-70b-instruct",
    input={"prompt": "Please write a haiku about llamas."},
    stream=True,
)

for event in prediction.stream():
    print(str(event), end="")
```

--------------------------------

### Run Model with Webhook Notification using Replicate Python

Source: https://github.com/replicate/replicate-python/blob/main/README.md

This Python example demonstrates how to run a model and receive a webhook notification upon its completion. It configures a prediction to send an event to a specified webhook URL when the 'completed' event occurs. This is ideal for event-driven architectures where external systems need to be notified without constant polling.

```python
model = replicate.models.get("ai-forever/kandinsky-2.2")
version = model.versions.get("ea1addaab376f4dc227f5368bbd8eff901820fd1cc14ed8cad63b29249e9d463")
prediction = replicate.predictions.create(
    version=version,
    input={"prompt":"Watercolor painting of an underwater submarine"},
    webhook="https://example.com/your-webhook",
    webhook_events_filter=["completed"]
)
```

--------------------------------

### Validate Webhook Signatures (Python with Flask)

Source: https://context7.com/replicate/replicate-python/llms.txt

Provides a basic Flask application setup for validating incoming webhook requests from Replicate using HMAC signatures. This is crucial for ensuring the authenticity and integrity of webhook data.

```python
import replicate
from replicate.exceptions import InvalidSignatureError
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    try:
        # Replicate SDK will automatically verify the signature
        # when you use webhook-related methods or objects.
        # For manual verification (if needed outside SDK context):
        # signature = request.headers.get('X-Replicate-Signature')
        # payload = request.data
        # secret = os.environ.get("REPLICATE_WEBHOOK_SECRET")
        # replicate.verify_webhook(secret, payload, signature)

        data = request.json
        print(f"Received webhook data: {data}")
        # Process the webhook data (e.g., update job status, notify users)

        return jsonify({"status": "success"}), 200

    except InvalidSignatureError:
        return jsonify({"error": "Invalid signature"}), 400
    except Exception as e:
        print(f"Error processing webhook: {e}")
        return jsonify({"error": "Internal server error"}), 500

if __name__ == '__main__':
    # Ensure you have REPLICATE_WEBHOOK_SECRET environment variable set
    # Example: export REPLICATE_WEBHOOK_SECRET='your-secret-key'
    app.run(port=5000, debug=True)
```

--------------------------------

### Opt-out of File Output - JavaScript (Conceptual)

Source: https://github.com/replicate/replicate-python/blob/main/README.md

Illustrates how to disable the default file output handling in Replicate by setting `use_file_output=False` when calling `replicate.run()`. This is a conceptual example, as the provided code snippet appears to be JavaScript syntax attempting to call a Python function, which is not directly supported.

```javascript
const replicate = replicate.run("acmecorp/acme-model", use_file_output=False);
```

--------------------------------

### Handle Replicate API Errors with Python

Source: https://context7.com/replicate/replicate-python/llms.txt

Demonstrates how to catch and handle various Replicate-specific exceptions, including ModelError for execution failures and ReplicateError for API issues. It shows how to access detailed error information from the prediction object and provides examples of validating inputs before running a model.

```python
import replicate
from replicate.exceptions import ModelError, ReplicateError, ReplicateException

try:
    output = replicate.run(
        "stability-ai/sdxl",
        input={"prompt": "test"}
    )
except ModelError as e:
    # Model execution failed (prediction.status == "failed")
    print(f"Model execution failed")
    print(f"Prediction ID: {e.prediction.id}")
    print(f"Error: {e.prediction.error}")
    print(f"Status: {e.prediction.status}")

    # Access full prediction object
    if e.prediction.logs:
        print(f"Logs:\n{e.prediction.logs}")

except ReplicateError as e:
    # API error (RFC 7807 Problem Details)
    print(f"API Error: {e.status}")
    print(f"Type: {e.type}")
    print(f"Title: {e.title}")
    print(f"Detail: {e.detail}")

    # Convert to dict for logging
    error_dict = e.to_dict()
    print(f"Error details: {error_dict}")

except ReplicateException as e:
    # Base exception for all Replicate errors
    print(f"Replicate error: {e}")

# Retry logic is built-in for transient errors
# (429 Too Many Requests, 503 Service Unavailable, 504 Gateway Timeout)
# No manual retry needed for these cases

# Validate input before running
try:
    model = replicate.models.get("owner/name")
    version = model.latest_version

    # Check input schema
    schema = version.openapi_schema
    required_inputs = schema["components"]["schemas"]["Input"].get("required", [])
    print(f"Required inputs: {required_inputs}")

    output = replicate.run(
        model.id,
        input={param: "value" for param in required_inputs}
    )
except Exception as e:
    print(f"Failed to run model: {e}")

```

--------------------------------

### List Available Hardware - Python

Source: https://github.com/replicate/replicate-python/blob/main/README.md

Demonstrates how to list all available hardware SKUs for running models on Replicate. The code iterates through the hardware list returned by `replicate.hardware.list()` and extracts the SKU for each hardware option.

```python
>>> [hw.sku for hw in replicate.hardware.list()]
['cpu', 'gpu-t4', 'gpu-a40-small', 'gpu-a40-large']
```

--------------------------------

### Create and Manage Deployments (Python)

Source: https://context7.com/replicate/replicate-python/llms.txt

Covers creating, updating, and deleting deployments for production workloads. This includes configuring auto-scaling, rolling out new model versions with zero downtime, and retrieving deployment information.

```python
import replicate

# Create a deployment with auto-scaling
deployment = replicate.deployments.create(
    name="my-app-generator",
    model="stability-ai/sdxl",
    version="39ed52f2a78e934b3ba6e2a89f5b1c712de7dfea535525255b1aa35c5565e08b",
    hardware="gpu-a40-large",
    min_instances=1,  # Always keep 1 instance warm
    max_instances=5   # Scale up to 5 instances under load
)

print(f"Deployment created: {deployment.id}")
print(f"Current release: {deployment.current_release.number}")
print(f"Hardware: {deployment.current_release.configuration.hardware}")

# Run predictions on deployment
prediction = replicate.predictions.create(
    deployment="your-username/my-app-generator",
    input={"prompt": "beautiful landscape"}
)

# Or use deployment object
prediction = deployment.predictions.create(
    input={"prompt": "beautiful landscape"}
)

# Update deployment (rolling update, no downtime)
updated = replicate.deployments.update(
    deployment_owner="your-username",
    deployment_name="my-app-generator",
    version="new-version-id",  # Roll out new version
    min_instances=2,  # Scale up minimum
    max_instances=10
)

print(f"Updated to release: {updated.current_release.number}")

# Get deployment info
deployment = replicate.deployments.get("your-username/my-app-generator")

# List all deployments
page = replicate.deployments.list()
for d in page:
    print(f"{d.id}: {d.current_release.configuration.hardware}")

# Delete deployment
deleted = replicate.deployments.delete(
    deployment_owner="your-username",
    deployment_name="my-app-generator"
)
```

--------------------------------

### Create, Update, and Delete Models (Python)

Source: https://context7.com/replicate/replicate-python/llms.txt

Illustrates the process of creating a new private or public model, managing its metadata, and deleting it. It covers creating a model with details like owner, name, visibility, hardware, description, and links to GitHub and paper.

```python
import replicate

# Create a new model
model = replicate.models.create(
    owner="your-username",
    name="my-custom-model",
    visibility="private",  # or "public"
    hardware="gpu-a40-large",
    description="My custom fine-tuned model",
    github_url="https://github.com/user/repo",
    paper_url="https://arxiv.org/abs/...",
    license_url="https://opensource.org/licenses/MIT",
    cover_image_url="https://example.com/cover.jpg"
)

print(f"Created model: {model.id}")

# Update model (through API, not shown in SDK)
# Use replicate.com/your-username/my-custom-model/settings

# Delete a model
deleted = replicate.models.delete("your-username/my-custom-model")
if deleted:
    print("Model deleted successfully")

# Delete specific version
model = replicate.models.get("your-username/my-model")
model.versions.delete("version-id-to-delete")

# Create predictions without specifying version (uses latest)
prediction = replicate.models.predictions.create(
    model="stability-ai/sdxl",
    input={"prompt": "test generation"}
)
```

--------------------------------

### Handle File Inputs for Models (Python)

Source: https://context7.com/replicate/replicate-python/llms.txt

Illustrates various methods for providing file inputs to Replicate models, including using direct URLs, local file paths, and `Path` objects. It also covers pre-uploading files for reuse, using Base64 encoding for smaller files, and managing uploaded files (listing and deleting).

```python
import replicate
from pathlib import Path

# URL input (most efficient, no upload needed)
output = replicate.run(
    "nightmareai/real-esrgan",
    input={
        "image": "https://example.com/photo.jpg",
        "scale": 4,
        "face_enhance": True
    }
)

# Local file input (automatically uploaded)
with open("photo.jpg", "rb") as f:
    output = replicate.run(
        "nightmareai/real-esrgan",
        input={"image": f}
    )

# Path object (automatically opened and uploaded)
output = replicate.run(
    "nightmareai/real-esrgan",
    input={"image": Path("photo.jpg")}
)

# Pre-upload file and reuse
file = replicate.files.create(
    "large_dataset.zip",
    metadata={"purpose": "training"}
)
print(f"Uploaded file: {file.id}")
print(f"URL: {file.urls['get']}")

# Use in multiple predictions
for prompt in ["style1", "style2"]:
    output = replicate.run(
        "some-model/with-file-input",
        input={
            "data": file.urls['get'],
            "prompt": prompt
        }
    )

# Base64 encoding for small files (< 10MB)
with open("small.png", "rb") as f:
    output = replicate.run(
        "some-model/image-processor",
        input={"image": f},
        file_encoding_strategy="base64"  # Embed in request
    )

# Manage files
files_list = replicate.files.list()
for file in files_list:
    print(f"{file.name}: {file.size} bytes")

# Delete file
replicate.files.delete(file.id)
```

--------------------------------

### Publish Beta Release for replicate-python

Source: https://github.com/replicate/replicate-python/blob/main/CONTRIBUTING.md

To publish a beta release, first update the version number in `pyproject.toml` to a beta version (e.g., `1.1.0b3`). Commit this change to the `beta` branch. Then, create and push a Git tag corresponding to the beta version. Monitor the release on the GitHub Actions page.

```toml
version = "1.1.0b3"

```

```console
git checkout beta
git fetch --all --tags
git tag 1.1.0b3
git push --tags

```

--------------------------------

### Run Model with FileOutput Handling in Python

Source: https://github.com/replicate/replicate-python/blob/main/README.md

Execute a Replicate model and process FileOutput objects returned by default in version 1.0.0+. Demonstrates iterating through outputs and writing them to local files.

```python
import replicate
outputs = replicate.run(
    "black-forest-labs/flux-schnell",
    input={"prompt": "astronaut riding a rocket like a horse"}
)
for index, output in enumerate(outputs):
    with open(f"output_{index}.webp", "wb") as file:
        file.write(output.read())
```

--------------------------------

### Pass File Input to Model in Python

Source: https://github.com/replicate/replicate-python/blob/main/README.md

Run a model with file input by passing either a URL to a publicly accessible file or a local file handle. Demonstrates opening and passing a local image file.

```python
output = replicate.run(
    "andreasjansson/blip-2:f677695e5e89f8b236e52ecd1d3f01beb44c34606419bcc19345e046d8f786f9",
    input={ "image": open("path/to/mystery.jpg") }
)
```

--------------------------------

### Load Output Files from Replicate Python

Source: https://github.com/replicate/replicate-python/blob/main/README.md

This Python code snippet illustrates how to handle output files returned by Replicate, which are provided as `FileOutput` objects. It shows a typical workflow where a model is run, and its output is then read and written to a local file. This is fundamental for accessing and utilizing the results of model predictions.

```python
import replicate
from PIL import Image # pip install pillow

output = replicate.run(
    "stability-ai/stable-diffusion:27b93a2413e7f36cd83da926f3656280b2931564ff050bf9575f1fdf9bcd7478",
    input={"prompt": "wavy colorful abstract patterns, oceans"}
    )

# The 'output' variable here is a FileOutput object.
# You can read its content and save it to a file:
# with open("output.png", "wb") as f:
#     f.write(output.read())
```

--------------------------------

### Fine-tune a Model with Replicate Python

Source: https://github.com/replicate/replicate-python/blob/main/README.md

This snippet demonstrates how to fine-tune a model using the Replicate Python SDK. It requires specifying the model, version, input data (including images and captions), and a destination for the trained model. Ensure the `REPLICATE_API_TOKEN` environment variable is set.

```python
import replicate

training = replicate.trainings.create(
    model="stability-ai/sdxl",
    version="39ed52f2a78e934b3ba6e2a89f5b1c712de7dfea535525255b1aa35c5565e08b",
    input={
      "input_images": "https://my-domain/training-images.zip",
      "token_string": "TOK",
      "caption_prefix": "a photo of TOK",
      "max_train_steps": 1000,
      "use_face_detection_instead": False
    },
    # You need to create a model on Replicate that will be the destination for the trained version.
    destination="your-username/model-name"
)
```

--------------------------------

### Configure Replicate Python Client with Custom Settings

Source: https://context7.com/replicate/replicate-python/llms.txt

Illustrates various ways to configure the Replicate Python client for advanced usage. This includes setting custom API tokens, configuring network timeouts (connect, read, write, pool), adding custom headers, specifying a custom base URL for proxies, and enabling advanced HTTP client options like HTTP/2.

```python
import replicate
from replicate import Client
import httpx

# Use environment variable (default)
# export REPLICATE_API_TOKEN="r8_..."
output = replicate.run("model/name", input={})

# Custom API token
client = Client(api_token="r8_custom_token_here")
output = client.run("model/name", input={})

# Custom timeout configuration
client = Client(
    api_token="r8_...",
    timeout=httpx.Timeout(
        connect=10.0,  # Connection timeout
        read=300.0,    # Read timeout (for long-running predictions)
        write=10.0,
        pool=10.0
    )
)

# Custom headers
client = Client(
    api_token="r8_...",
    headers={
        "User-Agent": "MyApp/1.0",
        "X-Custom-Header": "value"
    }
)

# Custom base URL (for proxies or testing)
client = Client(
    api_token="r8_...",
    base_url="https://custom-proxy.example.com"
)

# Advanced httpx client configuration
client = Client(
    api_token="r8_...",
    follow_redirects=True,
    verify=True,  # SSL verification
    http2=True,   # Enable HTTP/2
)

# Use client instance
model = client.models.get("stability-ai/sdxl")
prediction = client.predictions.create(
    version=model.latest_version.id,
    input={"prompt": "test"}
)

# Access different namespaces
files = client.files.list()
trainings = client.trainings.list()
deployments = client.deployments.list()

```

--------------------------------

### Compose Models into a Pipeline with Replicate Python

Source: https://github.com/replicate/replicate-python/blob/main/README.md

This Python snippet shows how to chain models together to create a pipeline, where the output of one model serves as the input for another. It retrieves two different models, runs the first with a prompt, and then uses its output image to upscale with the second model. This enables complex creative workflows by combining specialized models.

```python
laionide = replicate.models.get("afiaka87/laionide-v4").versions.get("b21cbe271e65c1718f2999b038c18b45e21e4fba961181fbfae9342fc53b9e05")
swinir = replicate.models.get("jingyunliang/swinir").versions.get("660d922d33153019e8c263a3bba265de882e7f4f70396546b6c9c8f9d47a021a")
image = laionide.predict(prompt="avocado armchair")
upscaled_image = swinir.predict(image=image)
```

--------------------------------

### Run a Replicate Model Synchronously (Python)

Source: https://context7.com/replicate/replicate-python/llms.txt

Executes a machine learning model on Replicate synchronously, blocking until completion and returning the output. Handles file downloads and streaming of image data. Requires the 'replicate' library. Inputs are model identifier and a dictionary of parameters; outputs are model-specific, often FileOutput objects for images.

```python
import replicate

# Set your API token (or use REPLICATE_API_TOKEN environment variable)
# export REPLICATE_API_TOKEN="r8_..."

# Run a model with the latest version
output = replicate.run(
    "stability-ai/sdxl",
    input={
        "prompt": "astronaut riding a horse on mars",
        "num_outputs": 1,
        "guidance_scale": 7.5
    }
)

# Output is automatically typed based on the model
# For image models, you get FileOutput objects
for image in output:
    # Download the image
    with open("generated.png", "wb") as f:
        f.write(image.read())

    # Or stream it in chunks
    with open("generated.png", "wb") as f:
        for chunk in image:
            f.write(chunk)

    # Get the URL as a string
    print(f"Image URL: {image.url}")

# Run a specific version of a model
output = replicate.run(
    "stability-ai/sdxl:39ed52f2a78e934b3ba6e2a89f5b1c712de7dfea535525255b1aa35c5565e08b",
    input={"prompt": "sunset over mountains"}
)
```

--------------------------------

### Customize Replicate Python Client Behavior

Source: https://github.com/replicate/replicate-python/blob/main/README.md

This snippet shows how to create a custom client instance for the Replicate Python SDK. This allows you to specify a different API token (ideally from an environment variable) and add custom headers, such as a User-Agent, to the client's requests. Avoid hardcoding sensitive credentials.

```python
import os
from replicate.client import Client

replicate = Client(
    api_token=os.environ["SOME_OTHER_REPLICATE_API_TOKEN"]
    headers={
        "User-Agent": "my-app/1.0"
    }
)
```

--------------------------------

### Set Replicate API Token Environment Variable

Source: https://github.com/replicate/replicate-python/blob/main/README.md

Configure authentication by setting the REPLICATE_API_TOKEN environment variable. Store your token from replicate.com/account without committing to source control.

```bash
export REPLICATE_API_TOKEN=<your token>
```

--------------------------------

### Stream Model Output with Replicate Python

Source: https://github.com/replicate/replicate-python/blob/main/README.md

This snippet demonstrates how to stream output from a language model using Replicate's Python client. It utilizes the `replicate.stream` method to consume tokens as they are generated, providing real-time output. This is particularly useful for interactive applications or when immediate feedback is desired.

```python
import replicate

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

--------------------------------

### Run a Replicate Model Asynchronously (Python)

Source: https://context7.com/replicate/replicate-python/llms.txt

Executes a machine learning model on Replicate asynchronously using asyncio for concurrent operations in async applications. Supports asynchronous file handling and concurrent model execution. Requires the 'replicate' and 'aiofiles' libraries. Inputs are model identifier and input parameters; outputs are handled asynchronously.

```python
import asyncio
import replicate
import aiofiles

async def main():
    # Run model asynchronously
    output = await replicate.async_run(
        "stability-ai/sdxl",
        input={"prompt": "cyberpunk cityscape"}
    )

    # Handle FileOutput objects
    for idx, image in enumerate(output):
        async with aiofiles.open(f"output_{idx}.png", "wb") as f:
            # Stream asynchronously
            async for chunk in image:
                await f.write(chunk)

            # Or read all at once
            # data = await image.aread()
            # await f.write(data)

    # Run multiple models concurrently
    tasks = [
        replicate.async_run(
            "stability-ai/sdxl",
            input={"prompt": "sunset"}
        ),
        replicate.async_run(
            "meta/meta-llama-3-70b-instruct",
            input={"prompt": "Write a haiku about sunsets"}
        )
    ]

    results = await asyncio.gather(*tasks)
    images, text = results
    print(f"Generated {len(images)} images and text: {text}")

asyncio.run(main())
```

--------------------------------

### List Collections and Models within Collections - Python

Source: https://github.com/replicate/replicate-python/blob/main/README.md

Shows how to retrieve collections of featured models on Replicate and access models within a specific collection. It demonstrates iterating through paginated collection lists and accessing model details like slug and description.

```python
collections = [collection for page in replicate.paginate(replicate.collections.list) for collection in page]
collections[0].slug
collections[0].description

replicate.collections.get("text-to-image").models
```

--------------------------------

### Publish Stable Release for replicate-python

Source: https://github.com/replicate/replicate-python/blob/main/CONTRIBUTING.md

To publish a stable release, update the version number in `pyproject.toml` to the stable version (e.g., `0.7.0`) and commit it to the `main` branch. Afterward, create and push a Git tag matching the stable version. The release process can be monitored via GitHub Actions.

```toml
version = "0.7.0"

```

```console
git checkout main
git fetch --all --tags
git tag 0.7.0
git push --tags

```

--------------------------------

### Run Model with Wait=False and Collect Outputs (Python)

Source: https://context7.com/replicate/replicate-python/llms.txt

Shows how to initiate a model prediction using `replicate.run` with `wait=False`. This returns a prediction object immediately, allowing subsequent processing or collection of outputs via `output_iterator` without blocking the main thread.

```python
outputs = []
prediction = replicate.run(
    "pixray/text2image",
    input={"prompts": "portrait"},
    wait=False  # Returns prediction object
)

for output in prediction.output_iterator():
    outputs.append(output)
    print(f"Collected {len(outputs)} outputs so far")
```

--------------------------------

### Automatically Sign Git Commits with a prepare-commit-msg Hook

Source: https://github.com/replicate/replicate-python/blob/main/CONTRIBUTING.md

Automate the signing of all your Git commits by creating a `prepare-commit-msg` git hook. This script retrieves your Git name and email and appends the 'Signed-off-by' trailer to the commit message, ensuring compliance with contribution guidelines.

```shell
#!/bin/sh

NAME=$(git config user.name)
EMAIL=$(git config user.email)

if [ -z "$NAME" ]; then
    echo "empty git config user.name"
    exit 1
fi

if [ -z "$EMAIL" ]; then
    echo "empty git config user.email"
    exit 1
fi

git interpret-trailers --if-exists doNothing --trailer \
    "Signed-off-by: $NAME <$EMAIL>" \
    --in-place "$1"

```

--------------------------------

### Run Model in Background (Async) with Replicate Python

Source: https://github.com/replicate/replicate-python/blob/main/README.md

This Python code illustrates how to run a model asynchronously in the background using the Replicate client. It shows how to retrieve a model and its version, create a prediction, check its status through polling (`reload`), and finally wait for completion and access the output. This is useful for long-running tasks where immediate results are not required.

```python
>>> 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"})

>>> prediction
Prediction(...)

>>> prediction.status
'starting'

>>> dict(prediction)
{"id": "...", "status": "starting", ...}

>>> prediction.reload()
>>> prediction.status
'processing'

>>> print(prediction.logs)
iteration: 0, render:loss: -0.6171875
iteration: 10, render:loss: -0.92236328125
iteration: 20, render:loss: -1.197265625
iteration: 30, render:loss: -1.3994140625

>>> prediction.wait()

>>> prediction.status
'succeeded'

>>> prediction.output
<replicate.helpers.FileOutput object at 0x107179b50>

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

--------------------------------

### Handle Replicate Webhooks with Python

Source: https://context7.com/replicate/replicate-python/llms.txt

Demonstrates how to set up a webhook endpoint in a Python Flask application to receive and validate notifications from Replicate. It covers retrieving the webhook secret, validating incoming signatures using different methods (raw data and httpx.Request), and processing prediction status updates.

```python
import replicate
from flask import request, jsonify
from replicate.exceptions import InvalidSignatureError

# Assume 'app' is your Flask application instance
# Assume 'secret' is obtained using replicate.webhooks.default.secret()

@app.route("/webhook", methods=["POST"])
def handle_webhook():
    # Validate the webhook signature
    try:
        replicate.webhooks.validate(
            headers=dict(request.headers),
            body=request.data.decode(),
            secret=secret,
            tolerance=300  # Allow 5 minutes clock skew
        )
    except InvalidSignatureError as e:
        print(f"Invalid signature: {e}")
        return jsonify({"error": "Invalid signature"}), 401

    # Process the webhook payload
    payload = request.json
    print(f"Prediction {payload['id']} status: {payload['status']}")

    if payload['status'] == 'succeeded':
        print(f"Output: {payload['output']}")
    elif payload['status'] == 'failed':
        print(f"Error: {payload['error']}")

    return jsonify({"success": True})

# Example of creating a prediction with a webhook
# prediction = replicate.predictions.create(
#     model="stability-ai/sdxl",
#     input={"prompt": "astronaut"},
#     webhook="https://your-app.com/webhook",
#     webhook_events_filter=["start", "completed"]
# )

@app.route("/webhook2", methods=["POST"])
def handle_webhook_httpx():
    import httpx

    req = httpx.Request(
        method=request.method,
        url=request.url,
        headers=request.headers,
        content=request.data
    )

    try:
        replicate.webhooks.validate(request=req, secret=secret)
    except InvalidSignatureError:
        return jsonify({"error": "Invalid"}), 401

    return jsonify({"success": True})

```

--------------------------------

### Read File Output Directly - Python

Source: https://github.com/replicate/replicate-python/blob/main/README.md

Demonstrates how to read the entire binary data from a file output generated by Replicate using the .read() method. This is suitable for smaller files that can fit into memory. It opens a file in binary write mode and writes the content obtained from output[0].read().

```python
with open("my_output.png", "wb") as file:
  file.write(output[0].read())
```

--------------------------------

### Stream Response with Flask

Source: https://github.com/replicate/replicate-python/blob/main/README.md

Shows how to stream Replicate file output using Flask's response_class and stream_with_context. This Flask route function executes a Replicate model and returns its output as a streamed response, suitable for large files.

```python
@app.route('/stream')
def streamed_response():
    output = replicate.run("black-forest-labs/flux-schnell", input={...}, use_file_output =True)
    return app.response_class(stream_with_context(output))
```

--------------------------------

### Sign Git Commits with Developer Certificate of Origin

Source: https://github.com/replicate/replicate-python/blob/main/CONTRIBUTING.md

To contribute to this project, each commit must be signed off with the Developer Certificate of Origin. This confirms you have the right to submit the work under the project's open source license. Add the 'Signed-off-by' line to your commit message with your name and email.

```text
Signed-off-by: Sam Smith <sam.smith@example.com>
```