# RocketRide Documentation > Build, run, and ship data + AI pipelines with the RocketRide toolchain. --- # RocketRide Documentation Route: / --- slug: / title: RocketRide Documentation sidebar_label: Home sidebar_position: 0 hide_table_of_contents: true hide_title: true pagination_next: null pagination_prev: null --- import { LuZap, LuCode, LuRocket, LuCloud, LuPlug, LuBlocks, LuBookOpen } from 'react-icons/lu'; import { SiPython, SiTypescript, SiDiscord, SiGithub } from 'react-icons/si'; import { VscVscode } from 'react-icons/vsc';
# RocketRide Documentation RocketRide is an open-source runtime for AI pipelines. Pipelines are portable JSON: version-controlled, shareable, and executed by a multithreaded C++ core that runs the same way on your laptop, your servers, or RocketRide Cloud.
Get Started Core Concepts
01 · Build ## Build your first pipeline New to RocketRide? Start here: run a working pipeline and pick up the core concepts along the way. }> Install the VS Code extension, deploy a local runtime, and assemble a pipeline visually in minutes. }> Drop pipelines into your TypeScript or Python application with a few lines of code. 02 · Run ## Choose how you run RocketRide }> Start building right away on the managed platform. Same pipeline JSON, zero infrastructure to manage. }> Run the runtime on your own infrastructure — Docker, on-prem, or anywhere the engine builds. 03 · Explore ## Explore more }> Deploy, run, and observe pipelines from your TypeScript app. }> The same client surface, native to Python. }> The full catalog: LLMs, vector stores, parsers, tools, and more. }> Expose any pipeline as an MCP tool for AI assistants. }> Build, run, and debug pipelines visually in your editor. }> Every field of the portable pipeline JSON format. 04 · Community ## Community }> Share what you're building, swap ideas, and get help from the community. }> RocketRide is open source. Report bugs, discuss in issues, and contribute code. --- # Quickstart Route: /quickstart --- sidebar_position: 2 sidebar_label: Overview title: Quickstart hide_table_of_contents: true --- import { LuLayoutDashboard, LuCode, LuTerminal } from 'react-icons/lu'; # Quickstart Pick the path that matches how you want to build. Each one takes you from zero to a running pipeline.
Build in your IDE Install the extension, deploy a local runtime, and wire up your first pipeline on the visual canvas. Integrate with an SDK Run the .pipe file you built from your own Python or TypeScript application. Run from the CLI Start, feed, and monitor pipelines from a terminal — no code required.
## New to RocketRide? Start with **[Build in your IDE](/quickstart/ide-walkthrough)**, it installs the extension, spins up a local runtime, and walks you through a `Chat → LLM` pipeline you can run in minutes. Once you have a pipeline, the **[SDK walkthrough](/quickstart/sdk-integration)** shows how to run it from your own [Python](/clients/python) or [TypeScript](/clients/typescript) application. Not sure where it should run long-term? See [Choose How to Run RocketRide](/operate). ## More examples For a curated, community-maintained list of RocketRide projects, templates, and integration examples, see [**awesome-rocketride**](https://github.com/rocketride-org/awesome-rocketride): real-world pipelines (RAG over your docs, document extraction (OCR/NER), PII anonymization, multi-provider LLM routing, and agent workflows), plus starter templates you can clone and run. --- # Run from the CLI Route: /quickstart/cli --- title: Run from the CLI --- # Run from the CLI The `rocketride` command-line tool runs pipelines from a terminal — the same operations the SDKs expose, no code required. It ships with both [clients](/clients): `pip install rocketride` or `npm install rocketride` puts `rocketride` on your path. The five steps below take you from a fresh install to a running pipeline. The full command and flag reference is on the [CLI page](/connect/cli). ## 1. Point at an engine Set your connection once with environment variables so you don't have to repeat them on every command: ```bash # Local engine (no API key needed) export ROCKETRIDE_URI=ws://localhost:5565 # RocketRide Cloud — generate an API key from the online editor export ROCKETRIDE_URI=wss://api.rocketride.ai export ROCKETRIDE_APIKEY=your-api-key ``` See [Choose How to Run RocketRide](/operate) for engine setup. ## 2. Start a pipeline Pass a `.pipe` file. The command starts the task, prints its token, and exits; the pipeline keeps running on the engine: ```bash rocketride start --pipeline ./my-pipeline.pipe ``` The CLI prints a **task token** when the run starts. Copy it, you'll use it in the next steps. ``` Starting pipeline from ./my-pipeline.pipe... Pipeline started. Token: ey... Stop it with: rocketride stop --token ey... ``` ## 3. Upload files through a pipeline Use `upload` to push one or more files through an extraction or processing pipeline: ```bash rocketride upload --pipeline ./extract.pipe ./document.pdf ``` Or feed files into a task that's already running by passing its token: ```bash rocketride upload --token ./report-q1.pdf ./report-q2.pdf ``` ## 4. Check what is running List the active tasks to confirm yours is still up and to find its token again: ```bash rocketride list ``` The CLI does not stream events; live monitoring belongs to the platform's monitor apps. ## 5. Stop a task When you're done, or need to cancel early: ```bash rocketride stop --token ``` ## Next steps - [CLI reference](/connect/cli): every command, flag, and the file-store operations. - [Integrate with an SDK](/quickstart/sdk-integration): the same operations, in code. - [Examples](/examples/rag-pipeline): full pipelines to run. --- # Build a pipeline in your IDE Route: /quickstart/ide-walkthrough --- title: Build a pipeline in your IDE sidebar_label: Build in your IDE --- import ThemedImage from '@theme/ThemedImage'; import { LuPlay } from 'react-icons/lu'; # Build a pipeline in your IDE The visual canvas in the VS Code extension is the fastest way to author a `.pipe` file. This walkthrough starts from zero and finishes with a running `Chat → LLM` pipeline you can talk to. ## 1. Install the extension Search for **RocketRide** in the VS Code Extension Marketplace and install it. The extension also works in VS Code forks (Cursor, Windsurf, VSCodium) via the [Open VSX Registry](https://open-vsx.org/extension/RocketRide/rocketride). ## 2. Deploy a server Click the RocketRide () icon in your IDE sidebar, then choose how to run the runtime. **Local** is the right choice here — it pulls the server straight into your IDE with no extra setup. (The other options are covered in [Choose How to Run RocketRide](/operate).) ## 3. Create a pipeline file Create a file ending in `.pipe` (e.g. `my-first-pipeline.pipe`). The extension opens it in the visual builder canvas. `.pipe` files are JSON under the hood, but you author them visually. ## 4. Build a simple chat pipeline Every pipeline starts with a **source node**: 1. Add a **Chat** source node: an interactive conversational interface. 2. Add an **LLM** node: pick a provider (OpenAI, Anthropic, Google, …) and set your API key. 3. Connect the Chat source's output lane to the LLM's input lane. The result is a `Chat → LLM` pipeline; the LLM's response routes back to the chat interface automatically. ## 5. Run it Press the ** Run button** on the source node, or launch from the **RocketRide sidebar**. Open the chat interface, send a message, and watch the LLM respond in real time. Use the **Server Monitor** page to trace call trees, token usage, and memory consumption. Save the `.pipe` file, you'll run it from code in the next walkthrough. ## Next - [Integrate a pipeline with an SDK](/quickstart/sdk-integration): run the `.pipe` file you just built from your own Python or TypeScript application. - [VS Code extension](/clients/vscode): the full extension guide (canvas, runtime management, tracing). --- # Integrate a pipeline with an SDK Route: /quickstart/sdk-integration --- title: Integrate a pipeline with an SDK sidebar_label: Integrate with an SDK --- # Integrate a pipeline with an SDK Once you have a `.pipe` file, run it from your own application with the [Python](/clients/python) or [TypeScript](/clients/typescript) SDK. Both connect to a running engine, a local server (`ws://localhost:5565`) or RocketRide Cloud (`https://api.rocketride.ai`), start the pipeline with `use()`, stream data with `send()`, and stop it with `terminate()`. If you do not have a `.pipe` file yet, build one first with the [IDE walkthrough](/quickstart/ide-walkthrough). ## Python ```bash pip install rocketride ``` ```python import asyncio from rocketride import RocketRideClient async def main(): async with RocketRideClient(uri='ws://localhost:5565', auth='my-key') as client: result = await client.use(filepath='my-first-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()) ``` See the [Python SDK reference](/clients/python) for chat, file uploads, streaming pipes, events, and persist-mode reconnection. ## TypeScript ```bash npm install rocketride ``` ```typescript import { RocketRideClient } from 'rocketride'; const client = new RocketRideClient({ uri: 'ws://localhost:5565', auth: process.env.ROCKETRIDE_APIKEY! }); await client.connect(); const { token } = await client.use({ filepath: './my-first-pipeline.pipe' }); const result = await client.send(token, 'Hello, pipeline!', { name: 'input.txt' }, 'text/plain'); console.log(result); await client.terminate(token); await client.disconnect(); ``` See the [TypeScript SDK reference](/clients/typescript) for chat, file uploads, streaming pipes, events, and persist-mode reconnection. ## Next - [Pipeline JSON reference](/reference/pipeline-reference): every field of a `.pipe` file. - [Troubleshooting](/support/troubleshooting): what to check when a run does not behave. --- # Understanding RocketRide Route: /concepts --- title: Understanding RocketRide sidebar_label: Understanding RocketRide --- # Understanding RocketRide RocketRide has a few moving parts. Once you know how they fit together, the rest of the docs map cleanly onto them. ## The pipeline A **pipeline** is a graph of nodes defined in a `.pipe` file (JSON). Data flows between nodes along typed **data lanes**: a node declares which input lanes it consumes and which output lanes it produces, and the engine routes data accordingly. See [Pipelines](/concepts/pipelines) and the [Execution model](/concepts/execution-model). ## Nodes **[Nodes](/nodes)** are the building blocks: LLM providers, vector stores, embedding models, preprocessors, OCR/NER, web tools, agents, and sources like Chat. Each node ships a schema (its config, inputs, and outputs) and runs inside the engine. Connectors are the nodes that read from and write to external systems. See [Nodes](/concepts/nodes) and [Agents & tools](/concepts/agents-tools-skills). ## The runtime engine Pipelines execute on a multithreaded **C++ engine** (the runtime). It loads the `.pipe` definition, instantiates the nodes, and streams data through the graph. The same engine runs locally, on-premises, and on RocketRide Cloud. See [Runtime & engine](/concepts/runtime-engine). ## Talking to the engine You start and feed pipelines through one of two protocols: - **[WebSocket](/connect/websocket)**: the native engine protocol (port 5565). The [TypeScript](/clients/typescript) and [Python](/clients/python) SDKs speak it for you: `use()` to start a pipeline, `send()`/`pipe()` to stream data, `chat()` for conversational flows, `terminate()` to stop. - **[MCP](/connect/mcp/stdio)**: expose a pipeline as a tool for AI assistants like Claude and Cursor. ## How you build - **Visually**: the VS Code [extension](/clients/vscode) opens `.pipe` files on a canvas; wire nodes by connecting lanes and press Run. - **In code**: author or run the same pipeline from your application with the SDKs. - **As an app**: wrap a pipeline in a UI that runs inside the RocketRide shell and deploy it to your team. See [Apps](/concepts/apps). ## Putting it together A typical flow: author a `.pipe` visually → run it locally to iterate → integrate it into your app via an SDK → deploy the engine on-prem or to [Cloud](/operate/cloud). The pipeline JSON never changes across those steps. See the [Quickstart](/quickstart) walkthroughs to do this end to end. --- # Agents & Tools Route: /concepts/agents-tools-skills --- title: Agents & Tools sidebar_position: 4 --- # Agents & tools Most [nodes](/concepts/nodes) pass data along a lane and move on. An **agent** is different: it reasons in a loop, deciding which model to call, which tools to use, and when it is done. To do that it needs a few helpers wired to it: an LLM, optionally tools, and (for some agent types) memory. ## Data lanes vs. control connections Agents introduce a second kind of wiring alongside data lanes: - **Data lanes** carry data _into_ and _out of_ the agent, a question arrives on an input lane, an answer leaves on an output lane. - **Control (`invoke`) connections** attach the agent's _capabilities_: the LLM it thinks with, the tools it can call, the memory it reads and writes. See the [Execution model](/concepts/execution-model) for how the two interact. ## Wiring: `control` lives on the helper The connection between an agent and its helpers is declared on the **helper**, not on the agent. Each LLM, tool, or memory node carries a `control` array whose `from` points back at the agent that invokes it. The agent itself has no `control` array, only its input lanes. The agent has input lanes only, no `control` array. The LLM declares it is controlled _by_ the agent, and the tool does likewise: ```json [ { "id": "agent_1", "provider": "agent_rocketride", "input": [{ "lane": "questions", "from": "chat_1" }] }, { "id": "llm_1", "provider": "llm_openai", "control": [{ "classType": "llm", "from": "agent_1" }] }, { "id": "tool_1", "provider": "tool_http_request", "control": [{ "classType": "tool", "from": "agent_1" }] } ] ``` A single LLM, tool, or memory node can serve several invokers: list each one as its own entry in the helper's `control` array. ## Tools A **tool** (class type `tool`) is a capability an agent can invoke at runtime: an HTTP request, a web search, a shell command, a filesystem or git operation, another pipeline, and many more. Tools have **no data lanes**: nothing streams through them. They sit idle until an agent decides to call one, then return a result to that agent. A tool joins a pipeline purely through its `control` connection. ## Memory Some agents keep state across turns through a **memory** node (`memory_internal` or `memory_persistent`), wired the same way as any other helper. | Agent | LLM | Memory | Tools | | ------------------ | -------------------- | -------------------- | -------- | | `agent_rocketride` | Required (exactly 1) | Required (exactly 1) | Optional | | `agent_crewai` | Required (min 1) | Not supported | Optional | | `agent_langchain` | Required (min 1) | Not supported | Optional | Only `agent_rocketride` has a memory port. Do not wire memory to `agent_crewai` or `agent_langchain`. ## Multi-agent pipelines An agent can invoke **another agent as a tool**. The sub-agent declares `control: [{ "classType": "tool", "from": "" }]` and takes no input lanes of its own, it is driven by its parent. The sub-agent's own helpers (its LLM and memory) point their `control` at the sub-agent, not at the parent. This lets you compose specialists under a coordinator. > The same `invoke`/`control` pattern applies beyond agents, any node whose > catalog entry declares an `invoke` field (for example `summarization` or > `extract_data`) is wired to its LLM the same way. ## Next steps - [Nodes](/nodes): every agent, tool, LLM, and memory provider, with its `invoke` requirements. - [Execution model](/concepts/execution-model): how control connections run alongside data lanes. - [Pipeline JSON reference](/reference/pipeline-reference): the `control` and `invoke` fields in full. --- # Apps Route: /concepts/apps --- title: Apps --- # Apps A **RocketRide app** is a user interface that runs inside the RocketRide shell against your engine. The shell provides sign-in, the engine connection, the workspace, and settings; the app brings the screens and the pipelines behind them. Apps run in the browser or in a VS Code webview; the pipelines they start always run on the server. ## Why apps To a user, an app turns a pipeline's work into a tool a teammate opens, not a script someone has to run themselves. To RocketRide, the same engine, node catalog, and deploy path serve both pipelines and the interfaces built on top of them, and the catalog of apps is how that work is shared inside an organization. ## Apps and pipelines A pipeline can belong to an app, or stand beside it. The difference decides who runs it and when. **Bundled with the app.** The `.pipe` file lives in the app folder and ships with the app. The app starts it when it needs it, for example when the user runs a query, and every signed-in user gets their own instance, automatically. A bundled pipeline is never shared between users and cannot run on a schedule. **Deployed on its own.** The pipeline is its own project, deployed separately from the app. One instance serves everyone on the team it is published to, and it can run on a schedule. The app refers to it by identity and attaches to whatever is running; it does not start it. A `.pipe` file outside the app folder is not deployed with the app. If the app needs it, deploy it separately. Secrets never ship in either case: pipeline configs carry `${ROCKETRIDE_*}` placeholders that the server fills from the signed-in user's stored keys. ## Next steps - [App Builder](/guides/apps/app-builder): build, preview, and deploy an app. - [Pipelines](/concepts/pipelines): what the app is running. - [Shell API](/guides/apps): the hooks and descriptor, when you code by hand. --- # Execution Model Route: /concepts/execution-model --- title: Execution Model sidebar_position: 5 --- # Execution model A [pipeline](/concepts/pipelines) describes _what_ to run; the execution model is _how_ the [engine](/concepts/runtime-engine) runs it. Two mechanisms move work through the graph: **data lanes** carry data between nodes, and **control connections** let agents invoke their helpers. ## Data lanes A **lane** is a typed channel between two nodes. A node declares the lanes it consumes in its `input` array; the engine routes each output lane to the inputs that ask for it. ```json "input": [{ "lane": "questions", "from": "qdrant_1" }] ``` This reads: _take the `questions` lane produced by `qdrant_1` as my input._ ### Lane types Lanes are typed, and the type must match across a connection. This is the complete lane table: lanes are a closed set fixed by the engine, and a node only chooses which of them it accepts (see each node's entry in [Nodes](/nodes)): | Lane | Carries | Accepted by | | ----------- | -------------------------------- | -------------------------------------------------------- | | `questions` | Queries flowing toward a model | LLMs, vector stores (for retrieval), agents | | `answers` | Model responses flowing back | `response` target, nodes expecting generated text | | `documents` | Vector-ready chunks | Vector store `documents` input | | `text` | Plain text content | Preprocessors | | `tags` | Structured metadata / parameters | Parsers | | `image` | Image content | Vision nodes | | `audio` | Audio streams | Audio nodes | | `video` | Video streams | Video nodes | | `table` | Structured / tabular data | Extractors, table-aware preprocessors | | `json` | JSON payloads | `response` target | | `words` | Word-level tokens | Engine-defined; no catalog node consumes it today | | `classifications` | Classification results | Engine-defined; no catalog node consumes it today | | `classificationContext` | Context for a classification pass | Engine-defined; no catalog node consumes it today | ### Lane flow rules - **Type compatibility**: the output lane of one node must match the input lane of the next. A `text` output feeds a `text` input; mismatched lanes are a pipeline error. - **Transformation**: many nodes change the lane type. A preprocessor turns `text` into `documents` (vector-ready chunks); a text embedding node enriches `documents` with vectors for a store (media embedding nodes turn `image`/ `video` into `documents`); an LLM turns `questions` into `answers`. - **Fan-in**: a node can consume the same lane from several upstream nodes by listing multiple entries in `input`. A `response` node, for example, can merge `answers` from several agents. ```json "input": [ { "lane": "answers", "from": "agent_rocketride_1" }, { "lane": "answers", "from": "agent_crewai_1" } ] ``` ## Control connections Data lanes are not the only wiring. Agents (and other nodes with an `invoke` field) reach their LLM, tools, and memory through **control connections** instead of lanes, see [Agents & tools](/concepts/agents-tools-skills). Control connections form a side channel: the engine resolves them at startup so an agent can call a tool mid-run without that tool ever sitting on a data lane. ## How a run flows The engine streams; it does not run the graph stage by stage. Once a pipeline starts: 1. Data enters at a **source** (a `webhook`, a `chat` stream, a file). 2. Each node processes data as it arrives and emits onto its output lanes. Independent branches run **concurrently** across threads. 3. Agents loop, calling their LLM and tools over one or more **waves** of reasoning, until they produce a result. For `agent_rocketride`, `max_waves` caps how many reasoning cycles it may take. 4. Results stream out through a **target** (typically a `response` node) back to the client as they are produced. Because data streams rather than buffering, results can begin returning before the whole input is consumed, which is what makes conversational `chat()` flows feel live. ## How the engine parallelises The engine is written in C++ and runs each pipeline run on its own thread pool. Concurrent requests to the same pipeline do not queue behind each other — the engine spawns an independent execution context for each incoming task. A slow request (a large document going through OCR, embedding, and an LLM call) does not block a fast one (a short question answered directly by the LLM). ### Streaming execution Nodes process data **as it arrives**, not after the full upstream output is available. When a preprocessor splits a 100-page document into 200 chunks, the embedding node starts embedding chunk 1 while the preprocessor is still producing chunk 2. The vector store starts upserting while the embedder is computing later chunks. This keeps memory usage low and reduces end-to-end latency, especially for large documents. ### Vector store batching Vector stores (Qdrant, Pinecone, Milvus, Weaviate, etc.) accumulate chunks and flush them in batches rather than upserting one at a time. A batch flushes when it reaches either a chunk-count limit or a payload-size limit, whichever comes first. The exact thresholds are backend-specific: for example, Qdrant flushes at 500 points or its payload limit, while Pinecone and Milvus use different chunk-count defaults. For small documents that produce few chunks, the flush happens at pipeline completion. For large document sets, flushing starts mid-run and reduces peak memory. Tuning batch size is covered in [Performance](/guides/performance). Every run is also recorded to a durable run log as it executes — chapters, traces, and console output you can replay after the run is gone. The [Observability guide](/guides/observability) covers that side of execution. ## Next steps - [Agents & tools](/concepts/agents-tools-skills): control connections in depth. - [Nodes](/concepts/nodes): what sits on each lane. - [WebSocket protocol](/connect/websocket): how clients feed and read a run. - [Observability](/guides/observability): the run-log DVR, trace levels, and every monitoring surface. - [Pipeline JSON reference](/reference/pipeline-reference): the `input`, `lane`, and `control` fields. --- # Nodes Route: /concepts/nodes --- title: Nodes sidebar_position: 3 --- # Nodes A [pipeline](/concepts/pipelines) is a graph, and **nodes** are its vertices. Every node is one component that does one job: call a model, embed text, query a vector store, parse a document, run a tool. You assemble nodes into a pipeline; the [engine](/concepts/runtime-engine) runs them. ## Anatomy of a node Each node in a `.pipe` file is an object with a stable identity and a behaviour: - **`id`**: a unique name for this node within the pipeline (e.g. `llm_1`). - **`provider`**: what the node _is_ (e.g. `llm_openai`, `qdrant`, `webhook`). The provider determines the node's behaviour and which lanes it supports. - **`config`**: provider-specific settings: API keys, model profiles, collection names, instructions. Swapping a provider or model is a config edit, not a code change. - **`input`**: the data lanes this node consumes and the nodes they come from. ```json { "id": "llm_1", "provider": "llm_openai", "config": { "profile": "openai-5-2" }, "input": [{ "lane": "questions", "from": "qdrant_1" }] } ``` ## Class types Every provider belongs to a **class type** that describes the kind of work it does. The class type also governs how the node is wired: data nodes connect through lanes, while `agent`, `tool`, `llm`, and `memory` nodes participate in control connections (see [Agents & tools](/concepts/agents-tools-skills)). > source · data · text · image · audio · video · embedding · llm · store · > database · graph · tool · agent · memory · guard · rerank · search · > infrastructure · target · preprocessor The list is not exhaustive — new class types are added as the catalog grows. ## Connectors **Connectors** are the nodes at the edges of the graph, the ones that read from or write to the world outside the pipeline: - **Sources** bring data in: a `webhook` that receives a request, a `chat` source that streams a conversation, a file or database reader. - **Targets** send results out: a `response` node that returns data to the caller, or a node that writes to a store or external system. Everything between a source and a target (embedding, retrieval, LLM calls, preprocessing) transforms data as it flows through. ## Swap providers, keep the pipeline Because behaviour lives in `provider` + `config`, you can often change _which_ LLM or vector store a pipeline uses without touching its shape — as long as the new provider has a compatible contract (same class type, supported lanes, and control connections). Point an `llm` node at a different provider, or repoint a `store` node at a different collection, and the surrounding graph is unchanged; swapping to a provider with a different contract can change required config, lanes, or control connections. Always [validate](/reference/pipeline-reference) the pipeline after changing a provider. ## The catalog Every available provider (100+ nodes across 15+ LLM providers, 9 vector databases, OCR, NER, PII anonymization, transcription, and web tools) is documented with its config, inputs, and outputs in **[Nodes](/nodes)**. ## Next steps - [Nodes](/nodes): every provider and its schema. - [Agents & tools](/concepts/agents-tools-skills): the control-plane nodes. - [Execution model](/concepts/execution-model): how lanes carry data between nodes. - [Pipeline JSON reference](/reference/pipeline-reference): every field of a node. --- # Pipelines Route: /concepts/pipelines --- title: Pipelines sidebar_position: 1 --- # Pipelines A **pipeline** is the unit of work in RocketRide: a directed graph of components that move and transform data. Pipelines are authored as `.pipe` files (JSON) and executed by the engine. ## Components and providers Each node in the graph is a **component** with a unique `id` and a `provider` that determines its behaviour (for example `webhook`, `response`, or an LLM provider). Provider-specific settings live in the component's `config`. See the [Nodes](/nodes) catalog for every available provider. ## Data lanes vs. invoke connections Components are wired together two ways: - **Data lanes**: a typed channel (e.g. `questions` → `answers`) carrying data from one component to the next. Declared as input connections. - **Invoke (control) connections**: a component calls another by class type (e.g. an agent invoking an `llm`), rather than streaming data through a lane. ## The `.pipe` JSON shape A `.pipe` file is JSON conforming to the pipeline schema. The full field-by-field reference is generated from the schema source and published at [Pipeline JSON reference](/reference/pipeline-reference). ## Minimal example ```json { "components": [ { "id": "in", "provider": "webhook", "config": { "mode": "Source" } }, { "id": "out", "provider": "response", "config": { "lanes": [{ "laneId": "questions", "laneName": "questions" }] }, "input": [{ "lane": "questions", "from": "in" }] } ] } ``` ## Next steps - [Quickstart](/quickstart): run your first pipeline. - [Pipeline JSON reference](/reference/pipeline-reference): every field. - [Nodes](/nodes): the component catalog. --- # Runtime & Engine Route: /concepts/runtime-engine --- title: Runtime & Engine sidebar_position: 2 --- # Runtime & engine Pipelines don't run themselves. The **engine** is the runtime that loads a [`.pipe` definition](/concepts/pipelines), brings its nodes to life, and moves data through the graph until the work is done. ## A multithreaded C++ core The engine is a native, multithreaded **C++ runtime**, not a thin wrapper around HTTP calls. It is built for throughput and reliability: nodes that have no dependency on one another run concurrently, and data streams between them rather than buffering the whole pipeline in memory. The same binary powers a quick local iteration loop and a production workload. ## What the engine does When a pipeline starts, the engine: 1. **Parses** the `.pipe` JSON and validates it against the pipeline schema. 2. **Instantiates** each component from its `provider`, applying the component's `config` (API keys, model profiles, collection names, and so on). 3. **Wires** the graph: connecting output [data lanes](/concepts/execution-model) to the input lanes that consume them, and resolving control (`invoke`) connections between agents and the LLMs, tools, and memory they drive. 4. **Streams** data through the graph, scheduling work across threads and emitting results as they are produced. 5. **Tears down** the run when the inputs are exhausted or the client calls `terminate()`. For the full picture of how data and control flow at step 4, see the [Execution model](/concepts/execution-model). ## One engine, anywhere The pipeline JSON never changes across environments — only where the engine lives does: locally behind the [VS Code extension](/clients/vscode) while you build, self-hosted in your own network, or managed on RocketRide Cloud. See [Choose How to Run RocketRide](/operate) for the comparison. ## Talking to the engine You never call the engine's internals directly. Clients connect over one of two protocols and the engine handles the rest: - **[WebSocket](/connect/websocket)**: the native engine protocol. The [TypeScript](/clients/typescript) and [Python](/clients/python) SDKs speak it for you. - **[MCP](/connect/mcp/stdio)**: exposes running pipelines as tools for AI assistants. As pipelines run, the engine reports call trees, token usage, and memory so you can observe what happened. See [Troubleshooting](/support/troubleshooting) for reading that signal. ## Next steps - [Execution model](/concepts/execution-model): how the engine schedules and streams a run. - [Nodes](/concepts/nodes): the components the engine instantiates. - [Self-hosting](/operate/self-hosting): run the engine in your own infrastructure. --- # Client Libraries Route: /clients --- title: Client Libraries --- # RocketRide Client Libraries Official client libraries for the RocketRide Engine. The TypeScript and Python clients communicate with the server over DAP (Debug Adapter Protocol) on WebSocket and offer the same capabilities. The MCP client provides AI assistant integration via the Model Context Protocol. The chat widget embeds a brandable pipeline chat UI in any web page. --- ## Overview - **Connect** with an API key; optional automatic reconnection (persist mode). - **Pipelines**: start with `use()`, get a token, then send data via `send()`, `sendFiles()` / `send_files()`, or `pipe()`. - **Chat** with AI via `chat()` and a `Question` object. - **Lifecycle**: `onConnected` / `on_connected`, `onDisconnected` / `on_disconnected`, `onConnectError` / `on_connect_error`, `onEvent` / `on_event`. - **Timeouts**: per-request timeout; optional max retry time for reconnects. URIs: clients accept `http`/`https` or `ws`/`wss` and convert to WebSocket (`http` to `ws`, `https` to `wss`) when needed. --- ## Client SDK Documentation | Client | Package | Document | | -------------- | ---------------- | ------------------------------------------- | | **TypeScript** | `rocketride` | [TypeScript SDK](/clients/typescript) | | **Python** | `rocketride` | [Python SDK](/clients/python) | | **VS Code** | `rocketride` (extension) | [VS Code Extension](/clients/vscode) | | **MCP** | `rocketride-mcp` | [MCP server](/connect/mcp/stdio) | Each document lists every constructor option, method, type, and usage example for that client. Both SDK packages also install the [`rocketride` CLI](/connect/cli) — the same operations from a terminal. The Python package additionally ships the [OpenTelemetry bridge](/clients/python/otel-bridge) (`rocketride otel`), which exports live pipeline traces and metrics over OTLP. --- ## Installation ### From PyPI / npm (public registry) ```bash # TypeScript npm install rocketride # Python pip install rocketride # MCP pip install rocketride-mcp ``` The chat widget is **not on a public registry yet**; build it from this repository and self-host the bundle: ```bash ./builder chat-widget:build # -> packages/chat-widget/dist/rocketride-chat.js ``` See the [chat widget README](https://github.com/rocketride-org/rocketride-server/blob/develop/docs/public/chat-widget/README.md#getting-the-bundle) for the embed snippets and for what changes once `rocketride-chat-widget` is published. ### From the Engine (self-hosted download) The engine serves the latest client packages via HTTP endpoints. Once the server is running, download them directly: | Endpoint | Package | Response | | ------------------------------- | ---------------------- | --------------------------------------- | | `GET /client/python/{filename}` | Python SDK wheel (curl: use `latest`) | `rocketride-{version}-py3-none-any.whl` | | `GET /client/typescript` | TypeScript SDK tarball | `rocketride-{version}.tgz` | | `GET /client/vscode` | VSCode extension | `rocketride-{version}.vsix` | | `GET /client/shell` | Shell platform package | `shell.tgz` | | `GET /client/docs` | Agent docs bundle | `docs.zip` (docs + stubs + manifest) | ```bash # Download and install Python client (use "latest" as filename for newest version) curl -o rocketride-latest.whl http://localhost:5565/client/python/latest pip install rocketride-latest.whl # Download and install TypeScript client curl -O http://localhost:5565/client/typescript npm install rocketride-*.tgz # Download and install VSCode extension curl -O http://localhost:5565/client/vscode code --install-extension rocketride-*.vsix ``` These endpoints are public (no authentication required) and automatically serve the latest version. Returns 404 with a JSON error if packages are not found. All `/client/*` responses carry `Cache-Control: no-cache` so HTTP caches revalidate instead of serving stale artifacts after a server upgrade. The agent docs bundle is what `rocketride init` and the VS Code extension install into a workspace's `.rocketride/docs/` — its `manifest.json` carries a content hash consumers use as their change stamp, so an unchanged bundle is a no-op to re-install. The recommended workspace bootstrap, per language: ```bash # TypeScript — the init shim, served by the server itself. No arguments # needed: the shim reads the server from its own install URL, downloads # that server's client into .rocketride/client/rocketride.tgz, installs # it as a content-hashed file: dependency, and runs `rocketride init`. pnpm install http://localhost:5565/client/typescript-init pnpm exec typescript-init # Python — install the server's wheel by its concrete filename (pip # requires the .whl name in the URL), then init: pip install http://localhost:5565/client/python/rocketride-1.3.0-py3-none-any.whl rocketride init ``` Once the `rocketride` package is published to npmjs/PyPI, the public bootstrap is simply `pnpm add rocketride` / `pip install rocketride` followed by `rocketride init`; the forms above remain the self-hosted path. Re-running `typescript-init` / `rocketride init` is the update path — the client and workspace refresh against the connected server. Two direct-URL forms to avoid, both verified broken: - `pnpm add http://.../client/typescript` — pnpm caches URL tarballs by URL and NEVER refetches, so after a server upgrade it silently reinstalls the old cached build. The shim's `file:` dependency is hashed by content, so a rebuilt server package always installs. (The shim itself is safe to cache — it is tiny, stable, and carries no server-versioned content.) - `pip install http://.../client/python/latest` — pip rejects URLs without a recognizable archive filename ("neither 'setup.py' nor 'pyproject.toml' found"). The `/latest` route is for curl/scripting only; pip installs use the concrete wheel filename. --- ## License MIT License -- see [LICENSE](https://github.com/rocketride-org/rocketride-server/blob/develop/LICENSE). --- # Python SDK Route: /clients/python --- title: Python SDK sidebar_position: 0 sidebar_label: Overview --- import { LuSettings, LuPlug, LuPlay, LuRocket, LuSend, LuFolderOpen, LuMessageSquare, LuActivity, LuShieldAlert, LuBookOpen, LuFlaskConical, LuChartLine, LuRadar } from 'react-icons/lu'; # Python SDK Build, run, and manage AI pipelines from Python. The `rocketride` package is async-first — built on `asyncio` and `websockets` — and speaks the engine's native [WebSocket protocol](/connect/websocket): you author a `.pipe` ([visually](/clients/vscode) or [by hand](/reference/pipeline-reference)) and run it against the engine from your own application. ## Install ```bash pip install rocketride ``` Python 3.10+. The [`rocketride` CLI](/connect/cli) is installed with the package. ## Quickstart ```python import asyncio from rocketride import RocketRideClient async def main(): async with RocketRideClient(uri='https://api.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()) ``` `send()` / `send_files()` target pipelines whose **source** is `webhook` or `dropper`; if your pipeline source is `chat`, use [`client.chat()`](/clients/python/chat). No pipeline yet? The [IDE walkthrough](/quickstart/ide-walkthrough) builds one in minutes. ## Explore the SDK
Configuration Constructor options, environment variables, timeouts, and reconnection. Connection The attach/login layers, context manager, persist mode, and lifecycle callbacks. Running Pipelines Start with use(), watch progress, validate, and terminate. Deployments Publish immutable versions, point team environments at them, and schedule runs. Sending Data One-shot sends, concurrent file uploads with progress, and chunked streaming. File Storage Read, write, and manage your server-side store; signed URLs for direct access. Chat Build a Question, stream the response, parse it with Answer. Run Logs The log continuum and the DVR session: live monitoring and replay with one API. Error Handling What the SDK raises, when, and how to catch it. API Reference Every public class and method, in one place. Examples Seven complete programs, from hello-pipeline to custom DAP requests. Analytics Report client-side analytics events through the SDK. OpenTelemetry Bridge Export live pipeline traces and metrics over OTLP with rocketride otel.
## Links - [GitHub](https://github.com/rocketride-org/rocketride-server) · [PyPI](https://pypi.org/project/rocketride/) · [Discord](https://discord.gg/PMXrtenMsY) - [Release notes](/support/release-notes) · [Get help](/support/get-help) --- # Analytics Route: /clients/python/analytics --- title: Analytics sidebar_position: 12 --- - [Overview](#overview) - [Import](#import) - [API](#api) - [Event Names](#event-names) - [What This Module Is Not](#what-this-module-is-not) ## **Overview** `rocketride.analytics` is the one shared event-report function, bare bones by design. Apps call ``report(event, props)`` with any string event name and a free-form props dict. There is no central event list: **each app owns its own taxonomy**. What the shared layer guarantees is that every reported event carries an ``app`` property identifying the emitting app (``home-ui``, ``rocket-ui``, …), so downstream analytics can always segment by app. The TypeScript mirror lives at ``rocketride/analytics`` in the TypeScript SDK. ## **Import** ```python from rocketride.analytics import init_report, report ``` ## **API** ```python # Once, at app init: wire the emitting app id + transport. The sink is # whatever callable your app uses to ship events (HTTP, queue, logger, ...). init_report('rocket-ui', lambda event, props: my_sink.send(event, props)) # Anywhere after that: report('pipeline:run', {'node_count': 4}) # → sink receives ('pipeline:run', {'app': 'rocket-ui', 'node_count': 4}) ``` - ``init_report(app, sink)`` — stores the app id and transport. Call once per app. - ``report(event, props=None)`` — forwards to the sink with ``app`` stamped into the props. Enforcement is string-ish only: a non-string or empty event name is a silent no-op, and nothing else is validated. Before ``init_report`` runs it is a safe no-op. It never raises — telemetry must never break the app. ## **Event Names** By convention event names are ``object:action`` — lowercase, colon-separated (``pipeline:run``, ``store:app_add``). The convention is documentation, not enforcement: ``report()`` accepts any string so an app can evolve its taxonomy without touching this module. Each app should keep its own documented event list next to its call sites. ## **What This Module Is Not** It is not a taxonomy and not a transport. There is no event-name ``Literal`` union, no typed property shapes, and no network I/O — the sink an app injects does the sending. An earlier revision centralised a strict cross-app event taxonomy here; that was removed in favour of per-app taxonomies and this loose core. --- # Chat Route: /clients/python/chat --- title: Chat sidebar_position: 7 --- # Chat Conversational pipelines: build a `Question`, send it with `client.chat()`, and parse the response with `Answer`. Class tables in the [API reference](/clients/python/reference#question). Chat is the conversational lane: it works against `chat`, `webhook`, and `dropper` pipeline sources. Under the hood the client opens a pipe with MIME type `application/rocketride-question`, writes the serialized `Question`, closes the pipe, and returns the server result. ## Build a Question ```python from rocketride.schema import Question 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.') ``` `Question(type=QuestionType.QUESTION, filter=DocFilter(), expectJson=False, role='')` — `QuestionType` is one of `QUESTION`, `SEMANTIC`, `KEYWORD`, `GET`, `PROMPT`. Steer the model with `addInstruction`, `addExample`, `addContext`, `addHistory` (for multi-turn), `addDocuments`, `addGoal`, and `addQuestion`. ## Send it ```python response = await client.chat(token=token, question=question) ``` `chat(*, token, question, on_sse=None)` is keyword-only; the optional `on_sse` callback streams server-sent events (token-by-token output) as they arrive. The final answer is in the result body. ## Parse the response with Answer `Answer` extracts structure from AI text, which often arrives wrapped in markdown or code fences. The client does **not** attach an `Answer` to the result — you read the body and feed it in: ```python from rocketride.schema import Answer answer_text = (response.get('answers') or [None])[0] answer = Answer(expectJson=True) answer.setAnswer(answer_text or '') if answer.isJson(): structured = answer.getJson() else: structured = answer.getText() ``` Semantics worth knowing: - `setAnswer(value)` stores the response, validating/parsing it as JSON when `expectJson` is `True`. - `isJson()` returns the `expectJson` flag — it does **not** inspect the content. - `getJson()` returns the parsed JSON; it returns `None` only when no answer has been set, and **raises `ValueError`** if the stored answer is not valid JSON. - `getText()` returns the answer as plain text; `parsePython(value)` extracts Python code from a code block. - `answer.tokens` carries the turn-total LLM token usage reported by the server. A complete chat program is [example 6](/clients/python/examples#6-chat-question-with-instructions-and-examples-parse-json-answer). --- # Configuration Route: /clients/python/configuration --- title: Configuration sidebar_position: 1 --- # Configuration Everything the `RocketRideClient` constructor accepts, the environment variables it reads, and how its timeouts, reconnection, and debug hooks behave. ## Constructor ```python from rocketride import RocketRideClient client = RocketRideClient( uri='https://api.rocketride.ai', auth='my-key', persist=True, ) ``` | Argument | Type | Default | Description | | --- | --- | --- | --- | | `uri` | `str` | RocketRide Cloud (`https://api.rocketride.ai`) | Server URI. Falls back to `ROCKETRIDE_URI` from the environment, then to the Cloud endpoint. See [URI scheme](/clients/python/connection#uri-scheme). | | `auth` | `str` | `None` | API key. Falls back to `ROCKETRIDE_APIKEY`. Omitting it leaves the client unauthenticated until [`connect(credential)`](/clients/python/connection) or `login()` supplies one. | | `env` | `dict` | — | Override the environment map used for `${ROCKETRIDE_*}` substitution and credential lookup. If omitted, the client copies `os.environ` and then merges `.env` underneath it — **process environment wins** over `.env` values. | | `module` | `str` | `CLIENT-0`, `CLIENT-1`, … | Client name for logging. | | `request_timeout` | `float` | — | Default timeout in **ms** for DAP requests. Prevents a single call from hanging. | | `max_retry_time` | `float` | — | Deprecated: accepted but **ignored** — reconnection never gives up. See [Reconnection](#reconnection). | | `persist` | `bool` | `False` | Automatic reconnection. Set `True` for long-lived scripts or UIs. See [Reconnection](#reconnection). | | `public` | `bool` | `False` | Reserved: declared but not currently consumed by the client. For unauthenticated public calls, [`attach()`](/clients/python/connection) without logging in. | | `ws_path` | `str` | `'/task/service'` | WebSocket path override. Model-server clients pass `'/models'`. | | `client_name` | `str` | `'Python SDK'` | Display name reported to the server. | | `client_version` | `str` | package version | Display version reported to the server. | There is no "missing uri/auth" error at construction time: an empty `uri` resolves to RocketRide Cloud and an empty `auth` stays `None`. The only constructor-time `ValueError` is a URI with no hostname. ## Callbacks All lifecycle callbacks are **awaited** — pass `async` functions (a plain `lambda` raises `TypeError` when the client awaits it). | Argument | Called | Description | | --- | --- | --- | | `on_event` | per server event | Receives each event `dict`. Subscribe per-task with [`add_monitor`](/clients/python/pipelines#events). | | `on_connected` | connection established | Receives connection info. | | `on_disconnected` | connection lost **after** being connected | Args: `reason`, `has_error`. Do not call `disconnect()` here if you want auto-reconnect. | | `on_connect_error` | each failed **reconnect** attempt (persist mode) | Args: `message: str`. An initial `connect()` failure raises to the caller instead. On auth failure the client stops retrying. | | `on_protocol_message` | per raw DAP message | Plain callable; for protocol debugging. | | `on_debug_message` | per debug line | Plain callable; for debug output. | | `on_trace` | around high-level SDK requests | Plain callable (NOT awaited): `(type, message)`. Unlike the TypeScript `onTrace`, the message is not credential-redacted. | ```python async def handle_event(event): print(event.get('event'), event.get('body')) async def handle_connect_error(message): print('Connect error:', message) client = RocketRideClient( uri='https://api.rocketride.ai', auth='my-key', persist=True, on_event=handle_event, on_connect_error=handle_connect_error, ) ``` ## Reconnection With `persist=True` the client reconnects with **linear backoff**: the delay grows by 0.25 s per consecutive failure, capped at 15 s, and reconnection **never gives up** — `on_connect_error` fires on each failed attempt so you can surface "still connecting…" in a UI. The one exception is an authentication failure: the client stops retrying so the app can fix credentials and call `connect()` again. `max_retry_time` is accepted for backward compatibility but **ignored**. ## Environment variables | Variable | Description | | --- | --- | | `ROCKETRIDE_URI` | Server URI (e.g. `wss://api.rocketride.ai` or `ws://localhost:5565`) | | `ROCKETRIDE_APIKEY` | API key for authentication | | `ROCKETRIDE_TOKEN` | User token, accepted by the [CLI](/connect/cli) as an alternative credential | The same map drives `${ROCKETRIDE_*}` substitution inside pipeline configs: the raw pipeline is sent to the server, which resolves variables from its merged environment. ## Timeouts `request_timeout` (constructor) sets the default for every DAP request; [`request(..., timeout=...)`](/clients/python/reference#advanced-low-level-dap) overrides it per call. [`connect(timeout=...)`](/clients/python/connection) bounds the connect + auth handshake in non-persist mode. All timeouts are in milliseconds. --- # Connection Route: /clients/python/connection --- title: Connection sidebar_position: 2 --- # Connection How the client attaches, authenticates, stays connected, and shuts down. The full method tables live in the [API reference](/clients/python/reference#connection). ## The layered model The connection has two layers you can drive together or separately: - **Attach** opens the WebSocket without authenticating — enough for public operations like catalog browsing. - **Login** performs the DAP auth handshake over an attached transport and returns a `ConnectResult` carrying your full identity (user, organizations, apps, teams). `connect()` does both in one call; `disconnect()` logs out and detaches. The state probes mirror the layers: `is_attached()` (socket open) and `is_authenticated()` (auth handshake succeeded). `is_connected()` is a backward-compatible alias with the same meaning as `is_attached()` — it does **not** imply authentication. ```python result = await client.connect() # attach + login print(result['displayName']) await client.logout() # drop auth, keep the socket await client.login('other-credential') # re-auth on the same connection await client.detach() # tear down the socket ``` Calling `login()` with a different credential logs out first (best-effort); with the same credential it is a no-op. `attach()` to a different URI detaches and re-attaches. ## Context manager (recommended) `async with` guarantees the connection closes even on exception — entering calls `connect()`, exiting calls `disconnect()`: ```python async with RocketRideClient(uri='wss://api.rocketride.ai', auth=os.environ['ROCKETRIDE_APIKEY']) as client: result = await client.use(filepath='pipeline.pipe') await client.send(result['token'], 'Hello, pipeline!') ``` > TypeScript's counterparts to the context manager are > [`withConnection()` and `await using`](/clients/typescript/connection#scoped-disconnect). ## 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://api.rocketride.ai` (or the equivalent `wss://api.rocketride.ai`); for a local engine use `ws://localhost:5565`. **Caution:** against a Cloud endpoint always use `https://` or `wss://` — an `http://` or `ws://` URI (or a bare `host:port`) silently downgrades to an unencrypted `ws://` connection. ## Staying connected With `persist=True` the client survives drops: it reconnects with linear backoff (+0.25 s per failure, 15 s cap) and never gives up, except on auth failure. Wire the [lifecycle callbacks](/clients/python/configuration#callbacks) to observe it: ```python async def on_connected(info): print('Connected:', info) async def on_disconnected(reason, has_error): # Fires only after a successful connection drops. # Do NOT call disconnect() here if you want auto-reconnect. print('Disconnected:', reason, has_error) async def on_connect_error(message): # Fires on each failed RECONNECT attempt; an initial connect() failure # raises to the caller instead. print('Connect error:', message) client = RocketRideClient( uri='https://api.rocketride.ai', auth='my-key', persist=True, on_connected=on_connected, on_disconnected=on_disconnected, on_connect_error=on_connect_error, ) await client.connect() ``` Use `on_disconnected` for "we were connected and then dropped"; use `on_connect_error` for "failed to connect". ## Inspecting state `get_connection_info()` returns `{'connected': bool, 'transport': str, 'uri': str}` — useful for a "Connected to …" display. `get_apikey()` returns the key in use (debugging only; avoid logging it). `set_env()` replaces the client's environment map used for `${ROCKETRIDE_*}` substitution and credential lookup. --- # Sending Data Route: /clients/python/data --- title: Sending Data sidebar_position: 5 --- # Sending Data Get data into a running pipeline: one-shot sends, file uploads with progress, and chunked streaming. Method tables in the [API reference](/clients/python/reference#data). `send()` / `send_files()` / `pipe()` target pipelines whose **source** is `webhook` or `dropper`. If your pipeline source is `chat`, use [`client.chat()`](/clients/python/chat) instead. ## One-shot: `send()` Use when you have the full payload in memory. It opens a pipe, writes once, closes, and returns the pipeline result: ```python result = await client.send(token, 'Hello, pipeline!', objinfo={'name': 'greeting.txt'}, mimetype='text/plain') ``` If `mimetype` is omitted the payload is sent as `application/octet-stream` — there is no auto-detection. An optional `on_sse` callback receives server-sent events for the transfer. ## Files: `send_files()` Uploads a list of files concurrently (all at once via `asyncio.gather`) and returns one `UPLOAD_RESULT` per file. Each entry is a path `str`, a `(path, objinfo)` tuple, or a `(path, objinfo, mimetype)` tuple: ```python 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')) ``` Two things to know: - `send_files` **requires an API key** on the client (it raises `RuntimeError` without one). - A missing file raises `ValueError` (`'File not found: …'`). Watch progress by subscribing to `apaevt_status_upload` events ([Events](/clients/python/pipelines#events)) — bodies carry `filepath`, `bytes_sent`, `file_size`. ## Streaming: `pipe()` Use `pipe()` when data arrives incrementally or is too large to hold in memory. One streaming upload is **open → write (one or more) → close**; `close()` returns the processing result. The pipe reads files best in ~1 MB chunks and enforces `bytes` payloads. ```python 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() ``` `DataPipe` is also an async context manager — entering calls `open()`, exiting calls `close()`: ```python async with await client.pipe(token, mime_type='application/json') as pipe: await pipe.write(b'{"key": "value1"}') await pipe.write(b'{"key": "value2"}') ``` Properties: `is_opened` and `pipe_id` (server-assigned after `open()`). `pipe()` and the pipe itself accept an `on_sse` callback for server-sent events, and `DataPipe.tool()` invokes a pipeline tool function through the pipe — see the [reference](/clients/python/reference#datapipe). ## Choosing | You have | Use | | --- | --- | | A string or bytes in memory | `send()` | | Files on disk, want per-file results + progress events | `send_files()` | | Chunked/incremental data, or very large payloads | `pipe()` | | A chat-source pipeline | [`chat()`](/clients/python/chat) | --- # Deployments Route: /clients/python/deploy --- title: Deployments sidebar_position: 4 --- # Deployments Persist pipelines server-side and run them on a schedule. Accessed via `client.deploy`; full method tables in the [API reference](/clients/python/reference#deploy-clientdeploy). ## Teams as environments `deploy.add` snapshots a pipeline as an **immutable, sha256-locked artifact version** in the org registry; `deploy.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 registry add and pointer change lands in an immutable audit history (`deploy.history`, rows carry `seq` as the stable append-order identity). ```python result = await client.deploy.add(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']) ``` `add(..., deploy_to=)` collapses add + deploy into one step. Listings (`deploy.list`, `deploy.versions`, `deploy.history`) return the standard `{rows, total, page, pageSize}` envelope, server-paged. `deploy.artifact(project_id, version)` fetches one immutable version's pipeline JSON, sha256-verified server-side. ## Schedules `deploy.set_schedule(project_id, source_id, schedule, team_id, ttl=None)` sets (or clears with `None`/`'manual'`) one source's 5-field cron schedule. `pause_schedule`/`resume_schedule` stop and restart a single source's firing without touching its cron. `deploy.preview(schedule, count=None)` is **the** single cron evaluator — validity plus next occurrences; never parse cron client-side. Scheduled runs execute **as the team** (no stored user credential); their logs land in the team's [run-log continuum](/clients/python/logs), readable by teammates via `client.log` with `team_id`. `deploy.run(project_id, source_id, team_id)` triggers one deployed source **now** — the same trusted, actor-free team dispatch the scheduler uses — returning `{token, version}`, and `deploy.set_source_config` sets per-source execution settings for deploy runs (trace level, debug output). ## States | State | Meaning | | --- | --- | | `enabled` | Schedules fire per cron. | | `disabled` | The kill switch (`deploy.disable`) — nothing runs until enabled again. | | `errored` | A scheduled dispatch failed — on permissions, or on an unusable artifact (missing or sha256-tampered) — and the scheduler stopped retrying. | | `removed` | Soft delete (`deploy.remove`): hidden from listings, history and artifacts survive; re-deploying revives it. | ## Permissions Mutations require `task.control` on the TARGET team. Reads follow the visibility model: an org admin sees every team and every personal space; a user sees their own personal space and the teams they are a member of. ## App publish ladder Typed wrappers over `rrext_deploy_app` — the publish ladder for RocketRide apps. **Deploy** copies code to the server as the next immutable registry version (`client.deploy.add`); a deployment carries the review lifecycle in its own `state` (`private` → `submit` → `ready` | `rejected`). **Publish** binds a deployment to an audience — `@me`, `@team/`, or `@public` — as a pure pointer (`@user` is a legacy input alias for `@me`, never displayed); repointing it covers first publish, update, promote, and rollback alike. The review state lives on the **deployment**, not the binding: an app deploys `private`, the developer `submit`s it, an admin approves (`ready`) or rejects (`rejected`). A `@public` binding may only point at a `ready` deployment; `@me`/`@team` accept any non-`failed` deployment. App ids are partitioned by the caller org's **developer id**: every app is `.` (globally unique), so an org can only deploy/publish ids inside its own namespace (the platform holds `rocketride`). Deploying or publishing an app requires the org to have claimed a developer id. `deploy.add` and `deploy.add_app` live on `client.deploy`; every other verb below is a method on the client itself (`client.list_deployments(...)`, `client.publish_app(...)`), not on `client.deploy`. | Method | Description | | --- | --- | | `deploy.add` | 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` | 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.verify_app` | 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` | 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_app` | Submit a deployed version for review — flips the deployment `private` → `submit`. | | `withdraw_app` | 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` | 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` | 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` | 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. | | `where_app` | 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//v/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). Full signatures: [API reference](/clients/python/reference#app-publish-ladder). See the [Shell API guide](/guides/apps) for the app model itself. --- # Error Handling Route: /clients/python/errors --- title: Error Handling sidebar_position: 9 --- # Error Handling What the SDK raises, when, and how to catch it. ## What actually raises | Situation | Raised | | --- | --- | | Bad API key / credentials during connect or login | `AuthenticationException` | | Connection and transport failures | builtin `ConnectionError` (timeouts: `asyncio.TimeoutError`) | | Data-pipe errors (open / write / close) | `PipeException` | | `use()` argument problems (neither `filepath` nor `pipeline`; both; bad types) | `ValueError` | | `use()` with a missing pipeline file | `FileNotFoundError` | | Server rejects a pipeline start or a DAP request | `RuntimeError` | | `get_service('')` / unknown service name | `ValueError` / `RuntimeError` | | `send_files` without an API key, or a missing file | `RuntimeError` / `ValueError` | | `Answer.getJson()` on non-JSON content | `ValueError` | ```python from rocketride import RocketRideClient, AuthenticationException from rocketride.core.exceptions import PipeException try: async with RocketRideClient(uri=uri, auth=auth) as client: result = await client.use(filepath='pipeline.pipe') await client.send(result['token'], data) except AuthenticationException: print('Bad credentials') except (ValueError, FileNotFoundError) as e: print(f'Bad pipeline arguments: {e}') except PipeException as e: print(f'Data transfer error: {e}') except ConnectionError as e: print(f'Transport failure: {e}') except RuntimeError as e: print(f'Server rejected the request: {e}') ``` `AuthenticationException` is raised on DAP auth failure. In [persist mode](/clients/python/configuration#reconnection) the client catches it, calls `on_connect_error`, and does **not** retry — fix credentials and call `connect()` again. ## The hierarchy ```text DAPException # Base DAP protocol error (has dap_result dict) └── RocketRideException # Base for all RocketRide errors ├── ConnectionException # Reserved: transport failures raise builtin ConnectionError │ └── AuthenticationException # Bad API key or credentials (actively raised) ├── PipeException # Data pipe errors (also subclasses RuntimeError) ├── ExecutionException # Reserved: defined but not currently raised └── ValidationException # Reserved: defined but not currently raised ``` All exceptions in the hierarchy expose a `dap_result` dict with detailed server error context, plus `code` and `hint`: - `code` is the server's machine-readable classification, or `None`. Task failures carry one: `TASK_NOT_REGISTERED` (the token names no live task — never started, terminated, replaced, or the engine restarted), `TASK_AMBIGUOUS`, `TASK_COMPLETED`, `TASK_STOPPED`. **Classify on `code`, not on the message text**, which is written for people and may be reworded. - `hint` is troubleshooting text the SDK attached for a developer, or `None`. It is kept out of `str(e)` so an application can show the message to an end user without the developer checklist. ```python except PipeException as e: if e.code == 'TASK_NOT_REGISTERED': await restart_pipeline() # the task is gone; start a new one else: print(e) # safe to show if e.hint: log.debug(e.hint) # developer detail ``` `PipeException` also subclasses `RuntimeError`, so a broad `except RuntimeError` catches pipe failures too. `ConnectionException`, `ExecutionException`, and `ValidationException` exist in `rocketride.core.exceptions` but the SDK does not currently raise them: transport failures surface as the builtin `ConnectionError`, and pipeline start and validation failures as the built-ins in the table above. Don't write handlers that rely on them. --- # Examples Route: /clients/python/examples --- title: Examples sidebar_position: 11 --- # Examples Complete, runnable programs covering the SDK surface end to end. For pipeline-level examples, see the site-wide examples — [RAG pipeline](/examples/rag-pipeline), [webhook pipeline](/examples/webhook-pipeline), and [document extraction](/examples/document-extraction). ## 1. Minimal: connect, run pipeline from file, send one string, disconnect ```python import asyncio from rocketride import RocketRideClient async def main(): client = RocketRideClient(uri='https://api.rocketride.ai', auth='my-key') await client.connect() 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) await client.disconnect() asyncio.run(main()) ``` ## 2. One-off script with context manager (recommended) ```python import asyncio from rocketride import RocketRideClient my_pipeline_config = {'components': []} # your pipeline dict, e.g. json.load(open('pipeline.pipe')) async def main(): async with RocketRideClient(uri='wss://api.rocketride.ai', auth='my-key') as client: result = await client.use(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 Lifecycle callbacks are awaited, so pass `async` functions: ```python import asyncio from rocketride import RocketRideClient async def on_connected(info): print('Connected:', info) async def on_disconnected(reason, has_error): # Do not call disconnect() here if you want auto-reconnect. print('Disconnected:', reason, has_error) async def on_connect_error(msg): print('Connect error:', msg) async def on_event(e): print(e.get('event'), e.get('body')) async def main(): client = RocketRideClient( uri='https://api.rocketride.ai', auth='my-key', persist=True, on_connected=on_connected, on_disconnected=on_disconnected, on_connect_error=on_connect_error, on_event=on_event, ) await client.connect() # Later: use(), send_files(), etc. If the connection drops, the client # retries forever (linear backoff) — except on auth failure. asyncio.run(main()) ``` ## 4. Upload multiple files and poll until pipeline completes ```python import asyncio from rocketride import RocketRideClient async def main(): client = RocketRideClient(uri='https://api.rocketride.ai', auth='my-key') await client.connect() result = await client.use(filepath='vectorize.pipe') 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 ```python import asyncio from rocketride import RocketRideClient async def main(): async with RocketRideClient(uri='https://api.rocketride.ai', auth='my-key') as client: result = await client.use(filepath='ingest.pipe') 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 ```python import asyncio from rocketride import RocketRideClient from rocketride.schema import Question, Answer async def main(): async with RocketRideClient(uri='https://api.rocketride.ai', auth='my-key') as client: result = await client.use(filepath='chat_pipeline.pipe') 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('answers') or [None])[0] answer = Answer(expectJson=True) answer.setAnswer(answer_text or '') if answer.isJson(): structured = answer.getJson() else: structured = answer.getText() print(structured) await client.terminate(token) asyncio.run(main()) ``` ## 7. Discover services and send a custom DAP request ```python import asyncio from rocketride import RocketRideClient async def main(): client = RocketRideClient(uri='https://api.rocketride.ai', auth='my-key') await client.connect() services = await client.get_services() print('Available:', list(services['services'].keys())) ocr = await client.get_service('ocr') # raises if the service is unknown print('OCR definition sections:', list(ocr.keys())) my_token = 'existing-task-token' # token from an earlier client.use() call 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()) ``` --- # Run Logs Route: /clients/python/logs --- title: Run Logs sidebar_position: 8 --- # Run Logs — `client.log` Every task writes a **run log**: one continuous JSONL event stream per task identity (`project_id` + `source`, plus the scope: a `team_id` addresses that team's DEPLOY continuum — [deploy runs](/clients/python/deploy) log into the team's tree, readable by any teammate with monitor rights — while omitting it addresses your own stream, where the optional `run_kind` picks between your dev stream (the default) and your personal deploy stream: `'deploy'` without a `team_id` is the only way to address that continuum). Individual runs are **chapters** (tracks) inside the stream — there are no per-run log files. The log survives disconnects and server restarts, powers replay of past runs through the same panels that render live monitoring, and is retained on a ring (last ~1 GB) plus a history age (7 days dev / 30 days deploy). Streams are addressed by the plain identity tuple — never by task token (tokens are credentials and appear nowhere in the log system). ## The DVR session — `open_event_stream()` Opens a **DVR session** over one source continuum — the recommended way to consume a run log. The session thinks in *positions* on the timeline; storage details (segments, keyframes, deltas) are invisible and every event it delivers is fully reconstructed. The protocol is **seed-then-stream**: `seek(pos)` positions the session, the `get_*()` calls seed your panels with state *as of* that position, and `play(pos, speed, cb)` streams events strictly *after* what the seeds covered — no gap, no duplicate. Speed `0` delivers as fast as possible, `1` is real time, `10` is 10×. Playing from a past position auto-pins to live on catching the wall clock; live is just the position pinned to now (`seek('live')`), not a separate mode. ```python # Own dev stream; pass team_id='team-prod' for a team's deploy continuum. session = client.log.open_event_stream('proj-1', 'chat_1') # Canonical startup: position, seed the panels, then roll. await session.seek('live') status = await session.get_status() # state as of the position console_lines = await session.get_console(500) # exactly what the console showed traces = await session.get_traces(50) # all in-flight + last 50 closed await session.play(None, 0, lambda item: fold(item['event'])) # Replay a past run at 10x from its beginning. chapters = await session.get_chapters() await session.play(chapters[0]['beginTime'], 10, lambda item: fold(item['event'])) # Drill into one trace (a call tree; fetched sparsely from exactly # the segments that contain it). detail = await session.get_trace(traces['closed'][0]['beginSeq']) session.pause() # freeze the position session.close_event_stream() # dispose ``` `get_traces(n)` errors when `n > 50` — the session exposes all in-flight traces plus a sliding window of the 50 most recently closed; any older trace is still reachable by seeking to a position inside its lifetime. `get_trace(trace_id)` resolves a trace by its begin event's continuum seq (pass the trace's own `beginSeq` from its `get_traces` summary, not the chapter's) — the permanent identity (slot ids recycle; `beginSeq` never does). Hosts that own a live subscription feed arriving events to the session via `ingest_live(event)`; while pinned, arrival paces delivery. ## `chapters()` Returns the stream's timeline in one small read: each run's begin/end date-time, starting sequence number and outcome, the activity spans for the timeline bar, the retained window, and the retention horizon. ```python timeline = await client.log.chapters('proj-1', 'chat_1') for track in timeline['chapters']: print(track['beginTime'], track.get('endTime'), track.get('outcome')) ``` ## `read()` Ranged, paged event read over the continuum. Range forms: a sequence range (`from_seq`/`to_seq`), a time range (`from_time`/`to_time`, omit `to_time` for "to now"), or time-to-segment (`from_time` + `to_segment`). Responses are paged (`max_events`/`max_bytes`, server-clamped): when `nextSeq` is present, pass it back as `cursor` to continue. `types` filters event types server-side; a `truncatedAtSeq` field means the request reached below the retention horizon. ```python cursor = None while True: page = await client.log.read('proj-1', 'chat_1', from_seq=0, cursor=cursor, types=['output']) for event in page['events']: print(event['body'].get('output', ''), end='') cursor = page.get('nextSeq') if cursor is None: break ``` Every event carries the continuum stamps in its **body** — the only place they exist: `body['eventTime']` (epoch seconds, stamped once at engine ingress) and `body['logSeq']` (catalog-seeded — a fresh stream starts at 1 and continues from the recorded `lastSeq + 1` across runs and restarts; strictly monotonic) — identical live and on replay. The DAP envelope's own `seq` is per-connection protocol bookkeeping and says nothing about the continuum. ## `segment()` Fetches one segment's raw JSONL bytes, chunked by byte offset — the bulk replay path. The server hands over the immutable segment content as-is, in **whole-line-aligned chunks** (every response ends on a newline, so each chunk parses standalone). Repeat with the returned `nextOffset` until `final`. The segment table comes from `chapters()`. ```python offset = 0 while True: chunk = await client.log.segment('proj-1', 'chat_1', 0, offset=offset) for line in chunk['data'].splitlines(): if line.strip(): handle_event(json.loads(line)) if chunk['final']: break offset = chunk['nextOffset'] ``` Prefer `segment()` over paged `read()` when consuming whole runs (replay, export); use `read()` for filtered or narrow ranged queries. ## `delete()` Destructive. `before_time` drops segments wholly older than the cutoff (chapters trimmed, horizon advanced); `all` removes the entire stream including its control file. ```python await client.log.delete('proj-1', 'chat_1', before_time=time.time() - 86400) await client.log.delete('proj-1', 'chat_1', all=True) ``` ## Wire surface and permissions All methods use the single `rrext_log` DAP command, dispatched by a `subcommand` argument (`chapters`, `read`, `segment`, `delete`). Reads require `task.monitor`; `delete` requires `task.control`. The scope the request addresses picks whose streams those rights are resolved against: without `team_id` you access your OWN dev streams; with `team_id` the permission is checked against the TARGET team — membership is the read/write right. `open_event_stream()` is client-side composition: it issues `chapters` and `segment` calls under the hood and registers a live `rrext_monitor` subscription while open — it adds no wire surface of its own. --- # OpenTelemetry Bridge Route: /clients/python/otel-bridge --- title: OpenTelemetry Bridge sidebar_position: 13 --- # OpenTelemetry Bridge (`rocketride otel`) The OpenTelemetry bridge exports **live pipeline traces and metrics** from a running RocketRide engine to any OpenTelemetry collector over OTLP — Jaeger, Grafana Tempo, Datadog, Langfuse, LangSmith, or anything else that ingests OTLP. It ships with the Python client as the `rocketride otel` CLI command and requires **zero engine or server changes**: the bridge is a pure consumer of the engine's documented [WebSocket monitor protocol](/connect/websocket/observability), subscribing to the `TASK`, `SUMMARY`, `FLOW`, and `SSE` event streams with the wildcard token scope (the documented scope for an ingestion service) and translating them into OTel spans and metrics on the fly. Point it at your engine and your collector, and every pipeline run visible to your API key shows up as a trace. ```text RocketRide engine ──(WebSocket monitor events)──▶ rocketride otel ──(OTLP)──▶ your backend ``` - [Install](#install) - [Quickstart: Jaeger end-to-end](#quickstart-jaeger-end-to-end) - [CLI reference](#cli-reference) - [Backend recipes](#backend-recipes) - [The span model](#the-span-model) - [Attributes](#attributes) - [Privacy: content is excluded by default](#privacy-content-is-excluded-by-default) - [Metrics](#metrics) - [Reconnection and shutdown](#reconnection-and-shutdown) - [Troubleshooting](#troubleshooting) - [Limitations](#limitations) --- ## Install The bridge's OpenTelemetry dependencies are an optional extra — the base `rocketride` package gains no new dependencies: ```bash pip install 'rocketride[otel]' ``` Running `rocketride otel` without the extra exits with code 2 and prints the install command above. ## Quickstart: Jaeger end-to-end Run Jaeger all-in-one (UI + OTLP receivers, in-memory storage): ```bash docker run --rm -d --name jaeger \ -p 16686:16686 -p 4317:4317 -p 4318:4318 \ jaegertracing/all-in-one:1.76.0 ``` Start the bridge — the defaults already point at `http://localhost:4318` (OTLP over HTTP/protobuf): ```bash export ROCKETRIDE_URI=ws://localhost:5565 # or wss://api.rocketride.ai + ROCKETRIDE_APIKEY rocketride otel ``` Now start a pipeline run **with a trace level** so it emits `FLOW` events (see [the callout below](#flow-spans-need-a-trace-level) — this is the step everyone misses): ```python import asyncio from rocketride import RocketRideClient async def main(): # Uses ROCKETRIDE_URI / ROCKETRIDE_APIKEY from the environment async with RocketRideClient() as client: result = await client.use(filepath='pipeline.pipe', pipelineTraceLevel='summary') token = result['token'] await client.send(token, 'hello traces', objinfo={'name': 'input.txt'}, mimetype='text/plain') await client.terminate(token) asyncio.run(main()) ``` Open [http://localhost:16686](http://localhost:16686), select the **rocketride-engine** service, and click **Find Traces**. You'll see one trace per run: a task root span, a pipe span per object, and component child spans underneath. ## CLI reference ```bash rocketride otel [--endpoint URL] [--protocol http|grpc] [--service-name NAME] [--headers k=v,k=v] [--include-content] [--no-metrics] [--trace-level none|metadata|summary|full] ``` | Flag | Default | Description | | ------------------- | --------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------- | | `--endpoint ` | `OTEL_EXPORTER_OTLP_ENDPOINT`, else the exporter default (`http://localhost:4318` for http, `localhost:4317` for grpc) | OTLP **base** URL. The signal paths `/v1/traces` / `/v1/metrics` are appended automatically unless already present, so pasting Langfuse's or LangSmith's ingest URL just works. Without the flag the endpoint environment variables are resolved by the OTel SDK exporters themselves (never pre-read by the bridge), so `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` / `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` keep their spec-defined precedence over the generic `OTEL_EXPORTER_OTLP_ENDPOINT` (which the SDK still treats as a base URL). | | `--protocol

` | `http` | OTLP transport: `http` (http/protobuf) or `grpc`. The gRPC exporter is **not** part of the `otel` extra — see [Troubleshooting](#troubleshooting). | | `--service-name `| `OTEL_SERVICE_NAME` or `rocketride-engine` | The `service.name` resource attribute your backend groups by. | | `--headers ` | `OTEL_EXPORTER_OTLP_HEADERS` | Comma-separated `key=value` pairs sent with every OTLP request. Values are split on the *first* `=` only, so base64 padding survives. **Prefer the env var for secrets** — command-line arguments are visible in shell history and `ps` output; keep `--headers` for non-secret headers. Without the flag, the OTel SDK resolves `OTEL_EXPORTER_OTLP_HEADERS` and the signal-specific `OTEL_EXPORTER_OTLP_TRACES_HEADERS` / `OTEL_EXPORTER_OTLP_METRICS_HEADERS` itself. | | `--include-content` | off | Include pipeline payload content in spans, truncated to 8 KB per attribute. **By default no payload text reaches any span.** | | `--no-metrics` | off | Export traces only; task status snapshots are not mapped to metrics. | | `--insecure` | off (`ROCKETRIDE_OTEL_ALLOW_INSECURE`) | Allow credential-bearing OTLP headers over cleartext transport to a **non-loopback** collector. Without it the bridge exits `2` rather than putting an `Authorization` / `x-api-key` value on the wire in the clear — see [Transport security](#transport-security). | | `--trace-level ` | — | **Informational only.** The bridge cannot change the trace level of runs it did not start; this flag just prints a reminder of how to start traced runs. | Plus the standard connection arguments shared by all subcommands: `--uri` (`ROCKETRIDE_URI`) and `--apikey` (`ROCKETRIDE_APIKEY`). No task token is needed — the bridge subscribes to every task your API key owns. **Configuration precedence:** explicit CLI flags > standard `OTEL_*` environment variables > built-in defaults. Only `OTEL_SERVICE_NAME` is read by the bridge; the endpoint and header variables are resolved by the OTel SDK exporters themselves (never pre-read by the bridge), so the signal-specific `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` / `_METRICS_ENDPOINT` / `OTEL_EXPORTER_OTLP_TRACES_HEADERS` / `_METRICS_HEADERS` keep their spec-defined precedence over the generic variables whenever `--endpoint` / `--headers` is not given. Pre-reading a generic variable and handing it to both exporters would turn it into an explicit value that silently overrides the signal-specific ones. ### Transport security OTLP exporters send whatever headers they are configured with to whatever endpoint they are given; they impose no TLS requirement of their own. So before any exporter is built, the bridge checks the *effective* endpoint of each signal (explicit `--endpoint` > signal-specific env var > generic env var > SDK default) against the *effective* headers (`--headers` plus all three `OTEL_EXPORTER_OTLP_*HEADERS` variables — names only; values are never read, logged or echoed): - A credential-looking header name (`Authorization`, `x-api-key`, `*-token`, `*-secret`, …) plus a **non-loopback** `http://` endpoint — or a gRPC endpoint with `OTEL_EXPORTER_OTLP_INSECURE` set — aborts startup with exit code `2`. - Loopback endpoints (`localhost`, `127.0.0.1`, `::1`) are exempt: that is the local collector / Jaeger case from the quickstart. - `--insecure` (or `ROCKETRIDE_OTEL_ALLOW_INSECURE=1`) overrides the check for a trusted network, and prints a warning to stderr. Two related hardening measures need no configuration: the OTLP/HTTP exporters use a session that **does not follow redirects** (`requests` strips `Authorization` only on a cross-*host* redirect, so a 3xx would otherwise replay `x-api-key` to the redirect target), and the startup line prints the endpoint **redacted** — userinfo and query string removed — because a signed collector URL is itself a credential. **Exit codes:** `0` graceful shutdown (Ctrl+C / SIGTERM), `1` unexpected runtime error, `2` missing dependency (the `otel` extra, or the gRPC exporter with `--protocol grpc`), a cleartext-credential refusal, or startup connection/subscribe failure. ### FLOW spans need a trace level `apaevt_flow` events — and therefore per-component spans — fire **only for runs started with a `pipelineTraceLevel`** (an argument of the `execute` request, i.e. `client.use(..., pipelineTraceLevel='summary')`). The bridge can only observe; it cannot turn tracing on for a run it did not start. Without a trace level you still get task lifecycle spans and all metrics, just no component breakdown. `summary` is the practical level: lane writes and final results without per-call noise. ## Backend recipes ### Jaeger / Grafana Tempo / any OTLP collector Covered by the [quickstart](#quickstart-jaeger-end-to-end): OTLP/HTTP on port 4318 (or `--protocol grpc` against 4317 with the gRPC exporter installed). Grafana Tempo listens on the same two ports once its `distributor.receivers.otlp` block is enabled, so `--endpoint http://tempo:4318` is the only change; pair it with Grafana Mimir or Prometheus for the metric stream. The same shape works for the OpenTelemetry Collector and any other standard OTLP receiver. ### Langfuse Langfuse ingests OTLP **traces over HTTP only** (no gRPC, no metrics) at `/api/public/otel`, with HTTP Basic auth built from your project keys: ```bash # Read the Langfuse project keys instead of typing them into the command: # a literal key in a command line lands in shell history and in `ps` output, # and `echo -n '' | base64` exposes it in the process list too. printf 'Langfuse public key (pk-lf-...): ' >&2; read -r LANGFUSE_PUBLIC_KEY printf 'Langfuse secret key (sk-lf-...): ' >&2; read -rs LANGFUSE_SECRET_KEY; echo >&2 # tr -d strips the line breaks GNU base64 inserts every 76 characters: real # project keys exceed that, and a wrapped value is an invalid header. export LANGFUSE_AUTH="$(printf '%s' "${LANGFUSE_PUBLIC_KEY}:${LANGFUSE_SECRET_KEY}" | base64 | tr -d '\r\n')" export OTEL_EXPORTER_OTLP_HEADERS="Authorization=Basic ${LANGFUSE_AUTH}" unset LANGFUSE_SECRET_KEY rocketride otel \ --endpoint https://cloud.langfuse.com/api/public/otel \ --no-metrics ``` (`printf` + `read -rs` is spelled the same in bash and zsh; bash users can shorten the second line to `read -rsp 'Langfuse secret key: ' LANGFUSE_SECRET_KEY`.) The auth header travels via `OTEL_EXPORTER_OTLP_HEADERS` rather than a `--headers` argument so the secret never lands in your shell history or shows up in `ps` output. Better still, source it from your secret manager — `export LANGFUSE_SECRET_KEY="$(op read ...)"` or the equivalent — and export only the variable reference. Use `https://us.cloud.langfuse.com/api/public/otel` for the US region, or `https:///api/public/otel` for self-hosted (Langfuse v3.22.0+). Pass `--no-metrics`: Langfuse's OTLP ingest is traces-only. Spans from LLM components carry `gen_ai.*` attributes, which Langfuse maps into its own data model. ### LangSmith LangSmith ingests OTLP traces over HTTP at `/otel`, authenticated with an `x-api-key` header (optional `Langsmith-Project` header to pick the project): ```bash # Prompt for the key (or read it from a secret manager) rather than typing it # into the command: the env var keeps it out of `ps`, but a literal value in # the export line is still recorded in shell history. printf 'LangSmith API key: ' >&2; read -rs LANGSMITH_API_KEY; echo >&2 export OTEL_EXPORTER_OTLP_HEADERS="x-api-key=${LANGSMITH_API_KEY},Langsmith-Project=your-project-name" rocketride otel \ --endpoint https://api.smith.langchain.com/otel \ --no-metrics ``` Regional hosts: `eu.api.smith.langchain.com` (EU); self-hosted follows `https:///api/v1/otel`. ### Datadog Send OTLP to a Datadog Agent (or an OpenTelemetry Collector with the Datadog exporter) that has OTLP ingestion enabled — the agent handles the Datadog API key, so the bridge needs no auth headers: ```bash rocketride otel --endpoint http://localhost:4318 ``` Datadog natively maps `gen_ai.*` semantic-convention attributes (semconv v1.37+) in its LLM Observability product, so LLM component spans light up without a Datadog SDK. ## The span model One trace per pipeline run. The hierarchy the bridge builds: ```text task task root span (INTERNAL), one per run │ opened on apaevt_task begin (or the seeded │ "running" snapshot when attaching mid-run) │ └── pipe root span, one per (run, pipe id) segment │ opened on flow op=begin, closed on op=end; │ named after the object flowing through │ ├── chat component span (llm_openai_1) │ │ SpanKind CLIENT, gen_ai.operation.name=chat, │ │ gen_ai.provider.name=openai │ └─ ● thinking apaevt_sse events attach as span events to the │ innermost open span of their pipe │ └── response_1 plain component span (INTERNAL) opened on op=enter, closed on op=leave — paired by component identity, one span per lane write (including control lanes) ``` Runs are correlated primarily on the wire correlation id (`__id`, `.`), falling back to `(project_id, source)`. A run first observed through events lacking `__id` is promoted to its canonical id when that id later arrives — never tracked twice. Details worth knowing: - **enter/leave pairing is by component identity**, never stack position (the monitor protocol documents why). Expect one short component span per lane write — including the `open`/`closing`/`close` control lanes. - **Component spans still open at run end or bridge shutdown** are closed with status `UNSET` and `rocketride.span.unclosed=true` — an honest "we never saw the leave", not an error. - **Attaching mid-run:** events for pipes the bridge never saw begin get an implicit root span marked `rocketride.span.implicit=true`. - **Errors:** a `trace.error` on a component `leave` event sets span status `ERROR`, records an `exception` span event, and sets `error.type` (`_OTHER` — the wire error is a free-form string). The error *text* is treated as payload: by default the status description is a generic `component error` and the exception event carries no message; with `--include-content` the wire error text is exported (8 KB cap). - **Task restarts** close the current spans and open a fresh task span with `rocketride.task.restarted=true`. - **GenAI conventions:** component ids map to GenAI semantic conventions where the id makes the role unambiguous — `llm_*` → `chat` (CLIENT), `embedding_*` → `embeddings` (CLIENT), `agent_*` → `invoke_agent` (INTERNAL), `tool_*` → `execute_tool` (INTERNAL, with `gen_ai.tool.name`). `gen_ai.provider.name` is set only for providers on the semconv well-known list (e.g. `llm_openai_*` → `openai`, `llm_anthropic_*` → `anthropic`); unknown providers omit the attribute rather than inventing a value. ## Attributes | Attribute | On | Meaning | | -------------------------------- | --------------------- | ------------------------------------------------------------------- | | `rocketride.project_id` | all spans and metrics | Pipeline project id | | `rocketride.source` | all spans and metrics | Pipeline source (e.g. `webhook_1`) | | `rocketride.run_id` | all spans | Wire correlation id of the run (`.`) | | `rocketride.task.name` | task spans | Task name from the lifecycle event | | `rocketride.task.restarted` | task spans | `true` when this span was opened by a task restart | | `rocketride.pipe_id` | pipe/component spans | Pipe index within the pipeline | | `rocketride.object` | pipe spans | Name of the object flowing through this segment | | `rocketride.component` | component spans | Component id (e.g. `llm_openai_1`) | | `rocketride.lane` | component spans | Lane being written (e.g. `text`, `open`, `close`) | | `rocketride.flow.result` | component spans | Flow result string on leave (e.g. `continue`) | | `rocketride.span.unclosed` | any span | `true`: closed at run end/shutdown without a matching leave | | `rocketride.span.implicit` | pipe spans | `true`: created for a pipe whose begin the bridge never saw | | `rocketride.flow.unmatched_leaves` | pipe spans | Count of leave events that matched no open component span | | `rocketride.sse.type` | span events | SSE message type (e.g. `thinking`, `tool_call`) | | `gen_ai.operation.name` | LLM/agent/tool spans | `chat`, `embeddings`, `invoke_agent`, or `execute_tool` | | `gen_ai.provider.name` | LLM/embedding spans | Well-known provider value derived from the component id | | `gen_ai.tool.name` | tool spans | Tool name derived from the component id | | `error.type` | failed spans | Always `_OTHER` (wire errors are free-form strings) | | `rocketride.trace.data` / `rocketride.result` / `rocketride.sse.data` | content-gated | Payload content — **only with `--include-content`**, 8 KB cap | `gen_ai.*` names follow the July 2026 snapshot of the `open-telemetry/semantic-conventions-genai` registry (Development stability, no tagged release); deprecated names such as `gen_ai.system` or `gen_ai.usage.prompt_tokens` are never emitted. ## Privacy: content is excluded by default By default **no pipeline payload content reaches any span** — not lane data (`trace.data`), not run-segment results, not SSE message bodies, and not the free-form error text of failed components (node errors routinely quote their input; the error *signal* — `ERROR` status, `error.type`, the `exception` span event — still exports, with the generic description `component error`). Only structural metadata (names, ids, lanes, counts, timings) is exported, so the bridge is safe to point at a shared collector out of the box. Opting in with `--include-content` copies payload content into the `rocketride.trace.data`, `rocketride.result`, and `rocketride.sse.data` attributes and exports the verbatim wire error text in span statuses and `exception` events — JSON-serialized and truncated to **8192 characters** per attribute. Treat the flag as what it is: pipeline inputs and outputs flowing into your telemetry backend. ## Metrics Unless `--no-metrics` is set, every `apaevt_status_update` snapshot (roughly every 500 ms per running task) is mapped to OTel metrics, exported over the same OTLP endpoint. All instruments carry `rocketride.project_id` / `rocketride.source` attributes. | Instrument | Type | Unit | Meaning | | -------------------------------- | --------------- | ------------ | ---------------------------------------- | | `rocketride.objects.total` | up-down counter | `{object}` | Objects seen by the pipeline run | | `rocketride.objects.completed` | up-down counter | `{object}` | Objects completed | | `rocketride.objects.failed` | up-down counter | `{object}` | Objects failed | | `rocketride.rate.count` | gauge | `{object}/s` | Instantaneous object processing rate | | `rocketride.rate.size` | gauge | `By/s` | Instantaneous byte processing rate | | `rocketride.cpu.percent` | gauge | `%` | Engine CPU utilization | | `rocketride.cpu.percent.peak` | gauge | `%` | Peak engine CPU utilization | | `rocketride.memory.cpu_mb` | gauge | `MBy` | Engine CPU memory usage | | `rocketride.memory.cpu_mb.peak` | gauge | `MBy` | Peak engine CPU memory usage | | `rocketride.memory.gpu_mb` | gauge | `MBy` | Engine GPU memory usage | | `rocketride.memory.gpu_mb.peak` | gauge | `MBy` | Peak engine GPU memory usage | Object counters are fed **per-run deltas** between snapshots, so re-sent snapshots don't double-count; a task restart resets the engine's counts, which legitimately produces negative deltas. The snapshot's `tokens.*` block is **compute credits (billing), not LLM tokens**, and is deliberately not exported as `gen_ai.usage.*`. ## Reconnection and shutdown - **Reconnects** use capped exponential backoff (1 s doubling up to 30 s). Monitor subscriptions are per-connection and not durable, but the SDK replays them on every reconnect, and the re-seeded "running"/status snapshots are handled idempotently — already-open spans are not duplicated. The seeded snapshot is also authoritative: tracked runs it no longer announces (their `end` was missed while disconnected) are closed with `rocketride.span.unclosed=true` and dropped. - **Ctrl+C / SIGTERM** closes all open spans (marked `rocketride.span.unclosed=true`), flushes both exporters, and exits `0`. - **Startup failure** (engine unreachable, subscribe rejected) prints a clean message to stderr and exits `2` so supervisors can tell "never started" from "was stopped". ### Embedding the bridge `rocketride.otelbridge.run_bridge()` runs the same loop inside an application that owns its own event loop (pass `install_signal_handlers=False` when the application owns process signals). Two ownership rules matter there: - **The bridge builds only the halves you did not supply.** Pass `mapper_factory` and it builds no `TracerProvider`; pass `metrics_factory` and it builds no `MeterProvider`. Each provider carries a background export thread, so a half that is built and never read is a leaked thread, not just wasted setup. - **Whatever the bridge builds, the bridge shuts down.** A `shutdown_fn` you supply is *chained in front of* the providers' own shutdown, never substituted for it: yours runs first, theirs runs in a `finally` so it still happens if yours raises (your exception is logged to stderr and the bridge still exits `0`). Supplying `shutdown_fn` therefore never orphans a provider the bridge created. ## Troubleshooting | Symptom | Cause / fix | | -------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | | Exits immediately with code 2 and an install hint | The `otel` extra is missing: `pip install 'rocketride[otel]'`. | | Bridge runs, tasks appear, but **no component spans** | The run was not started with a trace level. Start runs with `client.use(..., pipelineTraceLevel='summary')` — the bridge cannot enable it for you. | | Exits with code 2, "connection" in the message | Engine unreachable at startup: check `--uri` / `ROCKETRIDE_URI` and `--apikey` / `ROCKETRIDE_APIKEY`. | | Bridge runs but nothing arrives in the backend | Exporter can't reach the collector (exports fail in the background; the bridge keeps running). Check host/port — OTLP/HTTP is **4318**, gRPC is **4317** — and that `--protocol` matches the receiver. | | `--protocol grpc` fails with an install error | The gRPC exporter is not part of the extra: `pip install opentelemetry-exporter-otlp-proto-grpc`. Note Langfuse does not accept gRPC at all. | | Langfuse/LangSmith receive traces but metric exports error | Their OTLP ingest is traces-only — run with `--no-metrics`. | | Spans named `chat` carry no model name or token counts | Expected — see [Limitations](#limitations). | ## Limitations Honest edges of a protocol-level bridge: - **Span timestamps come from the engine; metric points do not.** Every forwarded event body carries `eventTime` (epoch seconds, stamped at ingress by the engine's run-log continuum), and spans and span events are stamped with it — so span durations are engine-measured and exclude WebSocket latency. Two edges remain: a span closed by a non-event path (bridge shutdown, snapshot reconciliation, or the tracked-run cap) and any event from an engine too old to stamp `eventTime` fall back to the bridge's own clock at that moment; and OpenTelemetry metric instruments take no explicit timestamp, so metric points are always recorded at arrival. A close is never stamped before its own span's start, so mixing the two sources cannot produce a negative duration. - **LLM token usage and model names appear only when nodes surface them in events.** Flow events at `summary` level carry neither, so `gen_ai.request.model` and `gen_ai.usage.*` are honestly omitted rather than guessed, and LLM span names degrade to the bare operation (`chat`, not `chat gpt-4.1`). The status snapshot's `tokens.*` are compute credits, not LLM tokens, and are never mapped to `gen_ai.usage.*`. - **`gen_ai.*` conventions are Development stability.** Attribute names follow the July 2026 snapshot of `open-telemetry/semantic-conventions-genai`; they are centralized in one constants module and may be revised as the spec evolves. - **Trace level is the run starter's choice.** `--trace-level` on the bridge is informational only; there is no protocol surface to change it for running tasks. - **Restart accounting.** Object up-down counters are per-run deltas, so a task restart (counts reset) produces a negative step by design. - **Spans export when they close.** The batch span processor exports a span only at `end()`, so a run whose `end` event is never observed holds its spans back until the bridge closes them — at the next reconnect's seeded snapshot (runs no longer announced are closed), when the tracked-run cap (1024) evicts the least-recently-eventful run, or at shutdown. Such spans are flagged `rocketride.span.unclosed=true`. There is deliberately no idle timeout: a quiet but alive task (e.g. a webhook service) keeps being announced and is never expired by a clock. Metric delta bookkeeping is likewise capped (4096 runs, least recently updated evicted first), so a bridge left running for weeks has bounded memory. ## See also - [Monitor protocol reference](/connect/websocket/observability) — the event stream the bridge consumes - [CLI reference](/connect/cli) — the `otel` flag table alongside the rest of the CLI - [Python SDK](/clients/python) — the client the bridge ships with - [Client libraries overview](/clients) — every official client --- # Running Pipelines Route: /clients/python/pipelines --- title: Running Pipelines sidebar_position: 3 --- # Running Pipelines Start a pipeline, watch its progress, and stop it. Method tables live in the [API reference](/clients/python/reference#pipeline-execution); this page covers the workflow. ## Start with `use()` `use()` starts a pipeline from a file or an in-memory config and returns a dict whose `'token'` identifies the running task — every data and control call takes it. ```python result = await client.use(filepath='pipeline.pipe') token = result['token'] ``` Beyond `filepath`/`pipeline`, `use()` accepts `source`, `threads`, `use_existing`, `args`, `ttl`, `pipelineTraceLevel` (trace verbosity for the [run log](/clients/python/logs)), `name` (a display name for the task), and `env` (per-run variable overrides). Pass the pipeline config **as-is** — the client sends it to the server, which resolves `${ROCKETRIDE_*}` variables from its merged environment. **Check `reused` before trusting the result.** `use_existing` returns the instance that is already running under that token rather than starting the one you submitted, and the result's `reused` flag is `True` when that happened. A reused instance keeps the configuration it was created with — the pipeline in this call is ignored, edits included — along with whatever state it has accumulated. Benchmarks and A/B comparisons are where an unnoticed reuse costs the most. Call `restart()` to apply new configuration to a live token. **Why a token:** the server runs each pipeline as a separate task. The token targets `send()`, `send_files()`, `pipe()`, `chat()`, `get_task_status()`, and `terminate()` at the correct pipeline. ## Watch progress Poll `get_task_status(token)` — it returns `completedCount`, `totalCount`, `completed`, `state`, `exitCode`, and more: ```python 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) ``` ### Events For push-style progress instead of polling, add a monitor subscription; events arrive at your [`on_event` callback](/clients/python/configuration#callbacks): ```python await client.add_monitor({'token': token}, ['apaevt_status_upload', 'apaevt_status_processing']) # ... later: await client.remove_monitor({'token': token}, ['apaevt_status_upload', 'apaevt_status_processing']) ``` `add_monitor(key, types)` / `remove_monitor(key, types)` are reference-counted — adding the same key merges types, removing unsubscribes a type only when its count reaches zero. The key is `{'token': ...}` for a running task, or `{'project_id': ..., 'source': ...}` (optionally with `'pipe_id'` and/or `'team_id'` — a team ID addresses that team's deployed run). The older `set_events(token, event_types, pipe_id=None)` still works but is deprecated in favor of the monitor pair. ## Validate before you run `validate(pipeline, source=None)` checks a pipeline config server-side without starting it and returns errors and warnings — cheap insurance before `use()`. ## Stop with `terminate()` `terminate(token)` stops the pipeline and frees server resources. Long-lived tasks without a `ttl` run until terminated. ## Discover services `get_services()` returns lightweight **summaries** of every service the server supports (plus a deduplicated icon table and the server version). For a full definition — config schema included — fetch one by name with `get_service(name)`. Note `get_service` **raises** on failure (`ValueError` for an empty name, `RuntimeError` for an unknown service); it never returns `None`. ```python services = await client.get_services() ocr = await client.get_service('ocr') # raises if unknown ``` ## Liveness `ping()` performs a liveness check against the server and raises on failure. > Deploying a pipeline so it persists server-side and runs on a schedule is a > separate surface — see [Deployments](/clients/python/deploy). --- # API Reference Route: /clients/python/reference --- title: API Reference sidebar_position: 10 --- # API Reference The core public surface of the Python SDK. Constructor options and environment variables are on [Configuration](/clients/python/configuration); exceptions on [Error Handling](/clients/python/errors). ## 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](/clients/python/chat). | ### Store (file access) Paths are **relative** to the store root; absolute-like paths are rejected. See [File Storage](/clients/python/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 Sequelize ORM binding over this surface. | Method | Signature | Description | | --- | --- | --- | | `database.query` | `async def query(*, token, sql, node_id='', session_id='', params=None) -> dict` | Execute raw SQL through the pipeline's `execute` tool function; 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](/clients/python/deploy) 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.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](/clients/python/deploy#app-publish-ladder) 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(...)`). | 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.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 carries its deployment `state`, its `buildStatus` ('ok' = servable), and the `rungs` naming the audiences bound to it. | | `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. | | `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//v/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("", {"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](/clients/python/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)`. ```python Question( type: QuestionType = QuestionType.QUESTION, filter: DocFilter = None, expectJson: 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). | ## Answer From `rocketride.schema`. Parses chat response content — see [Chat](/clients/python/chat#parse-the-response-with-answer) 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` (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` | | 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`). | ```python # 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')) ``` --- # File Storage Route: /clients/python/storage --- title: File Storage sidebar_position: 6 --- # File Storage 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"`); `..` traversal is rejected client-side, and `fs_rmdir`, `fs_rename`, `fs_get_url`, and `fs_read_many` additionally reject absolute-like paths (leading `/` or `\`). Method tables in the [API reference](/clients/python/reference#store-file-access). ## Strings and JSON (start here) The convenience wrappers manage the handle lifecycle for you: ```python 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 ```python listing = await client.fs_list_dir('reports') # {entries: [{name, type, size?, modified?}], count} for entry in listing['entries']: print(entry['name'], entry['type']) meta = await client.fs_stat('reports/q3.pdf') # {exists, type, size, modified} await client.fs_mkdir('reports/2026') await client.fs_rename('reports/q3.pdf', 'archive/q3.pdf') await client.fs_delete('archive/q3.pdf') await client.fs_rmdir('reports/2026', recursive=True) ``` `fs_rename` moves files or directories (copy+delete on object stores, recursive for directories). `fs_rmdir` raises `ValueError` on empty or absolute-like paths. ## Binary I/O (handles) For large or binary files, use the explicit handle lifecycle — `fs_open` → `fs_read`/`fs_write` → `fs_close`, in up-to-4 MB chunks. `fs_close` must receive the same mode as `fs_open`. ```python 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') ``` Read mode's `fs_open` result also includes `'size'`; an empty `bytes` from `fs_read` means EOF. ## Batch reads `fs_read_many(paths)` fetches many small files in **one** round trip (max 256 paths / 32 MiB total per call). Missing or unreadable files come back as per-entry results (`ok: False` + `error`), never a call failure; results arrive in request order with `data` as `bytes`. ## Direct URLs `fs_get_url(path, expires_in=3600, download_name=None)` returns a 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 — right for streaming and ``/`