openbridge-buildlisted

# openbridge-build — author a new pool producer This is the library reference for authors. If you're driving an existing pool as a worker, load the **openbridge** skill, not this one. ## What you write A normal `async def main():` that calls `await bridge.ask(...)` to publish work into a pool. Each ask() is one work item, served by one worker. Concurrent ask()s fan out across workers in parallel. ```python import asyncio # Install OpenBridge first: `pip install -e .` from the OpenBridge repo root from openbridge import Bridge bridge = Bridge(name="my-producer", pool="my-pool") # name = this producer's unique ID (for `openbridge list`, history) # pool = the work-distribution group workers consume from async def process(item): result = await bridge.ask( item_id=item["id"], prompt=render_prompt(item), template={"summary": "", "scope": ""}, validate=lambda d: "summary required" if not d.get("summary") else None, ) if result.skipped: ... else: apply(item, result.data) async def main(): items = discover_items() progress = bridge.checkpoint("progress", default={"done": []}) pending = [i for i in items if i["id"] not in progress["done"]] # Fan out across the worker pool — N items processed in parallel sem = asyncio.Semaphore(5) async def gated(item): async with sem: await process(item) progress["done"].append(item["id"]) bridge.save("progress", p