Python SDK

Connect once

Everything is in the drkm module. Inside an agent sandbox no connection step is needed. From anywhere else, connect() sends the job and fleet calls to an outpost’s HTTPS API. The outpost is the address you pass, else $DRKM_OUTPOST, else the active drkm console profile. The credential is token (an API key an administrator minted under Settings → API keys, or an ID token), else session (a password sign-in’s cookie), else $DRKM_TOKEN, else what drkm console login stored. HTTPS only, except localhost.

1
2
3
4
5
6
7
8
9
10
11
12
13import drkm

# 1. Reuse the active drkm console profile: address, credential and CA.
drkm.connect()

# 2. Name the outpost and sign in with an API key.
drkm.connect("https://outpost.example", token="drkm_ak_...")

# 3. A private certificate authority.
drkm.connect("https://outpost.example", ca_file="ca.pem")

print(drkm.outposts())   # the fleet: name, outpost_id, version, local, stale
print(drkm.jobs())       # your 50 most recent jobs, newest first

Records, workers and bundles

A job runs one operation over input records. Each record is {"id", "data", "files"}: an ID that is unique within the job, any JSON as data, and up to 16 workspace-relative file paths. A job holds at most 100,000 records.

package() describes a script operation: the entrypoint module.py:function, the workspace files sealed with the job (the entrypoint module among them, 64 at most), the execution mode, the models it needs and whether it is safe to replay.

• map_records calls run(record, context) once per record; the return value is that record’s output. The default.
• map_batches calls run(records, context) once per partition with an iterator over its records and yields one output per record, in order. It may be an async generator. Today these jobs run on the submitting outpost.
• single calls run(records, context) once over the whole list and is never split.

Inside an attempt, context.read(path) returns a staged input file as bytes, context.models["alias"] a model you declared, and context.sources["alias"] a declared read-only source.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25# worker.py: map_records, once per record
def run(record, context):
    reading = record["data"]["temperature"]
    return {"id": record["id"], "alert": reading > 80}


# detect.py: map_batches, once per partition, one output per record
import itertools

def detect(records, context):
    detector = context.models["detector"]
    for batch in itertools.batched(records, 16):
        images = [context.read(r["files"][0]) for r in batch]
        yield from detector.predict_batch(images)


# The operations those workers describe
import drkm

worker = drkm.package("worker.py:run", ["worker.py"])
batched = drkm.package(
    "detect.py:detect", ["detect.py"],
    execution="map_batches",
    models={"detector": "overwatch-obb-r3-int8"},
)

Submit, wait, collect

submit() returns the job ID at once. Its keyword arguments:

name

What operators read in the job list, at most 120 characters. Say what it does and where.

inputs

The records inline, or inputs_file naming a workspace JSON file when the manifest is large.

key

Makes a retry idempotent: the same key with the same work returns the same job; the same key with different work is refused.

limits

Default first, then the ceiling: partition_size 16 (up to 512), max_attempts 3 (up to 10), attempt_timeout_s 600 (up to 86400), concurrency 64 (up to 256).

allowed, preferred

A hard limit and a soft hint, as outpost IDs from outposts(). preferred=["self"] prefers the submitting outpost.

wait() polls until the job settles and returns its snapshot: state, coverage (expected, completed, failed, running, pending), error, blocked. results() writes the manifest and every accepted partition under jobs/<id>/; rows() reads them back as {"record_id", "output"} in partition order. Until a job settles, rows() is empty.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20import drkm

drkm.connect()
job = drkm.Job("your-job-id")

print(job.status())               # state, coverage, error, clarification, blocked
snapshot = job.wait(timeout_s=60) # returns the unsettled snapshot on timeout
if snapshot["state"] in ("completed", "partial"):
    print(job.rows())
# job.cancel()

# Limits and an idempotent key
job_id = drkm.submit(
    drkm.package("worker.py:run", ["worker.py"], replay_safe=False),
    inputs=[{"id": "a", "data": 1}],
    key="nightly-2026-10-27",
    limits={"partition_size": 32, "max_attempts": 2,
            "attempt_timeout_s": 300, "concurrency": 8},
    name="Nightly sensor check",
)

The dataset interface

Batched work reads like a dataframe pipeline: describe every input and the function in one chain, and the scheduler decides partitions, placement and retries. parallelize() takes inline records and read.json() a workspace file of them. A dataset has one stage: .map(), .map_batches() or .predict(). .explain() answers what the scheduler would do and runs nothing; .submit() returns a Job.

.map() and .map_batches() ship the function’s module as source, so it must be a top-level function of a module under the workspace, not the script being run. That rule is enforced against the sandbox’s workspace, so use them from code the agent runs. From a laptop, describe the same work with package() and submit(); .predict() works in both places.

1
2
3
4
5
6
7
8
9
10
11
12
13import drkm
from detect import detect   # a top-level function in a workspace module

records = [{"id": "tile-01", "files": ["uploads/tile-01.png"]}]

job = (drkm.parallelize(records)          # or drkm.read.json("inputs.json")
       .map_batches(detect, models={"detector": "overwatch-obb-r3-int8"})
       .submit(name="Ships in the uploaded harbor tiles"))
job.wait()
rows = job.rows()

# A served detector over every record's image, with no code of yours
plan = drkm.parallelize(records).predict("overwatch-obb-r3-int8").explain()

Inside a sandbox: tasks, tools, sources

Code the agent runs on its computer can do more than submit jobs. task() pushes a prompt to this outpost’s agent, to a named peer from outposts(), or to the operator-configured remote, and returns a handle; gather() waits for one or several and returns their output. Configured connections expose typed functions under drkm.tools, generated from the outpost’s current settings; the host performs each call with the turn’s authority, and the program never holds a credential or an address. drkm.sources(alias).read(...) reads bounded data from a declared source.

The same calls are available from bash through the staged drkm.py, printed as JSON.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16import drkm
from drkm import tools

help(tools)                       # the configured connector modules
# help(tools.compass)             # one module's functions

handle = drkm.task("self", prompt="Summarize the rows in report.csv")
lines = drkm.gather(handle)       # [{"t": "text", "text": "..."}, ...]

peers = drkm.outposts()           # peers a task may target: name, version, model, tools

# From bash, beside the workspace
# python3 ../drkm.py help
# python3 ../drkm.py call compass.feed '{"video_id": 7}'
# python3 ../drkm.py jobs
# python3 ../drkm.py job ID         # also: wait ID, cancel ID, results ID

Errors you will meet

• BridgeError carries the broker’s reason verbatim: a refusal, a failed task, a lost connection, or an HTTP error from the outpost.
• BridgeUnavailable means this machine has no bridge: outside a sandbox, connect() found no outpost or no credential, and the message names the fix.
• not signed in after connect(): the terminal stored its credential in the OS keyring and the script’s environment lacks the keyring package. Sign in with --credential-file or set DRKM_TOKEN.
• outside the workspace from .map(): the function is not in a workspace module. Use package() and submit() from a laptop.
• Empty rows(): the job has not settled. Call wait() first.
• A 409 on submit: the key was used for different work.

import drkm always succeeds, so help(drkm) answers everywhere.