Skip to content

README.md

Latest commit

 

History

History
359 lines (306 loc) · 19 KB

README.md

File metadata and controls

359 lines (306 loc) · 19 KB
Raw

grida

The Python SDK for Grida FX. Import it as grida.fx.

It does three things: it authors node types and workflows (node, Workflow), it hosts Python node bodies, builders, stand-ins and project adapters when the engine runs them (python -m grida.fx.host, started by the engine; TypeScript ones run in @grida/fx's host on Node), and it drives the grida-fx engine from Python. The engine is the only place with engine logic: planning, identities, the cache and budgets all happen in grida-fx, never in this package.

Install

pip install grida

Python releases use stable versions starting at 0.1.0 (uv pip install grida also works). Each wheel carries the engine for its platform: macOS on Apple silicon (11 or later) and on Intel (10.12 or later), and Linux with glibc 2.28 or later on x86_64 and aarch64. Windows is not supported: the engine's runner uses Unix process groups and signals. There is no source distribution, and the wheels install no console command: the engine runs as python -m grida.fx.

From a checkout of the repository, python3 tools/build_engine.py builds the engine and puts it where a wheel carries it: python/src/grida/fx/_bin/grida-fx in the checkout (ignored by git). An editable install of the checkout's python/ folder (uv sync --project python, or a path dependency in another project) then finds it. Run it again after every pull.

The engine from the command line

python -m grida.fx plan workflows/gallery.yaml --poster inputs/poster.png
python -m grida.fx run workflows/gallery.yaml --poster inputs/poster.png --live --max-usd 10

python -m grida.fx <verb> … takes the same arguments as grida-fx and exits with its status (on POSIX it becomes the grida-fx process), so a Python installation needs no Node unless the project has TypeScript nodes, builders, stand-ins or adapters: those need Node 24 or later (GRIDA_FX_NODE or node on PATH) and the project's own @grida/fx (Getting started). The binary is the first of:

  1. GRIDA_FX_BIN, when set (a path; relative to the current directory);
  2. the one inside the package (grida/fx/_bin/grida-fx): shipped in the wheel, or built into a checkout by tools/build_engine.py;
  3. grida-fx on PATH.

With none of them, the command prints grida-fx: … naming GRIDA_FX_BIN and exits 2.

Planning and running from Python

from grida.fx import plan, run

planned = plan("workflows/gallery.yaml", inputs={"poster": "inputs/poster.png"}, max_usd=10)
print(planned.ok, planned.problems, planned.estimate())  # estimate: (low, high) in US dollars
for phase in planned.phases():
    print(phase["phase"], phase["steps"], phase["high_usd"])

result = run(planned, live=True)  # runs the plan with the target and options it was planned with
print(result.ok, result.cost, result.incomplete, result.failed)
for key, image in result.outputs["images"].items():  # a keyed collection
    verdict = result.steps[f"entity['{key}'].review"].facts["verdict"]
    print(key, image.path, verdict)
missing = result.deliver({"images": "out/{key}.png"})
  • Targets. A workflow file, a workflow id, a builder file.py:function or file.ts:function (its arguments as arguments={"level": "docks"}, the command line's --arg), or a Plan. A Workflow object is refused with TypeError: name the builder that returns it.
  • Options. inputs (a mapping, written to a temporary inputs file in cwd, so relative paths in it mean what they mean on the command line), input_files (--inputs), routes (--routes), max_usd (--max-usd); run adds live, yes_up_to and one of name (create-only), resume (require-existing) or run_dir (explicit-folder create/resume). cwd is the engine's working directory (default: the current one); every relative path is relative to it. A Plan takes no planning options when it runs: plan again to change them.
  • plan asks the engine for the plan (grida-fx plan --json) and its price (grida-fx price). Plan.document is the fx-graph-v1 document; ok, problems (where, message), estimate() and phases() read it. Planning never spends.
  • run plans first, and raises PlanRefused (the plan is refused: and one line per problem) before any run folder exists. A failed step does not raise: check result.ok and result.failed. The result is read from the run folder's events.jsonl: ok, incomplete, cost (the folder's charge over every invocation, in US dollars), run_dir, failed, failures, stopped, stand_in, outputs and steps.
  • Failures. result.failures holds a Failure for each failed instance id: its path, message, code (the protocol's name of the error, such as node_failure or call_failed; None for a timeout, an assertion or a skip), the node facts it reported, and skipped (it never ran, because something it reads failed). result.stopped says why a run stopped early.
  • Files. An output file has path (its copy in the planning project's store, <cache>/files/…: the project above the workflow file for a .yaml target, else above cwd; read it, never write it), digest, kind, name, size, key, read_bytes() and copy_to(). A keyed collection is a dict, a list a list, an absent output None.
  • Delivering. result.deliver({output: path}) copies output files out of the run; {key} in the path names each element of a collection or list (keys are made safe as run folders make them). A file that already holds the same bytes is left alone. It returns the outputs that hold no file.
  • Errors. The engine's own errors (exit status 2: a workflow it cannot read, a bad option) raise FxError with its message; so does a run it refuses to start (refused: …).
  • Async. plan_async and run_async do the same inside an event loop; plan and run cannot be called inside a running one. Cancelling run_async requests graceful engine cancellation once with SIGTERM and waits for cleanup before unwinding; repeated task cancellation does not force the engine. A stuck cleanup can keep the task pending.

Saved runs and observation

load_run(target, verify=False, cwd=None) returns a RunRecord containing the engine's inspection and absolute run_dir. It accepts a folder, workflow/name, or workflow ID for its newest-created run. The selected folder stays pinned even if a newer run appears. It has no invocation exit code.

Use record.snapshot() for a consistent plan/event prefix and record.events(after=cursor, limit=256) for bounded reads. Their async forms are snapshot_async and events_async; loading has load_run_async. Documents retain wire field names. record.follow(after=cursor, poll_interval=1.0) is an async generator of nonempty batches, continuing through completion and later resumes until broken or cancelled. Observation cancellation never cancels execution.

Reader failures raise RunObservationError with code and document; malformed responses raise FxError. Snapshot and inspection are separate samples. See Working with runs from an SDK for cursor-safe attachment, named execution/resume and reader lifetime. This describes current source, not a guarantee that an older published package has these APIs.

Run control

inspect_control(run) and cancel(run, invocation=None, wait=False, timeout=None) inspect or cancel an exact run folder or WORKFLOW_ID/NAME. Their async counterparts are inspect_control_async and cancel_async. RunControlResult preserves the wire contract's lower_snake_case fields. Structured failures raise RunControlError with code and result; unsupported binaries or malformed responses raise FxError. Choose run_dir to control a pending SDK invocation through these separate operations.

See Stopping and continuing a run for guarded requests, wait timeouts and verified local completion. Local completion never verifies remote provider completion. These APIs describe current source; check your installed version.

Listing and removing runs

list_runs(workflow=None, cwd=None) returns the planning project's runs, newest first, as RunEntry objects with the wire contract's fields and run_dir, the folder resolved against cwd. remove_runs(runs, preview=False, cwd=None) removes exactly the named runs (folders, run_dir values or WORKFLOW_ID/NAME); preview=True only reports what would go. There are no filters: choose from the list yourself. The async forms are list_runs_async and remove_runs_async. A run that is active or the last holding a pick is refused in the RunRemoval result, not raised; a target that is not a run raises RunRemovalError with code and run, and nothing is removed. An applied removal whose result cannot be read raises FxError saying it may already have taken effect, with the engine's stdout.

Stand-ins: running without a provider

A test runs a workflow's paid calls offline by answering them itself:

from collections import Counter
from pathlib import Path

from grida.fx import DECLINE, Answer, CallRefused, run

asked = Counter()


def answer(call):  # a plain function, or an async def
    asked[call.capability] += 1
    if call.capability == "image.generate":
        return Answer(files={"image": Path(f"tests/pictures/{call.request['size']}.png")})
    if call.capability == "structured.generate":
        return Answer.json({"title": "The lighthouse"})
    if call.capability == "agent.turn":
        return Answer.submit(caption="a lighthouse at dusk")
    if call.capability == "image.edit":
        raise CallRefused("no edits in this test")
    return DECLINE


result = run("workflows/gallery.yaml", run_dir="runs/test", stand_in=answer)
assert asked["image.generate"] == 2
assert result.failures["touch_up#1"].code == "capability_refused"
  • The call. answer gets a StandInCall for each paid call the cache cannot answer: capability, route.id and route.fingerprint, key, takes and take, instance.id, .path and .step, request (every file in it an InputFile: digest, path, read_bytes(), facts), files (the same files by digest) and params (the call as the engine sent it, files as {"file": digest}).
  • The answer. Answer(files={name: …}, data=…): a file is bytes (it takes the kind the capability names), an Output of bytes or an InputFile (its own kind), or a path (the kind of its suffix). Answer.json(value), Answer.turn(text, tool_calls) and Answer.submit(**arguments) build the data of structured.generate and agent.turn. Return DECLINE to leave a call unanswered (it fails not_live); raise CallRefused or CallFailed to refuse or fail it. The engine checks every answer as it checks a provider's: an image of the wrong size fails the call.
  • One at a time. The function runs in this process, so counters and scripted answers work. Calls are answered one at a time, in the order they come.
  • Faults. Anything else it raises, or a return of None, stops the run; run then raises that exception, noting the run folder.
  • Nothing is spent, and the real store is left alone. A stand-in run is never live and takes no yes_up_to; run does not plan it first. It keeps everything in the store's stand-in/ folder, where its result's files are read, so a later stand-in run, or a resumed one, replays its answers; delete that folder to forget them. A folder resumes only in the mode it was started in, and reroll and pick refuse it.
  • From the command line: grida-fx run … --stand-in tests/stand_in.py#answer.

Project adapters: a provider of your own

A project serves a provider FX does not ship with its own adapter, which the engine runs in a Python node host of the project (adapters.md §9; the whole job, from the declaration to checking it offline, is the guide's Community providers). fx.yaml declares the provider and names the adapter, <path>#<name>:

providers:
  acme:
    credential: { variable: ACME_API_KEY }
    base_url: { default: "https://api.acme.test/v1" }
    download_hosts: [files.acme.test]
    adapter: adapters/acme.py#acme
# adapters/acme.py
import asyncio

from grida.fx import (
    Accepted,
    Answered,
    Ended,
    Failed,
    FetchError,
    NotReceived,
    Refused,
    Uncertain,
    Unreachable,
    adapter,
)


async def send(call):  # one structured.generate request, sent once
    try:
        response = await call.fetch(
            f"{call.base_url}/chat",
            method="POST",
            json={"model": call.route.model, "prompt": call.request["prompt"]},
            credential=True,
            timeout_s=60,
        )
    except FetchError as error:  # refused, not_sent or after_send (adapters.md §3.2)
        if error.phase == "refused":
            return Refused(error.reason)
        if error.phase == "not_sent":
            return NotReceived(error.reason)
        return Failed(error.reason, retryable=True)
    if response.status != 200:
        return Failed(f"acme returned HTTP {response.status}", retryable=response.status >= 500)
    body = response.json()
    return Answered.json(body["answer"], cost_usd=body.get("cost_usd"))


async def submit(call):  # a video.generate job, submitted once
    try:
        response = await call.fetch(
            f"{call.base_url}/videos",
            method="POST",
            json={"prompt": call.request["prompt"]},
            credential=True,
            timeout_s=60,
        )
    except FetchError as error:
        if error.phase == "refused":
            return Refused(error.reason)
        if error.phase == "not_sent":
            return NotReceived(error.reason)
        return Uncertain(error.reason)  # acme may have taken the job
    if response.status != 202:
        return Uncertain(f"acme returned HTTP {response.status}")
    return Accepted({"id": response.json()["id"]})


async def collect(call):  # polls with its own clock, until its own deadline
    for _ in range(120):
        try:
            response = await call.fetch(
                f"{call.base_url}/videos/{call.handle['id']}", credential=True, timeout_s=30
            )
        except FetchError as error:
            return Unreachable(error.reason)
        job = response.json()
        if job["status"] == "failed":
            return Ended("acme could not make the video", cost_usd=job.get("cost_usd"))
        if job["status"] == "done":
            clip = await call.fetch(job["url"], lane="download", timeout_s=120)
            return Answered(files={"video": clip.body}, cost_usd=job.get("cost_usd"))
        await asyncio.sleep(5)
    return Unreachable("acme's video is still being made")


acme = adapter(
    requests=["structured.generate"],
    jobs=["video.generate"],
    send=send,
    submit=submit,
    collect=collect,
    version=1,
)
  • The declaration. adapter(requests=…, jobs=…, send=…, submit=…, collect=…, version=…) at a module's top level. send serves each capability of requests; submit and collect each of jobs. version is an integer of at least 0, which grida-fx lock locks as a node type's; without one, every edit to the module or a project module it imports changes the identity of its routes. Planning loads the module, so its top-level code runs then too.
  • The call. send, submit and collect each take an AdapterCall: capability, route.id, .model, .provider and .contract, base_url, key, attempt, takes and take, handle (in collect), request (every file in it an AdapterFile: digest, kind, size, path, read_bytes()) and files (the same files by digest). A plain function or an async def; calls are answered at once, in any order.
  • The outcome. send returns Answered, NotReceived (provably not received: sent again), Refused (nothing was sent) or Failed (retryable says whether to try again); submit returns Accepted(handle), Uncertain, or one of the last three; collect returns Answered, Ended or Unreachable. cost_usd is the provider's own figure: left out, the engine charges the whole hold or the route's cap. An answer's files are as a stand-in's (Answered.json(value) for structured.generate), and the engine checks them as it checks a provider's. Anything raised, or another value, ends the call as if the host were lost.
  • The network. await call.fetch(url, method=, headers=, json=, body=, content_type=, lane=, credential=, timeout_s=, max_response_bytes=) makes one exchange through the engine: never retried, no redirect followed, at most 300 seconds and 64 MiB. With credential=True the engine attaches the declaration's key, only to the base URL's origin, so the code never handles it; downloads (lane="download") are held to download_hosts. It returns a FetchResponse (status, headers, body, .json(), .text()) or raises FetchError (phase, reason). An adapter may use its own HTTP client instead, keeping the same rules itself, and sending nothing while GRIDA_FX_NETWORK is off, which the engine cannot stop for it; its host's environment has its own declaration's key and no other's.
  • Testing. grida-fx run … --live --max-usd 1 with GRIDA_FX_NETWORK=off runs the adapter with every fetch refused before it leaves; AdapterCall(params, fetch=…) builds a call for a unit test, with a coroutine function standing in for the engine.

Standard bodies

grida.fx.std holds the Python bodies of FX's free standard node types. The engine declares them, and runs them in this package's node host when a workflow uses one (fx/image.resize@1, …); nobody imports them to use them.

Type What it does
fx/image.mirror_repeat@1 mirrors a picture onto itself on x, y or both, at most 16384 px a side
fx/image.check_alpha@1 judge: transparent wants a fully transparent pixel, anything else every pixel opaque
fx/image.check_size@1 judge: the picture's width and height against the ones asked for
fx/image.resize@1 longest_side, or width and height; Lanczos
fx/image.crop@1 a box in fractions, or the first box mark of a region, grown by padding
fx/image.pad@1 centres a picture on a transparent canvas
fx/json.merge@1 a shallow merge of JSON objects, a later key winning
fx/files.copy@1 the same bytes and kind
fx/package@1 lays files out by destination path ({key} takes a keyed collection) with a manifest

Every picture the bodies write comes from Pillow's own PNG encoder, so pillow is pinned to one exact version: another build may write other bytes, which would change every identity and request downstream of them. grida.fx.std.body_of("fx/<name>@1") returns a body; grida.fx.std.pictures holds the Pillow reads and writes behind ctx.read.image and ctx.out.png.