Skip to content
Skip to content

Read Compute jobs synchronously with Python

How to read a Compute job with the synchronous Python client, handle paging and errors, and recover an uncertain submission safely.

Updated

Use LithiClient for one synchronous job read. Your existing agent keeps its model, goal, project context and final decision; this client exposes the current API operations. Inspect the returned projection and recover an uncertain POST with its original idempotency key.

The example below injects a transport callable, so it stays local and uses synthetic response data. A client built with your credential and no injected transport sends requests to your account.

1. Construct LithiClient

Import LithiClient from lithi and construct it with base_url and a credential callback. The callback may return the key string or {"type": "bearer", "token": ...}. Keep the credential in your protected store. The client allows HTTPS and exact localhost HTTP for an injected local transport.

The constructor is:

LithiClient(base_url, credentials, transport=None, timeout=30, max_retries=2, sleep=time.sleep)

The request body limit is 64 KiB, the response limit is 4 MiB, and max_retries may be from 0 through 8. The client retries 408, 425, 429, 500, 502, 503, and 504.

2. Make one synchronous read

This complete example uses an injected transport. It constructs the same synchronous call as a normal integration, but it never opens a connection.

import json
from lithi import LithiClient

KEY = "lthi_cmp_" + "0123456789abcdef" + "." + "0123456789012345678901234567890123456789012"
JOB = "550e8400-e29b-41d4-a716-446655440000"

class Response:
    status = 200
    headers = {"Content-Type": "application/json"}
    body = json.dumps({"ok": True, "data": {
        "job_id": JOB, "state": "queued", "state_version": 1,
        "digest": "a" * 64, "p_next_action": "none",
    }}).encode()

def transport(method, url, headers, body, timeout):
    assert method == "GET"
    assert url.endswith("/v1/jobs/" + JOB)
    assert body is None
    return Response()

with LithiClient("https://api.example.test", lambda: KEY,
                 transport=transport, max_retries=0) as client:
    job = client.getJob(id=JOB)
    print(job["state"], job["state_version"])

The output is queued 1. That is job metadata from the synthetic response, not completed work. getJob performs one synchronous read. It doesn't wait for a terminal state or return a result artifact.

3. Use operation names and exact arguments

Methods use the registered camel-case names: createQuote, createJob, getJob, cancelJob, listJobEvents, getJobResult, createBatch, getBatch, listBatchResults, prepareArtifact, completeArtifact, listCapabilities, listUsage, getReceipt, listAgentEvents, getImpactReport, and the four compatibility methods.

Pass a POST body as the first positional argument or as body=, and pass its stable idempotency_key= separately when required. createJob takes its key from the body and rejects a conflicting keyword value. Resource reads use id=, while paged reads use limit= and optional cursor=. Unknown arguments are rejected.

4. Read metadata, then recover safely

The returned job projection includes its ID, state, state version, digest, and next action. A result projection contains encrypted artifact metadata and digests. Retrieve authorized bytes through the documented artifact flow, then verify their digest and media type; a JSON projection isn't the artifact itself.

LithiComputeError separates the SDK code from HTTP status, server code, and next action. A timeout becomes SDK_TIMEOUT; caller code remains synchronous. A timeout or network failure doesn't prove that a submitted job did not exist. Preserve the original idempotency key and body, then read the existing job before considering a retry. Use cancelJob for an explicit cancellation request.

getImpactReport(period_id=..., revision=..., snapshot_digest=...) is a schema-backed reporting read. The client validates the selected report's period and pinned revision or snapshot digest under the finance-automation credential boundary and approved reporting access; it does not submit Compute work or add a sixth MCP tool.

See the compatibility matrix for matching versions and the Work API reference for wire contracts.

Use the selected local Python helpers

The Python package supplies the wire client plus selected local data and result helpers. It does not carry the complete TypeScript preparation, workflow or delivery steps, and its package name does not imply a public package-registry install. Check the client matrix before building around those boundaries.

Local taskPython surfaceBoundary
Validate records and stream supported formatsvalidate_schema, validate_records, iter_jsonl, write_jsonl, iter_csv, encode_csv, iter_parquet, iter_arrowThe adapters enforce their declared limits and return local data; they do not approve transfer or execution.
Read and verify a result bundleverify_result_bundle, compute_result_bundle_digest, parse_partition_manifest, read_artifact_rowsThese helpers inspect the bundle and its parts; they do not replace the complete TypeScript delivery path.
Check a local destinationLocalDirectoryDestination, check_destination_file_setThe destination is selected by your application and remains local.
Read a schema-backed impact reportgetImpactReportThe client validates the ComputeImpactReportV1 response and the requested period, revision or snapshot digest.

The Python client therefore supplies useful data and result boundaries, while pseudonymization, workflow planning, sealing and full delivery remain caller responsibilities. Your agent keeps its goal, context and final decision; the selected local helpers support preparation and result handling. For a concrete support-ticket flow, start with What leaves your machine, then compare result verification with result delivery. Use the Kit helper reference when you need the exact TypeScript and Python surfaces side by side.

See Kit contents for the supplied modules and workflow guides.

Was this page helpful?