Python SDK reference.
cosmonapse Python package - verified against packages/python-sdk. If something here disagrees with the code, the code wins.
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.
cosmonapse.AxonWraps a Neuron function with the metadata and validation needed to put it on the bus.
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
| Parameter | Type | Description |
|---|---|---|
| neuron_id | str | The address other processes use to reach this Neuron. Must be unique within a namespace. |
| neuron_fn | async (input, context) → dict | The Neuron itself. Receives the TASK payload and resolved context; must return a JSON-serialisable dict. |
| capabilities | list[str] | None | Tags advertised in REGISTER for capability-based routing. Defaults to an empty list. |
| version | str | None | Optional version string surfaced in REGISTER so callers can target a specific revision. Defaults to None. |
| context_fetcher | async (context_ref) → list | Resolver for payload.context_ref. Defaults to a no-op returning []. |
Methods
Axon.handle_task(task: Signal) -> SignalCalled 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.
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
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.