Quickstart

This page gets two agents talking. One agent handles a request, a caller sends it one, and the caller prints the reply. Then the same exchange runs inside a recorded session. Pick Rust, Python, or TypeScript in the tabs. Your choice carries across the SDK docs.

1. Start a target

Every step on this page works on any of these targets:

Laser Stack runs Apache Iggy and the managed plane on your machine. Use it for local development.
LaserData Cloud gives you a connection string on the Credentials tab of your deployment. Make sure an access rule allows your client IP.
Plain Apache Iggy runs messaging, agents, and sessions. Managed features such as queries and key-value state return Unsupported.

To start Laser Stack:

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

The script waits for Iggy and the plane, then prints a LASER_CONNECTION_STRING export. Copy it into your shell. For Cloud or your own Iggy server, export that server's connection string the same way:

export LASER_CONNECTION_STRING='iggy:[email protected]:8090'

Laser Stack covers requirements, persistence, and troubleshooting. Connect covers connection strings, tokens, and TLS.

2. Install the SDK

npm install @laserdata/laser-sdk

Requires Node.js 22.14 or later.

cargo add laser-sdk --features agent
cargo add tokio --features macros,rt-multi-thread

Requires Rust 1.98.0 or later. The agent feature adds agents and sessions. Add managed for queries, key-value state, and the other managed features.

pip install laser-sdk

Requires Python 3.10 or later. With uv, run uv pip install laser-sdk instead.

3. Have one agent answer another

The program connects, creates the agent topics on a stream named quickstart, and starts an agent called triage. Then a caller called intake sends triage a request and prints the reply.

import {
  Agent,
  AgentId,
  AgentTopic,
  Laser,
  TopicRetention,
  agentMessageBody,
  routeTo
} from "@laserdata/laser-sdk"

const text = new TextEncoder()
const decode = (bytes: Uint8Array) => new TextDecoder().decode(bytes)

await using root = await Laser.connectEnv()
const laser = root.withDefaultStream("quickstart")
await laser.bootstrap(1, TopicRetention.expireAfter(86_400_000))

await using triage = Agent.builder()
  .id(AgentId.new("triage"))
  .listenOn(AgentTopic.Sessions)
  .respondOn(AgentTopic.Sessions)
  .handler({
    handle: (message, ctx) =>
      ctx.respond(text.encode(`triaged: ${decode(agentMessageBody(message))}`))
  })
  .build()
  .spawn(laser)
await triage.ready()

const outcome = await laser
  .contract(routeTo(AgentId.new("triage")))
  .from(AgentId.new("intake"))
  .payload(text.encode("disk full on node-7"))
  .inboxRoute({ kind: "fixed", topic: AgentTopic.Sessions })
  .send()

if (outcome.kind === "completed") {
  console.log(decode(agentMessageBody(outcome.reply)))
} else {
  console.log(`no reply: ${outcome.kind}`)
}

use laser_sdk::prelude::full::*;
use std::time::Duration;

struct Triage;

impl AgentHandler for Triage {
    async fn handle(&self, message: &AgentMessage, ctx: &AgentCtx<'_>) -> Result<(), LaserError> {
        let ticket = String::from_utf8_lossy(message.body());
        ctx.respond(format!("triaged: {ticket}")).await
    }
}

#[tokio::main]
async fn main() -> Result<(), LaserError> {
    let laser = Laser::connect_env().await?.with_default_stream("quickstart");
    let retention = TopicRetention::expire_after(Duration::from_secs(86_400));
    laser.bootstrap(1, retention).await?;

    let mut triage = Agent::builder()
        .id("triage".parse()?)
        .listen_on(AgentTopic::Sessions)
        .respond_on(AgentTopic::Sessions)
        .handler(Triage)
        .build()
        .spawn(laser.clone());
    triage.ready().await?;

    let outcome = laser
        .contract(Router::to("triage".parse()?))
        .from("intake".parse()?)
        .payload("disk full on node-7")
        .inbox_route(InboxRoute::Fixed(AgentTopic::Sessions))
        .send()
        .await?;

    match outcome {
        Contract::Completed(reply) => println!("{}", String::from_utf8_lossy(reply.body())),
        other => println!("no reply: {other:?}"),
    }
    triage.shutdown().await
}

import asyncio

import laser_sdk as ls


async def handle(ctx, message):
    ticket = bytes(message.body()).decode()
    await ctx.respond(f"triaged: {ticket}".encode())


async def main():
    async with await ls.Laser.connect_env() as root:
        laser = root.with_default_stream("quickstart")
        await laser.bootstrap(1, ls.TopicRetention.expire_after(86_400_000))

        async with laser.spawn_agent(
            "triage",
            ls.AgentTopic.Sessions,
            handle,
            respond_on=ls.AgentTopic.Sessions,
        ):
            outcome = await laser.contract(
                None,
                b"disk full on node-7",
                agent="triage",
                source="intake",
                fixed_inbox=ls.AgentTopic.Sessions,
            )

        if isinstance(outcome, ls.Contract.Completed):
            print(bytes(outcome[0].body()).decode())
        else:
            print(f"no reply: {outcome!r}")


asyncio.run(main())

The program prints:

triaged: disk full on node-7

What each part does:

bootstrap creates the agent topics on the stream, one partition each, and keeps agent.sessions records for one day. It is safe to run again.
ready() waits until the agent reads its topic, so the request cannot arrive before it listens. Python's async with waits for you.
contract sends one request to one agent and waits for the answer, 30 seconds by default. It ends as completed, failed, not consumed, or timed out.
The fixed inbox sends the request to agent.sessions, where triage listens. Without it, the SDK looks up the agent's inbox from live presence, which needs an agent that advertises capabilities and a server that serves presence.

4. Record the exchange as a session

A session is one unit of work with a start, an end, and every record in between, kept on the log. Open a session as intake, send the request inside it, end the session, and read it back. Replace the request in step 3 with this code.

import { sessionTurnText } from "@laserdata/laser-sdk"

const { session } = await laser.sessions().start().agent(AgentId.new("intake")).begin()

await laser
  .contract(routeTo(AgentId.new("triage")))
  .from(AgentId.new("intake"))
  .payload(text.encode("disk full on node-7"))
  .inboxRoute({ kind: "fixed", topic: AgentTopic.Sessions })
  .conversation(session.conversation)
  .send()
await session.end()

for (const turn of await session.context()) {
  console.log(turn.display, turn.display === "agent.message" ? sessionTurnText(turn) : "")
}

let (session, _lease) = laser
    .sessions()
    .start()
    .agent("intake".parse::<AgentId>()?)
    .begin()
    .await?;

laser
    .contract(Router::to("triage".parse()?))
    .from("intake".parse()?)
    .payload("disk full on node-7")
    .inbox_route(InboxRoute::Fixed(AgentTopic::Sessions))
    .conversation(session.conversation())
    .send()
    .await?;
session.end().await?;

for turn in session.context().await? {
    let kind: &str = turn.display.into();
    match kind {
        "agent.message" => println!("{kind}: {}", turn.text()),
        _ => println!("{kind}"),
    }
}

session, _lease = await laser.sessions().start().agent("intake").begin()

await laser.contract(
    None,
    b"disk full on node-7",
    agent="triage",
    source="intake",
    fixed_inbox=ls.AgentTopic.Sessions,
    conversation=session.conversation,
)
await session.end()

for turn in await session.context():
    print(turn.display, turn.text() if turn.display == "agent.message" else "")

The program prints the session as it was recorded:

session.started
agent.message disk full on node-7
agent.message triaged: disk full on node-7
session.completed

The request and the reply carry the session ID as their conversation, so context() returns them with the session's start and end. This works on plain Apache Iggy, because the SDK reads the records back from agent.sessions. Laser Stack and LaserData Cloud also index sessions, which adds listing, status, and timeline reads across the stream. Sessions shows both.

Quickstart

1. Start a target

2. Install the SDK

3. Have one agent answer another

4. Record the exchange as a session

5. Publish and read your own messages

Next

Agents

Sessions

Concepts

Laser Stack

Connect

On this page