A Choreographed Agent. No Loop.

One dispatch, no loop  -  the Dendrites chain the run themselves

  agent-loop (orchestrator) --- ONE TASK ["planner"] ---> awaits the trace's FINAL

  planner-node   @on_agent_output("planner")   -> TASK ["research"|"coding"],
                                                 or assemble + imprint + FINAL
  research-node  @on_agent_output("research")  -> TASK ["planner"] (next step)
  coding-node    @on_agent_output("coding")    -> TASK ["planner"] (next step)
  engram host    @on_imprint_signal            -> mirror "answer" imprints to disk
  tool nodes     @on_tool_call(<capability>)   -> run the MCP, reply TOOL_RESULT

  tools (workers)  web-node (websearch + fetch Axons) | clock | files
  planner -> ["time"]   research -> ["websearch"] ["fetch"]   coding -> ["filesystem"]

# Needs a real HF_TOKEN in cosmonapse-examples/.env. fetch/time run via
# uvx (install uv); filesystem runs via npx (Node 18+).
$ pip install cosmonapse httpx python-dotenv mcp fastapi uvicorn
$ pip install duckduckgo-mcp-server   # the websearch tool (launched via uvx)
$ uvicorn app:app --port 8000

$ curl -X POST localhost:8000/run -H 'content-type: application/json' \
       -d '{"goal": "Research X and write a Python script that does Y"}'
$ curl localhost:8000/memory

$ cosmo init agent -n agent

  Scaffolded agent in ./agent
    + config.py   + neurons/hello.py   + effector/tools.py
    + brain.py    + demo.py            + README.md

# run_agent - dispatch ONE TASK, wait for FINAL. No loop; the nodes chain.
async def run_agent(orchestrator, goal, *, max_steps=MAX_STEPS, timeout_s=600.0):
    tag = new_trace_id()        # the run's trace; also labels engram entries
    sig = await orchestrator.dispatch_and_wait(
        capabilities=["planner"],
        input={"goal": goal, "tag": tag, "step": 1, "max_steps": max_steps,
               "steps": [], "sources": []},
        trace_id=tag,
        scope="terminal",       # resolve on FINAL / ERROR only
        finalize=False,         # the planner node owns FINAL, not a worker
        timeout_s=timeout_s,
    )
    if sig.type is SignalType.ERROR:
        raise RuntimeError(f"run failed: {sig.payload.get('message')}")
    return sig.payload["result"]

# The neuron_fn is a STOCK Neuron - an HF-router chat model with zero
# protocol knowledge. Everything else is decorator-specified on its Axon.
AXON = Axon(neuron_id="planner", neuron_fn=llm(MODEL), capabilities=["planner"])

@AXON.before_task
async def situate(input):
    """Cache check + progress recall + clock grounding, then the prompt."""
    tid, _ = ambient_trace()
    d = AXON.dendrite

    # A previously remembered answer short-circuits the whole run.
    cached = await d.recall(engram_id=ENGRAM_ID,
                            query={"merge_key": f"answer:{input['goal']}"},
                            deadline_ms=2000)

    # Progress is not carried by any node - it is recalled from the engram.
    done = await d.recall(engram_id=ENGRAM_ID,
                          query={"tag": input["tag"], "top_k": 50},
                          deadline_ms=2000)

    # Ground the prompt in real time - via the planner's OWN clock tool.
    clock = await bus.mcp(d, ["time"], "get_current_time", {"timezone": "UTC"})
    return {"messages": [...]}   # the prompt the stock Neuron will run

@AXON.detects_output
def decide(raw):
    """Parse the model's JSON; force finish past max_steps; echo the chain."""
    decision = _first_json(raw.get("response")) or {}
    ...
    return {"route": route, "task": task, **chain}

@AXON.host.on_agent_output(neuron="planner")
async def chain(sig):
    """Declared at module level. The Axon applies it to the HOSTING
    Dendrite when announced (subscription ensured) - no on_connect
    boilerplate, no wiring helpers."""
    node = AXON.dendrite
    d = sig.payload.get("output", {})
    route, goal, tag = d.get("route"), d.get("goal"), d.get("tag")

    if route in ("research", "coding"):
        # The Dendrite CREATES the next TASK - same trace, so the
        # chain's FINAL resolves the caller's Pathway.
        await node.dispatch_task(
            capabilities=[route],
            input={"task": d.get("task", ""), "goal": goal, "tag": tag, ...},
            trace_id=sig.trace_id, parent_id=sig.id,
        )
        return

    # finish: assemble the report from this run's engram entries,
    # remember it under the goal, then conclude the trace.
    got = await node.recall(engram_id=ENGRAM_ID,
                            query={"tag": tag, "top_k": 50}, ...)
    await node.imprint(engram_id=ENGRAM_ID, op="upsert",
                       merge_key=f"answer:{goal}",
                       entry={"content": report, "tags": ["answer"]},
                       meta={"path": report_path},
                       await_ack=True, deadline_ms=2000, ...)
    await node.emit_final(
        trace_id=sig.trace_id, parent_id=sig.id,
        result={"report": report, "source": "web", ...},
    )

@AXON.before_task
async def gather(input):
    """Gather web context via MY tools, shaped into the prompt."""
    search = await bus.mcp(d, ["websearch"], "search",
                           {"query": task, "max_results": 5})
    page = await bus.mcp(d, ["fetch"], "fetch",
                         {"url": url, "max_length": 2500})
    return {"messages": [...]}   # subtask + web context

@AXON.detects_output
async def note_and_imprint(raw):
    """Shape the note, remember it (dendrite.imprint), echo the chain."""
    note = (raw.get("response") or "").strip()
    await AXON.dendrite.imprint(
        engram_id=ENGRAM_ID, op="append",
        entry={"content": note, "tags": ["research", st.get("tag", "")]},
        await_ack=True, deadline_ms=2000,
    )
    return {"note": note, **chain_state}

# @AXON.host.on_agent_output(neuron="research"): the chain handler hands
# back to the planner with step + 1 - same trace, same pattern as above.

# bus.py - MCP tools are NOT tasks. A tool invocation rides the cognition
# pair built for exactly this: TOOL_CALL out, TOOL_RESULT back.
async def mcp(dendrite, caps, tool, arguments, *, timeout=60.0):
    """Call one MCP tool over the TOOL_CALL / TOOL_RESULT signal pair."""
    call_id = new_event_id()
    tid = new_trace_id()            # own trace: never the agent chain's
    pw = await dendrite.observe_pathway(tid)
    async with pw:
        await dendrite.emit_tool_call(
            trace_id=tid, parent_id=call_id,
            tool=tool, args=arguments, call_id=call_id,
            neuron=caps[0],         # directed at the CAPABILITY
        )
        while True:
            sig = await pw.wait_for(SignalType.TOOL_RESULT, timeout_s=timeout)
            if sig.payload.get("call_id") != call_id:
                continue
            if sig.payload.get("error") is not None:
                raise RuntimeError(f"{caps}.{tool} error: ...")
            return sig.payload["result"]

# Each exchange gets a FRESH trace, so a TOOL_RESULT can never resolve the
# calling agent's pending chain Pathway - and doppler shows tool traffic as
# TOOL_CALL -> TOOL_RESULT, not nested TASKs. Tool calls are not role-gated
# (only TASK/STOP are): even worker-role nodes call their tools.

ENGRAM = InMemoryEngram(engram_id="agent-memory", engram_kind="context")

host = Dendrite(synapse=synapse, namespace=NAMESPACE,
                dendrite_id="agent-memory-host",
                role="worker")   # TOOL_CALLs aren't role-gated - no
                                  # orchestrator rights needed for tools
host.attach_engram(ENGRAM)

@host.on_imprint_signal
async def persist_answers(sig):
    """Mirror "answer" imprints carrying a path to disk (filesystem MCP)."""
    entry = sig.payload.get("entry") or {}
    path = (sig.meta or {}).get("path")
    if path and "answer" in (entry.get("tags") or []):
        await bus.mcp(host, ["filesystem"], "write_file",
                      {"path": str(ROOT / path),
                       "content": entry.get("content", "")})

@app.post("/run")
async def run(body: dict):
    """{"goal", "max_steps"?, "timeout_s"?} -> the trace's FINAL payload."""
    return await run_agent(
        state["orchestrator"], goal,
        max_steps=int(body.get("max_steps", MAX_STEPS)),
        timeout_s=float(body.get("timeout_s", 600.0)),
    )

@app.get("/memory")
async def memory():
    """What the agent remembers - answers, research notes, code snapshots."""
    ...

# A repeat goal returns instantly from the Engram ("source": "memory").

$ curl -X POST localhost:8000/run -H 'content-type: application/json' \
       -d '{"goal": "Research the Collatz conjecture, then write a Python CLI
            that prints the Collatz sequence for N"}'
{
  "report": "# Research the Collatz conjecture, ...",
  "code_path": "report/solution.py",
  "report_path": "report/answer.md",
  "source": "web",
  "steps": [{"route": "research", ...}, {"route": "coding", ...}]
}

# run the same goal again - answered from memory on step 1:
{ "report": "...", "source": "memory", "steps": [] }

# Set SYNAPSE_URL to a running synapse to split across processes -
# and to watch the chain live.

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

# terminal 2  -  Prism, the live browser view (http://127.0.0.1:7071)
$ cosmo doppler --prism --url=cosmo://127.0.0.1:7070 -n agent

# terminal 3  -  the agent, pointed at the bus
$ SYNAPSE_URL=cosmo://127.0.0.1:7070 uvicorn app:app --port 8000