RAG + MCP: a coding agent.

$ git clone https://github.com/Cosmonapse/cosmonapse-examples
$ pip install cosmonapse httpx python-dotenv 'mcp<2'
# Node 18+ on PATH - the filesystem MCP server runs via npx
# a Hugging Face token with read scope, in cosmonapse-examples/.env
$ echo "HF_TOKEN=hf_xxxxxxxxxxxxxxxx" >> cosmonapse-examples/.env

$ cd cosmonapse-examples/12-rag-mcp
$ python brain.py "Code a small command-line tool that prints the first N Fibonacci numbers, where N is a positional argument." \
      --filename fib.py --argv 10

  indexed house-style          2 chunks
  indexed review-checklist     2 chunks
wrote generated/fib.py  (grounded on: house-style#0, review-checklist#1, ...)
--- code --------------------------------------------------
import argparse
...
--- python fib.py 10 --------------------------------------
0
1
1
2
...
exit code: 0

12-rag-mcp/
  config.py              appends ../11-rag to sys.path (VectorEngram, embeddings)
  sample_docs/           house-style.md, review-checklist.md
  neurons/librarian.py   index the docs
  neurons/coder.py       recall the docs -> HF writes one script
  effector/files.py      the filesystem MCP server, as an Effector
  effector/runner.py     run the script, capture the result
  helpers.py             index_docs(), code_pipeline()
  receptors/terminal.py  code
  brain.py               docs-node, a node per Neuron and per Effector, terminal-node

def make_coder_neuron():
    llm = Neuron(
        source="huggingface",
        endpoint="https://router.huggingface.co",
        model=GEN_MODEL,
        api_key=hf_token(),
        use_chat_api=True,
        max_new_tokens=1024,
        temperature=0.1,
    )

    async def coder_neuron(input, context, *, recall):
        """{"request", "filename"} -> {"filename", "code", "sources"}."""
        request = input["request"]
        qvec = (await embed([request], api_key=hf_token()))[0]
        result = await recall(
            "docs", query={"embedding": qvec, "top_k": TOP_K},
            deadline_ms=2000,
        )
        guide = "\n\n---\n\n".join(h.entry["text"] for h in result.hits)
        sources = [f"{h.entry['doc_id']}#{h.entry['chunk_index']}"
                   for h in result.hits]

        messages = [
            {"role": "system", "content": CODER_SYSTEM},
            {"role": "user", "content": (
                f"House-style context:\n\n{guide}\n\n"
                f"Request: {request}"
            )},
        ]
        out = await llm({"messages": messages}, [])
        code = extract_code(out["response"])
        if code is None:
            return {"__error__": True,
                    "message": "coder produced no usable code block"}
        return {"filename": input["filename"], "code": code,
                "sources": sources}

    return coder_neuron


def make_axon() -> Axon:
    return Axon(
        neuron_id="coder", neuron_fn=make_coder_neuron(),
        capabilities=["codegen"],
        engrams=[EngramBinding(name="docs", directed_id="code-docs")],
    )

from cosmonapse import Effector, Neuron

from config import ROOT

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

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


@EFFECTOR.on_tool_call
async def forward(tool: str, args: dict):
    out = await _server({"tool": tool, "arguments": args}, [])
    if out.get("is_error"):
        raise RuntimeError(str(out.get("response") or f"{tool} failed"))
    return out

import asyncio
import sys
from pathlib import Path

from cosmonapse import Effector

from config import ROOT

EFFECTOR = Effector.serve(effector_id="runner-effector", effector_kind="shell")


@EFFECTOR.on_tool_call
async def run(tool: str, args: dict):
    """run: {"path", "argv"?} -> {"exit_code", "stdout", "stderr"}."""
    if tool != "run":
        return None   # unhandled -> "unhandled tool" error on TOOL_RESULT
    path = Path(args["path"])
    proc = await asyncio.create_subprocess_exec(
        sys.executable, str(path), *args.get("argv", []),
        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
        cwd=str(ROOT),
    )
    try:
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=10)
    except asyncio.TimeoutError:
        proc.kill()
        return {"exit_code": -1, "stdout": "", "stderr": "timeout after 10s"}
    return {
        "exit_code": proc.returncode,
        "stdout": stdout.decode(errors="replace"),
        "stderr": stderr.decode(errors="replace"),
    }

async def code_pipeline(orchestrator, request: str, filename: str,
                        argv: list[str] | None = None,
                        timeout_s: float = 90.0) -> dict:
    """coder TASK -> write_file TOOL_CALL -> run TOOL_CALL, one trace."""
    tid = new_trace_id()

    # 1. RAG-grounded generation - the only TASK. Neurons think.
    r = await orchestrator.dispatch_and_wait(
        neuron="coder", input={"request": request, "filename": filename},
        trace_id=tid, timeout_s=timeout_s,
    )
    if r.type.value == "ERROR":
        raise RuntimeError(f"coder failed: {r.payload.get('message')}")
    gen = r.payload["output"]

    # 2. Persist through the filesystem Effector. Effectors act. A tool
    #    error rides the TOOL_RESULT - it never terminates the trace.
    rel_path = f"{OUT_DIR}/{gen['filename']}"
    write = await orchestrator.call_tool(
        effector_id="files-effector", tool="write_file",
        args={"path": str(ROOT / rel_path), "content": gen["code"]},
        trace_id=tid, parent_id=r.id, deadline_ms=30_000,
    )
    if write.error:
        raise RuntimeError(f"write failed: {write.error}")

    # 3. Run it. A non-zero exit is data, not an error.
    ran = await orchestrator.call_tool(
        effector_id="runner-effector", tool="run",
        args={"path": rel_path, "argv": argv or []},
        trace_id=tid, parent_id=r.id, deadline_ms=30_000,
    )
    if ran.error:
        raise RuntimeError(f"run failed: {ran.error}")

    return {"filename": rel_path, "code": gen["code"],
            "sources": gen["sources"], "run": ran.result}

from cosmonapse import CliReceptor

from helpers import code_pipeline, index_docs

RECEPTOR = CliReceptor(
    prog="rag-mcp",
    description="A RAG-grounded coding agent: generate, save, run.",
    timeout_s=120.0,
)


@RECEPTOR.command(local=True, default=True, help="write, save and run a script")
async def code(request: str, filename: str = "script.py", argv: str = ""):
    for doc in await index_docs(RECEPTOR.dendrite):
        print(f"  indexed {doc['doc_id']:<20} {doc['chunks']} chunks")
    result = await code_pipeline(RECEPTOR.dendrite, request, filename,
                                 argv=argv.split())
    run = result["run"]
    return "\n".join([
        f"wrote {result['filename']}  (grounded on: {', '.join(result['sources'])})",
        "--- code " + "-" * 50, result["code"].rstrip(),
        f"--- python {filename} {argv} ".ljust(59, "-"),
        (run["stdout"] or run["stderr"]).rstrip(),
        f"exit code: {run['exit_code']}",
    ])

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, runner
from engram.vector_engram import VectorEngram       # from 11-rag
from neurons import coder, librarian
from receptors import terminal

DOCS = VectorEngram(engram_id="code-docs", engram_kind="semantic")


def build_docs(synapse) -> Dendrite:
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id="docs-node", role="worker")
    node.attach_engram(DOCS)
    return node


def build_neuron(synapse, axon) -> Dendrite:
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id=f"{axon.neuron_id}-node", role="worker")
    node.attach_axon(axon)
    return node


def build_effector(synapse, effector) -> Dendrite:
    """attach_effector alone makes this node service that tool's TOOL_CALLs."""
    node = Dendrite(synapse=synapse, namespace=NAMESPACE,
                    dendrite_id=f"{effector.effector_kind}-node", role="worker")
    node.attach_effector(effector)
    return node


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

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

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

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

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