SDK · 04

Axon - agent-side tool

The Axon owns the Neuron’s identity (neuron_id, capabilities, version) and the tool body (neuron_fn). It never touches the Synapse - it must be attached to a Dendrite to participate.

classcosmonapse.Axon

Wraps a Neuron function with the metadata and validation needed to put it on the bus.

axon.pyi
class Axon(LifecycleHooks):
    def __init__(
        self,
        *,
        neuron_id:       str,
        neuron_fn:       Callable[[dict, list], Awaitable[dict]],
        capabilities:    list[str] | None = None,
        version:         str | None      = None,
        neuron_kind:     str = "neuron",
        context_fetcher: Callable[[str], Awaitable[list]] | None = None,
        engrams:         list[EngramBinding] | None = None,
        output_parser:   OutputParser | None = None,
    ) -> None: ...

    # ── Source-paired factories ─────────────────────────────────
    # Create the Neuron AND the Axon in one call, wired with the
    # matching recogniser. Every factory returns a plain Axon.
    @classmethod
    def from_source(cls, source, *, neuron_id,
                    capabilities=None, version=None, neuron_kind="neuron",
                    context_fetcher=None, engrams=None,
                    recognize=True, teach_intents=None,
                    **source_kwargs) -> "Axon": ...
    # source: ollama | huggingface/hf | openai | anthropic | groq |
    #         openrouter | together | mistral | mcp

    # Shorthands over from_source():
    @classmethod
    def ollama(cls, neuron_id, **kw) -> "Axon": ...       # model= required
    @classmethod
    def huggingface(cls, neuron_id, **kw) -> "Axon": ...  # endpoint= required; alias Axon.hf
    @classmethod
    def openai(cls, neuron_id, **kw) -> "Axon": ...       # model= required; api_key or OPENAI_API_KEY
    @classmethod
    def anthropic(cls, neuron_id, **kw) -> "Axon": ...    # model= required; api_key or ANTHROPIC_API_KEY
    @classmethod
    def mcp(cls, neuron_id, **kw) -> "Axon": ...          # command= or server= (+ args, tool)

    async def handle_task(self, task: Signal) -> Signal: ...
    # Called by the Dendrite. Resolves context_ref, invokes neuron_fn,
    # wraps the result in AGENT_OUTPUT / CLARIFICATION / PERMISSION / ERROR.

    # Pre-task hook  -  transform / validate / reject the TASK input.
    # Return a dict to replace the input, None to pass through, or raise
    # to reject (surfaces as ERROR code NEURON_EXCEPTION).
    @axon.before_task

    # Detectors over the Neuron's RAW output  -  named detects_* to stay
    # distinct from the Dendrite's on_* (which consume inbound Signals).
    # Return the intent's fields (dict) to match, None to fall through.
    # Precedence: error -> clarification -> permission -> output.
    @axon.detects_output           # -> AGENT_OUTPUT payload
    @axon.detects_clarification    # -> {"question": ..., "context": ...}
    @axon.detects_permission       # -> {"action": ..., "scope": ..., "reason": ...}
    @axon.detects_error            # -> {"code": ..., "message": ..., "recoverable": ...}

    # Inherited from LifecycleHooks:
    @axon.on_connect          # after the hosting Dendrite emits REGISTER
    @axon.on_refresh          # each heartbeat tick (reason="heartbeat")
    @axon.on_schedule(every_s=N)  # periodic background coroutine

Constructor parameters

ParameterTypeDescription
neuron_idstrThe address other processes use to reach this Neuron. Must be unique within a namespace.
neuron_fnasync (input, context) → dictThe Neuron itself. Receives the TASK payload and resolved context; must return a JSON-serialisable dict.
capabilitieslist[str] | NoneTags advertised in REGISTER for capability-based routing. Defaults to an empty list.
versionstr | NoneOptional version string surfaced in REGISTER so callers can target a specific revision. Defaults to None.
context_fetcherasync (context_ref) → listResolver for payload.context_ref. Defaults to a no-op returning [].

Methods

async methodAxon.handle_task(task: Signal) -> Signal

Called by the Dendrite for each inbound TASK. Resolves context_ref, invokes neuron_fn, and returns the corresponding outbound Signal (AGENT_OUTPUT, CLARIFICATION, PERMISSION, or ERROR). Application code never calls this directly.

Source-paired factories

The second way to build an Axon. Axon.from_source(source, ...) and its shorthands - Axon.ollama(), Axon.huggingface() (alias Axon.hf), Axon.openai(), Axon.anthropic(), Axon.mcp() - create the provider-backed Neuron and the Axon in one call. Extra kwargs flow to the Neuron(source=...) factory; the Axon kwargs (capabilities, version, engrams, ...) keep their meaning. By default the Axon is also wired with the source family’s recogniser (recognize=True) and the model is taught the {"cosmo": ...} intent convention (teach_intents) where the source accepts a system= prompt.

factories.py
import os
from cosmonapse import Axon

# One call: Neuron factory + Axon wiring + recogniser. Equivalent to
# Axon(neuron_id=..., neuron_fn=Neuron(source=...), output_parser=...).
chat = Axon.huggingface(
    neuron_id="llama",
    endpoint="https://router.huggingface.co",
    model="meta-llama/Llama-3.1-8B-Instruct",
    api_key=os.environ["HF_TOKEN"],
    use_chat_api=True,
    capabilities=["chat"],
)

cloud = Axon.openai(neuron_id="gpt", model="gpt-4o")         # api_key falls back to OPENAI_API_KEY
local = Axon.ollama(neuron_id="chat", model="llama3")
files = Axon.mcp(neuron_id="files", server="filesystem", args=["/data"])

# recognize=True (default) wires the source-family recogniser: the LLM
# recogniser parses a {"cosmo": ...} intent block out of the model's text
# (so it can emit CLARIFICATION / PERMISSION / ERROR, not just output);
# the MCP recogniser maps is_error -> ERROR. teach_intents (default: on
# for system=-capable LLM sources) appends COSMO_INTENT_SYSTEM_PROMPT so
# the model knows the convention. Opt out of both:
raw = Axon.openai(neuron_id="raw", model="gpt-4o", recognize=False)

Example

answerer.py
from cosmonapse import Axon

async def answerer(input: dict, context: list) -> dict:
    return {"answer": input["q"].upper()}

axon = Axon(
    neuron_id    = "answerer",
    neuron_fn    = answerer,
    capabilities = ["text", "qa"],
    version      = "1.2.0",
)

@axon.on_connect
async def warmup(a):
    await preload_model_weights()

Have a feature in mind?

The protocol, SDKs, and CLI are still pre-1.0. If something here is missing, ambiguous, or wrong - open an issue and propose a change. Every breaking change is debated in DECISIONS.md first.