Python

Build, run, and manage AI pipelines from Python.

Quick Start

pip install rocketride

import asyncio
from rocketride import RocketRideClient


async def main():
    async with RocketRideClient(uri='https://cloud.rocketride.ai', auth='my-key') as client:
        result = await client.use(filepath='pipeline.pipe')
        token = result['token']
        out = await client.send(token, 'Hello, pipeline!', objinfo={'name': 'input.txt'}, mimetype='text/plain')
        print(out)
        await client.terminate(token)


asyncio.run(main())

URI scheme: the scheme selects the transport. The client normalizes the uri to a WebSocket address before connecting: https:// and wss:// both resolve to a secure wss:// connection, while http://, ws://, and a bare host:port resolve to plain ws://. For RocketRide Cloud use https://cloud.rocketride.ai (or the equivalent wss://cloud.rocketride.ai); for a local engine use ws://localhost:5565. Caution: against a Cloud endpoint always use https:// or wss://, because an http:// or ws:// URI (or a bare host:port) silently downgrades to an unencrypted ws:// connection.

Don't have a pipeline yet? Visit RocketRide on GitHub or download the extension directly in your IDE.

Install RocketRide extension

What is RocketRide?

RocketRide is an open-source, developer-native AI pipeline platform. It lets you build, debug, and deploy production AI workflows without leaving your IDE -- using a visual drag-and-drop canvas or code-first with TypeScript and Python SDKs.

50+ ready-to-use nodes - 13 LLM providers, 8 vector databases, OCR, NER, PII anonymization, and more
High-performance C++ engine - production-grade speed and reliability
Deploy anywhere - locally, on-premises, or self-hosted with Docker
MIT licensed - fully open-source, OSI-compliant

You build your .pipe - and you run it against the fastest AI runtime available.

Features

Pipeline execution - Start with use(), send data via send(), send_files(), or pipe()
Deployments - Persist pipelines server-side and run them on a cron schedule via client.deploy
Chat - Conversational AI via chat() and Question
Event streaming - Real-time events via on_event and set_events()
File upload - send_files() with progress; streaming with pipe()
Connection lifecycle - Optional persist mode, reconnection, and callbacks (on_connected, on_disconnected, on_connect_error)
Project storage - Save, retrieve, and version-control pipelines on the server
Async-first - Built on asyncio and websockets; supports async with context manager
CLI included - Manage pipelines from the command line
Telemetry reporting - The shared loose report() core via rocketride.analytics; each app owns its own event taxonomy (Analytics / Telemetry Reporting)

RocketRideClient

Constructor

RocketRideClient(
    uri: str = "",
    auth: str = "",
    *,
    env: dict = None,
    module: str = None,
    request_timeout: float = None,
    max_retry_time: float = None,
    persist: bool = False,
    on_event = None,
    on_connected = None,
    on_disconnected = None,
    on_connect_error = None,
    on_protocol_message = None,
    on_debug_message = None,
)

Why the options matter: uri and auth tell the client where and how to authenticate. persist and max_retry_time control what happens when the connection fails or the server is not ready yet: with persist=True the client retries with exponential backoff and calls on_connect_error on each failure, so you can show "Still connecting..." or "Connection failed" without implementing retry logic yourself. Use on_disconnected only for "we were connected and then dropped"; use on_connect_error for "failed to connect" or "gave up after max retry time."

Argument	Type	Required	Description
`uri`	`str`	Yes*	Server URI. *Can be empty if `ROCKETRIDE_URI` is set in env/`.env`.
`auth`	`str`	Yes*	API key. *Can be empty if `ROCKETRIDE_APIKEY` is set.
`env`	`dict`	No	Override env; if omitted, `.env` is loaded. Use when passing config in code instead of env files.
`module`	`str`	No	Client name for logging.
`request_timeout`	`float`	No	Default timeout in ms for requests. Prevents a single DAP call from hanging.
`max_retry_time`	`float`	No	Max time in ms to keep retrying connection. Use (e.g. 300000) so the app can show "gave up" after a bounded time.
`persist`	`bool`	No	Enable automatic reconnection. Default: `False`. Set `True` for long-lived scripts or UIs.
`on_event`	async callable	No	Called with each server event dict. Use for progress or status updates.
`on_connected`	async callable	No	Called when connection is established.
`on_disconnected`	async callable	No	Called when connection is lost only if connected first; args: `reason`, `has_error`. Do not call `disconnect()` here if you want auto-reconnect.
`on_connect_error`	callable `(message: str)`	No	Called on each failed connection attempt. On auth failure the client stops retrying.
`on_protocol_message`	callable `(message: str)`	No	Optional; for logging raw DAP messages. Helpful when debugging protocol issues.
`on_debug_message`	callable `(message: str)`	No	Optional; for debug output.

Raises ValueError if both uri and ROCKETRIDE_URI are empty or if auth is missing and not in env.

Example - client with persist and callbacks:

client = RocketRideClient(
    uri='https://cloud.rocketride.ai',
    auth='my-key',
    persist=True,
    max_retry_time=300000,
    on_connect_error=lambda msg: print('Connect error:', msg),
    on_event=handle_event,
)

Context manager

Method	Signature	Returns	Description
`__aenter__`	`async def __aenter__(self)`	`self`	Enters context; calls `connect()`.
`__aexit__`	`async def __aexit__(self, exc_type, exc_val, exc_tb)`	-	Exits context; calls `disconnect()`.

How to use: Prefer async with RocketRideClient(...) as client: so the connection is always closed when you leave the block, even on exception. No need to call disconnect() manually.

Example:

import asyncio
import os

from rocketride import RocketRideClient


async def main():
    async with RocketRideClient(uri='wss://cloud.rocketride.ai', auth=os.environ['ROCKETRIDE_APIKEY']) as client:
        result = await client.use(filepath='pipeline.json')
        token = result['token']
        await client.send(token, 'Hello, pipeline!')


asyncio.run(main())

Connection

Method	Signature	Returns	Description
`connect`	`async def connect(self, uri: str = None, auth: str = None, timeout: float = None) -> None`	-	Opens the WebSocket and performs DAP auth. Optional `uri`/`auth` override the constructor values 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. Call when the user disconnects or the script is done.
`is_connected`	`def is_connected(self) -> bool`	`bool`	Whether the client is connected. Check before calling `use()` or `send()` if needed.
`set_connection_params`	`async def set_connection_params(self, uri: str = None, auth: str = None) -> None`	-	Updates server URI and/or auth at runtime. If currently connected, disconnects and reconnects with the new params (in persist mode, reconnection is scheduled; otherwise reconnects once). Use when the user changes server or credentials without creating a new client.
`get_connection_info`	`def get_connection_info(self) -> dict`	`dict`	Current connection state and URI. Returns `{ 'connected': bool, 'transport': str, 'uri': str }`. Useful for debugging or displaying "Connected to ..." in the UI.
`get_apikey`	`def get_apikey(self) -> Optional[str]`	`str \| None`	The API key in use. For debugging only; avoid logging in production.

Low-level DAP

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. Use for custom commands not covered by `use()`, `send()`, etc.
`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 a request and sends it in one call. Equivalent to `build_request()` + `request()`.
`did_fail`	`def did_fail(self, request: dict) -> bool`	`bool`	Returns `True` when the response indicates failure (`success === False`).

Example:

# 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'))

Pipeline execution

Method	Signature	Returns	Description
`use`	`async def use(self, *, token: str = None, filepath: str = None, pipeline: dict = None, source: str = None, threads: int = None, use_existing: bool = None, args: list = None, ttl: int = None) -> dict`	`dict`	Starts a pipeline. Requires `filepath` or `pipeline`. The client substitutes `${ROCKETRIDE_*}` from its env. Returns a dict with at least `'token'`; use that token for all data and control operations.
`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`	Returns current task status (e.g. completed count, total, state). Poll until `completed` or use for progress display.

Why a token: The server runs each pipeline as a separate task. The token identifies that task so send(), send_files(), pipe(), chat(), and get_task_status() target the correct pipeline.

Data

Method	Signature	Returns	Description
`pipe`	`async def pipe(self, token: str, objinfo: dict = None, mime_type: str = None, provider: str = None) -> DataPipe`	`DataPipe`	Creates a streaming pipe: open, then one or more write, then close. Use for large or chunked data. Default MIME: `'application/octet-stream'`.
`send`	`async def send(self, token: str, data: str \| bytes, objinfo: dict = None, mimetype: str = None) -> PIPELINE_RESULT`	`PIPELINE_RESULT`	Sends data in one shot (open pipe, write once, close). Use when you have the full payload in memory.
`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. Each item: path `str`, or `(path, objinfo)`, or `(path, objinfo, mimetype)`. Progress via `on_event` as `apaevt_status_upload`.

When to use pipe vs send: Use send() for a single string or bytes. Use pipe() when you read a file in chunks, or when data arrives incrementally.

Example - send a string:

result = await client.send(token, 'Hello, pipeline!', objinfo={'name': 'greeting.txt'}, mimetype='text/plain')

Example - stream with a pipe:

pipe = await client.pipe(token, mime_type='application/json')
await pipe.open()
await pipe.write(b'{"key": "value1"}')
await pipe.write(b'{"key": "value2"}')
result = await pipe.close()

Store (file access)

Read, write, and manage files in your account's server-side store. All paths are relative to the store root (e.g. "docs/readme.md"); absolute-like paths (starting with / or \) are rejected. Binary I/O uses an explicit handle lifecycle (fs_open → fs_read / fs_write → fs_close, 4 MB chunks); for most cases prefer the string/JSON convenience wrappers.

Handle I/O (low-level binary)

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`.

Convenience wrappers (open/read/write/close handled internally)

Method	Signature	Returns	Description
`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.

Directory & metadata

Method	Signature	Returns	Description
`fs_list_dir`	`async def fs_list_dir(self, path: str = '') -> dict`	`dict`	List immediate children. Returns `{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. Rejects empty / absolute-like paths.
`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.

Direct URL

Method	Signature	Returns	Description
`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. Cloud backends (S3/Azure) return a presigned/SAS URL; the local filesystem backend returns a JWT-signed `/task/fetch` URL. Served inline by default (for streaming / `<img>`/`<video>` sources). Pass `download_name` to force a download with that filename via `Content-Disposition: attachment` — the only reliable way to set the download filename for cross-origin cloud URLs (where the browser `<a download>` hint is ignored). `expires_in` is in seconds (default 3600).
`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 call) — for many-small-file access patterns where per-file open/read/close is too chatty. Missing/unreadable files are per-entry results (`ok: False` + `error`), never a call failure; results come back in request order with `data` as `bytes`.

Examples:

# Strings and JSON (wrappers manage the handle for you)
await client.fs_write_string('notes/todo.txt', 'buy milk')
text = await client.fs_read_string('notes/todo.txt')
await client.fs_write_json('config/app.json', {'debug': True})
cfg = await client.fs_read_json('config/app.json')

# Browse and inspect
listing = await client.fs_list_dir('reports')
for entry in listing['entries']:
    print(entry['name'], entry['type'])

# Streaming binary upload via a write handle (4 MB chunks)
info = await client.fs_open('uploads/video.mp4', 'w')
handle = info['handle']
try:
    with open('video.mp4', 'rb') as f:
        while chunk := f.read(4_194_304):
            await client.fs_write(handle, chunk)
finally:
    await client.fs_close(handle, 'w')

# Inline URL for streaming in a browser (<video>/<img> src)
stream_url = await client.fs_get_url('uploads/video.mp4', expires_in=600)

# Force a download with a friendly filename (works cross-origin on S3/Azure too)
download_url = await client.fs_get_url('uploads/video.mp4', download_name='my video.mp4')

App publish ladder

Typed wrappers over rrext_app_deploy — the publish ladder for RocketRide apps. Publish snapshots an immutable version (never activates anything); Deploy pins a rung (@user, @team/<name>, @org) to a version — first publish, update, promote, and rollback are all this one verb.

Method	Signature	Description
`app_publish`	`async def app_publish(self, app_id, version, bundle, message='', module_id=None, name=None) -> dict`	Publish an immutable version to the org registry (single-file `remoteEntry.js` bundle; commit-style `message` shows on the version card).
`app_versions`	`async def app_versions(self, app_id) -> list[dict]`	The version rail, newest first; each entry carries `rungs` naming the rungs currently pinned to it.
`app_deploy`	`async def app_deploy(self, app_id, registry_version, target) -> dict`	Pin a rung to a version. Personal deploys resolve into your own manifest immediately.
`app_where`	`async def app_where(self, app_id) -> list[dict]`	The reverse index: `{rung, handle, version, appVersion, state, deployedAt}` per rung.

Events

Method	Signature	Returns	Description
`set_events`	`async def set_events(self, token: str, event_types: List[str]) -> None`	-	Subscribes this task to the given event types. After this, those events are delivered to `on_event`. Call after `use()` when you need upload or processing progress.

Services, validation, and ping

Method	Signature	Returns	Description
`get_services`	`async def get_services(self) -> dict`	`dict`	Returns all service definitions. Use to discover what the server supports.
`get_service`	`async def get_service(self, service: str) -> Optional[dict]`	`dict \| None`	Returns one service by name; `None` if not found or on error.
`validate`	`async def validate(self, pipeline: PipelineConfig, *, source: str = None) -> dict`	`dict`	Validates a pipeline configuration without starting it. Returns validation results (e.g. errors, warnings). Use to check pipeline correctness before `use()`.
`ping`	`async def ping(self, token: str = None) -> None`	-	Liveness check; raises on failure.

Chat

Method	Signature	Returns	Description
`chat`	`async def chat(self, *, token: str, question: Question) -> PIPELINE_RESULT`	`PIPELINE_RESULT`	Sends the `Question` to the AI for the given token and returns the pipeline result. The answer is in the result body; use the schema's answer helpers if you need to parse JSON from the AI text.

How it works: The client opens a pipe with the question MIME type, writes the serialized Question, closes the pipe, and returns the server result. The pipeline must support the chat provider.

Deploy

Accessed via client.deploy. Teams-as-environments deployments: publish snapshots a pipeline as an immutable, sha256-locked artifact version in the org registry; deploy points a team (the environment — Staging, Production, ...) at a version. Promotion and rollback are the same pointer move. Deploy targets are always explicit — there is no default-team fallback. Every publish and pointer change lands in an immutable audit history.

Method	Signature	Returns	Description
`deploy.publish`	`async def publish(self, pipeline, *, comment=None, deploy_to=None) -> PublishResult`	`PublishResult`	Snapshots the pipeline as the next registry version. `deploy_to` also points that team at it (one-step).
`deploy.deploy`	`async def deploy(self, project_id, version, team_id) -> Deployment`	`Deployment`	Points the team at a published version — promotion and rollback alike.
`deploy.list`	`async def list(self, *, team_id=None, page=None, page_size=None, search=None, filters=None, sort=None)`	`DeployListResult`	Deployments visible to the caller, standard `{rows, total, page, pageSize}` envelope.
`deploy.get`	`async def get(self, project_id, team_id) -> Deployment`	`Deployment`	One team's deployment, registry-joined.
`deploy.versions`	`async def versions(self, project_id, *, page=None, ...) -> DeployVersionsResult`	`DeployVersionsResult`	The registry versions (the version strip), newest first, standard envelope.
`deploy.history`	`async def history(self, project_id, *, team_id=None, page=None, ...) -> DeployHistoryResult`	`DeployHistoryResult`	The immutable audit trail, newest first; rows carry `seq` (the stable append-order identity). Server-paged.
`deploy.disable`	`async def disable(self, project_id, team_id) -> Deployment`	`Deployment`	Disables the team deployment — the kill switch: nothing runs until enabled again.
`deploy.enable`	`async def enable(self, project_id, team_id) -> Deployment`	`Deployment`	Enables a disabled team deployment.
`deploy.remove`	`async def remove(self, project_id, team_id) -> Deployment`	`Deployment`	Soft remove: hidden from listings; history and artifacts survive forever. Re-deploying revives it.
`deploy.set_schedule`	`async def set_schedule(self, project_id, source_id, schedule, team_id, *, ttl=None) -> Deployment`	`Deployment`	Sets (or clears with `None`/`'manual'`) one source's 5-field cron schedule; the paused flag is untouched.
`deploy.pause_schedule`	`async def pause_schedule(self, project_id, source_id, team_id) -> Deployment`	`Deployment`	Pauses ONE source's schedule — cron/ttl kept, it just stops firing.
`deploy.resume_schedule`	`async def resume_schedule(self, project_id, source_id, team_id) -> Deployment`	`Deployment`	Resumes a paused source schedule.
`deploy.preview`	`async def preview(self, schedule, count=None) -> SchedulePreview`	`SchedulePreview`	THE single cron evaluator: validity + next occurrences. Never parse cron client-side.

States: state is 'enabled' (schedules fire per cron), 'disabled' (the kill switch), 'errored' (a scheduled dispatch failed on permissions and the scheduler stopped retrying), or 'removed' (soft delete). Scheduled runs execute AS THE TEAM (no stored user credential); their logs land in the team's run-log continuum, readable by teammates via client.log with team_id.

Example:

result = await client.deploy.publish(my_pipeline, comment='v2 prompt fix')
await client.deploy.deploy('proj-1', result['artifact']['version'], 'team-staging')
await client.deploy.set_schedule('proj-1', 'webhook_1', '*/15 * * * *', 'team-staging')

# Promote the same version to Production later — the identical gesture.
await client.deploy.deploy('proj-1', result['artifact']['version'], 'team-prod')

live = await client.deploy.list()
for dep in live['rows']:
    print(dep['teamId'], dep['projectId'], 'v', dep['version'], dep['state'])

DataPipe

Returned by await client.pipe(...). One streaming upload: open -> write (one or more) -> close. You can also use it as 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.
`close`	`async def close(self) -> PIPELINE_RESULT`	`PIPELINE_RESULT`	Closes the pipe and returns the processing result.
`__aenter__`	`async def __aenter__(self)`	`self`	Enters context; calls `open()`.
`__aexit__`	`async def __aexit__(self, exc_type, exc_val, exc_tb)`	-	Exits context; calls `close()`.

Question

From rocketride.schema. Build a question for client.chat(token=..., question=question). Add instructions, examples, context, history, and documents to steer the AI.

Constructor

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

QuestionType: QUESTION, SEMANTIC, KEYWORD, GET, PROMPT. Default type is QUESTION. Default filter and expectJson=False, role='' if omitted.

Methods

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.
`getPrompt`	`getPrompt(self, has_previous_json_failed: bool = False) -> str`	Returns the full prompt (internal).

Answer

From rocketride.schema. Used to parse chat response content. The client does not attach an Answer instance to the pipeline result; you read the response body and, if needed, use these helpers to extract JSON or code from AI text (which often includes markdown or code fences).

Method	Signature	Description
`getText`	`getText(self) -> str`	Get the answer as plain text.
`getJson`	`getJson(self) -> Optional[dict]`	Get the answer as parsed JSON; returns `None` if not valid JSON.
`isJson`	`isJson(self) -> bool`	Whether the answer contains valid JSON.
`parseJson`	`parseJson(self, value: str) -> Any`	Parses JSON from AI text (strips markdown/code blocks).
`parsePython`	`parsePython(self, value: str) -> Any`	Extracts Python code from a code block in the response.

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.
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.
DeployArtifact: one immutable registry version — version, sha256, bytes, pipelineName, publishedBy, publishedAt, comment.
Deployment: one team's deployment, registry-joined — teamId, projectId, version, state ('enabled' | 'disabled' | 'errored' | 'removed'), schedules, actor/timestamp fields.
DeployHistoryEntry: one audit row — seq (stable append-order identity), at, action, teamId, version, actor.
PublishResult / DeployListResult / DeployVersionsResult / DeployHistoryResult / SchedulePreview: method result shapes (list results are the standard {rows, total, page, pageSize} envelope).
QuestionHistory: { 'role': str, 'content': str }.
QuestionInstruction: { 'subtitle': str, 'instructions': str }.
QuestionExample: { 'given': str, 'result': str }.

Exceptions

The exception hierarchy provides fine-grained error handling:

DAPException                    # Base DAP protocol error (has dap_result dict)
└── RocketRideException         # Base for all RocketRide errors
    ├── ConnectionException     # Connection/network issues
    │   └── AuthenticationException  # Bad API key or credentials
    ├── PipeException           # Data pipe errors (open/write/close)
    ├── ExecutionException      # Pipeline start/run failures
    └── ValidationException     # Invalid input/config

All exceptions expose a dap_result dict with detailed server error context.

AuthenticationException is thrown on DAP auth failure. In persist mode the client catches it, calls on_connect_error, and does not retry so the app can fix credentials and call connect() again.

Example:

from rocketride import RocketRideClient, AuthenticationException
from rocketride.core.exceptions import PipeException, ExecutionException

try:
    async with RocketRideClient(uri=uri, auth=auth) as client:
        result = await client.use(filepath='pipeline.json')
        await client.send(result['token'], data)
except AuthenticationException:
    print('Bad credentials')
except ExecutionException as e:
    print(f'Pipeline failed: {e}')
except PipeException as e:
    print(f'Data transfer error: {e}')

Examples (Full API Usage)

1. Minimal: connect, run pipeline from file, send one string, disconnect

import asyncio
from rocketride import RocketRideClient


async def main():
    client = RocketRideClient(uri='https://cloud.rocketride.ai', auth='my-key')
    await client.connect()
    result = await client.use(filepath='pipeline.json')
    token = result['token']
    out = await client.send(token, 'Hello, pipeline!', objinfo={'name': 'input.txt'}, mimetype='text/plain')
    print(out)
    await client.terminate(token)
    await client.disconnect()


asyncio.run(main())

2. One-off script with context manager (recommended)

import asyncio
from rocketride import RocketRideClient


async def main():
    async with RocketRideClient(uri='wss://cloud.rocketride.ai', auth='my-key') as client:
        result = await client.use(pipeline={'pipeline': my_pipeline_config})
        token = result['token']
        await client.send(token, '{"data": 1}')
        status = await client.get_task_status(token)
        print(status)
        await client.terminate(token)


asyncio.run(main())

3. Long-lived app: persist mode, callbacks, and status handling

import asyncio
from rocketride import RocketRideClient


async def main():
    client = RocketRideClient(
        uri='https://cloud.rocketride.ai',
        auth='my-key',
        persist=True,
        max_retry_time=300000,
        on_connected=lambda info: print('Connected:', info),
        on_disconnected=lambda reason, has_error: print('Disconnected:', reason, has_error),
        on_connect_error=lambda msg: print('Connect error:', msg),
        on_event=lambda e: print(e.get('event'), e.get('body')),
    )
    await client.connect()
    # Later: use(), send_files(), etc. If connection drops, client retries; do not call disconnect() in on_disconnected.


asyncio.run(main())

4. Upload multiple files and poll until pipeline completes

import asyncio
from pathlib import Path
from rocketride import RocketRideClient


async def main():
    client = RocketRideClient(uri='https://cloud.rocketride.ai', auth='my-key')
    await client.connect()
    result = await client.use(filepath='vectorize.json')
    token = result['token']
    await client.set_events(token, ['apaevt_status_upload', 'apaevt_status_processing'])

    files = ['doc1.md', 'doc2.md', ('doc3.json', {'tag': 'export'}, 'application/json')]
    upload_results = await client.send_files(files, token)
    for r in upload_results:
        if r['action'] == 'complete':
            print('OK', r['filepath'])
        else:
            print('Failed', r['filepath'], r.get('error'))

    while True:
        status = await client.get_task_status(token)
        print(f'Progress: {status.get("completedCount", 0)}/{status.get("totalCount", 0)}')
        if status.get('completed'):
            break
        await asyncio.sleep(2)
    await client.terminate(token)
    await client.disconnect()


asyncio.run(main())

5. Streaming large data with a pipe

import asyncio
from rocketride import RocketRideClient


async def main():
    async with RocketRideClient(uri='https://cloud.rocketride.ai', auth='my-key') as client:
        result = await client.use(filepath='ingest.json')
        token = result['token']
        pipe = await client.pipe(token, objinfo={'name': 'large.csv'}, mime_type='text/csv')
        await pipe.open()
        with open('large.csv', 'rb') as f:
            while True:
                chunk = f.read(64 * 1024)
                if not chunk:
                    break
                await pipe.write(chunk)
        result = await pipe.close()
        print(result)
        await client.terminate(token)


asyncio.run(main())

6. Chat: question with instructions and examples, parse JSON answer

import asyncio
from rocketride import RocketRideClient
from rocketride.schema import Question, Answer


async def main():
    async with RocketRideClient(uri='https://cloud.rocketride.ai', auth='my-key') as client:
        result = await client.use(filepath='chat_pipeline.json')
        token = result['token']
        question = Question(expectJson=True)
        question.addInstruction('Format', 'Return a JSON object with keys: summary, keywords.')
        question.addExample('Summarize X', {'summary': '...', 'keywords': ['a', 'b']})
        question.addQuestion('Summarize the main points and list keywords.')
        response = await client.chat(token=token, question=question)
        answer_text = response.get('data', {}).get('answer') or (response.get('answers') or [None])[0]
        structured = Answer().parseJson(answer_text) if answer_text else None
        print(structured)
        await client.terminate(token)


asyncio.run(main())

7. Discover services and send a custom DAP request

import asyncio
from rocketride import RocketRideClient


async def main():
    client = RocketRideClient(uri='https://cloud.rocketride.ai', auth='my-key')
    await client.connect()
    services = await client.get_services()
    print('Available:', list(services.keys()))
    ocr = await client.get_service('ocr')
    if ocr:
        print('OCR schema:', ocr.get('schema'))
    req = client.build_request('rrext_ping', token=my_token)
    res = await client.request(req, timeout=5000)
    if client.did_fail(res):
        raise RuntimeError(res.get('message', 'Ping failed'))
    await client.disconnect()


asyncio.run(main())

CLI

The rocketride command is installed automatically with the package.

rocketride start pipeline.json              # Start a pipeline
rocketride upload *.pdf --token <token>      # Upload files to a running pipeline
rocketride status --token <token>            # Monitor task progress
rocketride stop --token <token>              # Terminate a running task
rocketride list                              # List all active tasks
rocketride events ALL --token <token>        # Stream task events
rocketride rrext_store get_all_projects      # List stored projects

All commands accept --uri and --apikey flags, or read from environment variables.

Configuration

Variable	Description
`ROCKETRIDE_URI`	Server URI (e.g. `wss://cloud.rocketride.ai` or `ws://localhost:5565`)
`ROCKETRIDE_APIKEY`	API key for authentication

License

MIT - see LICENSE.

Quick Start​

What is RocketRide?​

Features​

RocketRideClient​

Constructor​

Context manager​

Connection​

Low-level DAP​

Pipeline execution​

Data​

Store (file access)​

App publish ladder​

Events​

Services, validation, and ping​

Chat​

Deploy​

DataPipe​

Question​

Constructor​

Methods​

Answer​

Types​

Exceptions​

Examples (Full API Usage)​

1. Minimal: connect, run pipeline from file, send one string, disconnect​

2. One-off script with context manager (recommended)​

3. Long-lived app: persist mode, callbacks, and status handling​

4. Upload multiple files and poll until pipeline completes​

5. Streaming large data with a pipe​

6. Chat: question with instructions and examples, parse JSON answer​

7. Discover services and send a custom DAP request​

CLI​

Configuration​

Links​

License​

Quick Start

What is RocketRide?

Features

RocketRideClient

Constructor

Context manager

Connection

Low-level DAP

Pipeline execution

Data

Store (file access)

App publish ladder

Events

Services, validation, and ping

Chat

Deploy

DataPipe

Question

Constructor

Methods

Answer

Types

Exceptions

Examples (Full API Usage)

1. Minimal: connect, run pipeline from file, send one string, disconnect

2. One-off script with context manager (recommended)

3. Long-lived app: persist mode, callbacks, and status handling

4. Upload multiple files and poll until pipeline completes

5. Streaming large data with a pipe

6. Chat: question with instructions and examples, parse JSON answer

7. Discover services and send a custom DAP request

CLI

Configuration

Links

License