reference.md

title	API Reference
sidebar_position	10

API Reference

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

RocketRideClient

Connection

Method	Signature	Returns	Description
attach	async 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.
detach	async def detach(self) -> None	-	Closes the transport and leaves the client detached.
is_attached	def is_attached(self) -> bool	bool	Whether the WebSocket transport is open, regardless of authentication.
login	async def login(self, credential: Optional[str] = None, *, uri: Optional[str] = None, timeout: Optional[float] = None) -> ConnectResult	ConnectResult	Authenticates 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.
logout	async def logout(self) -> None	-	Deauthenticates (sends deauth) and clears client auth state while keeping the attachment.
is_authenticated	def is_authenticated(self) -> bool	bool	Whether the auth handshake has succeeded on the current connection.
connect	async def connect(self, credential: Optional[str] = None, *, timeout: Optional[float] = None) -> ConnectResult	ConnectResult	Opens 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.
disconnect	async def disconnect(self) -> None	-	Closes the connection and cancels reconnection.
is_connected	def is_connected(self) -> bool	bool	Backward-compatible alias for is_attached() — True when the WebSocket is open; does not imply authentication.
get_connection_info	def get_connection_info(self) -> dict	dict	Returns { 'connected': bool, 'transport': str, 'uri': str }.
get_apikey	def get_apikey(self) -> Optional[str]	str | None	The API key in use. For debugging only; avoid logging in production.
set_env	def 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

Method	Signature	Returns	Description
use	async 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) -> dict	dict	Starts 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.
terminate	async def terminate(self, token: str) -> None	-	Stops the pipeline and frees server resources.
get_task_status	async def get_task_status(self, token: str) -> dict	dict	Current task status (completedCount, totalCount, completed, state, exitCode, …).

Data

Method	Signature	Returns	Description
pipe	async def pipe(self, token: str, objinfo: dict = None, mime_type: str = None, provider: str = None, on_sse=None) -> DataPipe	DataPipe	Creates a streaming pipe: open, then one or more writes, then close. Default MIME: 'application/octet-stream'. on_sse receives server-sent events.
send	async def send(self, token: str, data: str | bytes, objinfo: dict = None, mimetype: str = None, on_sse=None) -> PIPELINE_RESULT	PIPELINE_RESULT	Sends data in one shot (open, write once, close). No MIME auto-detection — default is 'application/octet-stream'.
send_files	async 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

Method	Signature	Returns	Description
add_monitor	async 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_monitor	async 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_events	async 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

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

Chat

Method	Signature	Returns	Description
chat	async def chat(self, *, token: str, question: Question, on_sse=None) -> PIPELINE_RESULT	PIPELINE_RESULT	Sends 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.

Method	Signature	Returns	Description
fs_open	async def fs_open(self, path: str, mode: str = 'r') -> dict	dict	Open a handle. Returns {'handle': str}; read mode also includes 'size' (int).
fs_read	async def fs_read(self, handle: str, offset: int = 0, length: int = 4_194_304) -> bytes	bytes	Read up to length bytes (default 4 MB) from offset. Empty bytes = EOF.
fs_write	async def fs_write(self, handle: str, data: bytes) -> int	int	Write raw bytes to a write handle. Returns the number of bytes written.
fs_close	async def fs_close(self, handle: str, mode: str = 'r') -> None	-	Close a handle. mode must match the mode passed to fs_open.
fs_read_string	async def fs_read_string(self, path: str, encoding: str = 'utf-8') -> str	str	Read an entire file as a decoded string.
fs_write_string	async def fs_write_string(self, path: str, text: str, encoding: str = 'utf-8') -> None	-	Write a string to a file (overwrites).
fs_read_json	async def fs_read_json(self, path: str) -> Any	Any	Read and parse a JSON file.
fs_write_json	async def fs_write_json(self, path: str, obj: Any) -> None	-	Serialize an object to JSON and write it.
fs_list_dir	async def fs_list_dir(self, path: str = '') -> dict	dict	List immediate children: {entries: [{name, type, size?, modified?}], count}.
fs_stat	async def fs_stat(self, path: str) -> dict	dict	Metadata: {exists, type, size, modified} (size/modified for files only).
fs_mkdir	async def fs_mkdir(self, path: str) -> None	-	Create a directory.
fs_rmdir	async def fs_rmdir(self, path: str, *, recursive: bool = False) -> None	-	Remove a directory. recursive=True deletes contents.
fs_rename	async 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_delete	async def fs_delete(self, path: str) -> None	-	Delete a file.
fs_get_url	async def fs_get_url(self, path: str, expires_in: int = 3600, download_name: str = None) -> str	str	Time-limited HTTP(S) URL for direct browser access; inline by default, download_name forces Content-Disposition: attachment.
fs_read_many	async 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 Drizzle ORM driver over this surface (rocketride/drizzle), plus a deprecated Sequelize binding.

Method	Signature	Description
database.query	async def query(*, token, sql, node_id='', session_id='', params=None) -> dict	Execute raw SQL or Cypher through the pipeline's execute tool function; params is a positional list for SQL nodes, or a dict keyed by placeholder name for the rocketride_graph node (graph_neo4j / graph_falkordb reject params); returns {rows, affected_rows}.
database.begin_transaction	async def begin_transaction(*, token, node_id='') -> dict	Open a transaction (begin tool function); returns {session_id}.
database.commit	async def commit(*, token, session_id, node_id='') -> dict	Commit the open transaction.
database.rollback	async def rollback(*, token, session_id, node_id='') -> dict	Roll back the open transaction.
database.dialect	async def dialect(*, token, node_id='') -> DatabaseDialect	The target node's SQL dialect.

Deploy (client.deploy)

See Deployments for the model.

Method	Signature	Returns
deploy.add	async def add(self, pipeline=None, *, kind='pipe', data=None, metadata=None, comment=None, deploy_to=None) -> PublishResult	PublishResult
deploy.add_app	async def add_app(self, app_root, *, workspace_root=None, comment=None, metadata=None, on_progress=None) -> PublishResult	PublishResult
deploy.create_app	async def create_app(self, slug, *, workspace_root=None, template='Blank', display_name=None, developer_id=None, sidebar=False, status_footer=True, doc_tabs=False, install=True, server_base_url=None, on_progress=None) -> dict	{appId, folder, files, vendored, installed}
deploy.verify_app	async def verify_app(self, app_root, *, workspace_root=None) -> AppVerifyReport	AppVerifyReport
deploy.deploy	async def deploy(self, project_id, version, team_id) -> Deployment	Deployment
deploy.list	async def list(self, *, team_id=None, page=None, page_size=None, search=None, filters=None, sort=None) -> DeployListResult	DeployListResult
deploy.get	async def get(self, project_id, team_id) -> Deployment	Deployment
deploy.versions	async def versions(self, project_id, *, page=None, ...) -> DeployVersionsResult	DeployVersionsResult
deploy.history	async def history(self, project_id, *, team_id=None, page=None, ...) -> DeployHistoryResult	DeployHistoryResult
deploy.disable	async def disable(self, project_id, team_id) -> Deployment	Deployment
deploy.enable	async def enable(self, project_id, team_id) -> Deployment	Deployment
deploy.remove	async def remove(self, project_id, team_id) -> Deployment	Deployment
deploy.set_schedule	async def set_schedule(self, project_id, source_id, schedule, team_id, *, ttl=None) -> Deployment	Deployment
deploy.pause_schedule	async def pause_schedule(self, project_id, source_id, team_id) -> Deployment	Deployment
deploy.resume_schedule	async def resume_schedule(self, project_id, source_id, team_id) -> Deployment	Deployment
deploy.set_source_config	async def set_source_config(self, project_id, source_id, team_id, ...) -> Deployment	Deployment
deploy.run	async def run(self, project_id, source_id, team_id) -> dict	{token, version}
deploy.artifact	async def artifact(self, project_id, version) -> PipelineConfig	PipelineConfig
deploy.preview	async def preview(self, schedule, count=None) -> SchedulePreview	SchedulePreview

App publish ladder

See Deployments for the model. The deploy.-prefixed verbs below (add, add_app, create_app, verify_app) live on client.deploy; the unprefixed verbs are methods on the client itself (client.publish_app(...)).

Method	Signature	Description
deploy.add	async def add(self, pipeline=None, *, kind='pipe', data=None, metadata=None, comment=None, deploy_to=None) -> PublishResult	The 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_app	async def add_app(self, app_root, *, workspace_root=None, comment=None, metadata=None, on_progress=None) -> PublishResult	Pack 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.create_app	async def create_app(self, slug, *, workspace_root=None, template='Blank', display_name=None, developer_id=None, sidebar=False, status_footer=True, doc_tabs=False, install=True, server_base_url=None, on_progress=None) -> dict	Scaffold a new app in the workspace — the programmatic twin of the App Builder's New App wizard, rendering the identical templates. Writes ./apps/<slug>, vendors the connected server's shell + client packages, and runs the workspace install. The id becomes <developerId>.<slug>. Scaffolding only — the lifecycle (verify_app → add_app → publish_app) follows. Returns the TypeScript CreatedApp keys (appId, folder, files, vendored, installed).
deploy.verify_app	async def verify_app(self, app_root, *, workspace_root=None) -> AppVerifyReport	The 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_deployments	async 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 (registryVersion, appVersion, sha256, publishedAt, author, message, state, buildStatus, buildPhase, buildEndedAt, rungs) carries its deployment state, its build lifecycle (buildStatus — 'ok' = servable — plus the buildPhase it reached and buildEndedAt, which is always present: epoch seconds, or None until the build ends), and the rungs naming the audiences bound to it. No error text rides the rail: read it with build_log.
submit_app	async def submit_app(self, app_id, registry_version) -> dict	Submit a deployed version for review — flips the deployment private → submit.
withdraw_app	async def withdraw_app(self, app_id, registry_version) -> dict	Withdraw a pending review — the developer's own cancel: flips the deployment submit → private, the version leaves the admin queue and history records withdrawn. Only a version in submit withdraws. Developer-org + namespace gated, like submit.
reply_app	async def reply_app(self, app_id, message, registry_version=None) -> dict	Append 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_log	async def build_log(self, app_id, registry_version) -> dict	One 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_app	async def publish_app(self, app_id, registry_version, target) -> dict	Bind 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.
remove_app_publish	async def remove_app_publish(self, app_id, target) -> dict	Remove an audience binding — the app stops serving to that audience and the row leaves the where-live listing. SOFT: registry versions and audit history survive; publishing to the audience again revives it. Returns the final publish binding row (state removed).
disable_app_publish	async def disable_app_publish(self, app_id, target) -> dict	Disable an audience binding — serving stops, but the row STAYS in the where-live listing marked disabled (a visible off switch), unlike remove_app_publish. Publishing any version to that audience re-enables it. Returns the publish binding row (state disabled).
where_app	async 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 family	Subcommands	Guard	Purpose
developer_*	developer_register · developer_stripe · developer_dashboard · developer_status	org.admin (register)	Claim the org's developer id slug + Stripe Connect onboarding.
submit	submit	developer org + namespace	Submit a deployed version for review — flips the DEPLOYMENT private → submit.
register_dev	register_dev	self	Per-user live dev overlay (App Builder hot-reload); OSS-capable.
catalog	list · get · list_mine · desktop_add · desktop_remove	authenticated	Browse reachable apps, the developer's own rail view, and desktop membership.
admin_*	admin_queue · admin_approve · admin_reject · admin_reply · admin_reseed	sys.admin	Store 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_delete	developer org	Manage 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().

Property	Type	Description
is_opened	bool	Whether the pipe is open.
pipe_id	int | None	Server-assigned pipe ID after open().

Method	Signature	Returns	Description
open	async def open(self) -> DataPipe	self	Opens the pipe; required before write().
write	async def write(self, buffer: bytes) -> None	-	Writes a chunk. Pipe must be open; payload must be bytes.
close	async def close(self) -> PIPELINE_RESULT	PIPELINE_RESULT	Closes the pipe and returns the processing result.
tool	async def tool(self, *, tool: str, node_id: str = '', input: dict = None) -> Any	Any	Invokes 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,
    cachePrefix: bool = False,
    role: str = '',
)

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

Method	Signature	Description
addInstruction	addInstruction(self, title: str, instruction: str)	Adds an instruction (e.g. "Use bullet points").
addExample	addExample(self, given: str, result: dict | list | str)	Adds an example input/output; result can be dict/list (JSON-serialized).
addContext	addContext(self, context: str | dict | List[str] | List[dict])	Adds context.
addHistory	addHistory(self, item: QuestionHistory)	Adds a history item for multi-turn chat.
addQuestion	addQuestion(self, question: str)	Appends the question text.
addDocuments	addDocuments(self, documents: Doc | List[Doc])	Adds documents for the AI to reference.
addGoal	addGoal(self, goal: str)	Adds a goal statement for the AI.
getPrompt	getPrompt(self, has_previous_json_failed: bool = False) -> str	Returns the full prompt (internal).
cachePrefix	field, default False	Set to True when the question will be sent again with only its context, documents, goals or questions changed (an agent loop). A provider with prompt caching (the Anthropic node, and Claude models on the Bedrock node) may then cache the unchanging start of the prompt: role, instructions, examples and history. Other providers ignore it.

Answer

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

Member	Signature	Description
setAnswer	setAnswer(self, value: str | dict | list)	Stores the response value, validating/parsing it as JSON when expectJson is True.
getText	getText(self) -> str	The answer as plain text.
getJson	getJson(self) -> Optional[dict]	The parsed JSON. Returns None only when no answer has been set; raises ValueError on invalid JSON.
isJson	isJson(self) -> bool	Returns the expectJson flag (does not inspect content).
parsePython	parsePython(self, value: str) -> Any	Extracts Python code from a code block in the response.
tokens	field	Turn-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, AppVerifyReport, AppVerifyCheck (from rocketride.types).

Additional client surface

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

Area	Methods
Generic invoke	call(command, ...) — any DAP command; tool(...) — invoke a pipeline tool function
Task helpers	get_task_token, get_task_pipeline, restart
Identity	get_account_info; static get_server_info, normalize_uri
Monitors	clear_all_monitors, identify (plus add_monitor/remove_monitor above)
Template storage	save_template, get_template, delete_template, get_all_templates
Log storage	save_log, get_log, delete_log, list_logs
Dashboard	get_dashboard, list_connections, list_tasks
Profiling	cprofile_start, cprofile_stop, cprofile_status, cprofile_report, cprofile_report_tree, cprofile_threads
Namespaces	client.account, client.billing (account and billing APIs)

Advanced: low-level DAP

For commands not covered by the typed surface.

Method	Signature	Returns	Description
build_request	def build_request(self, command: str, *, token: str = None, arguments: dict = None, data: bytes | str = None) -> dict	dict	Builds a DAP request message.
request	async def request(self, request: dict, timeout: float = None) -> dict	dict	Sends the request and returns the response. timeout in ms overrides the default for this call. Use did_fail(response) before trusting body.
dap_request	async def dap_request(self, command: str, arguments: dict = None, token: str = None, timeout: float = None) -> dict	dict	Shorthand: builds and sends in one call. Python-only — in TypeScript, compose buildRequest() + request().
did_fail	def did_fail(self, request: dict) -> bool	bool	True 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'))