Building a Neuron.
Clone the examples repo, install, and run from inside the example's folder.
$ git clone https://github.com/Cosmonapse/cosmonapse-examples $ pip install cosmonapse 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/02-building-a-neuron $ python brain.py "Say hello to a project called Cosmonapse in one line." $ python brain.py --stream "..." # every Signal on the trace as it lands $ python brain.py # a REPL on the same brain
[AGENT_OUTPUT] Hello, Cosmonapse! Glad to have you on the bus.
Exact text varies - the model is stochastic.
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.
02-building-a-neuron/ config.py settings + hf_token() neurons/greeter.py the Neuron and its Axon receptors/terminal.py the interface - a prompt becomes a TASK brain.py greeter-node + terminal-node, run_brain
An LLM, behind the same interface as a function.
A Neuron is anything that satisfies async fn(input, context) → output. Neuron(source="huggingface", ...) returns an async callable of that shape around any OpenAI-compatible chat endpoint. Switch source="ollama" and nothing else in the program changes.
The Axon gives the Neuron an addressable id and capabilities, and turns its return value into a protocol-valid AGENT_OUTPUT Signal. It never touches the Synapse.
from cosmonapse import Axon, Neuron from config import hf_token greeter = 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=128, temperature=0.7, ) # The Axon declares identity + capabilities and owns the Neuron. AXON = Axon( neuron_id="greeter", neuron_fn=greeter, capabilities=["text-generation", "chat", "greet"], ) # Swapping providers - the endpoint is the only HF-specific line: # endpoint="https://<your-endpoint>.endpoints.huggingface.cloud" # dedicated HF # endpoint="http://localhost:8080" # local TGI / vLLM / LM Studio # For Ollama, switch source - same Axon, same Dendrite: # greeter = Neuron(source="ollama", model="llama3")
A command becomes a TASK.
A Receptor is the interface primitive. The command function returns the TASK input; the argparse tree, the REPL, --stream and --send are derived from it. The default shape is request/reply: dispatch, open a Pathway on the trace, wait for the terminal Signal, render it.
from cosmonapse import CliReceptor, SignalType RECEPTOR = CliReceptor( neuron="greeter", prog="greeter", description="Ask the greeter Neuron for a one-liner.", timeout_s=30.0, ) @RECEPTOR.command(help="send a prompt to the greeter") def ask(prompt: str): return {"prompt": prompt} # <- the TASK input @RECEPTOR.on_result def render(sig): # The HF Neuron returns {"response": "<text>", "meta": <raw>}. if sig.type is SignalType.ERROR: return f"[ERROR] {sig.payload.get('message')}" return f"[{sig.type.value}] {sig.payload['output']['response'].strip()}"
Two nodes, one bus.
A Dendrite is a node's attachment to the Synapse: it hosts components, emits REGISTER and HEARTBEAT for them, and routes inbound TASKs. The greeter gets a role="worker" node, which may answer TASKs but not emit them. The Receptor originates TASKs, so it mounts on an orchestrator node of its own.
run_brain starts every node before serving any interface, so a command can never reach an Axon whose REGISTER has not gone out.
import asyncio import contextlib import signal from cosmonapse import Dendrite, MemorySynapse, connect_synapse, run_brain from config import NAMESPACE, SYNAPSE_URL from neurons import greeter from receptors import terminal def build_greeter(synapse) -> Dendrite: """Hosts the greeter Axon. role="worker" is a protocol guard: a worker serves TASKs and bids, but cannot emit TASK / FINAL / STOP.""" node = Dendrite(synapse=synapse, namespace=NAMESPACE, dendrite_id="greeter-node", role="worker") node.attach_axon(greeter.AXON) return node def build_terminal(synapse) -> Dendrite: """Hosts the Receptor. It originates TASKs, so it needs an orchestrator-role node of its own.""" node = Dendrite(synapse=synapse, namespace=NAMESPACE, dendrite_id="terminal-node", heartbeat_s=0) node.attach_receptor(terminal.RECEPTOR) return node
Below the builders, main() opens the synapse (in-process unless SYNAPSE_URL is set) and calls run_brain(build_greeter(synapse), build_terminal(synapse)). That part is the same in every example.
The endpoint is the only provider-specific line.
endpoint="https://router.huggingface.co" # default endpoint="https://<your-endpoint>.endpoints.huggingface.cloud" # dedicated HF endpoint endpoint="http://localhost:8080" # local TGI / vLLM / LM Studio # For Ollama, switch source - same Axon, same brain. greeter = Neuron(source="ollama", model="llama3")
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 1 - a dev synapse (TCP, single host) $ cosmo synapse start memory --namespace=demo # terminal 2 - the same brain, over that synapse $ SYNAPSE_URL=cosmo://127.0.0.1:7070 python brain.py "Say hello in one line." # terminal 3 - Prism, the live view (http://127.0.0.1:7071) $ cosmo prism --url=cosmo://127.0.0.1:7070 -n demo # production transports - only the URL changes $ SYNAPSE_URL=nats://127.0.0.1:4222 python brain.py "Say hello in one line." $ SYNAPSE_URL=kafka://127.0.0.1:9092 python brain.py "Say hello in one line."