Agents

An agent is an ID plus a handler that reads one topic. Other agents and services send it work through the log, and it replies through the log. Use agents when several services or models must hand work to each other and you need a record of every request and reply.

Agents run on plain Apache Iggy, Laser Stack, and LaserData Cloud.

Quick example

The triage agent advertises the triage-ticket capability. The caller asks for that capability, not for a named agent, and prints the reply.

import {
  Agent,
  AgentId,
  AgentTopic,
  Laser,
  TopicRetention,
  agentMessageBody,
  routeToCapable
} 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("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: "triage-ticket" }])
  .ackOnPickup()
  .handler({
    handle: (message, ctx) =>
      ctx.respond(text.encode(`triaged: ${decode(agentMessageBody(message))}`))
  })
  .build()
  .spawn(laser)
await triage.ready()

const outcome = await laser
  .contract(routeToCapable("triage-ticket", { kind: "any" }))
  .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)))
}

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> {
        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("support");
    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: "triage-ticket".to_owned(),
            ..Default::default()
        }])
        .ack_on_pickup(true)
        .handler(Triage)
        .build()
        .spawn(laser.clone());
    triage.ready().await?;

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

    if let Contract::Completed(reply) = outcome {
        println!("{}", String::from_utf8_lossy(reply.body()));
    }
    triage.shutdown().await
}

import asyncio

import laser_sdk as ls


async def triage(ctx, message):
    await ctx.respond(b"triaged: " + bytes(message.body()))


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

        async with laser.spawn_agent(
            "triage",
            ls.AgentTopic.Sessions,
            triage,
            respond_on=ls.AgentTopic.Sessions,
            capabilities=["triage-ticket"],
            ack_on_pickup=True,
        ):
            outcome = await laser.contract(
                "triage-ticket",
                b"disk full on node-7",
                source="intake",
                fixed_inbox=ls.AgentTopic.Sessions,
            )
            if isinstance(outcome, ls.Contract.Completed):
                print(bytes(outcome[0].body()).decode())


asyncio.run(main())

An agent with capabilities publishes a card to the registry when it starts, so callers can find it. ack_on_pickup makes the agent report Working as soon as it takes a request. The fixed inbox sends the request to agent.sessions. Without it, the SDK sends each request to the inbox the agent advertises in its live presence, which needs a server that serves presence.

Send a request to one agent

Name the agent with Router::to in Rust, routeTo in TypeScript, or agent= in Python. Set deadline to change how long the caller waits, 30 seconds by default. Set expire_if_not_consumed to learn that no agent picked the request up. Handle all four outcomes.

const outcome = await laser
  .contract(routeTo(AgentId.new("triage")))
  .from(AgentId.new("intake"))
  .payload(text.encode("node-7 unreachable"))
  .inboxRoute({ kind: "fixed", topic: AgentTopic.Sessions })
  .expireIfNotConsumed(5_000)
  .deadline(60_000)
  .send()

switch (outcome.kind) {
  case "completed":
    console.log(`done: ${decode(agentMessageBody(outcome.reply))}`)
    break
  case "failed":
    console.log(`failed: ${decode(agentMessageBody(outcome.reply))}`)
    break
  case "notConsumed":
    console.log("no agent picked it up")
    break
  case "timedOut":
    console.log("picked up, no answer in time")
    break
}

let outcome = laser
    .contract(Router::to("triage".parse()?))
    .from("intake".parse()?)
    .payload("node-7 unreachable")
    .inbox_route(InboxRoute::Fixed(AgentTopic::Sessions))
    .expire_if_not_consumed(Duration::from_secs(5))
    .deadline(Duration::from_secs(60))
    .send()
    .await?;

match outcome {
    Contract::Completed(reply) => println!("done: {}", String::from_utf8_lossy(reply.body())),
    Contract::Failed(reply) => println!("failed: {}", String::from_utf8_lossy(reply.body())),
    Contract::NotConsumed => println!("no agent picked it up"),
    Contract::TimedOut => println!("picked up, no answer in time"),
}

outcome = await laser.contract(
    None,
    b"node-7 unreachable",
    agent="triage",
    source="intake",
    fixed_inbox=ls.AgentTopic.Sessions,
    expire_if_not_consumed_ms=5_000,
    deadline_ms=60_000,
)
match outcome:
    case ls.Contract.Completed(reply):
        print(f"done: {bytes(reply.body()).decode()}")
    case ls.Contract.Failed(reply):
        print(f"failed: {bytes(reply.body()).decode()}")
    case ls.Contract.NotConsumed():
        print("no agent picked it up")
    case ls.Contract.TimedOut():
        print("picked up, no answer in time")

NotConsumed means that neither a pickup report nor a reply arrived before the consumption expiry. Trust it only when the agent runs with ack_on_pickup, because without pickup reports a slow handler also reads as not consumed. Without an expiry, an unanswered request ends as TimedOut.

Hand off work without waiting

submit starts a session for an agent and puts the input in front of it. The call returns once both records are written. The agent's handler reaches the session through ctx.session() and ends it when the work is done.

const summarizer = {
  handle: async (message: AgentMessage, ctx: AgentCtx) => {
    console.log(`summarizing ${decode(agentMessageBody(message))}`)
    await ctx.session().end()
  }
}

const submitted = await laser
  .sessions()
  .submit(AgentId.new("summarizer"), text.encode("summarize incident 7"))
  .from(AgentId.new("intake"))
  .send()

struct Summarizer;

impl AgentHandler for Summarizer {
    async fn handle(&self, message: &AgentMessage, ctx: &AgentCtx<'_>) -> Result<(), LaserError> {
        println!("summarizing {}", String::from_utf8_lossy(message.body()));
        ctx.session().end().await
    }
}

let submitted = laser
    .sessions()
    .submit("summarizer".parse::<AgentId>()?, "summarize incident 7")
    .from("intake".parse::<AgentId>()?)
    .send()
    .await?;

async def summarize(ctx, message):
    print(f"summarizing {bytes(message.body()).decode()}")
    await ctx.session().end()

submitted = await (
    laser.sessions()
    .submit("summarizer", b"summarize incident 7")
    .from_("intake")
    .send()
)

submitted.session is the session ID. Read the session back with laser.sessions().open(id).context().

Find agents by capability

The registry holds the cards that agents publish. resolve returns the cards that advertise a skill, are fresh, do not mark the skill unavailable, and are not quarantined.

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

const registry = await laser.agentRegistry()
const now = new SystemClock().nowMicros()
await registry.refresh(now)
for (const card of registry.resolve("triage-ticket", now)) {
  console.log(`${card.agent.asStr()} serves triage-ticket`)
}

use laser_sdk::agent::{Clock, SystemClock};

let mut registry = laser.agent_registry()?;
let now = SystemClock.now_micros();
registry.refresh(now).await?;
for card in registry.resolve("triage-ticket", now) {
    println!("{} serves triage-ticket", card.agent);
}

registry = laser.agent_registry()
await registry.refresh()
for card in registry.resolve("triage-ticket"):
    print(f"{card.agent} serves triage-ticket")

A capability route picks one of these agents. The default policy takes any of them. Other policies pick the cheapest, the fastest, the least loaded, or a preferred agent, or run your own ranking. Advanced agents lists them.

Stop an agent

The handle returned by spawn owns the running agent. Keep it until you stop the agent.

shutdown() stops new reads, finishes the message in flight, and reports any consumer error. It waits up to the shutdown grace, 30 seconds by default.
join() waits for the agent to stop on its own.
abort() stops it at once.

TypeScript also stops the agent at the end of an await using block. Python stops it at the end of an async with block. Dropping the Rust handle, or letting Python collect it, also stops the agent, but you will not see an error from the final drain.

Good to know

Run bootstrap(partitions, retention) once per stream before agents start. It creates agent.sessions and the other agent topics.
Delivery is at least once. The agent commits a message only after the handler succeeds, so a crash before that means the agent sees the message again. Make external effects safe to repeat.
A handler that fails with a retryable error gets 5 attempts in total. The first retry waits 200 ms and each later wait doubles. After the last attempt, or after a non-retryable error, the message moves to agent.dlq and the agent moves on.
Each agent ID reads through its own consumer group and handles only commands addressed to it or to every agent. Replies and other agents' work are skipped. Run more copies of the same agent to share its load, and never give two different agent IDs the same group.
Live presence belongs to a connection. On a server that serves presence, give each agent that advertises capabilities its own connection. A second advertising agent on the same connection fails to start with a presence conflict error.
Exclusive workflow steps need the managed plane. Session listing and status reads need the managed session index. Everything else on this page works on plain Apache Iggy.

Agents

Quick example

Send a request to one agent

Hand off work without waiting

Find agents by capability

Stop an agent

Good to know

Advanced agents

Sessions

Interop

Governance

Example on GitHub

On this page

Agents

Quick example

Send a request to one agent

Hand off work without waiting

Find agents by capability

Stop an agent

Good to know

Related

Advanced agents

Sessions

Interop

Governance

Example on GitHub

On this page