falaw

Agent-friendly Python facade over fal.ai for generating and managing AI media (images, video, audio).

from falaw import generate_image, list_models, journal

r = generate_image("a tiger eye, macro, 35mm", quality="fast")
r.first.download(to="./tiger.png")

[m.id for m in list_models(category="image_to_video")]
journal.note("schnell at quality='fast' defaults to 1024x1024")

Why

fal-client already gives you 100+ models behind a uniform call. What agents (and humans) still struggle with is which model to use, what parameters it takes, and what to do with the URL it returns. falaw adds:

• Task-level verbs (generate_image, text_to_speech, ...) with smart model selection by quality tier.
• A queryable model registry --- no more grepping docs for IDs.
• Result / Asset objects that download, name, and organize outputs.
• A journal so each session leaves notes for the next one.
• A Claude skill, plus stub bridges for MCP and HTTP services --- all derived from the same tool registry.

Install

pip install 'falaw[fal]'   # to actually call fal.ai (pulls in fal-client)
export FAL_KEY="your-fal-api-key"

pip install falaw alone installs no fal.ai client. That is enough to build, price, hash, serialize and cache Plans (and to read cache hits), which is all a caller needs if it never makes a paid call. The first real network call raises falaw.FalClientNotInstalled (an ImportError) naming pip install 'falaw[fal]'. Anything that executes plans declares falaw[fal] as its requirement.

From the browser

The planning half of this package ships to npm as falaw-client (the ts/ directory): the catalogue with its prices, the cost rules, CallPlan and plan_hash are generated from and pinned to this package, and execution goes through a server relay because fal.ai forbids browser-held keys.

import { planGenerateImage, makePlan, totalCostUsd, execute, queueTransport } from 'falaw-client';
const plan = makePlan([planGenerateImage({ prompt: 'a tiger eye', quality: 'fast' })]);
totalCostUsd(plan);  // before any network
await execute(plan, { transport: queueTransport({ proxyUrl: '/api/fal/proxy', key }) });

See ts/README.md. After changing a dataclass, the catalogue, a cost rule, parse_response or the canonical byte-form, run python -m falaw export-schema, then cd ts && npm run codegen, and commit both — tests/test_schema_export.py and ts/src/codegen.test.ts fail until you do.

Core surface

Function	Purpose
generate_image(prompt, *, quality, image_size, model_id, extra)	Text-to-image, picks FLUX by quality tier.
text_to_speech(text, *, quality, voice, model_id, extra)	TTS, picks a voice model by tier.
list_models(*, category, quality_tier)	Browse the catalog.
pick_model(*, category, quality_tier)	Pick a sensible default.
call_fal(application, arguments, *, on_event)	Escape hatch to any fal model. Emits ProgressEvents + auto-journals on error.
cached_call_fal(...)	Same, plus content-addressed cache; emits cache_hit events on reuse.
execute_plan(plan, *, concurrency=N)	Run a Plan → list[Artifact]. Raises on the first failure, unwrapped.
execute_plan_isolated(plan, *, concurrency=N)	Run a Plan → ExecutionReport: one outcome per call, so one bad call does not discard the rest.
plan_dependencies(plan)	The Plan's "<from N>" dependency DAG, and its structural validator.
render_scene(scene, *, concurrency=N) / iter_render_scene(...)	Render every shot+beat; thread-pooled, with yield-as-done iterator.
estimate_scene_cost(scene)	Walk a Scene, return a CostRollup with per-line USD breakdown.
subscribe(callback)	Attach a global subscriber to the ProgressEvent bus.
journal.note / issue / improvement(...)	Leave a trace for future sessions.
Session(output_dir=...)	Optional stateful controller.

Structured progress events

call_fal and cached_call_fal emit ProgressEvents at every lifecycle transition (queued, progress, log, done, error, cache_hit, and provider_queued when fal reports the call waiting in its line). Subscribe per-call (on_event=), globally (falaw.subscribe(...)) or for one block and its threads (with falaw.listening(callback): ...); the legacy on_log=print is still honored for backward compatibility.

fal's own word on where a call is rides the events as provider_phase ("in_queue" with queue_position, or "in_progress"); it is None when fal said nothing (a cache hit, a provider that reports no queue state), never a guess. The done event carries fal's run metrics (e.g. inference_time) when it reported any, and the run's last log lines, which fal sends with its final status, arrive as log events. The TypeScript client's onEvent speaks the same kinds and phases (providerPhase).

from falaw import subscribe, generate_image

subscribe(lambda ev: print(f"[{ev.kind}] {ev.application} {ev.elapsed_s:.2f}s"))
generate_image("a tiger eye", quality="fast")

Cost estimation

ModelRecord.cost_estimate: CostEstimate | None carries a structured {kind, amount, currency} price (kinds: per_call | per_image | per_second | per_token | per_megapixel). estimate_scene_cost(scene) sums per-call costs and returns a CostRollup with per-line breakdown. Models without a populated cost_estimate appear in the rollup's skipped list so audits surface drift.

Pricing a direct Anthropic / OpenAI call

falaw prices calls made straight to the model vendor too, from one dated table (falaw/data/first_party_llm_prices.json), so an app does not keep its own copy:

from falaw import llm_call_cost_usd

usd, known = llm_call_cost_usd("anthropic", "claude-opus-4-7", response.usage)

usage is the vendor's own (a dict or the SDK object; OpenAI's Responses and Chat Completions shapes both work, cached tokens included). A model the table does not list is priced at the provider's dearest listed rates with known=False: never a free call, but not a ceiling either (an older model can cost more), so a caller that must never under-count refuses when known is false.

To record the token counts beside the cost, use llm_token_counts("anthropic", response.usage): it returns the very counts the cost was computed from (fresh_input, cache_read, cache_write, output, reasoning, and input for the total), so the two cannot disagree.

Re-pricing a saved plan

A CallPlan's estimated_cost_usd is frozen at plan time — right for a quote, wrong for a gate, because rate tables move (0.0.46 re-quoted every premium LLM call tenfold upward). Every plan_* therefore records how it priced the call in CallPlan.cost_basis: which pricer, which rate table and version, and the quantity hints (duration_s, token bounds) that price the call but deliberately never enter arguments. reprice_plan(plan) reads that back and re-quotes at today's rates — pure data, no network, no cache peek:

from falaw import plan_from_dict, reprice_plan

out = reprice_plan(plan_from_dict(saved_row))
out.plan  # same calls, same plan_hash, today's costs
out.known_delta_usd  # how much the priced part moved
out.changed  # per-call diffs: index, status, old, new, basis_changed
out.unpriced  # calls today's rates cannot price — refuse the gate

Per-call status is unchanged, changed, unknown (a basis today's tables cannot price) or no_basis (a plan saved before this existed). The last two clear the call's cost to None and light up Plan.has_unknown_costs: a stale figure re-presented as a current quote is the failure this fixes, so an unknown basis is reported, never trusted. cost_basis is descriptive only — it is omitted from the serialized dict when unset, and never enters plan_hash or the cache key, so existing plans and cassettes are unmoved.

A basis is only re-quoted by a pricer reading the same table it names; a mismatch is unknown, never a re-quote. Two tables are two sets of books, and re-pricing a caller's reconciled $0.50 row at falaw's published $0.01 is a 50x under-quote wearing the clothes of a price drop. So a call you quoted with your own llm_rates= needs your own Pricer to re-quote it: reprice_plan(plan, pricers={**DFLT_PRICERS, "llm_rates": mine}), where mine.table is the identity the basis carries.

Fan-out: partial results, bounded concurrency, per-call isolation

A Plan is a fan-out — 200 panels is 200 CallPlans in one Plan — so execute_plan's list[Artifact] has no room to say "call 7 failed, here are the other 199". execute_plan_isolated does:

from falaw import execute_plan_isolated

report = execute_plan_isolated(plan, concurrency=8)

for outcome in report.outcomes:  # always one per call, in plan order
    if outcome.ok:
        save(outcome.artifact)
    elif outcome.status == "failed":
        retry_later(outcome.call, outcome.error)  # retry this call verbatim
    else:
        replan(outcome.call, outcome.reason)  # blocked: its input does not exist

report.estimated_spend_usd, report.has_unknown_costs, report.cache_hit_savings_usd

Three states, not two. Failed means the call raised and can be retried as it is. Blocked means it never ran — a "<from N>" placeholder whose producer did not succeed — so it has to be re-planned, not retried. A caller that cannot tell them apart retries a call whose input does not exist.

concurrency bounds how many calls are in flight; independent calls run in parallel, and a call is never started beside the producer it consumes. It defaults to 1 (sequential) everywhere, here and in render_scene, because every call is a paid request: parallelism is opt-in. Weigh the vendor's rate limit and memory — materializing a media result peaks at roughly twice the asset's size, and concurrency multiplies that.

execute_plan keeps the plain list[Artifact] contract and re-raises the first failure unwrapped, so falaw's typed error hierarchy still reaches the caller. It is execute_plan_isolated(...).artifacts_or_raise() — one engine, two policies.

render_scene(..., concurrency=4) runs shots and beats in parallel through the same kind of thread pool (fal calls are HTTP-bound). Use iter_render_scene(...) to yield (kind, result) pairs as each unit completes — handy for live UI updates.

Three cache modes, because a re-run is not a cache bypass

execute_plan reads the cache and writes to it. Those are separate decisions, and it takes two flags rather than one so a caller can ask for either half.

	cache read	cache write	ask for it with
reuse, then keep — the default	yes	yes	execute_plan(plan)
re-run, but keep	no	yes	execute_plan(plan, refresh=True)
do not touch the cache	no	no	execute_plan(plan, use_cache=False)

refresh=True is what belongs behind a force / "re-verify this" switch. use_cache=False re-runs and discards, so the next consumer of the same plan pays for the result the forced run already bought — a guaranteed double charge, noise on an LLM plan and a real charge for nothing on an image or video one. Reach for use_cache=False only when you genuinely want no cache interaction at all.

use_cache=False, refresh=True raises: there is no key to write under when the cache is off.

Testing code that uses falaw

falaw content-addresses every media result, so execute_plan reads the bytes behind each result URL. A suite that stubs the fal response but not the asset transport is resolving its made-up URLs for real — and it passes while doing so, because a failed fetch degrades to a URL-only artifact with a UserWarning rather than raising.

falaw ships the fake, so nobody has to write one. Three autouse fixtures, each re-exportable into a conftest.py in one line:

# conftest.py
from falaw.testing import (  # noqa: F401
    fake_assets,  # url -> bytes, served from memory
    isolated_falaw_cache,  # cache / content store / url-index under tmp_path
    no_outbound_network,  # refuses AND records any non-loopback connection
)

def test_two_urls_one_content_address(fake_assets):
    shared = fake_assets.serve("http://cdn/a.png", b"identical bytes")
    fake_assets.serve("http://cdn/b.png", shared)  # pin explicit bytes
    fake_assets.fail("http://cdn/gone.png")  # 404, as an expired asset does
    ...
    assert fake_assets.fetched == [...]  # assert on the record

fake_assets installs itself through falaw.content.using_url_fetcher, the public transport seam, so it covers execute_plan, execute_plan_isolated, materialize_asset and content_ref_for_url at once — including calls your own public API makes, which have no asset_fetcher= argument to pass. An explicitly passed fetcher= still wins. file:// URLs are not faked (they are not the network); pin one explicitly if you want it served from memory.

using_url_fetcher is not test-only: it is also how you give falaw an authenticated, retrying, or mirrored transport in production, in one place instead of at every call site.

Two rules that are easy to get wrong:

• Don't reach for FALAW_FETCH_ARTIFACT_BYTES=0 to make a suite offline. It works by turning content addressing off — the suite goes hermetic and stops testing the feature.
• Assert on the record, not on an exception. falaw absorbs a failed fetch by design, so a guard that only raises is invisible. That is why both the fake and the network guard record every attempt, and why no_outbound_network fails the test at teardown.

Marker live_api opts a test back into the real transport and no network guard. A suite with different marker names builds its own fixtures from make_fake_assets_fixture / make_no_outbound_network_fixture.

Architecture

Single source of truth: a ToolSpec dataclass per tool. From it we derive every external surface:

falaw.registry  ──► bridges/skill.py    ──►  .claude/skills/falaw/SKILL.md
                ──► bridges/mcp.py      ──►  MCP server          (planned)
                ──► bridges/service.py  ──►  qh HTTP service     (planned)
                ──► (UI)                                          (planned)

Adding a new surface is a new bridge module, never a re-implementation of the operations.

Self-improvement loop

Every session can read and write the agent journal at ~/.config/falaw/journal/. The Claude skill instructs Claude to:

1. Read recent entries before novel work.
2. Write a note / issue / improvement when something surprises it.

call_fal auto-journals failures with the application id and arguments, so the next session recognizes the trap.

Layout

falaw/
  base.py            ToolSpec, ModelRecord
  core.py            call_fal: subscribe + auto-journal
  registry.py        register_tool, list/get/pick model
  reprice.py         reprice_plan: re-quote a saved Plan at today's rates
  results.py         Asset, Result, parse_response
  session.py         Session
  journal.py         file-backed journal
  operations/
    images.py        generate_image
    audio.py         text_to_speech
  bridges/
    skill.py         render Claude SKILL.md from registry
    mcp.py           (stub)
    service.py       (stub)
  data/
    models.json      seed catalog
    skills/falaw/    generated skill files (shipped with package)
misc/
  docs/              aggregated fal.ai docs (3MB md, llms.txt, llms-full.txt)
  regenerate_skill.py
tests/

Regenerate the skill after adding a tool

python misc/regenerate_skill.py

Writes falaw/data/skills/falaw/SKILL.md and .claude/skills/falaw/SKILL.md.

Status

v0 --- functional core, real Claude skill, stubs for MCP and HTTP service. The bridges share the same registry, so filling in the stubs is additive.

Roadmap

See misc/docs/roadmap.md for the ordered work and the standing constraints. In short, four tracks:

1. Content addressing --- landed (falaw#14). Artifact.asset_id is the SHA-256 of the artifact's bytes and the per-call cache key holds upstream content hashes, not fal CDN URLs. fal states that every upload gets a unique URL and that expired files are permanently deleted, so URL-keying meant a byte-identical regeneration missed the cache and a stored response decayed into a dead link. Bytes now route through lacing.ArtifactStore.put_blob_stream. Still open: the plan-time cache peek keys on unresolved arguments (D2), which lands with track 2.
2. Backend-parametric Plan --- CallPlan.backend + registry dispatch in execute_plan, so a second execution backend is still priced, cached and dry-runnable.
3. Licence-and-terms ledger --- per (model, backend), queried at plan time, unknown means refuse.
4. Cost data from the vendor --- read fal's pricing API instead of hand-maintaining data/models.json (19 of 40 records currently carry no structured price at all).