Install & run

Clone the examples repo, install, and run from inside the example's folder.

terminal

$ git clone https://github.com/Cosmonapse/cosmonapse-examples
$ pip install cosmonapse

$ cd cosmonapse-examples/09-capability-routing
$ python brain.py demo
$ python brain.py summarize "Neurons are plain functions that..."
$ python brain.py who translate

-> summarize summarizer <- summary: Cosmonapse is an event-driven substrate ...
-> translate translator <- [fr] Hello, world
-> summarize summarizer <- summary: Neurons are plain functions that......

Layout

One component per file, one Dendrite per node.

The skeleton cosmo init scaffolds. Modules under neurons/, engram/, effector/ and receptors/ declare behaviour; brain.py owns deployment - which node hosts what.

09-capability-routing

09-capability-routing/
  neurons/roles.py       summarizer + translator, different capabilities
  receptors/terminal.py  route by capability; who lists providers
  brain.py               a node per role + terminal-node

01 · The workers

Capabilities are advertised, not configured.

Each Axon's capabilities ride its REGISTER. Nothing on the calling side keeps a list of workers.

neurons/roles.py

from cosmonapse import Axon


async def summarize(input, context):
    return {"result": f"summary: {input['text'][:40]}..."}


async def translate(input, context):
    return {"result": f"[fr] {input['text']}"}


# role -> (capabilities advertised in REGISTER, neuron_fn)
ROLES = {
    "summarizer": (["summarize", "text"], summarize),
    "translator": (["translate", "text"], translate),
}


def make_axon(role: str) -> Axon:
    capabilities, fn = ROLES[role]
    return Axon(neuron_id=role, neuron_fn=fn, capabilities=capabilities)

02 · Routed dispatch

ask(..., capabilities=[...]).

The terminal holds no ids. Identical capability profiles share a queue group, so two summarizers would load-balance and each TASK is still delivered once. The rendered reply shows sig.directed.id: who actually answered.

receptors/terminal.py

RECEPTOR = CliReceptor(
    input_key="text",
    prog="routing",
    description="Route by capability, never by worker id.",
    timeout_s=30.0,
)


@RECEPTOR.on_result
def render(sig):
    who = sig.directed.id if sig.directed else "?"
    if sig.type is SignalType.ERROR:
        return f"{who}: ERROR {sig.payload.get('message')}"
    return f"{who:<10} <- {sig.payload['output']['result']}"


@RECEPTOR.command(local=True, help="route text to whoever can summarize")
async def summarize(text: str):
    return await RECEPTOR.ask({"text": text}, capabilities=["summarize"])


@RECEPTOR.command(local=True, help="route text to whoever can translate")
async def translate(text: str):
    return await RECEPTOR.ask({"text": text}, capabilities=["translate"])


@RECEPTOR.command(local=True, help="run three capability-tagged tasks")
async def demo():
    lines = []
    for capability, text in TASKS:
        reply = await RECEPTOR.ask({"text": text}, capabilities=[capability])
        lines.append(f"-> {capability:<9} {reply}")
    return "\n".join(lines)

03 · Discovery

Who can do this, right now?

Routing does not need a registry; listing does. A node with a registry_store folds REGISTER, HEARTBEAT and DEREGISTER into a live view that find_neurons reads. REGISTER rides the bus like any Signal, so a fresh process polls briefly.

receptors/terminal.py

@RECEPTOR.command(local=True, help="which live Neurons advertise a capability")
async def who(capability: str):
    for _ in range(20):                    # REGISTER rides the bus - poll briefly
        found = await RECEPTOR.dendrite.find_neurons(capability=capability)
        if found:
            return {capability: [n.neuron_id for n in found]}
        await asyncio.sleep(0.05)
    return {capability: []}

brain.py

import asyncio
import contextlib
import signal

from cosmonapse import (Dendrite, MemoryRegistryStore, MemorySynapse,
                        connect_synapse, run_brain)

from config import NAMESPACE, SYNAPSE_URL
from neurons import roles
from receptors import terminal


def build_worker(synapse, role: str) -> Dendrite:
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id=role, role="worker")
    node.attach_axon(roles.make_axon(role))
    return node


def build_terminal(synapse) -> Dendrite:
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id="terminal-node", heartbeat_s=0,
                    registry_store=MemoryRegistryStore())   # powers find_neurons()
    node.attach_receptor(terminal.RECEPTOR)
    return node

Over a real synapse

Same code, different transport.

With SYNAPSE_URL unset the brain runs on an in-process bus. Point it at a synapse and nothing in the modules changes. Prism attaches to any shared synapse and animates every Signal as it crosses.

terminal

# terminal 1 - a dev synapse (TCP, single host)
$ cosmo synapse start memory --namespace=quickstart

# terminal 2 - the same brain, over that synapse
$ SYNAPSE_URL=cosmo://127.0.0.1:7070 python brain.py demo

# terminal 3 - Prism, the live view (http://127.0.0.1:7071)
$ cosmo prism --url=cosmo://127.0.0.1:7070 -n quickstart

# production transports - only the URL changes
$ SYNAPSE_URL=nats://127.0.0.1:4222 python brain.py demo
$ SYNAPSE_URL=kafka://127.0.0.1:9092 python brain.py demo

http://127.0.0.1:7071 · -n quickstart

Prism renders every Signal on the bus as it fires - REGISTER, TASK, AGENT_OUTPUT, FINAL.

→

Capability routing.

One component per file, one Dendrite per node.

Capabilities are advertised, not configured.

ask(..., capabilities=[...]).

Who can do this, right now?

Same code, different transport.

Bidding

Round robin

Agent