Real-world Neurons.

$ git clone https://github.com/Cosmonapse/cosmonapse-examples
$ pip install 'cosmonapse[receptor]' 'mcp<2'
# Node 18+ on PATH - the filesystem MCP server runs via npx

$ cd cosmonapse-examples/08-real-world-neurons
$ python brain.py
$ curl -s localhost:8000/summarise -H 'content-type: application/json' \
       -d '{"text": "Cosmonapse is an event-driven substrate for agents."}'
$ curl -s localhost:8000/files

{"summary":"Cosmonapse is an event-driven substrate for agents.","length":51}
{"listing":"[FILE] README.md\n[FILE] brain.py\n[DIR] effector\n[DIR] neurons\n[DIR] receptors"}

08-real-world-neurons/
  neurons/summary.py     a plain async function behind an Axon
  effector/files.py      the filesystem MCP server, as an Effector
  receptors/api.py       POST /summarise (a TASK) + GET /files (a TOOL_CALL)
  brain.py               summary-node + files-node + api-node

from cosmonapse import Axon


async def summarise(input, context):
    text = input.get("text", "")
    return {"summary": text[:120], "length": len(text)}


AXON = Axon(
    neuron_id="summary", neuron_fn=summarise, capabilities=["summarise"],
)

from cosmonapse import Effector, Neuron

_server = Neuron(source="mcp", server="filesystem", args=["."])

EFFECTOR = Effector.serve(effector_id="files-effector", effector_kind="filesystem")


@EFFECTOR.on_tool_call
async def forward(tool: str, args: dict):
    """Any tool name the server knows: list_directory, read_text_file, ...

    The return value becomes the TOOL_RESULT; raising puts the message on
    the TOOL_RESULT's error instead - the trace carries on either way.
    """
    out = await _server({"tool": tool, "arguments": args}, [])
    if out.get("is_error"):
        raise RuntimeError(str(out.get("response") or f"{tool} failed"))
    return out

from cosmonapse import ApiReceptor

RECEPTOR = ApiReceptor(neuron="summary", path="/summarise", timeout_s=15.0)


@RECEPTOR.route("/files")
async def files(path: str = "."):
    outcome = await RECEPTOR.dendrite.call_tool(
        effector_id="files-effector",
        tool="list_directory",
        args={"path": path},
        deadline_ms=15_000,
    )
    if outcome.error:
        return {"error": outcome.error}
    return {"listing": outcome.result["response"]}

import asyncio
import contextlib
import signal

from cosmonapse import Dendrite, MemorySynapse, connect_synapse, run_brain

from config import NAMESPACE, SYNAPSE_URL
from effector import files
from neurons import summary
from receptors import api


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


def build_files(synapse) -> Dendrite:
    """attach_effector is all it takes: this node now services TOOL_CALLs
    addressed to files-effector."""
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id="files-node", role="worker")
    node.attach_effector(files.EFFECTOR)
    return node


def build_api(synapse) -> Dendrite:
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id="api-node", heartbeat_s=0)
    node.attach_receptor(api.RECEPTOR)
    return node

# 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

# 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
$ SYNAPSE_URL=kafka://127.0.0.1:9092 python brain.py