Python SDK reference.

from cosmonapse import Effector, EffectorBinding, Axon, Dendrite

# ── The Effector side: what actually runs ──────────────────────────
FX = Effector.serve(effector_id="fs-effector",
                    effector_kind="filesystem")

@FX.on_tool_call
async def handle(tool, args):
    # Return value IS the TOOL_RESULT  -  no manual publish.
    if tool == "read":
        return {"content": open(args["path"]).read()}
    return None          # fall through to the next handler

worker = Dendrite(synapse=synapse, role="worker")
worker.attach_effector(FX)

# ── The Neuron side: binding + native dialect ──────────────────────
# THE GATE: effectors= without tool_standard= raises ValueError.
brain = Axon(
    neuron_id="brain",
    neuron_fn=Neuron(source="openai", model="gpt-4o"),
    effectors=[EffectorBinding(name="fs",
                              directed_id="fs-effector")],
    tool_standard="codex",
)

# A Neuron that declares call_tool= gets the helper injected, already
# scoped to the running trace  -  the action-side twin of recall/imprint.
async def think(inp, ctx, *, call_tool):
    out = await call_tool("fs", tool="read",
                          args={"path": "/etc/hosts"})
    return {"response": out.result if out.error is None else out.error}

class Effector(ABC):
    effector_id:   str
    effector_kind: str
    capabilities:  list[str]          # tool names; [] means "everything"
    version:       str | None

    async def connect(self) -> None: ...   # open subprocess / HTTP pool
    async def close(self) -> None: ...     # release them

    async def can_serve(self, tool: str) -> bool: ...
    # Default: tool in capabilities (empty list = serve everything).

    async def invoke(
        self, tool: str, args: dict, *,
        call_id: str | None = None,
        deadline_ms: int | None = None,
        trace_id: str | None = None,
    ) -> ToolOutcome: ...
    # Tool-level failure (bad args, missing file, non-zero exit) is
    # ToolOutcome(error=...)  -  NOT a raise. Raise only for a backend
    # fault; the Dendrite maps it onto TOOL_RESULT error either way.

    @property
    def host(self) -> _EffectorHostProxy: ...
    # @fx.host.on_<signal> queues a Dendrite decorator at module level,
    # replayed onto the HOSTING Dendrite once it connects this Effector.

@dataclass(frozen=True)
class ToolOutcome:
    tool:        str
    result:      Any = None
    error:       str | None = None
    call_id:     str | None = None
    took_ms:     int | None = None
    effector_id: str | None = None