Skip to main content

API Reference

View as Markdown

API Reference

The core public surface of the Python SDK. Constructor options and environment variables are on Configuration; exceptions on Error Handling.

RocketRideClient

Connection

MethodSignatureReturnsDescription
attachasync def attach(self, uri: Optional[str] = None, *, timeout: Optional[float] = None) -> None-Opens the WebSocket without authenticating. If uri is provided and differs from the current URI, detaches first; if already attached to the same URI, this is a no-op.
detachasync def detach(self) -> None-Closes the transport and leaves the client detached.
is_attacheddef is_attached(self) -> boolboolWhether the WebSocket transport is open, regardless of authentication.
loginasync def login(self, credential: Optional[str] = None, *, uri: Optional[str] = None, timeout: Optional[float] = None) -> ConnectResultConnectResultAuthenticates over an attached transport (attaching first if needed). A differing uri detaches and re-attaches; a differing credential logs out (best-effort) before logging in; logging in again with the same credential is a no-op.
logoutasync def logout(self) -> None-Deauthenticates (sends deauth) and clears client auth state while keeping the attachment.
is_authenticateddef is_authenticated(self) -> boolboolWhether the auth handshake has succeeded on the current connection.
connectasync def connect(self, credential: Optional[str] = None, *, timeout: Optional[float] = None) -> ConnectResultConnectResultOpens the WebSocket and performs DAP auth. Optional credential overrides the constructor auth for this connection attempt. Optional timeout (ms) bounds the connect + auth handshake (non-persist only). In persist mode, on failure the client calls on_connect_error and retries; on auth failure it does not retry.
disconnectasync def disconnect(self) -> None-Closes the connection and cancels reconnection.
is_connecteddef is_connected(self) -> boolboolBackward-compatible alias for is_attached()True when the WebSocket is open; does not imply authentication.
get_connection_infodef get_connection_info(self) -> dictdictReturns { 'connected': bool, 'transport': str, 'uri': str }.
get_apikeydef get_apikey(self) -> Optional[str]str | NoneThe API key in use. For debugging only; avoid logging in production.
set_envdef set_env(self, env: Dict[str, str]) -> None-Replaces the client's environment map, used for ${ROCKETRIDE_*} substitution and credential lookup.

Context manager: async with RocketRideClient(...) as client: — entering calls connect(), exiting calls disconnect().

Pipeline execution

MethodSignatureReturnsDescription
useasync def use(self, *, token: str = None, filepath: str = None, pipeline: PipelineConfig = None, source: str = None, threads: int = None, use_existing: bool = None, args: List[str] = None, ttl: int = None, pipelineTraceLevel: str = None, name: str = None, env: Dict[str, str] = None) -> dictdictStarts a pipeline. Requires filepath or pipeline. pipelineTraceLevel sets run-log trace verbosity, name a task display name, env per-run variable overrides. Returns a dict with at least 'token', plus 'reused': True when use_existing handed back an already-running instance instead of starting this pipeline.
terminateasync def terminate(self, token: str) -> None-Stops the pipeline and frees server resources.
get_task_statusasync def get_task_status(self, token: str) -> dictdictCurrent task status (completedCount, totalCount, completed, state, exitCode, …).

Data

MethodSignatureReturnsDescription
pipeasync def pipe(self, token: str, objinfo: dict = None, mime_type: str = None, provider: str = None, on_sse=None) -> DataPipeDataPipeCreates a streaming pipe: open, then one or more writes, then close. Default MIME: 'application/octet-stream'. on_sse receives server-sent events.
sendasync def send(self, token: str, data: str | bytes, objinfo: dict = None, mimetype: str = None, on_sse=None) -> PIPELINE_RESULTPIPELINE_RESULTSends data in one shot (open, write once, close). No MIME auto-detection — default is 'application/octet-stream'.
send_filesasync def send_files(self, files: List[str | Tuple[str, dict] | Tuple[str, dict, str]], token: str) -> List[UPLOAD_RESULT]List[UPLOAD_RESULT]Uploads files concurrently (unbounded asyncio.gather). Requires an API key (RuntimeError without one); a missing file raises ValueError. Progress via on_event as apaevt_status_upload.

Events

MethodSignatureReturnsDescription
add_monitorasync def add_monitor(self, key: Dict[str, Any], types: List[str]) -> None-Adds a reference-counted monitor subscription; events are delivered to on_event. Key: {'token': ...} or {'project_id': ..., 'source': ...} (+ optional 'pipe_id', 'team_id').
remove_monitorasync def remove_monitor(self, key: Dict[str, Any], types: List[str]) -> None-Removes a monitor subscription; a type unsubscribes from the server only when its reference count reaches zero.
set_eventsasync def set_events(self, token: str, event_types: List[str], pipe_id: int = None) -> None-Deprecated — use add_monitor/remove_monitor. Subscribes the task to the given event types.

Services, validation, and ping

MethodSignatureReturnsDescription
get_servicesasync def get_services(self) -> dictdictLightweight summaries of every service, plus a deduplicated icons table and the server version. Full definitions come from get_service.
get_serviceasync def get_service(self, service: str) -> dictdictOne service's full definition (config schema included). Raises ValueError (empty name) or RuntimeError (unknown service); never returns None.
validateasync def validate(self, pipeline: PipelineConfig, *, source: str = None) -> dictdictValidates a pipeline configuration without starting it; returns errors and warnings.
pingasync def ping(self) -> None-Liveness check; raises on failure. (A legacy token argument is accepted but not sent.)

Chat

MethodSignatureReturnsDescription
chatasync def chat(self, *, token: str, question: Question, on_sse=None) -> PIPELINE_RESULTPIPELINE_RESULTSends the Question to the pipeline and returns the result. on_sse streams server-sent events (e.g. token-by-token output). See Chat.

Store (file access)

Paths are relative to the store root; absolute-like paths are rejected. See File Storage for the workflow.

MethodSignatureReturnsDescription
fs_openasync def fs_open(self, path: str, mode: str = 'r') -> dictdictOpen a handle. Returns {'handle': str}; read mode also includes 'size' (int).
fs_readasync def fs_read(self, handle: str, offset: int = 0, length: int = 4_194_304) -> bytesbytesRead up to length bytes (default 4 MB) from offset. Empty bytes = EOF.
fs_writeasync def fs_write(self, handle: str, data: bytes) -> intintWrite raw bytes to a write handle. Returns the number of bytes written.
fs_closeasync def fs_close(self, handle: str, mode: str = 'r') -> None-Close a handle. mode must match the mode passed to fs_open.
fs_read_stringasync def fs_read_string(self, path: str, encoding: str = 'utf-8') -> strstrRead an entire file as a decoded string.
fs_write_stringasync def fs_write_string(self, path: str, text: str, encoding: str = 'utf-8') -> None-Write a string to a file (overwrites).
fs_read_jsonasync def fs_read_json(self, path: str) -> AnyAnyRead and parse a JSON file.
fs_write_jsonasync def fs_write_json(self, path: str, obj: Any) -> None-Serialize an object to JSON and write it.
fs_list_dirasync def fs_list_dir(self, path: str = '') -> dictdictList immediate children: {entries: [{name, type, size?, modified?}], count}.
fs_statasync def fs_stat(self, path: str) -> dictdictMetadata: {exists, type, size, modified} (size/modified for files only).
fs_mkdirasync def fs_mkdir(self, path: str) -> None-Create a directory.
fs_rmdirasync def fs_rmdir(self, path: str, *, recursive: bool = False) -> None-Remove a directory. recursive=True deletes contents.
fs_renameasync def fs_rename(self, old_path: str, new_path: str) -> None-Rename or move a file/directory (copy+delete on object stores; recursive for directories).
fs_deleteasync def fs_delete(self, path: str) -> None-Delete a file.
fs_get_urlasync def fs_get_url(self, path: str, expires_in: int = 3600, download_name: str = None) -> strstrTime-limited HTTP(S) URL for direct browser access; inline by default, download_name forces Content-Disposition: attachment.
fs_read_manyasync def fs_read_many(self, paths: List[str]) -> List[Dict[str, Any]]List[Dict]Batch-read many small files in one round trip (max 256 paths / 32 MiB total). Per-entry failures (ok: False + error), request order, data as bytes.

Database

Raw SQL through a pipeline database node (requires allow_execute: true on the node). The TypeScript SDK additionally offers a Sequelize ORM binding over this surface.

MethodSignatureDescription
database.queryasync def query(*, token, sql, node_id='', session_id='', params=None) -> dictExecute raw SQL through the pipeline's execute tool function; returns {rows, affected_rows}.
database.begin_transactionasync def begin_transaction(*, token, node_id='') -> dictOpen a transaction (begin tool function); returns {session_id}.
database.commitasync def commit(*, token, session_id, node_id='') -> dictCommit the open transaction.
database.rollbackasync def rollback(*, token, session_id, node_id='') -> dictRoll back the open transaction.
database.dialectasync def dialect(*, token, node_id='') -> DatabaseDialectThe target node's SQL dialect.

Deploy (client.deploy)

See Deployments for the model.

MethodSignatureReturns
deploy.addasync def add(self, pipeline=None, *, kind='pipe', data=None, metadata=None, comment=None, deploy_to=None) -> PublishResultPublishResult
deploy.add_appasync def add_app(self, app_root, *, workspace_root=None, comment=None, metadata=None, on_progress=None) -> PublishResultPublishResult
deploy.verify_appasync def verify_app(self, app_root, *, workspace_root=None) -> AppVerifyReportAppVerifyReport
deploy.deployasync def deploy(self, project_id, version, team_id) -> DeploymentDeployment
deploy.listasync def list(self, *, team_id=None, page=None, page_size=None, search=None, filters=None, sort=None) -> DeployListResultDeployListResult
deploy.getasync def get(self, project_id, team_id) -> DeploymentDeployment
deploy.versionsasync def versions(self, project_id, *, page=None, ...) -> DeployVersionsResultDeployVersionsResult
deploy.historyasync def history(self, project_id, *, team_id=None, page=None, ...) -> DeployHistoryResultDeployHistoryResult
deploy.disableasync def disable(self, project_id, team_id) -> DeploymentDeployment
deploy.enableasync def enable(self, project_id, team_id) -> DeploymentDeployment
deploy.removeasync def remove(self, project_id, team_id) -> DeploymentDeployment
deploy.set_scheduleasync def set_schedule(self, project_id, source_id, schedule, team_id, *, ttl=None) -> DeploymentDeployment
deploy.pause_scheduleasync def pause_schedule(self, project_id, source_id, team_id) -> DeploymentDeployment
deploy.resume_scheduleasync def resume_schedule(self, project_id, source_id, team_id) -> DeploymentDeployment
deploy.set_source_configasync def set_source_config(self, project_id, source_id, team_id, ...) -> DeploymentDeployment
deploy.runasync def run(self, project_id, source_id, team_id) -> dict{token, version}
deploy.artifactasync def artifact(self, project_id, version) -> PipelineConfigPipelineConfig
deploy.previewasync def preview(self, schedule, count=None) -> SchedulePreviewSchedulePreview

App publish ladder

See Deployments for the model. Only deploy.add and deploy.add_app live on client.deploy; the unprefixed verbs below are methods on the client itself (client.publish_app(...)).

MethodSignatureDescription
deploy.addasync def add(self, pipeline=None, *, kind='pipe', data=None, metadata=None, comment=None, deploy_to=None) -> PublishResultThe ONE rail door (on the client.deploy namespace): deploy any kind of object as the next immutable registry version. kind='pipe' (default) takes a pipeline dict; kind='app' takes ONE data zip of the app's SOURCE (the server performs the build; client-produced binaries are never trusted), retained and unpacked at receipt, born deployment-state private. The app id must be inside your developer namespace.
deploy.add_appasync def add_app(self, app_root, *, workspace_root=None, comment=None, metadata=None, on_progress=None) -> PublishResultPack an app folder's source and deploy it as the next registry version — the one call behind the App Builder's Deploy button and CI scripts. Packs by the App Builder rules (workspace-rooted zip, appManifest.include, hierarchical gitignore + the hard node_modules/dist/.git baseline, symlink containment, 50MB zipped / 512MB uncompressed caps); on_progress narrates one line per step. Deploying activates nothing — bind an audience with publish_app afterwards.
deploy.verify_appasync def verify_app(self, app_root, *, workspace_root=None) -> AppVerifyReportThe no-side-effect precheck for add_app — purely local, no server call: manifest shape and id grammar, declared icon/README assets, appManifest.include entries, and a pack dry run against the size caps. Server-side concerns (the build, store review) are out of scope.
list_deploymentsasync def list_deployments(self, app_id) -> list[dict]The version rail, newest first — the developer org sees its FULL rail (published or not), other callers only their visible versions. Each entry carries its deployment state, its buildStatus ('ok' = servable), and the rungs naming the audiences bound to it.
submit_appasync def submit_app(self, app_id, registry_version) -> dictSubmit a deployed version for review — flips the deployment privatesubmit.
withdraw_appasync def withdraw_app(self, app_id, registry_version) -> dictWithdraw a pending review — the developer's own cancel: flips the deployment submitprivate, the version leaves the admin queue and history records withdrawn. Only a version in submit withdraws. Developer-org + namespace gated, like submit.
reply_appasync def reply_app(self, app_id, message, registry_version=None) -> dictAppend a developer message to the app's review thread — rides deployment_history as a reply row (side 'developer'), the same stream deploy.history() reads. Developer-org + namespace gated, like submit.
build_logasync def build_log(self, app_id, registry_version) -> dictOne version's durable server build log — the full phase-by-phase output stored beside the version's artifacts (no error text rides the rail rows). Long logs serve their tail; empty log = none. Developer-org gated.
publish_appasync def publish_app(self, app_id, registry_version, target) -> dictBind a deployment to '@me', '@team/', or '@public' ('@user' = legacy input alias). The binding is a pure pointer born 'enabled'. '@public' requires the deployment be ready; '@me'/'@team' accept any non-failed deployment. Pinning ANOTHER org's public app to '@me'/'@team' is the version selector; publishing your own app requires the id to be in your namespace.
where_appasync def where_app(self, app_id) -> list[dict]The reverse index: {rung, handle, version, appVersion, state, deployedAt} per audience — state is the bound deployment's review state.

Serving needs no verb: a version's bundle loads from the stable /apps/<app_id>/v<N>/remoteEntry.js URL constructed from its registry version number, with entitlement enforced by the serve route on every request (registry ints ONLY — semver is display).

App marketplace + developer verbs

Two raw DAP commands carry this surface (call via client.call("<command>", {"subcommand": ...})):

  • rrext_deploy_app — the developer-account + review verbs (claiming a developerId is a deploy PREREQUISITE, not a marketplace action): the developer_* family, submit, and register_dev.
  • rrext_app — the pure marketplace: browse (list/get/list_mine), install (desktop_add/desktop_remove), admin review (admin_*), and pricing (pricing_*).

Grouped families (the developer_*/submit/register_dev rows are on rrext_deploy_app; the rest on rrext_app):

Subcommand familySubcommandsGuardPurpose
developer_*developer_register · developer_stripe · developer_dashboard · developer_statusorg.admin (register)Claim the org's developer id slug + Stripe Connect onboarding.
submitsubmitdeveloper org + namespaceSubmit a deployed version for review — flips the DEPLOYMENT privatesubmit.
register_devregister_devselfPer-user live dev overlay (App Builder hot-reload); OSS-capable.
cataloglist · get · list_mine · desktop_add · desktop_removeauthenticatedBrowse reachable apps, the developer's own rail view, and desktop membership.
admin_*admin_queue · admin_approve · admin_reject · admin_reply · admin_reseedsys.adminStore review over the DEPLOYMENTS: the queue is deployments in submit; admin_approve(appId, version)ready, admin_reject(appId, version)rejected.
pricing_*pricing_list · pricing_create · pricing_deletedeveloper orgManage Stripe price tiers for a monetized app.

Review model. The review state lives on the DEPLOYMENT. Going public is a three-step flow: submit (deployment → submit, enters the admin queue) → admin_approve (→ ready) → publish_app @public (point the public binding at the ready version). A reject flips the deployment rejected; the developer fixes and deploys a NEW version. @me/@team bindings need no approval.

Run logs (client.log)

See Run Logs for the continuum model and the DVR session.

DataPipe

Returned by await client.pipe(...). One streaming upload: open → write (one or more) → close. Also an async context manager: entering calls open(), exiting calls close().

PropertyTypeDescription
is_openedboolWhether the pipe is open.
pipe_idint | NoneServer-assigned pipe ID after open().
MethodSignatureReturnsDescription
openasync def open(self) -> DataPipeselfOpens the pipe; required before write().
writeasync def write(self, buffer: bytes) -> None-Writes a chunk. Pipe must be open; payload must be bytes.
closeasync def close(self) -> PIPELINE_RESULTPIPELINE_RESULTCloses the pipe and returns the processing result.
toolasync def tool(self, *, tool: str, node_id: str = '', input: dict = None) -> AnyAnyInvokes a pipeline tool function through the pipe.

open() retries once automatically if it hits a transient "Connect call failed" while the pipeline's data listener is still starting up (worst case adds ~1.75s); a PipeException from open() means it kept failing past that retry budget.

Question

From rocketride.schema. Build a question for client.chat(token=..., question=question).

Question(
type: QuestionType = QuestionType.QUESTION,
filter: DocFilter = None,
expectJson: bool = False,
role: str = '',
)

QuestionType: QUESTION, SEMANTIC, KEYWORD, GET, PROMPT.

MethodSignatureDescription
addInstructionaddInstruction(self, title: str, instruction: str)Adds an instruction (e.g. "Use bullet points").
addExampleaddExample(self, given: str, result: dict | list | str)Adds an example input/output; result can be dict/list (JSON-serialized).
addContextaddContext(self, context: str | dict | List[str] | List[dict])Adds context.
addHistoryaddHistory(self, item: QuestionHistory)Adds a history item for multi-turn chat.
addQuestionaddQuestion(self, question: str)Appends the question text.
addDocumentsaddDocuments(self, documents: Doc | List[Doc])Adds documents for the AI to reference.
addGoaladdGoal(self, goal: str)Adds a goal statement for the AI.
getPromptgetPrompt(self, has_previous_json_failed: bool = False) -> strReturns the full prompt (internal).

Answer

From rocketride.schema. Parses chat response content — see Chat for semantics.

MemberSignatureDescription
setAnswersetAnswer(self, value: str | dict | list)Stores the response value, validating/parsing it as JSON when expectJson is True.
getTextgetText(self) -> strThe answer as plain text.
getJsongetJson(self) -> Optional[dict]The parsed JSON. Returns None only when no answer has been set; raises ValueError on invalid JSON.
isJsonisJson(self) -> boolReturns the expectJson flag (does not inspect content).
parsePythonparsePython(self, value: str) -> AnyExtracts Python code from a code block in the response.
tokensfieldTurn-total LLM token usage reported by the server. The TypeScript Answer carries no usage field.

Types

  • PIPELINE_RESULT: TypedDict with name, path, objectId, optional result_types, and dynamic fields.
  • UPLOAD_RESULT: Per-file result with action, filepath, error?, result?, upload_time?, etc.
  • TASK_STATUS: Task status with completedCount, totalCount, completed, state, exitCode, and many more fields.
  • ConnectResult: Identity payload returned by connect()/login()userToken, userId, displayName, organizations, apps, teams (all optional).
  • DAPMessage: Dict with type, seq, and optional command, arguments, body, success, message, event, token, etc.
  • PipelineConfig: Pipeline definition with name, description, version, components, source, project_id.
  • QuestionHistory: { 'role': str, 'content': str }.
  • QuestionExample: { 'given': str, 'result': str }.
  • QuestionType / QuestionText: question kind enum and text wrapper from rocketride.schema.
  • Deploy types: DeployArtifact, Deployment, DeploymentSchedule, DeployActor, DeployHistoryEntry, PublishResult, DeployListResult, DeployVersionsResult, DeployHistoryResult, SchedulePreview (from rocketride.types).

Additional client surface

Further public methods, present in both SDKs, in brief:

AreaMethods
Generic invokecall(command, ...) — any DAP command; tool(...) — invoke a pipeline tool function
Task helpersget_task_token, get_task_pipeline, restart
Identityget_account_info; static get_server_info, normalize_uri
Monitorsclear_all_monitors, identify (plus add_monitor/remove_monitor above)
Template storagesave_template, get_template, delete_template, get_all_templates
Log storagesave_log, get_log, delete_log, list_logs
Dashboardget_dashboard, list_connections, list_tasks
Profilingcprofile_start, cprofile_stop, cprofile_status, cprofile_report, cprofile_report_tree
Namespacesclient.account, client.billing (account and billing APIs)

Advanced: low-level DAP

For commands not covered by the typed surface.

MethodSignatureReturnsDescription
build_requestdef build_request(self, command: str, *, token: str = None, arguments: dict = None, data: bytes | str = None) -> dictdictBuilds a DAP request message.
requestasync def request(self, request: dict, timeout: float = None) -> dictdictSends the request and returns the response. timeout in ms overrides the default for this call. Use did_fail(response) before trusting body.
dap_requestasync def dap_request(self, command: str, arguments: dict = None, token: str = None, timeout: float = None) -> dictdictShorthand: builds and sends in one call. Python-only — in TypeScript, compose buildRequest() + request().
did_faildef did_fail(self, request: dict) -> boolboolTrue when the response indicates failure (success === False).
# Two-step (build then request)
req = client.build_request('rrext_monitor', token=token, arguments={'types': ['apaevt_status_upload']})
res = await client.request(req, timeout=5000)

# One-step with dap_request
res = await client.dap_request('rrext_services', {}, timeout=5000)

if client.did_fail(res):
raise RuntimeError(res.get('message', 'Request failed'))