Fabric

Fabric coordinates agents through the log. Agents advertise capabilities, receive work, and process uncommitted tasks again after failure. Contracts set deadlines. Workflows add budgets and compensation for partial work.

Built for

Use Fabric for multi-agent systems, durable pipelines, and approval workflows.

How it works

Agent::builder() creates a handler that reads one topic and replies on another. Python uses spawn_agent(..). Configure an agent through these operations:

• id(..) sets its identity.
• listen_on(..) and respond_on(..) select topics.
• capabilities([...]) advertises the work that it can perform.
• handler(..) supplies a Rust struct, TypeScript object, or Python function.
• build().spawn(laser) starts it. ready() waits until it consumes messages.

Callers use laser.contract(..) to address work by capability. A router selects an agent that advertises that capability. The caller does not need a fixed agent instance.

deadline(duration) limits the wait. Outcomes are Completed, Failed, NotConsumed, or TimedOut. NotConsumed means that no agent picked up the task. TimedOut means that an agent picked it up but did not reply before the deadline. Handle both as ordinary outcomes.

The runtime records work on the log. If a handler crashes before commit, it receives the task again and restarts. Save checkpoints explicitly when work must resume within a handler.

Deduplication uses message IDs. External effects still need an idempotency key or fenced write before commit. An unhandled message moves to a dead-letter path rather than blocking the topic indefinitely.

A handler can open the conversation on an inbound message as a session with laser.sessions().open(conversation). It reads the typed turns so far and appends its own. See Context.

Publishes from agents use the connection's publish timeout and retries. See Connect.

Before agents join a stream, run bootstrap(partitions) once to create their topics.

Fabric uses these supporting features:

• Governance applies roles and policy before actions, with access denied by default.
• AGDX defines agent records and their log behavior.
• Interop exposes agents through A2A, MCP, and AG-UI.

Core agent handling works with Iggy without laser-plane.

Quick example

await using triage = Agent.builder()
  .id(AgentId.new("triage"))
  .listenOn(AgentTopic.Commands)
  .respondOn(AgentTopic.Responses)
  .capabilities([{ skillId: "resolve-ticket" }])
  .ackOnPickup()
  .handler({
    handle: (_message, context) => context.respond(utf8("on it"))
  })
  .spawn(laser)
await triage.ready()

const contract = await laser
  .contract(routeToCapable("resolve-ticket", ANY_ROUTE_POLICY))
  .from(AgentId.new("orchestrator"))
  .payload(utf8("ticket #42 is stuck"))
  .inboxRoute({ kind: "fixed", topic: AgentTopic.Commands })
  .deadline(60_000)
  .send()

if (contract.kind === "completed") {
  console.log(decodeUtf8(agentMessageBody(contract.reply)))
}

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

let mut triage = Agent::builder()
    .id("triage".parse()?)
    .listen_on(AgentTopic::Commands)
    .respond_on(AgentTopic::Responses)
    .capabilities(vec![CapabilityDescriptor {
        skill_id: "resolve-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("resolve-ticket", RoutePolicy::Any))
    .from("orchestrator".parse()?)
    .payload("ticket #42 is stuck")
    .inbox_route(InboxRoute::Fixed(AgentTopic::Commands))
    .deadline(Duration::from_secs(60))
    .send()
    .await?;

match outcome {
    Contract::Completed(reply) => {
        println!("{}", String::from_utf8_lossy(reply.body()));
    }
    other => println!("contract ended: {other:?}"),
}

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

triage = laser.spawn_agent(
    "triage",
    COMMANDS,
    handle,
    respond_on=RESPONSES,
    capabilities=["resolve-ticket"],
    ack_on_pickup=True,
)
await triage.ready()

reply = await laser.contract(
    "resolve-ticket",
    b"ticket #42 is stuck",
    source="orchestrator",
    fixed_inbox=COMMANDS,
    deadline_ms=60_000,
)
print(reply.decode() if reply else "<no reply>")

All clients advertise capabilities and acknowledge task pickup. Rust and TypeScript use builders. Python uses direct calls and returns reply bytes. A crash before commit causes a retry.

Complete examples: Rust, Python, and TypeScript.

The orchestra example covers discovery, workflows, quarantine, and deadline recovery.

Key operations

Running it

Exclusive workflows use revocable fenced leases, ownership grants with increasing tokens. Use Rust's exclusive_in(namespace) or the corresponding language API for writes through a coordination namespace. Keep renewal active while making sure that results are correct and recording completion. If the lease is lost, stop protected work. State explains holder and namespace requirements.

Fabric works on Laser Stack, LaserData Cloud, and standalone Iggy. Managed deduplication, fenced leases, and runs require their advertised capabilities.

Related