Python Workers

查看 Markdown查看 Markdown

This page shows how to start a Python Worker. It lists the Python names for the Worker settings and diagnostics. The Workers and the task queue page describes their behavior for every SDK.

Start a Worker

This example serves a Greeting Flow through a pull Worker. Its Step stores a greeting, and the greeting RPC reads it. The Worker allows eight concurrent tasks and keeps one slot for RPCs. See the Python SDK README for startup details.

async def start_pull_greeting(
    cache: BlobCache, server_address: str
) -> tuple[GreetingFlow, AsyncWorker, AsyncClient, asyncio.Task[None]]:
    flow = GreetingFlow()
    registry = Registry((flow,), allow_async_handlers=True)
    worker = AsyncWorker(
        registry,
        cache,
        WorkerOptions(
            server_address=server_address,
            pull=PullWorkerOptions(
                max_concurrent_tasks=8,
                # Keep one slot for RPCs so Flows stay responsive.
                reserved_rpc_tasks=1,
            ),
        ),
    )
    worker_task = asyncio.create_task(worker.start())
    client = AsyncClient(
        registry,
        cache,
        ClientOptions(server_address=server_address),
    )
    try:
        await worker.wait_pull_registered(timeout=30)
    except BaseException:
        await stop_pull_greeting(worker, client, worker_task)
        raise
    return flow, worker, client, worker_task


async def stop_pull_greeting(
    worker: AsyncWorker, client: AsyncClient, worker_task: asyncio.Task[None]
) -> None:
    await client.close()
    await worker.close()
    await worker_task

Example: examples/python/dex_examples/production/pull_workers.py

Readiness

Python provides wait_pull_registered(timeout) and pull_registration_state() on both Worker and AsyncWorker in SDK 1.14.0; await the AsyncWorker wait. Python's stop_on_permanent_stream_rejection provides the same opt-in shutdown behavior as Go.

A PullRegistrationError reports STEP_AND_RPC or RPC_ONLY as its slot kind.

Call the boolean helper from your probe endpoint and return 200 when true or 503 when false.

def is_pull_greeting_ready(worker: AsyncWorker) -> bool:
    return worker.pull_registration_state().is_ready

Example: examples/python/dex_examples/production/pull_workers.py

Capacity and shutdown

Setting	Python name
Concurrent tasks	max_concurrent_tasks
Reserved RPC slots	reserved_rpc_tasks
Stop drain timeout	stop_drain_timeout
Shutdown	stop

The idle stream option, idle_streams_per_queue, is deprecated. It is still accepted and has no effect.

Reserved RPC capacity

The reserve-decline counter, pull_reserved_rpc_declines, is deprecated. It remains for source compatibility and always reads zero.

This diagnostics helper exposes registration readiness. A full Worker stays ready.

@dataclass(frozen=True)
class PullGreetingDiagnostics:
    is_ready: bool


def get_pull_greeting_diagnostics(worker: AsyncWorker) -> PullGreetingDiagnostics:
    return PullGreetingDiagnostics(
        is_ready=worker.pull_registration_state().is_ready,
    )

Example: examples/python/dex_examples/production/pull_workers.py