visus-sdk-runpilot 0.2.1

In-process Python SDK for the RunPilot RPA orchestrator (the client bots use to talk back to the server).

runpilot — Python SDK

In-process Python SDK for the RunPilot RPA orchestrator. This is the library your bot imports to talk back to the RunPilot server from inside a run: report progress, stream logs, fetch credentials, upload artifacts, poll interrupt/kill signals and pull DataPool items.

Think of it as the runtime bridge between an RPA script and the orchestrator that scheduled it. The runner injects the connection details as environment variables; the SDK reads them and gives you a single Orchestrator object.

Installation

pip install runpilot

Requires Python 3.9+. The only runtime dependency is requests.

Quick start

Inside a run, build the client from the environment the runner injected:

from runpilot import Orchestrator

bot = Orchestrator.from_env()          # reads RUNPILOT_SERVER + run token / api key

ctx = bot.get_run()                    # the run + task context
bot.report_progress(10, "starting")

api_key = bot.get_credential("stripe", "SECRET_KEY")

pool = bot.datapool("invoices")        # claim-and-report DataPool items
while (item := pool.next_item()) is not None:
    process(item.data)
    item.report_done()                 # or item.report_error(business=True, message=...)

art = bot.upload_artifact("out/report.pdf")

signals = bot.get_signals()            # cooperative interrupt / kill
if signals.kill_requested or signals.interrupt_requested:
    bot.finish("failure", "interrupted")
    raise SystemExit(0)

bot.finish("success", "processed 42 rows")   # canonical: success | failure

Environment variables

Orchestrator.from_env() reads:

Variable	Purpose
RUNPILOT_SERVER	Base URL of the orchestrator (no path), e.g. https://api.runpilot.dev
RUNPILOT_RUN_TOKEN	Per-run token (RUN_TOKEN mode — the in-run path)
RUNPILOT_RUN_ID	Run ID that accompanies the run token
RUNPILOT_API_KEY	Organization API key (management mode)

RUN_TOKEN mode is what you get inside a run; the in-run methods (report_progress, get_credential, datapool, upload_artifact, finish, …) require it.

Testing your bot — fully offline

Orchestrator.mock() returns a client that never performs a network request. Every in-run call is recorded on .calls and returns a sensible canned value, so you can unit-test a bot without a live server:

from runpilot import Orchestrator

bot = Orchestrator.mock(
    run={"run": {"id": "run_1"}, "task": {"name": "demo"}},
    credentials={("stripe", "SECRET_KEY"): "sk_test_123"},
    datapool={"invoices": [{"id": "inv_1", "data": {"amount": 10}}]},
)

bot.report_progress(50, "halfway")
assert bot.get_credential("stripe", "SECRET_KEY") == "sk_test_123"
assert bot.calls[-1]["method"] == "get_credential"

API surface

from runpilot import ...

Export	What it is
Orchestrator	The client. from_env(), mock(), get_run(), report_progress(), log(), get_credential(), upload_artifact(), get_signals(), datapool(), finish()
DataPool, DataPoolItem	Claim-and-report work queue; pool.next_item(), then item.report_done() / item.report_error()
Run, RunContext, Task, Artifact, Signals	Typed models returned by the client
RunPilotError and subclasses	ConfigurationError, AuthModeError, AuthenticationError, NotFoundError, ApiError, RunAlreadyFinishedError, RunTerminalError

License

MIT