sbp-client 0.1.0

Stigmergic Blackboard Protocol - Python Client SDK

SBP Python Client

Python SDK for the Stigmergic Blackboard Protocol.

Installation

pip install sbp-client

Quick Start

from sbp import SbpClient, ThresholdCondition

# Connect to the blackboard
with SbpClient("http://localhost:3000") as client:
    # Emit a pheromone
    client.emit(
        trail="market.signals",
        type="volatility",
        intensity=0.8,
        payload={"symbol": "BTC-USD", "vix": 45.2},
    )

    # Sniff the environment
    result = client.sniff(trails=["market.signals"])
    for p in result.pheromones:
        print(f"{p.trail}/{p.type}: {p.current_intensity:.2f}")

Local Mode (No Server Required)

You can run SBP entirely in-memory within a single process. This is useful for testing, simulations, or simple multi-agent scripts where you don't want to manage a separate server process.

from sbp import SbpClient, SbpAgent

# Client in local mode
with SbpClient(local=True) as client:
    client.emit("local.test", "signal", 0.9)

# Agent in local mode
# Note: In a single process, all local=True instances share the same blackboard state.
agent = SbpAgent("my-agent", local=True)

@agent.when("local.test", "signal", value=0.5)
async def handle(trigger):
    print("Received signal locally!")

Async Usage

import asyncio
from sbp import AsyncSbpClient, ThresholdCondition

async def main():
    async with AsyncSbpClient() as client:
        # Emit
        await client.emit("signals", "event", 0.7)

        # Register a scent (trigger condition)
        await client.register_scent(
            "my-scent",
            condition=ThresholdCondition(
                trail="signals",
                signal_type="event",
                aggregation="max",
                operator=">=",
                value=0.5,
            ),
        )

        # Subscribe to triggers via WebSocket
        async def on_trigger(trigger):
            print(f"Triggered! {trigger.scent_id}")

        await client.subscribe("my-scent", on_trigger)

asyncio.run(main())

Declarative Agent Framework

from sbp import SbpAgent, TriggerPayload, run_agent

agent = SbpAgent("my-agent", "http://localhost:3000")

@agent.when("tasks", "new_task", operator=">=", value=0.5)
async def handle_task(trigger: TriggerPayload):
    print(f"Got task: {trigger.context_pheromones}")
    await agent.emit("tasks", "completed", 1.0)

run_agent(agent)

API Reference

SbpClient / AsyncSbpClient

Method	Description
emit(trail, type, intensity, ...)	Deposit a pheromone
sniff(trails, types, ...)	Read environment state
register_scent(scent_id, condition, ...)	Register a trigger
deregister_scent(scent_id)	Remove a trigger
subscribe(scent_id, handler)	Listen for triggers (async only)
inspect(include)	Get blackboard metadata
evaporate(...)	Force cleanup

Condition Types

# Simple threshold
ThresholdCondition(
    trail="market.signals",
    signal_type="volatility",
    aggregation="max",  # sum, max, avg, count, any
    operator=">=",      # >=, >, <=, <, ==, !=
    value=0.7,
)

# Composite (AND/OR/NOT)
CompositeCondition(
    operator="and",
    conditions=[condition1, condition2],
)

# Rate-based
RateCondition(
    trail="events",
    signal_type="click",
    metric="emissions_per_second",
    window_ms=10000,
    operator=">=",
    value=5.0,
)

Decay Models

from sbp.types import exponential_decay, linear_decay, immortal

# Exponential (default) - half-life in milliseconds
decay = exponential_decay(half_life_ms=300000)  # 5 minutes

# Linear - rate per millisecond
decay = linear_decay(rate_per_ms=0.0001)

# Immortal - never decays
decay = immortal()

License

MIT