stbass

Formal concurrency primitives for the agentic era.

stbass is a Python process algebra library that brings Occam-2's proven concurrency model — PAR, SEQ, ALT, typed channels, placement, and failure semantics — to modern AI agent orchestration.

pip install stbass

Get Started Read the Docs

Derek Smalls from Spinal Tap playing a bass guitar

The multi-agent problem

stbass is an homage to Spinal Tap Bassist Derek Smalls. Modern AI agent orchestration is struggling with the exact same problems as coordinating a bass guitar with two fretboards, or getting a cucumber wrapped in aluminium foil through airport security. We are repeating mistakes that parallel computing solved forty years ago: agents call other agents through ad hoc message passing, subagents spawn without lifecycle management, and tool calls race with no priority semantics.

stbass fixes this with formal primitives, not another framework. By modeling agents as rigorous concurrent processes communicating over typed channels, you regain control over complex orchestration graphs.

The Core Primitives

Process

Any async unit of work. LLM calls, API requests, database writes.

@Process
async def agent(ctx: ProcessContext):
    return ProcessResult.ok("done")

Chan

Typed, zero-buffered communication between exactly two processes.

ch = Chan(Query)
await ch.send(Query(text="hello"))
val = await ch.recv()

SEQ

Sequential composition. Run A, then B, then C.

result = await SEQ(
    parse, validate, save
).run(input_value=data)

PAR

Parallel composition. Run A, B, C concurrently.

result = await PAR(
    agent_a, agent_b,
    on_failure=FailurePolicy.HALT
).run()

ALT

Alternation. Wait for whichever channel is ready first.

result = await ALT(
    Guard(ch_a, handler=...),
    Guard(ch_b, handler=...)
)

Placement

Bind processes to specific execution backends.

@Process(placement=Placement(model="opus"))
async def heavy_agent(ctx):
    ...

Quickstart

hello_agents.py

import asynciofrom stbass import PAR, Process, Chan, ProcessContext, ProcessResult
@Processasync def producer(ctx: ProcessContext) -> ProcessResult:    await ctx.send("data_chan", { "query": "hello" })    return ProcessResult.ok("sent")
@Processasync def consumer(ctx: ProcessContext) -> ProcessResult:    data = await ctx.recv("data_chan")    return ProcessResult.ok(f"processed {data}")
async def main():    ch = Chan(dict, name="data_chan")    result = await PAR(        producer.with_channels(data_chan=ch),        consumer.with_channels(data_chan=ch)    ).run()    print(result.value)
asyncio.run(main())

Create two processes, connect them with a typed channel, and run them concurrently using PAR. The send will rendezvous with the recv transparently. No race conditions, no deadlocks.

Standing on the Transputer's shoulders

stbass is built around the core primitives and syntax of the Occam-2 programming language, created in 1987 by David May and the team at INMOS for the Transputer architecture.

The insight behind stbass is that the problems Occam-2 solved — parallel process composition, typed inter-process communication, priority-based alternation, and formal failure semantics — are identical to the problems facing modern AI agent orchestration.

While we pay tribute to the Occam-2 lineage, stbass adapts the core process algebra for Python's async ecosystem. We've added dynamic replication, Pydantic-typed channels, MCP integration, placement-aware heterogeneous execution, and production-grade failure semantics to support the scale of modern AI workloads.