An HTTP edge for a Neuron.

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

$ cd cosmonapse-examples/05-orchestrator-api
$ python brain.py
$ curl -s localhost:8000/ask -H 'content-type: application/json' \
       -d '{"prompt": "What is a synapse?"}'
$ curl -s localhost:8000/neurons

{"response":"A synapse is the junction where ...","trace_id":"trc_01J..."}
{"neurons":["worker"]}

05-orchestrator-api/
  neurons/chat.py        the HF worker
  receptors/api.py       ApiReceptor at POST /ask, plus GET /neurons
  brain.py               worker-node + api-node, and build_edge() for existing apps
  flask_app.py           an app that is already Flask
  wsgi_app.py            raw WSGI, no framework
  handlers_reference.py  orchestrator-side @on_* handlers, for reference

from cosmonapse import Axon, Neuron

from config import hf_token

AXON = Axon(
    neuron_id="worker",
    neuron_fn=Neuron(
        source="huggingface",
        endpoint="https://router.huggingface.co",
        model="meta-llama/Llama-3.1-8B-Instruct",
        api_key=hf_token(),
        use_chat_api=True,
        max_new_tokens=256,
    ),
    capabilities=["text-generation", "chat"],
)

from cosmonapse import ApiReceptor

RECEPTOR = ApiReceptor(
    neuron="worker",
    path="/ask",
    timeout_s=30.0,
)


@RECEPTOR.on_result
def render(sig):
    """Terminal Signal -> response body. ERROR is re-raised as a 500."""
    if sig.type.value == "ERROR":
        raise RuntimeError(sig.payload.get("message", "worker failed"))
    return {"response": sig.payload["output"]["response"],
            "trace_id": sig.trace_id}


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

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


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


def build_api(synapse) -> Dendrite:
    """registry_store powers find_neurons() for the /neurons route."""
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id="api-node", heartbeat_s=0,
                    registry_store=MemoryRegistryStore())
    node.attach_receptor(api.RECEPTOR)
    return node


def build_edge(synapse, dendrite_id: str) -> Dendrite:
    """An orchestrator node for a framework you already run (Flask, WSGI)."""
    return Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id=dendrite_id, heartbeat_s=0)

import asyncio
import threading

from flask import Flask, jsonify, request

from cosmonapse import connect_synapse

from brain import build_edge
from config import SYNAPSE_URL

if not SYNAPSE_URL:
    raise SystemExit("flask_app.py dispatches to brain.py over a shared synapse - set SYNAPSE_URL")

_loop = asyncio.new_event_loop()
_edge = None
_ready = threading.Event()


async def _connect() -> None:
    global _edge
    synapse = await connect_synapse(SYNAPSE_URL)
    _edge = build_edge(synapse, "flask-edge")
    await _edge.start()


def _run_loop() -> None:
    asyncio.set_event_loop(_loop)
    _loop.run_until_complete(_connect())
    _ready.set()
    _loop.run_forever()


threading.Thread(target=_run_loop, daemon=True).start()
_ready.wait()

app = Flask(__name__)


@app.post("/ask")
def ask():
    prompt = (request.get_json(silent=True) or {}).get("prompt", "")
    if not prompt:
        return jsonify({"error": "prompt required"}), 400
    future = asyncio.run_coroutine_threadsafe(
        _edge.dispatch_and_wait(neuron="worker", input={"prompt": prompt},
                                timeout_s=30.0),
        _loop,
    )
    try:
        reply = future.result(timeout=32)
    except TimeoutError:
        return jsonify({"error": "worker timed out"}), 504
    return jsonify({"response": reply.payload["output"]["response"]})


if __name__ == "__main__":
    app.run(port=5000)

$ cosmo synapse start memory --namespace=api-demo
$ SYNAPSE_URL=cosmo://127.0.0.1:7070 python brain.py        # the worker (and /ask)
$ SYNAPSE_URL=cosmo://127.0.0.1:7070 python flask_app.py    # :5000
$ curl -s localhost:5000/ask -H 'content-type: application/json' -d '{"prompt": "hi"}'

def attach_handlers(orchestrator: Dendrite) -> None:
    # -- AGENT_OUTPUT --------------------------------------------------
    # Fires when any worker finishes a TASK. dispatch_and_wait already
    # resolves the caller on the same Signal; use decorators for
    # side-effects (logging, metrics, webhooks).
    @orchestrator.on_agent_output
    async def _log_output(sig):
        print(f"[{sig.trace_id[:8]}] output from {sig.directed.id if sig.directed else '?'}")

    @orchestrator.on_agent_output(neuron="worker")  # narrow by id
    async def _worker_done(sig): ...

    @orchestrator.on_agent_output(capability="text-generation")  # by capability
    async def _gen_done(sig): ...

    # -- CLARIFICATION -------------------------------------------------
    # The Neuron asks a follow-up question. respond_to_clarification
    # re-dispatches a TASK with the answer attached, same trace_id.
    @orchestrator.on_clarification
    async def _handle_clarification(sig):
        print(f"[{sig.trace_id[:8]}] clarification: {sig.payload.get('question', '')}")
        await orchestrator.respond_to_clarification(
            sig, answer="Please use plain English, no technical jargon.",
        )

    # -- ERROR -----------------------------------------------------------
    @orchestrator.on_error_signal
    async def _handle_error(sig):
        print(f"[{sig.trace_id[:8]}] error: {sig.payload.get('message')}")

    # -- ESCALATION ------------------------------------------------------
    @orchestrator.on_escalation
    async def _handle_escalation(sig):
        print(f"[{sig.trace_id[:8]}] escalated: {sig.payload.get('reason')}")
        await orchestrator.respond_to_escalation(
            sig, input={"override": "proceed with best effort"},
        )

    # -- THOUGHT_DELTA - streamed reasoning tokens -------------------------
    @orchestrator.on_thought_delta
    async def _stream_thought(sig):
        print(sig.payload.get("delta", ""), end="", flush=True)

    # -- TOOL_CALL / TOOL_RESULT - audit logs. Effector servicing does NOT
    # consume these signals, so trace observers like this still fire. ------
    @orchestrator.on_tool_call
    async def _on_tool_call(sig):
        print(f"  tool -> {sig.payload.get('tool')}")

    @orchestrator.on_tool_result
    async def _on_tool_result(sig):
        print(f"  tool <- {sig.payload.get('tool')}")

    # -- REGISTER / HEARTBEAT - live worker roster -------------------------
    @orchestrator.on_register_signal
    async def _on_register(sig):
        print(f"worker joined: {sig.directed.id if sig.directed else '?'}  caps={sig.payload.get('capabilities')}")

    @orchestrator.on_heartbeat_signal
    async def _on_heartbeat(sig):
        print(f"heartbeat: {sig.directed.id if sig.directed else '?'}")

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

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

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

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