AgentProcess

Base class for all agents. Subclass it and implement handle().

See Core Concepts and Getting Started for usage examples.

---

civitas.process.AgentProcess(name, mailbox_size=1000, max_retries=3, shutdown_timeout=30.0)

Base class for all agent processes in Civitas.

Developers subclass this and override lifecycle hooks: - on_start(): called once before the first message - handle(message): called for every incoming message - on_error(error, message): called when handle() raises - on_stop(): called on graceful shutdown (always — even on crash)

Messaging methods available inside hooks: - send(recipient, payload, message_type): fire-and-forget - ask(recipient, payload, message_type, timeout): request-reply - send_capable(capability, payload, message_type): route to any capable agent - broadcast(pattern, payload): send to all matching agents - reply(payload): return from handle() for request-reply

Observability helpers (call from inside handle()): - llm_span(model, attrs): context manager for LLM call spans - tool_span(tool_name, attrs): context manager for tool call spans

Capability declaration (class-level, inherited and overridable): capabilities: list[str] = ["text.summarize", "text.translate"] capability_metadata: dict[str, Any] = { "text.summarize": {"description": "...", "version": "1"} }

Source code in civitas/process.py

def __init__(
    self,
    name: str,
    mailbox_size: int = 1000,
    max_retries: int = 3,
    shutdown_timeout: float = 30.0,
) -> None:
    self.name = name
    self.id: str = _uuid7()
    self.state: dict[str, Any] = {}
    self._status = ProcessStatus.INITIALIZING
    self._mailbox = Mailbox(maxsize=mailbox_size)
    self._task: asyncio.Task[None] | None = None
    self._max_retries = max_retries
    self._shutdown_timeout = shutdown_timeout

    # Injected by Runtime/Worker during setup
    self._bus: MessageBus | None = None
    self._tracer: Tracer | None = None
    self._registry: Registry | None = None
    self.llm: ModelProvider | None = None
    self.tools: ToolRegistry | None = None
    self.store: StateStore | None = None
    self.config: dict[str, Any] = {}

    # Per-agent credential map — populated from topology credentials: block.
    # Keys are provider names (e.g. "anthropic"); values are credential strings.
    self._credentials: dict[str, str] = {}

    # Audit sink — injected by ComponentSet, None when auditing is disabled.
    self._audit_sink: AuditSink | None = None

    # Metrics sink — injected by ComponentSet, None when no collector is attached.
    self._metrics: MetricsSink | None = None

    # Set by Runtime to the nearest DynamicSupervisor ancestor name (if any)
    self._dynamic_supervisor_name: str | None = None

    # MCP clients opened via connect_mcp() — keyed by server name
    self._mcp_clients: dict[str, Any] = {}

    # Current message context for reply/tracing
    self._current_message: Message | None = None
    self._current_handle_span: Span | None = None

    # Signalled when the message loop enters RUNNING
    self._running_event: asyncio.Event | None = None

    # _reached_loop gates restart eligibility (D8): only children that entered
    # the dispatch loop restart. _start_phase names where a start failure hit.
    self._reached_loop = False
    self._start_phase = "restore"

    # Non-blocking suspend intent (S2): set by suspend() / _agency.suspend,
    # actioned at the next message-loop boundary. Reason carried alongside.
    self._suspend_requested = False
    self._suspend_reason = ""

    self._pending_streams: dict[str, StreamSink] = {}
    self._stream_producers: dict[str, str] = {}
    self._out_streams: dict[str, _OutStream] = {}

on_start() async

Called once before the first message. Initialize self.state here.

Runs synchronously during startup — for a dynamically spawned agent it executes inside the spawn call, so await self.spawn(...) does not return until on_start() finishes (and raises TimeoutError if it outlasts the ask timeout). Keep it fast: do slow / I/O-bound work (LLM calls, browser sessions) in handle(), kicked off by a fire-and-forget message the spawner sends right after the spawn is confirmed. Spawn-time config is available here as self.config.

Source code in civitas/process.py

async def on_start(self) -> None:
    """Called once before the first message. Initialize self.state here.

    Runs synchronously during startup — for a dynamically spawned agent it
    executes *inside* the spawn call, so ``await self.spawn(...)`` does not
    return until ``on_start()`` finishes (and raises ``TimeoutError`` if it
    outlasts the ask timeout). Keep it fast: do slow / I/O-bound work (LLM
    calls, browser sessions) in ``handle()``, kicked off by a fire-and-forget
    message the spawner sends right after the spawn is confirmed. Spawn-time
    ``config`` is available here as ``self.config``.
    """

handle(message) async

Called for every incoming message.

Return self.reply(...) for request-reply. Return None for fire-and-forget.

Source code in civitas/process.py

async def handle(self, message: Message) -> Message | None:
    """Called for every incoming message.

    Return self.reply(...) for request-reply. Return None for fire-and-forget.
    """
    return None

on_error(error, message) async

Called when handle() raises an exception.

Return an ErrorAction. Default: ESCALATE (crash, let supervisor decide).

Source code in civitas/process.py

async def on_error(self, error: Exception, message: Message) -> ErrorAction:
    """Called when handle() raises an exception.

    Return an ErrorAction. Default: ESCALATE (crash, let supervisor decide).
    """
    return ErrorAction.ESCALATE

on_stop() async

Called on graceful shutdown. Always called — even on crash.

Source code in civitas/process.py

async def on_stop(self) -> None:
    """Called on graceful shutdown. Always called — even on crash."""

on_child_terminated(name, reason) async

Called when a dynamically spawned child is permanently removed.

reason is one of: "restarts_exhausted", "despawned", "clean_exit". Default implementation logs a warning. Override to re-spawn, alert, etc.

Source code in civitas/process.py

async def on_child_terminated(self, name: str, reason: str) -> None:
    """Called when a dynamically spawned child is permanently removed.

    reason is one of: "restarts_exhausted", "despawned", "clean_exit".
    Default implementation logs a warning. Override to re-spawn, alert, etc.
    """
    logger.warning("[%s] dynamic child '%s' terminated: %s", self.name, name, reason)

send(recipient, payload, message_type='message') async

Fire-and-forget: send a message to another agent by name.

Source code in civitas/process.py

async def send(
    self,
    recipient: str,
    payload: dict[str, Any],
    message_type: str = "message",
) -> None:
    """Fire-and-forget: send a message to another agent by name."""
    self._reject_reserved_type(message_type)
    if self._bus is None:
        raise RuntimeError("AgentProcess not wired to a MessageBus")
    trace_id = ""
    parent_span_id: str | None = None
    if self._current_message is not None:
        trace_id = self._current_message.trace_id
        parent_span_id = self._current_message.span_id

    message = Message(
        type=message_type,
        sender=self.name,
        recipient=recipient,
        payload=payload,
        trace_id=trace_id,
        span_id=_new_span_id(),
        parent_span_id=parent_span_id,
    )
    await self._bus.route(message)
    if self._metrics is not None:
        self._metrics.message_sent(self.name)

ask(recipient, payload, message_type='message', timeout=30.0) async

Request-reply: send a message and await a response.

Source code in civitas/process.py

async def ask(
    self,
    recipient: str,
    payload: dict[str, Any],
    message_type: str = "message",
    timeout: float = 30.0,
) -> Message:
    """Request-reply: send a message and await a response."""
    self._reject_reserved_type(message_type)
    if self._bus is None:
        raise RuntimeError("AgentProcess not wired to a MessageBus")
    trace_id = ""
    parent_span_id: str | None = None
    if self._current_message is not None:
        trace_id = self._current_message.trace_id
        parent_span_id = self._current_message.span_id

    correlation_id = _uuid7()
    message = Message(
        type=message_type,
        sender=self.name,
        recipient=recipient,
        payload=payload,
        correlation_id=correlation_id,
        trace_id=trace_id,
        span_id=_new_span_id(),
        parent_span_id=parent_span_id,
    )
    if self._metrics is not None:
        self._metrics.message_sent(self.name)
    return await self._bus.request(message, timeout=timeout)

broadcast(pattern, payload) async

Send a message to all agents matching a glob pattern.

Source code in civitas/process.py

async def broadcast(self, pattern: str, payload: dict[str, Any]) -> None:
    """Send a message to all agents matching a glob pattern."""
    if self._bus is None:
        raise RuntimeError("AgentProcess not wired to a MessageBus")
    targets = self._bus.lookup_all(pattern)
    for target in targets:
        await self.send(target.name, payload)

reply(payload)

Create a reply message. Return this from handle() for request-reply.

Source code in civitas/process.py

def reply(self, payload: dict[str, Any]) -> Message:
    """Create a reply message. Return this from handle() for request-reply."""
    if self._current_message is None:
        raise RuntimeError("reply() called outside of handle()")
    msg = self._current_message
    return Message(
        type=payload.get("type", "reply"),
        sender=self.name,
        recipient=msg.reply_to or msg.sender,
        payload=payload,
        correlation_id=msg.correlation_id,
        trace_id=msg.trace_id,
        span_id=_new_span_id(),
        parent_span_id=msg.span_id,
    )

checkpoint() async

Save self.state to the configured StateStore.

Call this from handle() after completing a meaningful unit of work. On restart, self.state is automatically restored from the last checkpoint. Agents that never call checkpoint() incur zero overhead.

Source code in civitas/process.py

async def checkpoint(self) -> None:
    """Save self.state to the configured StateStore.

    Call this from handle() after completing a meaningful unit of work.
    On restart, self.state is automatically restored from the last checkpoint.
    Agents that never call checkpoint() incur zero overhead.
    """
    if self.store is not None:
        await self.store.set(self.name, self.state)

spawn(agent_class, name, config=None, *, wait=True) async

Spawn a dynamic agent via the nearest ancestor DynamicSupervisor.

The nearest-ancestor special case of :meth:spawn_into — it resolves the ancestor DynamicSupervisor name wired at startup and delegates. Sends a civitas.dynamic.spawn message and awaits confirmation. Raises SpawnError if no DynamicSupervisor ancestor exists or spawn is denied. Returns the agent name on success.

With wait=True (default) the call returns only after the child reaches RUNNING/SUSPENDED and raises SpawnError if its start fails. With wait=False it returns as soon as the child's task exists; a later start failure is delivered via on_child_terminated (R1 · D2).

Source code in civitas/process.py

async def spawn(
    self,
    agent_class: type,
    name: str,
    config: dict[str, Any] | None = None,
    *,
    wait: bool = True,
) -> str:
    """Spawn a dynamic agent via the nearest ancestor DynamicSupervisor.

    The nearest-ancestor special case of :meth:`spawn_into` — it resolves the
    ancestor DynamicSupervisor name wired at startup and delegates. Sends a
    civitas.dynamic.spawn message and awaits confirmation. Raises SpawnError if
    no DynamicSupervisor ancestor exists or spawn is denied. Returns the agent
    name on success.

    With ``wait=True`` (default) the call returns only after the child reaches
    RUNNING/SUSPENDED and raises SpawnError if its start fails. With
    ``wait=False`` it returns as soon as the child's task exists; a later start
    failure is delivered via ``on_child_terminated`` (R1 · D2).
    """
    if self._dynamic_supervisor_name is None:
        raise SpawnError("No DynamicSupervisor ancestor found in supervision tree")
    return await self.spawn_into(
        self._dynamic_supervisor_name, agent_class, name, config, wait=wait
    )

despawn(name) async

Hard-stop a dynamic child immediately.

Cancels the agent's task. on_stop() still fires. Pending ask() callers into the agent receive SpawnError. The slot is freed immediately.

Source code in civitas/process.py

async def despawn(self, name: str) -> None:
    """Hard-stop a dynamic child immediately.

    Cancels the agent's task. on_stop() still fires. Pending ask() callers
    into the agent receive SpawnError. The slot is freed immediately.
    """
    if self._dynamic_supervisor_name is None:
        raise SpawnError("No DynamicSupervisor ancestor found in supervision tree")
    reply = await self.ask(
        self._dynamic_supervisor_name,
        {"name": name},
        message_type="civitas.dynamic.despawn",
    )
    if reply.payload.get("status") != "ok":
        raise SpawnError(reply.payload.get("reason", "despawn failed"))

stop(name, drain='current', timeout=30.0) async

Soft-stop a dynamic child. Awaitable — returns when fully stopped.

drain="current" — finishes the message currently being handled, then stops. drain="all" — drains the full mailbox, then stops. timeout — fallback hard stop if drain isn't complete in time.

Source code in civitas/process.py

async def stop(
    self,
    name: str,
    drain: str = "current",
    timeout: float = 30.0,
) -> None:
    """Soft-stop a dynamic child. Awaitable — returns when fully stopped.

    drain="current" — finishes the message currently being handled, then stops.
    drain="all"     — drains the full mailbox, then stops.
    timeout         — fallback hard stop if drain isn't complete in time.
    """
    if self._dynamic_supervisor_name is None:
        raise SpawnError("No DynamicSupervisor ancestor found in supervision tree")
    reply = await self.ask(
        self._dynamic_supervisor_name,
        {"name": name, "drain": drain, "timeout": timeout},
        message_type="civitas.dynamic.stop",
        timeout=timeout + 5.0,
    )
    if reply.payload.get("status") != "ok":
        raise SpawnError(reply.payload.get("reason", "stop failed"))

---

civitas.process.ProcessStatus

Bases: Enum

Lifecycle states for an AgentProcess.

---

civitas.process.Mailbox(maxsize=1000)

Bounded async queue for incoming messages with priority support.

High-priority system messages (priority > 0) are placed at the front. Normal messages follow FIFO order. Backpressure is applied when the mailbox is full — the sender awaits until space is available.

Source code in civitas/process.py

def __init__(self, maxsize: int = 1000) -> None:
    self._queue: asyncio.Queue[Message] = asyncio.Queue(maxsize=maxsize)
    self._priority_queue: asyncio.Queue[Message] = asyncio.Queue(maxsize=100)
    self._notify: asyncio.Event = asyncio.Event()

put(message) async

Enqueue a message. Priority messages bypass the normal queue.

Source code in civitas/process.py

async def put(self, message: Message) -> None:
    """Enqueue a message. Priority messages bypass the normal queue."""
    if message.priority > 0:
        await self._priority_queue.put(message)
        self._notify.set()
    else:
        await self._queue.put(message)
        self._notify.set()

get() async

Dequeue the next message. Priority messages are served first.

Messages whose ttl has elapsed are discarded with a warning instead of being returned; the search continues for the next message.

Source code in civitas/process.py

async def get(self) -> Message:
    """Dequeue the next message. Priority messages are served first.

    Messages whose ttl has elapsed are discarded with a warning instead
    of being returned; the search continues for the next message.
    """
    while True:
        message: Message | None = None
        if not self._priority_queue.empty():
            message = self._priority_queue.get_nowait()
        elif not self._queue.empty():
            message = self._queue.get_nowait()

        if message is not None:
            if self._drop_if_expired(message):
                continue
            return message

        # Wait for a notification
        self._notify.clear()
        # Double-check after clearing (avoid race)
        if not self._priority_queue.empty() or not self._queue.empty():
            continue
        await self._notify.wait()

get_priority() async

Dequeue the next priority-queue message only, leaving normal messages buffered.

Used while an agent is SUSPENDED (durable-suspension S3): control messages (priority > 0) are actioned while business messages stay in the normal queue, preserving FIFO order and backpressure for resume.

Concurrency footgun (Oracle finding #10): _notify is shared and is also set by normal-queue puts. We therefore clear it and re-check the priority queue before awaiting, so a normal put only produces a bounded spurious wakeup (no busy-loop) and a priority put racing with the clear is never lost.

Source code in civitas/process.py

async def get_priority(self) -> Message:
    """Dequeue the next priority-queue message only, leaving normal messages buffered.

    Used while an agent is SUSPENDED (durable-suspension S3): control
    messages (priority > 0) are actioned while business messages stay in
    the normal queue, preserving FIFO order and backpressure for resume.

    Concurrency footgun (Oracle finding #10): ``_notify`` is shared and is
    also set by normal-queue puts. We therefore clear it and re-check the
    priority queue *before* awaiting, so a normal put only produces a
    bounded spurious wakeup (no busy-loop) and a priority put racing with
    the clear is never lost.
    """
    while True:
        if not self._priority_queue.empty():
            message = self._priority_queue.get_nowait()
            if self._drop_if_expired(message):
                continue
            return message

        self._notify.clear()
        if not self._priority_queue.empty():
            continue
        await self._notify.wait()

empty()

Return True if both priority and normal queues are empty.

Source code in civitas/process.py

def empty(self) -> bool:
    """Return True if both priority and normal queues are empty."""
    return self._priority_queue.empty() and self._queue.empty()

drain()

Remove and return all buffered messages, priority queue first.

Source code in civitas/process.py

def drain(self) -> list[Message]:
    """Remove and return all buffered messages, priority queue first."""
    drained: list[Message] = []
    while not self._priority_queue.empty():
        drained.append(self._priority_queue.get_nowait())
    while not self._queue.empty():
        drained.append(self._queue.get_nowait())
    return drained