LaserData Cloud
Laser SDK

Messages

Publish typed messages to a topic, read them back, and process them live in a consumer group

A topic is a durable, ordered log of messages. Producers append to it, and any number of readers can read it live, resume from a saved position, or replay it from the start. Use it for agent events, tool-call records, telemetry, and any feed that more than one service needs.

Quick example

An agent records each tool call it makes. An auditor reads every record back as a typed value.

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

interface ToolCall {
  readonly agent: string
  readonly tool: string
  readonly ms: number
}

// Types do not exist at runtime, so the codec checks each decoded value.
const TOOL_CALL = new Json<ToolCall>((value) => {
  const { agent, tool, ms } = value as Record<string, unknown>
  if (typeof agent !== "string" || typeof tool !== "string" || typeof ms !== "number") {
    throw new TypeError("not a tool call")
  }
  return { agent, tool, ms }
})

await using laser = await Laser.connectEnv()
const topic = laser.stream("agents").topic("tool-calls")
await topic.ensure(2)

const calls = topic.json(TOOL_CALL)
await calls.publish({ agent: "researcher", tool: "search", ms: 120 })
await calls.publish({ agent: "researcher", tool: "summarize", ms: 840 })

const reader = await calls.records("audit")
for (let result = await reader.next(); result !== undefined; result = await reader.next()) {
  if (result.kind === "error") throw result.error
  const call = result.record.value
  console.log(call.agent, call.tool, call.ms)
}
use laser_sdk::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
struct ToolCall {
    agent: String,
    tool: String,
    ms: u32,
}

#[tokio::main]
async fn main() -> Result<(), LaserError> {
    let laser = Laser::connect_env().await?;
    let topic = laser.stream("agents").topic("tool-calls");
    topic.ensure(2).await?;

    let calls = topic.json::<ToolCall>();
    for (tool, ms) in [("search", 120), ("summarize", 840)] {
        let call = ToolCall { agent: "researcher".into(), tool: tool.into(), ms };
        calls.publish(&call)?.send().await?;
    }

    let mut reader = calls.records("audit")?;
    while let Some(next) = reader.next().await {
        let call = next?.value;
        println!("{} {} {}", call.agent, call.tool, call.ms);
    }
    Ok(())
}
import asyncio
from dataclasses import dataclass

import laser_sdk as ls


@dataclass
class ToolCall:
    agent: str
    tool: str
    ms: int


async def main() -> None:
    laser = await ls.Laser.connect_env()
    topic = laser.stream("agents").topic("tool-calls")
    await topic.ensure(2)

    calls = topic.json(ToolCall)
    for tool, ms in (("search", 120), ("summarize", 840)):
        await calls.publish(ToolCall("researcher", tool, ms)).send()

    reader = calls.records("audit")
    while (record := await reader.next()) is not None:
        print(record.value.agent, record.value.tool, record.value.ms)
    await laser.close()


asyncio.run(main())

topic.json(..) returns a typed handle. Its publish encodes the value and its records reader decodes it. The reader starts at offset 0. When it reaches the end of the topic, next() returns None in Rust and Python and undefined in TypeScript.

Process messages live in a consumer group

A consumer group shares a topic's partitions across processes. Each partition goes to one member at a time. The server stores the group's progress, so a restarted member resumes where the group left off. Commit a message after you finish handling it.

await using consumer = await topic.consumerGroup("auditors").consumer({
  commitPolicy: { kind: "disabled" },
  startAt: { kind: "first" }
})

for await (const message of consumer) {
  const call = TOOL_CALL.decode(message.payload)
  console.log(message.position.offset, call.tool)
  await consumer.commit(message)
}
let mut consumer = topic
    .consumer_group("auditors")
    .consumer()
    .commit_policy(CommitPolicy::Disabled)
    .start_at(ConsumerStart::First)
    .build()
    .await?;

while let Some(message) = consumer.next().await {
    let message = message?;
    let call: ToolCall = message.json()?;
    println!("{} {}", message.position.offset, call.tool);
    consumer.commit(&message).await?;
}
consumer.shutdown().await?;
consumer = topic.consumer_group("auditors").consumer(
    auto_commit="disabled",
    polling="first",
)

async for message in consumer:
    call = message.json()
    print(message.position.offset, call["tool"])
    await consumer.commit(message)

await consumer.shutdown()

The loop waits for new messages until you stop it. To commit automatically, pick another commit policy. See Consumers and commit policies.

Order holds only within one partition. Give messages that must stay in order the same key, such as an agent or session ID. The key picks the partition.

const key = new TextEncoder().encode("researcher")
await calls.publish({ agent: "researcher", tool: "search", ms: 120 }, { key })
let call = ToolCall { agent: "researcher".into(), tool: "search".into(), ms: 120 };
calls.publish(&call)?.partition_key("researcher").send().await?;
call = ToolCall("researcher", "search", 120)
await calls.publish(call).partition_key("researcher").send()

Resume a reader from saved offsets

A typed reader tracks the next offset of each partition. Save the offsets, then hand them back to resume after a restart. Without saved offsets, a reader starts at offset 0.

const saved = reader.offsets // Map of partition to bigint offset
const resumed = (await calls.records("audit")).fromOffsets(saved)
let saved = reader.offsets().to_vec(); // one offset per partition
let resumed = calls.records("audit")?.from_offsets(saved);
saved = reader.offsets  # one offset per partition
resumed = calls.records("audit", from_offsets=saved)

A typed reader is a cursor that you own. It is not a consumer group, and the server does not store its offsets. Use a consumer group for production processing.

Publish at volume

publish sends one message and waits for the server. For throughput, use a long-lived producer, a background producer, or a batching producer. See Producing at volume.

Good to know

  • Delivery is at least once. A crash between handling and commit delivers the message again, so handlers must be safe to repeat.
  • There is no order across partitions. More partitions give more parallel consumers. One key keeps its messages in order.
  • When a group member joins or leaves, the server moves partitions between members. Each partition has one member at a time.
  • A message that does not decode fails with a decode error that names its position, and the next read moves past it. In Python, an async for loop over a typed reader ends on that error, so use next() in a loop to skip it.
  • Payloads are bytes. JSON, CBOR, MessagePack, BSON, and Avro helpers encode values into those bytes.
  • Messages work on plain Apache Iggy, Laser Stack, and LaserData Cloud.

Messages in depth

Producers, batching, publish failures, commit policies, offsets, and statistics

Filters

Let the server send each group only the messages it needs

Sessions

Record a unit of agent work on the log

Connect

Connection strings, timeouts, and publish retries

log example

Write and read back typed messages in Rust, Python, and TypeScript

Sessions

Record every agent interaction as a durable session you can inspect, replay, and control

Memory

Remember facts for an agent or a user, recall them from any agent, and trace each one back to the log

On this page

Quick exampleProcess messages live in a consumer groupKeep related messages in orderResume a reader from saved offsetsPublish at volumeGood to know