Run pipelines from your own code

RocketRide is an open-source runtime for AI pipelines. The Python and TypeScript clients let you start a pipeline, push data through it, and follow the run as it happens, from an application you have already built.

Python 3.10+
$pip install rocketride

01 · Your first pipeline

A pipeline is a JSON file.

Three components here: something that takes input, a model, something that returns the result. Each one names the lane it reads and the component it reads from, which is the whole wiring model.

chat.pipe
{
  "project_id": "00000000-0000-4000-8000-000000000001",
  "version": 1,
  "source": "webhook_1",
  "components": [
    {
      "id": "webhook_1",
      "provider": "webhook",
      "config": { "hideForm": true, "mode": "Source", "parameters": {}, "type": "webhook" }
    },
    {
      "id": "llm_openai_1",
      "provider": "llm_openai",
      "config": {
        "profile": "openai-4o-mini",
        "openai-4o-mini": { "apikey": "${ROCKETRIDE_OPENAI_KEY}" }
      },
      "input": [{ "lane": "questions", "from": "webhook_1" }]
    },
    {
      "id": "response_answers_1",
      "provider": "response_answers",
      "config": { "laneName": "answers" },
      "input": [{ "lane": "answers", "from": "llm_openai_1" }]
    }
  ]
}

Save it, then run it from your app. Both clients open one WebSocket and speak the full pipeline protocol over it.

import asyncio
import os
from rocketride import RocketRideClient

async def main():
    async with RocketRideClient(
        uri="ws://localhost:5565",
        auth=os.environ["ROCKETRIDE_APIKEY"],
    ) as client:
        result = await client.use(filepath="chat.pipe")
        token = result["token"]
        try:
            out = await client.send(
                token,
                "What is the capital of France?",
                objinfo={"name": "input.txt"},
                mimetype="text/plain",
            )
            print(out)
        finally:
            await client.terminate(token)

asyncio.run(main())

Point the client at ws://localhost:5565 for an engine on your machine, or https://api.rocketride.ai for Cloud. The pipeline JSON does not change between them. Cloud adds an API key and an encrypted scheme: a plain ws:// or http:// URI stays unencrypted.

02 · What the client does

One connection, the whole protocol.

Both clients speak the full pipeline protocol over a single WebSocket, so all of this is the same client and the same session rather than separate integrations.

01

Run

Start from a .pipe file or an inline config and get a task token back. Push data with send(), stream incrementally with pipe(), drive a conversation with chat(), stop with terminate(). Restart a run without rebuilding it.

02

Move data

Upload many files in one call and follow byte-level progress as each one lands. Read, write, list, stat, and rename files in the pipeline file store directly from your code.

03

Observe

Subscribe to a run and receive events as the engine produces them. Poll status for state, counts, rates, and exit code. Persistent mode reconnects with backoff after a dropped socket, so a long job survives a flaky network.

03 · While it runs

Watch a run while it runs.

Most integrations give you a result and a stack trace. A pipeline that fails in component four after eleven minutes needs more than that, and building the more is what eats the sprint.

Scope a subscription to one task, a single pipe, your own dev run, or everything your token owns. Subscribing seeds the current state immediately, so there is no gap between connecting and knowing. Start a run with a trace level and every component reports its entry and exit with the lane data that passed through it, which is enough to reconstruct why a pipeline produced what it did rather than only that it stopped.

result = await client.use(
    filepath="chat.pipe",
    pipelineTraceLevel="summary",
)
token = result["token"]

# A monitor key scopes the subscription: one task here, or a
# project + source + pipe to follow a whole deployment.
await client.add_monitor({"token": token}, ["TASK", "SUMMARY", "FLOW"])

status = await client.get_task_status(token)
print(status["state"], status["completedCount"], status["totalCount"])

04 · From laptop to production

The pipeline does not change. Only where it runs.

Locally while you build, on your own hardware with Docker, or on RocketRide Cloud. Same JSON, same client, same calls.

You do not have to move all of it at once. On a self-hosted engine a heavy component can run on another machine while the rest of the pipeline stays on yours, so offloading the CPU and GPU work is a smaller first step than relocating anything. Components that hold local state, such as the SQL, graph, and filesystem nodes, stay put.

On Cloud, a deployed pipeline belongs to a team rather than to whoever last edited the file. Changing what is deployed is a permission, so the behavior behind a URI your application calls does not move without someone deciding it should.

05 · Command line

A CLI comes with it.

Installing either package puts a rocketride command on your path. Start a pipeline, push files at it, watch the run, and browse the file store without writing any code.

Pass a directory rather than a shell glob. Both CLIs expand paths themselves, so ./docs recurses the same way in bash, PowerShell, and cmd.

export ROCKETRIDE_URI=https://api.rocketride.ai
export ROCKETRIDE_APIKEY=your-api-key

rocketride start chat.pipe
rocketride upload ./docs --pipeline_path chat.pipe
rocketride status --token <token>
rocketride store dir /
rocketride stop --token <token>

MIT, and it runs on your hardware.

Pipelines are portable JSON you control. The client you integrate today works against a runtime you host yourself, indefinitely, with no cloud account. Cloud is where you go when a pipeline stops being yours alone, not the condition of using it.

Every pipeline you start is registered automatically as an MCP tool, with no per-pipeline configuration, so an assistant like Claude or Cursor can call it once the MCP server is pointed at your engine.