Receptors: the interface layer.

$ git clone https://github.com/Cosmonapse/cosmonapse-examples
$ pip install 'cosmonapse[receptor]' httpx python-dotenv
# a Hugging Face token with read scope, in cosmonapse-examples/.env
$ echo "HF_TOKEN=hf_xxxxxxxxxxxxxxxx" >> cosmonapse-examples/.env

$ cd cosmonapse-examples/17-receptors
$ python smoke_test.py                    # offline - all edges, no token
$ python brain.py "what is a synapse"      # one-shot   -> dispatch_and_wait
$ python brain.py --stream "..."           # one-shot   -> dispatch_and_subscribe
$ python brain.py --send "..."             # one-shot   -> dispatch_task
$ python brain.py ping                     # local command, nothing dispatched
$ python brain.py                          # REPL + http://127.0.0.1:8000

$ curl -s  localhost:8000/run -H 'content-type: application/json' -d '{"input": "what is a synapse"}'
$ curl -sN localhost:8000/run -H 'content-type: application/json' -d '{"input": "...", "mode": "stream"}'

cli    : ['echo(0 prior): hello from the cli']
send   : TASK trc_01J...
stream : ['AGENT_OUTPUT']
chat   : 'echo(0 prior): one' then 'echo(2 prior): two'
api    : wait='echo(0 prior): http' send=True stream=True

OK

17-receptors/
  neurons/assistant.py   the Neuron - {prompt | message, history?} -> {reply}
  receptors/terminal.py  CliReceptor
  receptors/api.py       ApiReceptor - one endpoint, send / wait / stream
  receptors/chat.py      ChatReceptor - one turn, one dispatch, voice optional
  brain.py               a worker node + a node per interface
  smoke_test.py          offline proof that every edge reaches the Neuron
  RECIPES.md             copy-paste setups for every backend and hook

shape    Receptor        Dendrite                  you get
send     rx.send(x)      dispatch_task             the emitted TASK
wait     rx.ask(x)       dispatch + Pathway.wait   the rendered result
stream   rx.stream(x)    dispatch_and_subscribe    a live Pathway

import asyncio

from cosmonapse import CliReceptor, SignalType

RECEPTOR = CliReceptor(
    neuron="assistant",
    prog="assistant",
    description="Talk to a Cosmonapse Neuron from a terminal.",
    timeout_s=120.0,
)


@RECEPTOR.command(help="ask the assistant")
def ask(prompt: str, tokens: int = 400):
    """The return value IS the TASK input - that is the whole contract."""
    return {"prompt": prompt, "max_new_tokens": tokens}


@RECEPTOR.command("ping", local=True, help="is the worker registered?")
async def ping():
    """local=True: answered here, nothing crosses the wire. REGISTER rides
    the bus like every other Signal, so poll briefly in a fresh process."""
    for _ in range(20):
        found = await RECEPTOR.dendrite.find_neurons()
        if found:
            return {"neurons": [n.neuron_id for n in found]}
        await asyncio.sleep(0.05)
    return {"neurons": []}


@RECEPTOR.on_result
def render(sig):
    return sig.payload["output"]["reply"]


@RECEPTOR.on_print
def show(reply: str):
    print(f"\n{reply}\n")


@RECEPTOR.on_signal(SignalType.TOOL_CALL, SignalType.PLAN)
async def progress(sig):
    # Observation only - delete this and the answers are identical.
    print(f"  . {sig.type.value.lower()}")

from cosmonapse import ApiReceptor

RECEPTOR = ApiReceptor(
    neuron="assistant",
    path="/run",
    input_key="prompt",
    timeout_s=120.0,
)


@RECEPTOR.on_result
def render(sig):
    return {"reply": sig.payload["output"]["reply"], "trace_id": sig.trace_id}


@RECEPTOR.route("/neurons")
async def neurons():
    """An ordinary route on the same router."""
    found = await RECEPTOR.dendrite.find_neurons()
    return {"neurons": [n.neuron_id for n in found]}

from cosmonapse import ChatReceptor

RECEPTOR = ChatReceptor(
    neuron="assistant",
    title="Cosmonapse Assistant",
    greeting="Ask me anything - or press mic and talk.",
    voice=True,
    history_turns=8,
    timeout_s=120.0,
)

from cosmonapse import Axon, Neuron

from config import MODEL, hf_token

_chat = Neuron(
    source="huggingface",
    endpoint="https://router.huggingface.co",
    model=MODEL,
    api_key=hf_token(),
    use_chat_api=True,
    max_new_tokens=400,
)


async def assistant(payload: dict, context: dict | None = None) -> dict:
    """{prompt|message, history?} -> {reply}"""
    text = payload.get("prompt") or payload.get("message") or ""
    turns = payload.get("history") or []
    if turns:
        transcript = "\n".join(f"{t['role']}: {t['content']}" for t in turns)
        text = f"{transcript}\nuser: {text}"
    out = await _chat({"prompt": text}, context)
    reply = out.get("response") if isinstance(out, dict) else str(out)
    return {"reply": reply}


AXON = Axon(
    neuron_id="assistant",
    neuron_fn=assistant,
    capabilities=["chat", "text-generation"],
)

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 assistant
from receptors import api, chat, terminal


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


def build_interface(synapse, name: str, receptor) -> Dendrite:
    """One node per interface. registry_store powers find_neurons() - the
    terminal's `ping` and the api's /neurons."""
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id=f"{name}-node", heartbeat_s=0,
                    registry_store=MemoryRegistryStore())
    node.attach_receptor(receptor)
    return node

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

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

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

# production transports - only the URL changes
$ SYNAPSE_URL=nats://127.0.0.1:4222 python brain.py "what is a synapse"
$ SYNAPSE_URL=kafka://127.0.0.1:9092 python brain.py "what is a synapse"