Build agents and data-driven systems on one durable log. Ultra-low-latency streaming, a query layer, key-value state, copy-on-write forks, a knowledge graph, and a full agent fabric (memory, discovery, contracts, workflows), all over a single Apache Iggy connection. By LaserData, Inc.

One connection replaces four systems. The stream you already publish to becomes the store you query, the state you coordinate on, and the fabric your agents discover, route, and reason over. No second database, no cache, no orchestration server, nothing to keep in sync. The log is the single source of truth, and every other surface is a read model you can rebuild from offset 0. A support task, say, streams its messages, keeps its working memory, and resolves the dependencies between them, all in one place.

Prerelease (0.0.1-rc.8). The wire contract and the public API may still change between release candidates, so pin an exact version.

Rust (the reference SDK) and Python (foreign/python, native bindings over the same core). The wire contract is a standalone, language-neutral crate (laser-wire) pinned byte-for-byte by a cross-language conformance suite, so more language SDKs follow without drift.

Every feature is a primitive you reach by one accessor on the connected client, and every action is a verb on that primitive. One shape, object.verb(input).await, across the whole platform:

Learn the pattern once and the whole platform reads the same way. In Rust:

let laser = Laser::connect_with_stream("iggy:[email protected]:8090", "app").await?;

// Log: streams group topics, topics carry your records.
laser.topic("orders").publish().json(&order)?.send().await?;
laser.stream("audit").topic("events").publish().json(&event)?.send().await?;
let mut replay = laser.topic("orders").replay()?;

// Views: declared projections answer queries, the graph answers traversals.
let rows = laser
    .query("orders_v1")
    .where_eq("status", "paid")
    .limit(10)
    .fetch()
    .await?;
let nearby = laser.graph("kg").neighbors(node, EdgeDir::Out, None, 2).await?;
let mut feed = laser.watch().index("orders_v1").records()?; // await-then-query

// State: point reads and writes, optimistic concurrency, branches.
laser.kv("sessions").set("user:42").json(&session)?.ttl(300).send().await?;
let draft = laser.fork("what-if");

// Fabric: identities, context, memory, coordination, runs.
let reply = laser.agent(id).ask(commands, replies, task, &prov, timeout).await?;

// Context: one task streams its messages, keeps its memory, resolves its deps.
let ctx = laser.context(conversation);
ctx.append(AgentTopic::Audit, b"step done").await?;
let facts = ctx.memory("support").recall().semantic("refund disputes").fetch().await?;
let deps = ctx.graph("services").neighbors(node, EdgeDir::Out, None, 2).await?;

laser.memory("notes").set("current-plan", plan_json).await?; // named point state, an event on the memory topic
let run = laser.workflow("refund").registered().step(/* .. */).run().await?;
let page = laser.runs().list().state(AgentRunState::Running).fetch().await?;

The same grammar in Python, one-to-one with the Rust accessors:

laser = await ls.Laser.connect("iggy:[email protected]:8090", stream="app")

# Log
await laser.topic("orders").publish().json(order).send()
await laser.stream("audit").topic("events").publish().json(event).send()

# Views + graph + change feed
rows = await laser.query("orders_v1").where_eq("status", "paid").limit(10).fetch()
nearby = await laser.graph("kg").neighbors(node, direction="out", depth=2)
feed = laser.watch(index="orders_v1")

# State
await laser.kv("sessions").set("user:42").json(session).ttl(300).send()

# Fabric: one task streams its messages, keeps its memory, resolves its deps
ctx = laser.context(conversation)
await ctx.append("audit", b"step done")
facts = await ctx.memory(laser.memory()).recall(semantic="refund disputes")
deps = await ctx.graph("services").neighbors(node, direction="out", depth=2)
run = await laser.runs().submit("refund", task)

The accessors are free to construct, IO happens at the terminal verb (.send() writes, .fetch() reads), options are always fluent, and the raw substrate stays one call away (topic.iggy_producer()). What follows is the same platform, primitive by primitive.

Data platform (the core, stands on its own):

Agent fabric (opt in with the agent feature):

The agent fabric is the part most systems bolt on as a separate service. Here, routing, contracts, fan-out, and workflows are conventions over the log, thin client-side state machines over offsets, deadlines, leases, and replies. There is no orchestration server in the path. The substrate stays a log, which means your agents inherit its durability, replay, and ordering for free.

• One connection, one mental model. Everything is records on a log. Publish, query, KV, forks, graph, and coordination share the same connection and the same provenance, so there is nothing to wire together and nothing to keep consistent.
• Replayable by construction. Every read model rebuilds from offset 0. A bad projection, a new index, a fresh agent joining late: all just replay the log.
• Typed end to end. Serde in, codec stamped on the wire, decoded back to your struct. One typed handle per topic when you want the contract pinned: laser.topic("orders").json::<Order>() publishes and replays Order values (a schema-bound form validates against the registered writer schema before a byte leaves the process), and a record that stops decoding surfaces with its exact log position. Batched both directions, so throughput is a flag, not a rewrite.
• Open core, no lock-in. Publish, consume, the agent runtime, memory, and all coordination run on stock Apache Iggy. The managed surfaces light up against LaserData Cloud through capability negotiation, with the exact same code.
• At-most-once when it matters. The .exclusive() fenced step gives a single-holder guarantee for an external effect, with reassignment on timeout, so a zombie worker cannot double-execute.
• Rust and Python in lockstep. The Python SDK is native bindings over the Rust core against one byte-pinned wire contract, so the two never diverge.

Run a server, then publish:

docker run -p 8090:8090 apache/iggy:latest

Rust

laser-sdk = { version = "0.0.1-rc.8", features = ["query"] }

use laser_sdk::prelude::*;

#[tokio::main]
async fn main() -> Result<(), LaserError> {
    let laser = Laser::connect_with_stream("iggy:[email protected]:8090", "telemetry").await?;
    laser
        .topic("inferences")
        .publish()
        .json(&serde_json::json!({ "model": "gpt-4o", "latency_ms": 420 }))?
        .send()
        .await?;
    Ok(())
}

Python (pip install laser-sdk)

import asyncio, laser_sdk as ls

async def main():
    laser = await ls.Laser.connect("iggy:[email protected]:8090", stream="telemetry")
    await laser.topic("inferences").publish().json({"model": "gpt-4o", "latency_ms": 420}).send()

asyncio.run(main())

Consume is just as first-class, and there is a rung for every need. Live processing rides an Iggy consumer group: a balanced, offset-committing futures::Stream you drive as an async iterator, with the full tuning surface (polling strategy, auto-commit mode, retries) on the builder:

use futures::StreamExt;

let mut consumer = laser
    .topic("inferences")
    .iggy_consumer_group("analytics")?
    .auto_commit(AutoCommit::When(AutoCommitWhen::PollingMessages))
    .create_consumer_group_if_not_exists()
    .auto_join_consumer_group()
    .build();
consumer.init().await?;
while let Some(message) = consumer.next().await {
    let event: serde_json::Value = serde_json::from_slice(&message?.payload)?;
    // your processing here, offsets commit as you poll
}

async def handle(ctx, message):
    print(message.payload.decode())

agent = laser.spawn_agent("analytics", "inferences", handle)   # the reliable handler loop

Replay is the positional rung: topic.replay() opens a resumable cursor that reads by explicit offset, starting at 0 and draining everything new on each poll. Persist cursor.offsets() and seed the next cursor with .from_offsets(..), and a restart re-reads nothing: that is how export jobs checkpoint and how read models rebuild.

let history = laser.topic("inferences");
let mut cursor = history.replay()?;
let first = cursor.poll().await?; // from offset 0
save(cursor.offsets()); // checkpoint
let mut resumed = history.replay()?.from_offsets(load());
let fresh = resumed.poll().await?; // only what landed since

The read ladder in one line: topic.replay() for positional history, topic.iggy_consumer_group(..) for live balanced consumption, and Agent::builder().handler(..) when you want dedup, retry, and dead-lettering handled for you.

Batch is the throughput lever: topic.publish_batch() accumulates records and ships them in one round-trip, and a topic.replay() cursor drains everything new in one poll. json / msgpack are conveniences over add_payload, which takes raw bytes the SDK never inspects.

An orchestrator over capability agents (the orchestra example, Rust and Python, an interactive walk-through you can watch live in the console):

// Agents advertise capability cards. The orchestrator never hard-codes who can do what.
let reply = laser
    .contract(Router::to_capable("diagnose", RoutePolicy::Any))
    .from("orchestrator".parse()?)
    .payload(b"checkout API latency spike")
    .deadline(Duration::from_secs(10))
    .send()
    .await?; // Completed / NotConsumed / TimedOut

laser.quarantine("operator".parse()?, &"bad-agent".parse()?).await?;
laser.unquarantine("operator".parse()?, &"bad-agent".parse()?).await?;

The open surface (publish, consume, the agent runtime, provenance, log-backed memory, AGDX, and all coordination) runs on raw Apache Iggy. The managed surface (query, projections, KV, forks, the knowledge graph, durable dedup, and the fenced lease behind .exclusive()) needs LaserData Cloud and returns LaserError::Unsupported on raw Iggy. The same code runs in both, and capability negotiation at connect decides what is available. The SDK never hides the Iggy client (topic.iggy_producer(), topic.iggy_consumer(..), laser.client()).

Access is governed in two layers. Iggy's native RBAC grants global, stream, and topic permissions, so it decides whether a credential can create streams, send records, poll topics, and see a stream at all. LaserData's governance RBAC grants managed-surface capabilities, so it decides whether the same server-stamped user can call query, KV, graph, projection, fork, agent, workflow, and authz operations. The layers are independent: creating a user does not bind any governance role, and an operator must explicitly bind roles to grant managed access. LaserError::is_permission_denied() and is_stream_or_topic_not_found() classify native permission misses; managed authorization failures return the unified unauthorized result.

• Tutorial: a progressive guide from one message to projections, queries, vector recall, codecs, multi-stream topologies, and the agent fabric.
• Building agents: a recipe guide that works one multi-agent scenario end to end, including governed agents, managed-surface RBAC, and concrete SDK calls.
• AGDX notes: an in-repo development reference for the Agent Data Exchange Protocol the SDK implements (the envelope, the Apache Iggy binding, the surfaces). The protocol home is agdxprotocol.ai.
• Interop: the A2A / MCP / AG-UI edge bridges over AGDX.
• Examples: complete runnable systems, each with its own README, runnable locally or against LaserData Cloud unchanged.
• wire/README.md: the contract crate and its compatibility rules.

just lint    # fmt + sort + machete + clippy -D warnings
just test    # workspace unit tests
just test-it # integration tests against an Iggy testcontainer (needs Docker)
just bdd     # cross-SDK BDD conformance (needs Docker)
just ci      # the full gate (lint, test, wasm, deny, advisories, fixtures)

Build profiles: --no-default-features --features query (substrate only), --features query (runtime + query), --all-features (everything).

At-least-once with idempotent operations, per-conversation (per-partition) ordering, and replay-friendly throughout. The materialized index rebuilds from offset 0.

Apache-2.0. Copyright LaserData, Inc. Apache and Apache Iggy are trademarks of the Apache Software Foundation, and use does not imply endorsement.