> ## Documentation Index
> Fetch the complete documentation index at: https://docs.blockx.chaintable.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Submit a task: examples

> Runnable examples of submitting tasks with blockx-py: a one-off computation that returns its result, writing a block table per block, writing a normal table, backfilling per bundle; plus the Go code for calling the WorkerService gRPC directly without the SDK.

This page turns [Task and Call](/en/concepts/task-and-call) and [Call Builder and Writer Plugin](/en/concepts/builders-and-handlers) into code. The first four examples use blockx-py (`python/pyproject.toml` in the blockx repository pins `v0.1.72`); the last section shows how to call `WorkerService` directly from Go without the SDK.

## Prerequisites

<Tabs>
  <Tab title="Local worker">
    Start the worker and `examples/local_test_service/local_proxy.py` as described in the [Quickstart](/en/quickstart). Example 1 runs locally as-is; examples 2 to 4 read and write BlockDB tables, so point `BLOCKDB_*_ADDR` at a reachable BlockDB and leave `STUB_BLOCKDB_DELAY_MS` unset.

    ```bash theme={null}
    cd examples/local_test_service
    uv sync                      # installs blockx-py and blockdb-py
    export PROXY_SOCKET_PATH=/tmp/blockx-proxy.sock
    ```
  </Tab>

  <Tab title="Production environment">
    In a notebook or function runtime, `PROXY_SOCKET_PATH` is injected by the platform; just `import blockx`. To install the packages yourself:

    ```bash theme={null}
    pip install git+https://github.com/Chaintable/blockx-py.git@v0.1.72
    pip install git+https://github.com/Chaintable/blockdb-py.git@v0.1.26
    ```
  </Tab>
</Tabs>

blockx-py only connects through the UDS proxy specified by `PROXY_SOCKET_PATH`. If the variable is not set, it raises `RuntimeError` immediately rather than silently connecting elsewhere.

## Example 1: run once and get the result back

`InputsCallConfig` + `ReturnValueHandler`. This is `examples/local_test_service/submit.py`:

```python theme={null}
import os
os.environ.setdefault("PROXY_SOCKET_PATH", "/tmp/blockx-proxy.sock")

from blockx import Function, InputsCallConfig, ReturnValueHandler, TaskBuilder

# The entry function is named `_`; each call receives one row of callList as positional arguments.
source_code = """
from blockx import LocalTestService

def _(name):
    return {"message": LocalTestService().helloworld(name)}
"""

task = TaskBuilder.build(
    InputsCallConfig(
        func=Function(source_code=source_code),
        callList=[["ct2"], ["blockx"]],   # two rows → the function is called twice
    ),
    ReturnValueHandler(),
)

# timeout is the task's execution limit on the worker (seconds), not how long the client waits.
result = task.submit(timeout=30)
if not result.task_result.success:
    raise RuntimeError(result.task_result.failure_code)
print(result.handler_result)
# [{'message': 'hello, ct2'}, {'message': 'hello, blockx'}]
```

Key points:

* `LocalTestService` is a sample capability backend built into the worker; no BlockDB is needed locally. Any self-contained pure-computation function works just as well.
* Results are collected in the completion order of the individual calls and are not guaranteed to match the row order of `callList`. When you need to map results back to inputs, have the return value carry the input.
* A task with `ReturnValueHandler` is routed to the block cluster by the SDK, because the bundle worker does not register this handler.

## Example 2: compute per block and write a block table

`BlockTableCallConfig` + `BlockTableWriteHandler(block=...)`. Subscribe to new blocks of the source table and submit one task per block:

```python theme={null}
from blockdb import BlockTable, Subscribe
from blockx import SOURCE_ROW, BlockTableCallConfig, BlockTableWriteHandler, TaskBuilder

SOURCE = "chain.trace.eth"
TARGET = "token.token_transfer.eth"


def trace_to_token_transfer(chain_id, record):
    # The function body must be self-contained: define constants inside the function and import modules inside it.
    NATIVE = "0x000000000000000000000000000000000000eeee"
    try:
        value = float(record.get("value") or 0)
    except (TypeError, ValueError):
        return None
    if value <= 0 or not record.get("tx_id"):
        return None                      # return None: this row is not written
    return {
        "id": record["id"],
        "token_id": NATIVE,
        "from_addr": record["from_addr"],
        "to_addr": record["to_addr"],
        "value": value,
        "tx_id": record["tx_id"],
    }


src = BlockTable(SOURCE)
for table_id, block in Subscribe(tables=[src]).listen():
    task = TaskBuilder.build(
        BlockTableCallConfig(
            block=block,
            triggerSources=[{
                "table": src,
                "func": trace_to_token_transfer,     # a callable has its source captured and `_ = trace_to_token_transfer` appended
                "params": ["eth", SOURCE_ROW],       # the SOURCE_ROW position is replaced with one row of the source table
            }],
        ),
        BlockTableWriteHandler(targetTable=TARGET, block=block),
    )
    res = task.submit(timeout=120)
    if res.task_result.success:
        print(f"h={block.height} written_rows={res.handler_result['written_rows']}")
    else:
        print(f"h={block.height} failed: {res.task_result.failure_code} retryable={res.task_result.retryable}")
```

Key points:

* When you pass a callable, the SDK captures the function body with `inspect.getsource` and appends the line `_ = trace_to_token_transfer`; you can also pass `Function(source_code=...)` directly, or a registered function ID string.
* `params` determines the function's positional arguments: here `chain_id="eth"`, and `record` is one row of the source table as a dict.
* `BlockTableWriteHandler` projects only the columns of the returned dict that belong to the target table; `written_rows` is the number of rows written this time.
* You can add `"operator": filter(...)` to `triggerSources` to filter and deduplicate in the Builder phase first and reduce the number of calls.

## Example 3: write a normal table

Swap the handler in example 2 for `NormalTableWriteHandler` and the target changes from a block table to a normal table (L1). It does not need `block`:

```python theme={null}
from blockx import NormalTableWriteHandler

# Plain upsert: overwrite by primary key id
handler = NormalTableWriteHandler(targetTable="token.token_transfer.eth")

# Conditional update: overwrite only when the new value is smaller in first_seen_at; out-of-order / replayed writes are naturally idempotent
handler = NormalTableWriteHandler(
    targetTable="token.token_transfer.eth",
    condition=("first_seen_at", "UpdateIfSmaller"),   # or "UpdateIfLarger"
)
```

`condition.column` cannot be the primary key `id`; an invalid value raises `ValueError` at construction time. This handler carries no block context, so the same instance can be handed directly to the backfill in example 4.

## Example 4: backfill per bundle

Historical backfill submits one task per bundle (a segment of 1000 blocks): `BlockBundleCallConfig` reads the parquet of the bundle partition, and `BlockTableWriteHandler(block_bundle=...)` overwrites the same bundle of the target table. `backfill_block_bundles` instantiates and submits bundle by bundle over a range for you:

```python theme={null}
from blockx import (
    SOURCE_ROW,
    BlockBundleCallConfigTemplate,
    BlockTableWriteHandler,
    backfill_block_bundles,
)

config = BlockBundleCallConfigTemplate(triggerSources=[{
    "table": "chain.trace.eth",
    "func": trace_to_token_transfer,      # the same function as in example 2
    "params": ["eth", SOURCE_ROW],
}])
handler = BlockTableWriteHandler(targetTable="token.token_transfer.eth")   # template form: no block / block_bundle

summary = backfill_block_bundles(18001, 18100, config, handler)
print(summary)                            # succeeded / failed bundle counts, failure code distribution, written_rows
```

To send a single bundle, construct the config and handler with the number directly:

```python theme={null}
from blockx import BlockBundleCallConfig, BlockTableWriteHandler, TaskBuilder

task = TaskBuilder.build(
    BlockBundleCallConfig(number=18001, triggerSources=[{
        "table": "chain.trace.eth",
        "func": trace_to_token_transfer,
        "params": ["eth", SOURCE_ROW],
    }]),
    BlockTableWriteHandler(targetTable="token.token_transfer.eth", block_bundle=18001),
)
result = task.submit()
```

<Note>
  The shape of `BlockTableWriteHandler` must follow the cluster: the bundle cluster needs `block_bundle=`, the block cluster needs `block=`. If you send the wrong shape, the server only rejects it in the Writer phase after all calls have run, so the SDK validates before `submit()` and raises `ValueError`.
</Note>

For day-to-day use, prefer blockx-py's higher-level `Pipeline`: one set of `triggers` + `target_table`, `backfill(block_start, block_end)` for historical backfill, and `update()` to fill gaps first and then switch to a real-time subscription. Bundle conversion, cluster routing, slot backoff, and multi-table alignment are all handled for you:

```python theme={null}
from blockx import Pipeline, Trigger, SOURCE_ROW

pipe = Pipeline(
    triggers=[Trigger(table="chain.trace.eth", func=trace_to_token_transfer, params=["eth", SOURCE_ROW])],
    target_table="token.token_transfer.eth",
)
pipe.backfill(block_start=18_000_001, block_end=18_100_000)
# pipe.update()   # long-running: fills gaps, then switches to real-time
```

## Reading the result

`task.submit()` returns a `TaskResult`:

| Field                      | Type                 | Meaning                                                                                                                                           |
| -------------------------- | -------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------- |
| `task_result.success`      | `bool`               | Whether the task succeeded. `True` only means the worker accepted it and ran it to completion; the table write is asynchronous in the background  |
| `task_result.failure_code` | `str`                | Failure code, see [Task and Call](/en/concepts/task-and-call#result)                                                                              |
| `task_result.retryable`    | `bool`               | Whether resubmitting the same task is worthwhile                                                                                                  |
| `handler_result`           | `dict / list / None` | The Writer Plugin's return value: the output array for `ReturnValueHandler`, `{"written_rows": n, "finish_time": ...}` for table-writing handlers |
| `raw`                      | gRPC response        | `GetTaskResultResponse` or an adapted `TaskUpdate`                                                                                                |

Decide whether to resubmit based on `retryable`; do not infer it from the error text:

```python theme={null}
res = task.submit(timeout=60)
if not res.task_result.success:
    if res.task_result.retryable:
        res = TaskBuilder.build(task.call_config, task.handler).submit(timeout=60)   # new task_id
    else:
        raise RuntimeError(f"{res.task_result.failure_code}: {res.error}")
```

`Task.task_id` is a fresh uuid on every `build`; log it when troubleshooting, since the worker's `task finished` log entries are searchable by it.

## Without the SDK: call gRPC directly

Go-side contributors often need to submit tasks directly to a Worker in tests or tools. The protocol is `WorkerService` in `api/grpc/worker/worker.proto`, with Go bindings in `api/grpc/worker/workerpb`. The program below goes through the three steps `RequestTaskSlot → SubmitTask → WatchTasks` against a local worker and submits the same task as example 1:

```go theme={null}
package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"time"

	"github.com/google/uuid"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"

	workerpb "github.com/Chaintable/blockx/api/grpc/worker/workerpb"
)

func main() {
	conn, err := grpc.NewClient("passthrough:///127.0.0.1:9221",
		grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	client := workerpb.NewWorkerServiceClient(conn)
	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
	defer cancel()

	taskID := uuid.NewString()

	// 1. Reserve a slot. In production, the Coordinator's ReserveWorkerSlot does this step and returns worker_addr + slot_id.
	slot, err := client.RequestTaskSlot(ctx, &workerpb.RequestTaskSlotRequest{TaskId: taskID, TtlMs: 5000})
	if err != nil {
		log.Fatal(err)
	}

	// 2. Submit the task. config is opaque JSON, parsed by the corresponding Builder / Plugin.
	submit, err := client.SubmitTask(ctx, &workerpb.SubmitTaskRequest{
		SlotId: slot.SlotId,
		Task: &workerpb.TaskInput{
			TaskId: taskID,
			FunctionCallConfig: &workerpb.FunctionCallConfig{
				Type: "InputsCallConfig",
				Config: []byte(`{"function":"","code":{"sourceCode":"def _(name):\n    return {\"message\": f\"hello, {name}\"}\n"},"callList":[["ct2"],["blockx"]]}`),
			},
			ResultHandler: &workerpb.ResultHandler{Type: "ReturnValueHandler", Config: []byte(`{}`)},
			TaskTimeoutMs: 30000,
		},
	})
	if err != nil {
		log.Fatal(err) // admission errors: InvalidArgument / ResourceExhausted / FailedPrecondition / Unavailable
	}
	fmt.Println("submitted:", submit.State) // RUNNING

	// 3. Subscribe to the terminal state. The stream first sends the current snapshot, then one update with is_terminal=true when the task converges.
	stream, err := client.WatchTasks(ctx, &workerpb.WatchTasksRequest{TaskIds: []string{taskID}})
	if err != nil {
		log.Fatal(err)
	}
	for {
		upd, err := stream.Recv()
		if err == io.EOF {
			return
		}
		if err != nil {
			log.Fatal(err)
		}
		if !upd.IsTerminal {
			continue
		}
		fmt.Println("state:", upd.State)
		for _, p := range upd.Result.GetExecuteResult().GetPluginResults() {
			fmt.Printf("%s success=%v result=%s\n", p.PluginName, p.Success, p.Result)
		}
	}
}
```

* `RequestTaskSlot` can be omitted: when `SubmitTask` carries no `slot_id`, the Worker requests a slot inline and returns `ResourceExhausted` when there is no capacity.
* When going through the Coordinator, first call `coordinator.v1.CoordinatorService.ReserveWorkerSlot(task_id)` (`api/grpc/coordinator/coordinator.proto`), then open a connection to the returned `worker_addr` and submit with `slot_id`.
* Repeating `SubmitTask` with the same `task_id` is idempotent: it returns `RUNNING` while the task is still running, and the terminal state while the result is still within the retention window.
* In tests you can use the `RequestTaskSlot` / `SubmitTask` / `PollUntilTerminal` helpers from `internal/testutil` directly; they wrap exactly the calls above.

## Common errors

| Symptom                                                                                           | Cause and fix                                                                                                                                |
| ------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------- |
| `RuntimeError: PROXY_SOCKET_PATH not set` or `UNAVAILABLE`                                        | The proxy is not running, or the socket path differs from the one in `local_proxy.py`                                                        |
| `ValueError: ... BlockTableWriteHandler carries no block_bundle`                                  | The handler shape does not match the target cluster, see the note in example 4                                                               |
| `InvalidArgument: plugin not registered: ReturnValueHandler`                                      | `cluster="bundle"` forced a task with `ReturnValueHandler` onto the bundle cluster; remove the `cluster` argument                            |
| Queueing messages keep printing / `ResourceExhausted`                                             | No idle slot. The SDK backs off and retries; when calling gRPC directly, back off yourself                                                   |
| `FAILED` with `failure_code` `CALL_FAILED`                                                        | The function raised an exception, or the code was rejected by the audit in `enforce` mode. Check the `call failed` entries in the worker log |
| `FAILED` with `failure_code` `BUILDER_FAILED`                                                     | The trigger table does not exist, `operator` validation failed, or reading BlockDB failed                                                    |
| `FAILED` with `failure_code` `PLUGIN_FAILED` and message `block deployment requires config.block` | The `BlockTableWriteHandler` shape was sent to the wrong cluster                                                                             |
| `FAILED` with `failure_code` `TIMED_OUT`                                                          | Exceeded `timeout` or the worker's default `TASK_DEADLINE_MS`; `retryable=true`                                                              |
| `InvalidArgument: task_timeout_ms ...`                                                            | `timeout` is negative or exceeds the worker's `MAX_TASK_TIMEOUT_MS` (default 8 hours)                                                        |

## Related docs

* [Quickstart](/en/quickstart): the shortest path to a local worker and proxy.
* [Task and Call](/en/concepts/task-and-call): `TaskInput` fields, states, and failure codes.
* [Call Builder and Writer Plugin](/en/concepts/builders-and-handlers): the wire shape of every config.
* [Protocols and interfaces](/en/architecture/protocols): the complete fields of `WorkerService` / `CoordinatorService`.
* [Testing](/en/development/testing): submission helpers in the process / system E2E tests.
* `docs/api_reference.md` in the blockx-py repository: all SDK parameters and the behavior details of `Pipeline`.