Quickstart

This guide connects to Laser, creates a topic, publishes one message, and reads it back. Choose a language tab. That choice persists across the SDK docs.

What this is built on

AGDX defines the data model and behavior for a durable partitioned log. Laser SDK implements AGDX on Apache Iggy. All clients use VSR (Viewstamped Replication Revisited).

The example publishes a regular JSON record. .json(...) adds the AGDX content type used during decoding. Agent code uses .agdx(...) for validated commands, responses, chunks, status, and errors.

1. Start a target

Choose one target:

• Laser Stack - the recommended local path. It starts the LaserData Apache Iggy fork and laser-plane, so every SDK primitive and managed example is available on your laptop.
• LaserData Cloud - use the connection string or token from the Console's Credentials tab.
• Standalone Apache Iggy with VSR - suitable for streaming and the log-backed agent fabric. Managed calls report Unsupported unless a compatible managed backend is attached.

Start Laser Stack:

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

The script waits for Iggy and laser-plane. It then prints the LASER_CONNECTION_STRING export used below. Run ./scripts/smoke to check both services. Laser Stack covers requirements, persistence, and troubleshooting.

This quickstart works against all three targets. Only LASER_CONNECTION_STRING changes. VSR has no flag. It is always enabled.

Prefer the CLI or the Console over writing code for this first step? See the platform Quick Start instead.

2. Install the SDK

npm install @laserdata/laser-sdk

cargo add laser-sdk

pip install laser-sdk

3. Publish and read one message

A topic is an append-only record split into partitions, ordered within a partition, not across the whole topic. See Log for what that means for routing and parallelism.

The example publishes one order and reads it through a named cursor. The cursor decodes records and tracks offsets while it is alive.

Persist offsets() and restore them after a restart. Use a live consumer for production workloads. Log covers consumers and commit policies.

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(..) is convenience over the raw .payload(bytes) primitive. .msgpack and .avro encode the same way if your wire format isn't JSON.

The streaming path runs on Laser Stack, LaserData Cloud, or standalone VSR-enabled Apache Iggy.

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

See Log for partitions, delivery guarantees, consumer groups, and commit policies in full, or jump straight to any of the other seven primitives.

Next