Fal.ai Setup¶

Configuration guide for the Fal.ai integration.

Prerequisites¶

Fal.ai account at fal.ai
API key from the dashboard

Environment Variables¶

# Required
FAL_KEY="your-fal-api-key"

# Optional - Custom workflow endpoints
FAL_ENDPOINT_ID=""

Compute Server Configuration¶

# app/core/config.py
class Settings(BaseSettings):
    # Fal.ai Configuration
    FAL_KEY: Optional[str] = None
    FAL_ENDPOINT_ID: Optional[str] = None

    # Task-specific endpoints
    DEFAULT_EDITING_TASK_FAL_APP_ID: str = "workflows/Shootify/editing-workflow"
    GENERATION_TASK_FAL_APP_ID: str = "workflows/Shootify/flux-krea-lora-i2i"
    VIDEO_GENERATION_FAL_APP_ID: str = "fal-ai/video-gen"

Dependencies¶

# pyproject.toml
[project.dependencies]
fal-client = "^0.5.0"
pillow = "^10.4.0"

Client Initialization¶

Basic Client¶

import fal_client
from app.core.config import settings

class FalClient:
    def __init__(self):
        if not settings.FAL_KEY:
            raise ValueError("FAL_KEY must be set")
        self.client = fal_client

With Status Callbacks¶

def submit_with_status(
    self,
    application_id: str,
    arguments: dict,
    on_status: Callable
) -> dict:
    handle = self.client.submit(application_id, arguments)

    for event in handle.iter_events(with_logs=False, interval=1):
        on_status(event)

    return handle.get()

API Patterns¶

response = fal_client.subscribe(
    "fal-ai/flux-pro/kontext/max",
    arguments={"prompt": prompt, "image_url": url},
    with_logs=False,
)

Method 2: Submit, Poll, Get Result¶

# Submit
handle = fal_client.submit("fal-ai/application-id", arguments)
request_id = handle.request_id

# Check status
status = fal_client.status("fal-ai/application-id", request_id)

# Get result when ready
result = fal_client.result("fal-ai/application-id", request_id)

Method 3: With Event Streaming¶

handle = fal_client.submit(application_id, arguments)

for event in handle.iter_events(with_logs=True, interval=1):
    if isinstance(event, fal_client.Queued):
        print(f"Position: {event.position}")
    elif isinstance(event, fal_client.InProgress):
        print(f"Progress: {event.progress}")
    elif isinstance(event, fal_client.Completed):
        print("Done!")

result = handle.get()

Image Upload¶

from PIL import Image

def upload_image(self, image: Image.Image, format: str = "jpeg") -> str:
    """Upload image and return URL."""
    return self.client.upload_image(image, format=format)

def upload_file(self, file_path: str) -> str:
    """Upload file and return URL."""
    return self.client.upload_file(file_path)

Status Handling¶

import fal_client

def handle_status(status: fal_client.Status):
    if isinstance(status, fal_client.Completed):
        print("Task completed")
    elif isinstance(status, fal_client.InProgress):
        print(f"In progress: {status.progress}")
    elif isinstance(status, fal_client.Queued):
        print(f"Queued at position: {status.position}")

Task Cancellation¶

def cancel_task(application_id: str, request_id: str):
    fal_client.cancel(application_id, request_id)

Verification¶

Test your configuration:

import fal_client

# Simple test - image generation
result = fal_client.subscribe(
    "fal-ai/flux/schnell",
    arguments={
        "prompt": "a cat",
        "image_size": "square"
    },
    with_logs=False,
)

print(f"Generated: {result['images'][0]['url']}")

Troubleshooting¶

Missing API Key¶

ValueError: FAL_KEY must be set in environment variables

Ensure FAL_KEY is set in your environment or .env file.

Task Timeout¶

Long-running tasks may timeout. Use async patterns:

handle = fal_client.submit(app_id, arguments)
# Store request_id for later retrieval
request_id = handle.request_id

# Later, get result
result = fal_client.result(app_id, request_id)

Rate Limits¶

Fal.ai has concurrency limits per account. Monitor queue depth and implement backoff if needed.