Data pipelines¶

Available · pip install agent-adapter (no extra)

Airflow, Prefect, Dagster, Ray and Spark run your code on workers, often in other processes. Build the executor on the worker, call tools from the task, return a small JSON result.

File	Does
`examples/backend_pipeline.py`	a per-process `backend()`, an async task, a sync task, a Spark partition body
`tests/docs/test_backend_handler_examples.py`	runs them offline

Data enters as task arguments, goes through run_many to the tools as one run, and leaves as the task's return value (rates and the run's cost).

One executor per worker process¶

@functools.cache
def backend() -> Backend:
    """One per worker process: never pickle an Executor into a task, build it there."""
    return build_backend()

An Executor or Registry holds locks, so it cannot be pickled. Never pass one into a task, a Ray remote function or a Spark closure: pass names and arguments, and call backend() inside.

Task bodies¶

async def shipping_rates(regions: list[str]) -> dict[str, Any]:
    """An async task body (Airflow 3 `@task`, Prefect 3 `@flow`/`@task`, Dagster async op)."""
    executor = backend().executor
    calls = [ToolCall(name="shipping_rate", arguments={"region": r}) for r in regions]
    async with executor.run_scope() as ctx:  # one run: one budget and one cost line
        results = await executor.run_many(calls, ctx, concurrency=8)
        cost = backend().ledger.total_usd(ctx.run_id)
    rates = {region: r.value if r.ok else None for region, r in zip(regions, results, strict=True)}
    return {"rates": rates, "cost_usd": cost}  # small and JSON-safe: fine for XCom/results


def shipping_rates_sync(regions: list[str]) -> dict[str, Any]:
    """The same for a sync task, a Ray task or a Dagster sync op."""
    return run_sync(shipping_rates(regions), timeout=TIMEOUT_S)

Tool	Write	Runs
Airflow 3	`@task async def ...` (async `PythonOperator`)	`await shipping_rates(...)` on the worker's loop
Airflow, sync `@task`	`def ...`	`shipping_rates_sync(...)`
Prefect 3	`@flow` / `@task` on `async def`	`await shipping_rates(...)`
Dagster	async `@op` / `@asset`	`await shipping_rates(...)`; sync ops `shipping_rates_sync`
Ray	`@ray.remote def ...`	`shipping_rates_sync(...)`; an actor can hold `backend()` as its tool host
Spark	`mapPartitions` / `pandas_udf`	`rates_for_partition` below

Spark partitions¶

def rates_for_partition(regions: Iterable[str]) -> Iterator[float | None]:
    """A Spark `mapPartitions` / `pandas_udf` body: one loop per partition, not per row."""
    rows = list(regions)
    rates = shipping_rates_sync(rows)["rates"]
    yield from (rates[region] for region in rows)

One run_sync per partition, not per row: each run_sync builds a loop. Each Python worker builds its own backend() once.

Rules¶

Rule	Why
build the executor in the worker (`functools.cache`)	executors do not pickle; forked workers must not share loop-bound clients
one run per task	`BudgetHook` caps it; the ledger sums its cost
return JSON-safe values	XCom, Prefect results and Dagster outputs serialise them
set `concurrency=` on `run_many`	a big batch must not open hundreds of connections
approvals in a pipeline	use `AutoDeny`, or park with `DeferredApproval` and a shared store, then resume in a later run
retries	the task framework retries the task; the tool's `retries=` retries one call