Laser SDK

Let agents and services message each other, share state and memory, and keep a replayable record of their work

Laser SDK lets agents and services exchange messages, share state and memory, and keep a durable, replayable record of every session, on one log. It has clients for Rust, Python, and TypeScript that behave the same way. It runs on Apache Iggy, locally with Laser Stack or in production on LaserData Cloud.

An agent and a caller

The triage agent advertises the resolve-ticket capability and replies to each request. The caller sends a request to whichever agent has that capability and waits for the reply. Both run on plain Apache Iggy. The connection string is the Laser Stack default.

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

await using laser = await Laser.connectWithStream("iggy:[email protected]:8090", "support")
await laser.bootstrap(1, TopicRetention.expireAfter(86_400_000))

await using triage = Agent.builder()
  .id(AgentId.new("triage"))
  .listenOn(AgentTopic.Sessions)
  .respondOn(AgentTopic.Sessions)
  .capabilities([{ skillId: "resolve-ticket" }])
  .handler({ handle: (_message, ctx) => ctx.respond(new TextEncoder().encode("on it")) })
  .build()
  .spawn(laser)
await triage.ready()

const result = await laser
  .contract(routeToCapable("resolve-ticket", { kind: "any" }))
  .from(AgentId.new("support-desk"))
  .payload(new TextEncoder().encode("ticket #42 is stuck"))
  .inboxRoute({ kind: "fixed", topic: AgentTopic.Sessions })
  .send()
if (result.kind === "completed") console.log(new TextDecoder().decode(agentMessageBody(result.reply)))

use laser_sdk::prelude::full::*;
use laser_sdk::wire::agent::CapabilityDescriptor;
use std::time::Duration;

struct Triage;

impl AgentHandler for Triage {
    async fn handle(&self, _message: &AgentMessage, ctx: &AgentCtx<'_>) -> Result<(), LaserError> {
        ctx.respond("on it").await
    }
}

#[tokio::main]
async fn main() -> Result<(), LaserError> {
    let laser = Laser::connect_with_stream("iggy:[email protected]:8090", "support").await?;
    laser.bootstrap(1, TopicRetention::expire_after(Duration::from_secs(86_400))).await?;

    let mut triage = Agent::builder()
        .id("triage".parse()?)
        .listen_on(AgentTopic::Sessions)
        .respond_on(AgentTopic::Sessions)
        .capabilities(vec![CapabilityDescriptor { skill_id: "resolve-ticket".into(), ..Default::default() }])
        .handler(Triage)
        .build()
        .spawn(laser.clone());
    triage.ready().await?;

    let result = laser
        .contract(Router::to_capable("resolve-ticket", RoutePolicy::Any))
        .from("support-desk".parse()?)
        .payload("ticket #42 is stuck")
        .inbox_route(InboxRoute::Fixed(AgentTopic::Sessions))
        .send()
        .await?;
    if let Contract::Completed(reply) = result {
        println!("{}", String::from_utf8_lossy(reply.body()));
    }
    triage.shutdown().await
}

import asyncio
import laser_sdk as ls

SESSIONS = ls.AgentTopic.Sessions

async def handle(ctx, message):
    await ctx.respond(b"on it")

async def main():
    laser = await ls.Laser.connect_with_stream("iggy:[email protected]:8090", "support")
    await laser.bootstrap(1, ls.TopicRetention.expire_after(86_400_000))

    triage = laser.spawn_agent("triage", SESSIONS, handle, respond_on=SESSIONS, capabilities=["resolve-ticket"])
    await triage.ready()

    result = await laser.contract("resolve-ticket", b"ticket #42 is stuck", source="support-desk", fixed_inbox=SESSIONS)
    if isinstance(result, ls.Contract.Completed):
        print(bytes(result[0].body()).decode())
    await triage.shutdown()
    await laser.close()

asyncio.run(main())

The request and the reply are messages on the agent.sessions topic, so they stay on the log after the program exits. The runnable agent example adds a deadline and pickup tracking. The Quickstart walks through a first program step by step.

What you can build

You need	Use	Needs Laser Stack or Cloud
Agents that find each other by capability, send requests, and get replies	Agents	No
A recorded unit of work with model calls, tool calls, state, and a final status	Sessions	Only for the session index and timelines
Facts an agent remembers and recalls across conversations	Memory	Only for the managed memory view
Keyed state with compare-and-swap, expiry, and branches	Key-value state	Yes
Queries over your messages that stay up to date, and a feed of what changed	Queries and views, Changes	Yes
A stream where each consumer receives only the records it asked for	Filters	Yes

npm install @laserdata/laser-sdk

The agent feature adds agents, sessions, and memory. The managed feature adds queries, key-value state, graph, forks, filters, and changes. Streaming is on by default.

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

Laser SDK needs Python 3.10 or later.

pip install laser-sdk

Packages are on npm, crates.io, and PyPI. Source and issues are at github.com/laserdata/laser-sdk.

Laser SDK

An agent and a caller

What you can build

Where to go next

Quickstart

Concepts

Agents

Sessions

Laser Stack

Advanced

Install

On this page