Client Library for Python

Introduction

The client for Python provides a seamless interface to interact with fal.

Installation

First, add the client as a dependency in your project:

pip install fal-client

Features

1. Call an endpoint

Endpoints requests are managed by a queue system. This allows fal to provide a reliable and scalable service.

The subscribe method allows you to submit a request to the queue and wait for the result.

Python

import fal_client

def on_queue_update(update):
    if isinstance(update, fal_client.InProgress):
        for log in update.logs:
           print(log["message"])

result = fal_client.subscribe(
    "fal-ai/flux/dev",
    arguments={
        "prompt": "a cat",
        "seed": 6252023,
        "image_size": "landscape_4_3",
        "num_images": 4
    },
    with_logs=True,
    on_queue_update=on_queue_update,
)

print(result)

Python (async)

import asyncio
import fal_client

async def subscribe():
    def on_queue_update(update):
        if isinstance(update, fal_client.InProgress):
            for log in update.logs:
                print(log["message"])

    result = await fal_client.subscribe_async(
        "fal-ai/flux/dev",
        arguments={
            "prompt": "a cat",
            "seed": 6252023,
            "image_size": "landscape_4_3",
            "num_images": 4
        },
        on_queue_update=on_queue_update,
    )

    print(result)


if __name__ == "__main__":
    asyncio.run(subscribe())

2. Queue Management

You can manage the queue using the following methods:

Submit a Request

Submit a request to the queue using the queue.submit method.

Python

import fal_client

handler = fal_client.submit(
    "fal-ai/flux/dev",
    arguments={
        "prompt": "a cat",
        "seed": 6252023,
        "image_size": "landscape_4_3",
        "num_images": 4
    },
)

request_id = handler.request_id

Python (async)

import asyncio
import fal_client

async def submit():
    handler = await fal_client.submit_async(
        "fal-ai/flux/dev",
        arguments={
            "prompt": "a cat",
            "seed": 6252023,
            "image_size": "landscape_4_3",
            "num_images": 4
        },
    )

    request_id = handler.request_id
    print(request_id)

This is useful when you want to submit a request to the queue and retrieve the result later. You can save the request_id and use it to retrieve the result later.

Webhooks

For long-running requests, such as training jobs, you can use webhooks to receive the result asynchronously. You can specify the webhook URL when submitting a request.

Check Request Status

Retrieve the status of a specific request in the queue:

Python

status = fal_client.status("fal-ai/flux/dev", request_id, with_logs=True)

Python (async)

status = await fal_client.status_async("fal-ai/flux/dev", request_id, with_logs=True)

Retrieve Request Result

Get the result of a specific request from the queue:

Python

result = fal_client.result("fal-ai/flux/dev", request_id)

Python (async)

result = await fal_client.result_async("fal-ai/flux/dev", request_id)

3. File Uploads

Some endpoints require files as input. However, since the endpoints run asynchronously, processed by the queue, you will need to provide URLs to the files instead of the actual file content.

Luckily, the client library provides a way to upload files to the server and get a URL to use in the request.

Python

url = fal_client.upload_file("path/to/file")

Python (async)

url = fal_client.upload_file_async("path/to/file")

4. Streaming

Some endpoints support streaming:

Python

import fal_client

def stream():
    stream = fal_client.stream(
        "fal-ai/flux/dev",
        arguments={
            "prompt": "a cat",
            "seed": 6252023,
            "image_size": "landscape_4_3",
            "num_images": 4
        },
    )
    for event in stream:
        print(event)


if __name__ == "__main__":
    stream()

Python (async)

import asyncio
import fal_client

async def stream():
    stream = fal_client.stream_async(
        "fal-ai/flux/dev",
        arguments={
            "prompt": "a cat",
            "seed": 6252023,
            "image_size": "landscape_4_3",
            "num_images": 4
        },
    )
    async for event in stream:
        print(event)


if __name__ == "__main__":
    asyncio.run(stream())

5. Realtime Communication

For the endpoints that support real-time inference via WebSockets, you can use the realtime client that abstracts the WebSocket connection, re-connection, serialization, and provides a simple interface to interact with the endpoint:

Python

Not implemented yet

This functionality is not available on this client yet.

Python (async)

Not implemented yet

This functionality is not available on this client yet.

6. Run

The endpoints can also be called directly instead of using the queue system.

Prefer the queue

We do not recommend this use most use cases as it will block the client until the response is received. Moreover, if the connection is closed before the response is received, the request will be lost.

Python

import fal_client

result = fal_client.run(
    "fal-ai/flux/dev",
    arguments={
        "prompt": "a cat",
        "seed": 6252023,
        "image_size": "landscape_4_3",
        "num_images": 4
    },
)

print(result)

Python (async)

import asyncio
import fal_client

async def submit():
    result = await fal_client.run_async(
        "fal-ai/flux/dev",
        arguments={
            "prompt": "a cat",
            "seed": 6252023,
            "image_size": "landscape_4_3",
            "num_images": 4
        },
    )

    print(result)


if __name__ == "__main__":
    asyncio.run(submit())

API Reference

For a complete list of available methods and their parameters, please refer to Python API Reference documentation.

Support

If you encounter any issues or have questions, please visit the GitHub repository or join our Discord Community.