salvor (Python)

A thin Python client for the Salvor control plane.

pip install salvor

from salvor import Client

with Client("http://127.0.0.1:8080") as client:
    agent = client.register_agent(open("agent.toml").read())
    run_id = client.start_run(agent, {"question": "..."})

    for event in client.stream_events(run_id):
        print(event.seq, event.kind)

    state = client.get_run(run_id)
    print(state.status.state)

You need a control plane to talk to: npm install -g salvor && salvor serve, or see the repository for other install routes.

What the control plane is

Salvor is a durable execution runtime for AI agents. A run is an append-only log of events: every model call and every tool call is recorded before the run moves on, so a process that dies mid-flight is recovered from the log and finished from exactly where it stopped, with no completed step run twice.

The control plane is a small HTTP and server-sent-events server that puts that runtime on a network. It owns one event store and drives runs in the background. You submit an agent definition and an input, then read the run's events as they land. The full contract is in crates/salvor-server/API.md.

Why the SDK is thin

The durability guarantees stay in one Rust process. Exact replay, crash-safe resume, and the write-ahead rule that parks a run whose write was recorded but never completed all live server-side, enforced by the same runtime the CLI uses. So this SDK is a few hundred lines: it submits data, reads events, and maps the server's error envelope to exceptions. It holds no agent loop, no run state, and no durability logic of its own. Because the server does all the work, the SDK stays consistent with it by construction.

Install

pip install salvor

The one runtime dependency is httpx. To work on the SDK itself, install it from a checkout instead: pip install -e sdks/python.

The client surface

from salvor import Client

client = Client("http://127.0.0.1:8080", token=None)

agent    = client.register_agent(toml_or_dict)      # -> agent hash
run_id   = client.start_run(agent, input=None)      # -> run id
state    = client.get_run(run_id)                   # -> RunState
runs     = client.list_runs()                       # -> list[RunSummary]
stream   = client.stream_events(run_id, from_seq=None)  # -> EventStream
result   = client.resume(run_id, input=None)        # -> ResumeResult
state    = client.resolve(run_id, output)           # record a dangling write
projected = client.replay(run_id)                   # -> ReplayState (dry run)

register_agent accepts a TOML string (sent as application/toml) or a dict of the same fields (sent as application/json). An agent is data, so it has a content hash; submit it once and reference it by that hash on every start.

The streaming and cursor model

stream_events returns an EventStream you iterate for Event objects in sequence order:

stream = client.stream_events(run_id)
for event in stream:
    print(event.seq, event.kind)
print(stream.end.status.state)   # the resting status the end frame carried

On connect the server replays every recorded event at or after the cursor, then tails new events as they land, then sends one terminal end frame and closes. A run's log has contiguous, ascending sequence numbers, so the stream is gap-free and duplicate-free by construction, and the client only has to track one number: the next sequence to expect.

That same number is what makes a dropped connection recoverable. If the socket drops mid-tail, the client reconnects with ?from_seq=<next> and the server resumes from there. Any event that arrived just before the drop is skipped by sequence number, so the merged stream stays gap-free and duplicate-free across the reconnect. Iteration stops at the end frame; its status (and a detached flag, set when the run is mid-step with no driver in this server process) is then on stream.end.

Errors

Every server error is decoded from the one JSON envelope ({"error": {"code", "message", "details?}}) into a SalvorAPIError carrying the stable code and the message. The one refusal with structured evidence, a resume blocked because a write was recorded but never completed, raises NeedsReconciliationError, whose .intent is the recorded write. Verify what that write did, then call resolve(run_id, output) to record its completion so replay never re-runs it.

from salvor import NeedsReconciliationError

try:
    client.resume(run_id)
except NeedsReconciliationError as e:
    print("stuck on write:", e.intent.get("tool"), e.intent.get("input"))
    client.resolve(run_id, output={"charged": True})
    client.resume(run_id)

The two modes

Salvor has two modes, and this SDK speaks both. The one above is server-driven: start_run hands the agent loop to the server, which drives it in a background task, and you read the events it produces. The second is client-driven: your code owns the loop and streams the events it produces, while the server still owns the durable log and, on every append, re-folds the log to confirm the incoming event is the one legal next event. The two never collide: a client-driven run and a server-driven run cannot share an id, and each surface serves only its own runs.

Open a client-driven run and drive it with a ClientRunDriver:

from salvor import Client

with Client("http://127.0.0.1:8080") as client:
    run = client.open_client_run(record_prompts=False)   # -> ClientRunDriver

    # The client emits its own control and context events through the guarded
    # append; the server confirms each is the legal next event before recording.
    run.append([run.envelope(0, "RunStarted", agent_def_hash=agent, input=task)])

    # The one side-effecting step the server must perform (it holds the key):
    result = run.model_step(1, request)          # -> ModelStepResult (response, usage)
    # or stream it, painting a live ticker:
    stream = run.model_step_stream(1, request)
    for delta in stream:
        ...                                      # {"type": "text_delta", ...}
    completion = stream.completion               # -> ModelStepResult

    # A tool the server's registry holds:
    output = run.tool_step(3, "render", {"doc": "plan.typ"})

    run.append([run.envelope(5, "RunCompleted", output=answer)])

The driver's full surface: open (also re-opens, i.e. resumes, an existing run), log(from_seq=0), append(events), model_step, model_step_stream, tool_step, and resolve(output). Re-opening a run returns its recorded log on run.log_envelopes and mints a fresh drive token (the single-writer lease every append presents), so a refreshed client rebuilds its cursor and re-drives from the log, paying nothing for a step the log already covers. A client-driven append the log rejects raises DivergenceError; a tool step that lands on a dangling write raises NeedsReconciliationError (whose .intent is the recorded write), which resolve(output) clears.

examples/browser-client-run drives this same client-driven surface from a browser page, and example/client_run_loop.py drives it from Python.

Runnable example

example/agent.toml is a model-only agent that answers one question. It is the Python mirror of examples/web-research, driven over the control plane instead of the CLI. Start a server with a key on its environment, then run the script:

npm install -g salvor          # or: cargo install salvor-cli

ANTHROPIC_API_KEY=sk-ant-... \
    salvor serve --bind 127.0.0.1:8080 --store /tmp/answer.db &

pip install salvor
python example/answer.py http://127.0.0.1:8080    # from sdks/python in a checkout

It registers the agent, starts a run, streams every event to completion, and prints the final answer, the event count, and the token usage.