fal-api-reference

快速参考

方法	使用场景	代码
fal.subscribe()	基于队列（推荐）	await fal.subscribe("model", { input })
fal.run()	快速端点（<30s）	await fal.run("model", { input })
fal.stream()	渐进式输出	for await (const event of stream) {}
fal.realtime.connect()	WebSocket 交互	fal.realtime.connect("model", callbacks)

队列方法	用途
fal.queue.submit()	提交任务，获取 request_id
fal.queue.status()	检查任务状态
fal.queue.result()	获取已完成的结果
fal.queue.cancel()	取消待处理的任务

身份验证	标头	格式
API Key	Authorization	Key YOUR_FAL_KEY

何时使用此技能

用于 API 集成基础知识：

• 在 JavaScript/TypeScript 中配置 fal.ai 客户端
• 在 Python 中配置 fal_client
• 在 subscribe、run 和 stream 方法之间进行选择
• 实现 webhook 回调
• 将文件上传到 fal.media CDN

相关技能：

• 选择模型：参见 fal-model-guide
• 优化性能：参见 fal-optimization
• 部署自定义模型：参见 fal-serverless-guide

---

fal.ai API 参考

fal.ai 客户端库和 REST 端点的完整 API 参考。

客户端库

JavaScript/TypeScript (@fal-ai/client)

npm install @fal-ai/client

配置

import { fal } from "@fal-ai/client";

// Configure credentials (reads FAL_KEY from environment by default)
fal.config({
  credentials: process.env.FAL_KEY,
  // Optional: custom proxy URL for browser apps
  proxyUrl: "https://your-server.com/api/fal-proxy"
});

核心方法

fal.subscribe(endpoint, options) 基于队列执行，并自动轮询。推荐用于大多数场景。

const result = await fal.subscribe("fal-ai/flux/dev", {
  input: {
    prompt: "A beautiful landscape"
  },
  logs: true,
  pollInterval: 1000,  // Poll every second (default: 1000)
  onQueueUpdate: (update) => {
    // update.status: "IN_QUEUE" | "IN_PROGRESS" | "COMPLETED"
    if (update.status === "IN_PROGRESS") {
      update.logs?.forEach(log => console.log(log.message));
    }
  }
});

fal.run(endpoint, options) 直接/同步执行。仅用于快速端点（< 30 秒）。

const result = await fal.run("fal-ai/fast-sdxl", {
  input: { prompt: "A cat" }
});

fal.stream(endpoint, options) 通过服务器发送事件实现渐进式输出。

const stream = await fal.stream("fal-ai/flux/dev", {
  input: { prompt: "A landscape" }
});

for await (const event of stream) {
  console.log("Progress:", event);
}

const finalResult = await stream.done();

fal.realtime.connect(endpoint, callbacks) 通过 WebSocket 连接构建实时交互式应用。

const connection = fal.realtime.connect("fal-ai/lcm-sd15-i2i", {
  connectionKey: "unique-session-id",
  throttleInterval: 128,  // Debounce inputs (ms)
  onResult: (result) => console.log("Generated:", result),
  onError: (error) => console.error("Error:", error),
  onOpen: () => console.log("Connected"),
  onClose: () => console.log("Disconnected")
});

// Send inputs
connection.send({
  prompt: "A cute cat",
  image_url: "https://example.com/base.jpg"
});

// Close when done
connection.close();

队列方法

用于高级控制的手动队列管理。

// Submit to queue
const { request_id } = await fal.queue.submit("fal-ai/flux/dev", {
  input: { prompt: "Test" },
  webhookUrl: "https://your-server.com/webhook"  // Optional
});

// Check status
const status = await fal.queue.status("fal-ai/flux/dev", {
  requestId: request_id,
  logs: true
});
// status.status: "IN_QUEUE" | "IN_PROGRESS" | "COMPLETED"

// Get result (blocks until complete)
const result = await fal.queue.result("fal-ai/flux/dev", {
  requestId: request_id
});

// Cancel request
await fal.queue.cancel("fal-ai/flux/dev", {
  requestId: request_id
});

存储方法

将文件上传到 fal.media CDN。

// Upload File object
const file = new File([blob], "image.png", { type: "image/png" });
const url = await fal.storage.upload(file);

// Upload from URL
const response = await fetch("https://example.com/image.jpg");
const blob = await response.blob();
const url = await fal.storage.upload(new File([blob], "image.jpg"));

Python (fal-client)

pip install fal-client

同步 API

import fal_client

# Simple run
result = fal_client.run(
    "fal-ai/flux/dev",
    arguments={
        "prompt": "A beautiful landscape",
        "image_size": "landscape_16_9"
    }
)

# Subscribe with status updates
def on_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": "Test"},
    with_logs=True,
    on_queue_update=on_update
)

# Manual queue management
handler = fal_client.submit(
    "fal-ai/flux/dev",
    arguments={"prompt": "Test"}
)
print(f"Request ID: {handler.request_id}")
status = handler.status()  # Check status
result = handler.get()     # Block until complete

异步 API

import asyncio
import fal_client

async def generate():
    # Async run
    result = await fal_client.run_async(
        "fal-ai/flux/dev",
        arguments={"prompt": "Test"}
    )

    # Async subscribe
    result = await fal_client.subscribe_async(
        "fal-ai/flux/dev",
        arguments={"prompt": "Test"},
        with_logs=True
    )

    # Async queue management
    handler = await fal_client.submit_async(
        "fal-ai/flux/dev",
        arguments={"prompt": "Test"}
    )
    status = await handler.status_async()
    result = await handler.get_async()

    return result

result = asyncio.run(generate())

文件上传

# Upload file from path
url = fal_client.upload_file("path/to/image.png")

# Upload bytes
with open("image.png", "rb") as f:
    url = fal_client.upload(f.read(), "image/png")

# Encode as data URL (small files only)
data_url = fal_client.encode_file("small_image.png")

REST API

基础 URL

用途	URL 模式
队列提交	https://queue.fal.run/{model_id}
队列状态	https://queue.fal.run/{model_id}/requests/{request_id}/status
队列结果	https://queue.fal.run/{model_id}/requests/{request_id}
队列取消	https://queue.fal.run/{model_id}/requests/{request_id}/cancel
直接运行	https://fal.run/{model_id}
WebSocket	wss://fal.run/{model_id}

身份验证

Authorization: Key YOUR_FAL_KEY

队列工作流

# 1. Submit to queue
curl -X POST "https://queue.fal.run/fal-ai/flux/dev" \
  -H "Authorization: Key $FAL_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "A beautiful landscape",
    "image_size": "landscape_16_9"
  }'

# Response:
# {
#   "request_id": "abc123-def456",
#   "status": "IN_QUEUE",
#   "queue_position": 0
# }

# 2. Check status
curl "https://queue.fal.run/fal-ai/flux/dev/requests/abc123-def456/status" \
  -H "Authorization: Key $FAL_KEY"

# Response (in progress):
# {
#   "status": "IN_PROGRESS",
#   "logs": [{"message": "Loading model...", "timestamp": "..."}]
# }

# Response (completed):
# {
#   "status": "COMPLETED"
# }

# 3. Get result
curl "https://queue.fal.run/fal-ai/flux/dev/requests/abc123-def456" \
  -H "Authorization: Key $FAL_KEY"

# Response:
# {
#   "images": [{"url": "https://fal.media/...", "width": 1024, "height": 576}],
#   "seed": 12345,
#   "prompt": "A beautiful landscape"
# }

Webhook

提交任务时提供 webhook URL，即可通过 POST 接收结果：

curl -X POST "https://queue.fal.run/fal-ai/flux/dev" \
  -H "Authorization: Key $FAL_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "Test",
    "webhook_url": "https://your-server.com/webhook"
  }'

Webhook 载荷：

{
  "request_id": "abc123",
  "status": "COMPLETED",
  "payload": {
    "images": [{"url": "https://fal.media/..."}]
  }
}

直接执行

适用于快速端点（< 30 秒）：

curl -X POST "https://fal.run/fal-ai/fast-sdxl" \
  -H "Authorization: Key $FAL_KEY" \
  -H "Content-Type: application/json" \
  -d '{"prompt": "A cat"}'

常见模型参数

FLUX 模型

interface FluxInput {
  prompt: string;                    // Required
  image_size?:
    | "square_hd"      // 1024x1024
    | "square"         // 512x512
    | "portrait_4_3"   // 768x1024
    | "portrait_16_9"  // 576x1024
    | "landscape_4_3"  // 1024x768
    | "landscape_16_9" // 1024x576
    | { width: number; height: number };
  num_inference_steps?: number;      // 1-50, default: 28
  guidance_scale?: number;           // 1-20, default: 3.5
  num_images?: number;               // 1-4, default: 1
  seed?: number;                     // For reproducibility
  enable_safety_checker?: boolean;   // Default: true
  output_format?: "jpeg" | "png";    // Default: jpeg
  sync_mode?: boolean;               // Wait for completion
}

interface FluxOutput {
  images: Array<{
    url: string;
    width: number;
    height: number;
    content_type: string;
  }>;
  seed: number;
  prompt: string;
  has_nsfw_concepts?: boolean[];
  timings?: {
    inference: number;
  };
}

视频模型

interface VideoInput {
  prompt: string;
  negative_prompt?: string;
  duration?: number;          // seconds
  aspect_ratio?: "16:9" | "9:16" | "1:1" | "4:3" | "21:9";
  cfg_scale?: number;         // 0.0-1.0
  seed?: number;
  // Model-specific options...
}

interface VideoOutput {
  video: {
    url: string;
    content_type: string;
    file_size?: number;
  };
  audio?: {
    url: string;
  };
  seed: number;
}

音频模型（Whisper）

interface WhisperInput {
  audio_url: string;
  task?: "transcribe" | "translate";
  language?: string;         // ISO code
  chunk_level?: "segment";
  version?: "3";
}

interface WhisperOutput {
  text: string;
  chunks?: Array<{
    text: string;
    timestamp: [number, number];
  }>;
}

错误响应

// 400 Bad Request - Validation error
{
  "detail": "Invalid input",
  "errors": [
    {"field": "prompt", "message": "Field is required"}
  ]
}

// 401 Unauthorized
{
  "detail": "Invalid API key"
}

// 429 Too Many Requests
{
  "detail": "Rate limit exceeded",
  "retry_after": 60
}

// 500 Internal Server Error
{
  "detail": "Internal server error",
  "request_id": "abc123"
}

速率限制

• 速率限制因订阅层级而异
• 对 429 响应实施指数退避
• 高流量应用请使用 webhook
• 如需企业级速率限制，请联系 fal.ai

SDK 版本

库	最新版本	安装
@fal-ai/client	0.15+	npm install @fal-ai/client
fal-client (Python)	0.4+	pip install fal-client
fal (Serverless)	0.13+	pip install fal

请始终查看 https://docs.fal.ai，获取最新的 API 文档和更新。