Real-world Neurons (API + MCP) — Examples

Pick your stack

The same topology, expressed across language and transport. Slide between tabs — the routing logic never changes; only the imports, the synapse you connect to, and how you launch it do.

01 · Install

Each Neuron(source=…) wrapper is a soft dependency — you only install what the sources you use require. The filesystem MCP server itself is a separate npm package, fetched on demand by npx.

# SDK + CLI, plus the soft deps the two sources need
pip install -e cosmonapse-core/packages/python-sdk
pip install flask mcp        # flask = API source · mcp = MCP source

# The filesystem MCP server is published on npm; npx fetches it on first run
$ node --version    # any Node 18+ provides npx

02 · Start the Synapse

One bus, one namespace. The CLI also streams every Signal that crosses it to stdout, so the synapse terminal doubles as a Doppler.

$ cosmo synapse start memory --namespace=quickstart

  URL:        cosmo://127.0.0.1:7070
  Namespace:  quickstart
  Transport:  TCP + NDJSON  (single-host dev only)
  ────────────────────────────────────────────────

03 · API worker — a Flask app as a Neuron

Neuron(source="flask") wraps an ordinary Flask app. The TASK’s input is replayed as an in-process HTTP request — no socket opened — and the JSON response becomes the Neuron output.

worker_api.py

import asyncio
from flask import Flask, request, jsonify
from cosmonapse import Axon, Neuron, Dendrite, connect_synapse

SYNAPSE_URL = "cosmo://127.0.0.1:7070"   # ← the only line that changes per transport
NAMESPACE   = "quickstart"

# An ordinary Flask app — zero Cosmonapse knowledge anywhere in it.
app = Flask(__name__)

@app.post("/summarise")
def summarise():
    body = request.get_json() or {}
    text = body.get("text", "")
    return jsonify(summary=text[:120], length=len(text))

async def main():
    # source="flask" replays each TASK as an in-process HTTP request.
    axon = Axon(
        neuron_id="summary-api",
        neuron_fn=Neuron(source="flask", app=app,
                         default_path="/summarise"),
        capabilities=["http", "summarise"],
    )
    synapse  = await connect_synapse(SYNAPSE_URL)
    dendrite = Dendrite(synapse=synapse, namespace=NAMESPACE,
                        dendrite_id="summary-api")
    dendrite.attach_axon(axon)
    try:
        async with dendrite:
            print("summary-api ready")
            await asyncio.Event().wait()
    finally:
        await synapse.close()

asyncio.run(main())

04 · MCP worker — a filesystem server as a Neuron

Neuron(source="mcp", server="filesystem") spawns the standard filesystem MCP server over stdio and exposes its tools. This is a wrapper — Cosmonapse ships no servers of its own. One subprocess is reused across tasks and torn down on deregister.

worker_files.py

import asyncio
from cosmonapse import Axon, Neuron, Dendrite, connect_synapse

SYNAPSE_URL = "cosmo://127.0.0.1:7070"
NAMESPACE   = "quickstart"

async def main():
    # Wrap the STANDARD filesystem MCP server (spawned via npx) as a Neuron.
    # Wrapper only — we do not implement the server. "." = allowed directory.
    axon = Axon(
        neuron_id="files",
        neuron_fn=Neuron(source="mcp", server="filesystem", args=["."]),
        capabilities=["mcp", "filesystem"],
    )
    synapse  = await connect_synapse(SYNAPSE_URL)
    dendrite = Dendrite(synapse=synapse, namespace=NAMESPACE,
                        dendrite_id="files")
    dendrite.attach_axon(axon)
    try:
        async with dendrite:
            print("files ready")
            await asyncio.Event().wait()
    finally:
        await synapse.close()

asyncio.run(main())

05 · The Cortex — one dispatch, two kinds of Neuron

The orchestrator is a plain Dendrite. Notice the two ask() calls are identical in shape — the Cortex never knows one worker is an HTTP API and the other an MCP server. That is the whole point: anything that maps a TASK to a result is a Neuron.

cortex.py

import asyncio
from cosmonapse import Dendrite, connect_synapse, new_trace_id

SYNAPSE_URL = "cosmo://127.0.0.1:7070"

async def main():
    synapse  = await connect_synapse(SYNAPSE_URL)
    dendrite = Dendrite(synapse=synapse, namespace="quickstart",
                        dendrite_id="cortex", heartbeat_s=0)
    pending = {}

    @dendrite.on_agent_output
    async def _on_output(sig):
        fut = pending.pop(sig.trace_id, None)
        if fut and not fut.done():
            fut.set_result(sig.payload.get("output", {}))

    async def ask(neuron, input, timeout=30.0):
        trace_id = new_trace_id()
        fut = asyncio.get_running_loop().create_future()
        pending[trace_id] = fut
        await dendrite.dispatch_task(neuron=neuron, input=input, trace_id=trace_id)
        return await asyncio.wait_for(fut, timeout=timeout)

    try:
        async with dendrite:
            # The same dispatch_task call — the Cortex can't tell an API
            # from an MCP server. Both are just Neurons behind Axons.
            summary = await ask("summary-api",
                {"text": "Cosmonapse turns any real-world thing into a neuron."})
            print("API →", summary)

            listing = await ask("files",
                {"tool": "list_directory", "arguments": {"path": "."}})
            print("MCP →", listing.get("response", "")[:200])
    finally:
        await synapse.close()

asyncio.run(main())

06 · Run the topology

Separate terminals, one Synapse shared by all. Start the bus first, then the workers, then the driver.

# terminal 1 — the bus
$ cosmo synapse start memory --namespace=quickstart

# terminal 2 — the API worker
$ python worker_api.py

# terminal 3 — the MCP worker
$ python worker_files.py

# terminal 4 — the cortex
$ python cortex.py

One TASK becomes an HTTP request; the other becomes an MCP tool call:

$ python cortex.py
API → {'status': 200, 'ok': True,
       'json': {'summary': 'Cosmonapse turns any real-world…', 'length': 51}, …}
MCP → [FILE] cortex.py
[FILE] worker_api.py
[FILE] worker_files.py
[DIR]  data

Extend the pattern

Any MCP server. Swap server="filesystem" for "git", "fetch", or "memory" — or point command/args at any other stdio MCP server you have.

Any API. The flask/wsgi source serves any WSGI callable; expressNeuron serves any Node request listener. Existing microservices become Neurons with no code changes.

Mix in LLMs. Add a third worker with Neuron(source="ollama") — the Cortex dispatches to it the same way. Route by capabilities to pick the right kind of Neuron per task.

Change transport. Every other tab is the same topology — only the install, the synapse you connect to, and the launch commands change. The Neuron wiring is byte-for-byte identical.

→

An API and an MCP server.
Both just Neurons.

Capability routing

Neuron sources

Concepts