Skip to main content

Federated Jobs Service

client.jobs submits local or federated Ray jobs. Existing callers may keep using run, submit_async, wait, and cancel without delegated access.

Governed delegated access​

Use DelegatedAccess to name exact receiver resources and operations. The SDK validates the same closed vocabulary as Core before making a request. Wildcards, duplicates, unknown operations, and an empty access block are rejected.

from kamiwaza_sdk import (
    DatasetDelegatedAccess,
    DelegatedAccess,
    ModelDelegatedAccess,
)

job_id = client.jobs.submit_async(
    target_cluster="receiver",
    entrypoint="python summarize.py",
    delegated_access=DelegatedAccess(
        datasets=(
            DatasetDelegatedAccess(
                urn=DATASET_URN,
                operations=("discover", "read", "retrieve"),
            ),
        ),
        models=(
            ModelDelegatedAccess(
                deployment_id=DEPLOYMENT_ID,
                operations=("discover", "chat"),
            ),
        ),
    ),
)

Passing a mapping with the same shape remains supported. Omitting delegated_access preserves the ordinary job wire format and execution path.

Approved Python dependencies​

An isolated delegated job can request exact Python packages from the receiver operator's approved catalog:

job_id = client.jobs.submit_async(
    target_cluster="receiver",
    entrypoint="python summarize.py",
    delegated_access=access,
    python_packages=["humanize==4.13.0"],
)

Only exact name==version coordinates are accepted. URLs, version ranges, extras, markers, duplicate package names, and more than 32 packages are rejected before the request is sent. The receiver resolves each coordinate to its operator-approved SHA-256 hashes and installs it from the configured private Python index. Repository URLs, credentials, and trust material are never part of the SDK request or the Ray job environment. Air-gapped deployments can point the operator-managed repository Secret at an internal PEP 503/691-compatible mirror without changing job code.

The older pip= convenience parameter belongs to ordinary Ray runtime environments. Use python_packages= for isolated delegated jobs.

In-job runtime client​

Managed job code uses JobRuntimeClient, which connects only to the private Unix socket created for that job:

from kamiwaza_sdk import JobRuntimeClient

with JobRuntimeClient.from_environment() as receiver:
    datasets = receiver.datasets.list_granted()
    models = receiver.models.list_granted()
    rows = receiver.retrieval.collect(dataset_urn=DATASET_URN)
    answer = receiver.models.chat(
        deployment_id=DEPLOYMENT_ID,
        messages=[{"role": "user", "content": summarize(rows)}],
    )

models.chat and models.stream_chat take optional generation options. Set enable_thinking=False to turn off a reasoning model's thinking section, and max_tokens (1 to 32,768) to cap the reply:

from kamiwaza_sdk import JobChatOptions

answer = receiver.models.chat(
    deployment_id=DEPLOYMENT_ID,
    messages=[{"role": "user", "content": summarize(rows)}],
    options=JobChatOptions(enable_thinking=False, max_tokens=2048),
)

A plain mapping such as {"enable_thinking": False} works too. Unset options keep the model's defaults. On a model that thinks, thinking counts against max_tokens, so a small cap without enable_thinking=False can return an empty reply. A receiver that predates generation options refuses a request that sets them, which raises JobAgentRequestRefusedError (see Errors below); a request without options works on any receiver. Options that set no field, such as {} or JobChatOptions(max_tokens=None), count as no options.

Streaming variants are retrieval.stream and models.stream_chat. Each operation re-reads the platform-owned agent identity and verifies the connected Unix peer before sending a request, so a restarted agent is used safely on the next operation. The client never falls back to a bearer token, password, API key, refresh token, or public HTTP endpoint.

Errors​

Denied resources raise DelegatedResourceNotFoundError; lifecycle revocation raises DelegationRevokedError; attestation failure raises JobIdentityUnavailableError; grant denial raises DelegatedOperationDeniedError; and authority, dependency, or replay-store outages raise DelegationUnavailableError.

A dataset read that would return more than 10,000 rows raises DelegatedRowLimitExceededError. The receiver refuses the read rather than truncating it, and retrying the same read does not help: read a smaller dataset or view instead, one the job's delegated_access names.

The error subclasses DelegationUnavailableError, so existing handlers still catch it, including one that retries outages. Catch it first:

from kamiwaza_sdk.exceptions import (
    DelegatedRowLimitExceededError,
    DelegationUnavailableError,
)

try:
    rows = receiver.retrieval.collect(dataset_urn=DATASET_URN)
except DelegatedRowLimitExceededError:
    raise  # refused, not truncated: read a smaller dataset or view instead
except DelegationUnavailableError:
    rows = retry_later()  # your own retry for outages

Against a receiver that predates this refusal, the error never occurs, and a read past the cap can come back cut to the first 10,000 rows, with no error. On such a receiver, a result of exactly 10,000 rows may be incomplete.

A request the job agent cannot read raises JobAgentRequestRefusedError, also in kamiwaza_sdk.exceptions: the agent rejected the request frame or timed out reading it. On a platform older than this SDK, the likely cause is a field the agent predates, such as chat options. A request refused for that reason is refused every time, so leave the field unset or upgrade the platform. The error subclasses DelegationUnavailableError, so existing handlers still catch it, and a handler that retries outages retries this refusal too, which helps only when reading the request timed out. To stop retrying a chat the platform refuses for its options, catch JobAgentRequestRefusedError first, as with the row-limit refusal above.