stbass — Formal concurrency primitives for the agentic era

Chan (Channels)

Import: from stbass import Chan Also: from stbass.channel import ChannelTypeError, ChannelClosedError, ChannelTopologyError, ChannelProtocolError, SequentialProtocol, VariantProtocol

Constructor

Chan(type_spec, *, name=None)

| Parameter | Type | Description | |-----------|------|-------------| | type_spec | type \| SequentialProtocol \| VariantProtocol | The type this channel enforces | | name | str \| None | Human-readable name (auto-generated if omitted) |

Type Specifications

Primitive types:

ch = Chan(int)                    # only int values
ch = Chan(str)                    # only str values
ch = Chan(dict)                   # only dict values

Pydantic models (with auto-coercion from dicts):

from pydantic import BaseModel
 
class Query(BaseModel):
    text: str
    limit: int = 10
 
ch = Chan(Query)
await ch.send(Query(text="hello"))              # direct model
await ch.send({"text": "hello"})                # dict auto-coerced to Query
await ch.send({"text": "hello", "limit": 5})    # full dict

SequentialProtocol (values must follow an exact type order):

from stbass.channel import SequentialProtocol
 
ch = Chan(SequentialProtocol(str, int, float))
await ch.send("hello")  # first must be str
await ch.send(42)        # second must be int
await ch.send(3.14)      # third must be float