receive_task

Part of Reusable workflows in the Python API reference.

Related guidance: shipped-tooling.

Runnable examples: receive_task.py.

from picogrid_ecn_client.workflows import receive_task

Register an echo handler and clean it up after task_limit requests.

async def receive_task(
    client: ECNClient,
    *,
    entity_id: UUID,
    command: str,
    task_limit: int = 0,
    on_event: Callable[[TaskRequestContext], None] | None = None,
    on_registered: Callable[[TaskRegistration], None] | None = None,
) -> ReceiveTaskResult

Parameters

Name	Type	Default	Description
client	ECNClient	—	Configured SDK client used to register the handler.
entity_id	UUID	—	Canonical identity that owns the task handler.
command	str	—	Exact command name accepted by the handler.
task_limit	int	0	Requests to handle before cleanup; zero waits until interrupted while retaining only the most recent configured watcher buffer.
on_event	Callable[[TaskRequestContext], None] | None	None	Optional callback invoked for every accepted request context.
on_registered	Callable[[TaskRegistration], None] | None	None	Optional callback invoked after registration succeeds.

Returns

Type	Description
ReceiveTaskResult	Registration metadata plus all positive-limit requests, or the most recently retained contexts and results from an unbounded receiver.

Raises

Exception	Description
ValidationError	If task_limit is negative or a request arrives after the configured limit has been accepted.
ECNClientError	If registration, request handling, or cleanup fails.

Source · workflows/tasks.py