Quickstart

Connect to a server, create a topic, publish an order, and read it back. Select Rust, Python, or TypeScript in the language tabs. Your selection carries across the SDK documentation.

If you already created a Cloud deployment, use its Credentials page and continue with Install the SDK. Docker, a local Iggy server, and Laser Stack are not required for this path. Make sure that your access rule permits your client IP before connecting.

What this is built on

AGDX defines the data model for a durable partitioned log, a stored message sequence divided into ordered parts. Laser SDK implements AGDX on Apache Iggy.

This example sends a JSON record. .json(...) adds the AGDX content type used for decoding. Agent applications use .agdx(...) for typed commands, responses, chunks, status, and errors.

1. Start a target

Choose a target for the example:

• Use Laser Stack for local development. It starts the LaserData Iggy fork and laser-plane, with all SDK primitives and managed examples available.
• Use LaserData Cloud with a connection string or token from the Console's Credentials tab.
• Use standalone Apache Iggy for streaming and agent Fabric backed by the log. Managed calls return Unsupported without a compatible managed backend.

To start Laser Stack:

git clone https://github.com/laserdata/laser-stack
cd laser-stack
./scripts/up

The script waits for Iggy and laser-plane, then prints the LASER_CONNECTION_STRING export. Run ./scripts/smoke to test Iggy health, plane readiness, AGDX capabilities, and a managed key-value round trip with the published SDK. See Laser Stack for requirements, persistence, and troubleshooting.

All three targets support this quickstart. Change LASER_CONNECTION_STRING for the selected target.

For a first example through the CLI or Console, use the platform Quick Start.

2. Install the SDK

npm install @laserdata/laser-sdk

cargo add laser-sdk

pip install laser-sdk

uv venv
uv pip install laser-sdk

3. Publish and read one message

A topic stores messages in partitions. Messages are ordered within each partition, without a total order across the topic. Log explains routing and parallel consumption.

The example reads an order through a named cursor, a reader that tracks its position. The cursor decodes records and keeps offsets while it exists. Save offsets() and restore them after a restart. Use a live consumer for production processing.

import { Laser, jsonCodec } from "@laserdata/laser-sdk"

interface Order {
  readonly id: number
  readonly total: number
}

const ORDER_CODEC = jsonCodec<Order>((value) => {
  if (typeof value !== "object" || value === null) {
    throw new TypeError("order must be an object")
  }
  const { id, total } = value as Record<string, unknown>
  if (typeof id !== "number" || typeof total !== "number") {
    throw new TypeError("order fields are invalid")
  }
  return { id, total }
})

await using laser = await Laser.connectEnv()
const topic = laser.stream("shop").topic("orders")
await topic.ensure(2)

await topic.publish().json({ id: 1, total: 99 }).send()

const records = await topic
  .json(ORDER_CODEC)
  .records("log-reader")

for (const result of await records.poll()) {
  if (result.kind === "record") {
    console.log(result.record.value.total)
  }
}

use laser_sdk::prelude::full::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
struct Order {
    id: u32,
    total: u32,
}

#[tokio::main]
async fn main() -> Result<(), LaserError> {
    let laser = Laser::connect_env().await?;

    let topic = laser.stream("shop").topic("orders");
    topic.ensure(2).await?;

    topic
        .publish()
        .json(&Order { id: 1, total: 99 })?
        .send()
        .await?;

    let mut reader = topic
        .json::<Order>()
        .records("log-reader")?;
    while let Some(next) = reader.next().await {
        println!("{}", next?.value.total);
    }
    Ok(())
}

import asyncio
import os
from dataclasses import dataclass

import laser_sdk as ls


@dataclass
class Order:
    id: int
    total: int


async def main():
    async with await ls.Laser.connect(
        os.environ["LASER_CONNECTION_STRING"]
    ) as laser:
        topic = laser.stream("shop").topic("orders", cls=Order)
        await topic.ensure(partitions=2)

        await topic.publish(Order(id=1, total=99)).send()

        records = topic.records("log-reader")
        while (record := await records.next()) is not None:
            print(record.value.total)

asyncio.run(main())

.json(..) encodes data for the raw .payload(bytes) operation. .msgpack and .avro provide other encodings.

Add the managed layer for projections, queries, key-value storage, forks, and graph. Laser Stack provides it locally through laser-plane. Add the agent layer for reliable handlers, memory, contracts, and workflows.

Read Log for delivery guarantees, groups, and offset commit policies, or continue with the other seven primitives.

Next