Runtime

Assembles and manages the full Civitas runtime. Wires transport, registry, serializer, tracer, plugins, and the supervision tree.

See Deployment and Topology & CLI for usage.

---

civitas.runtime.Runtime(supervisor=None, transport='in_process', serializer=None, model_provider=None, tool_registry=None, state_store=None, metrics=None, exporters=None, zmq_pub_addr='tcp://127.0.0.1:5559', zmq_sub_addr='tcp://127.0.0.1:5560', zmq_start_proxy=True, nats_servers='nats://localhost:4222', nats_jetstream=False, nats_stream_name='AGENCY', components=None)

Assembles and manages the full Civitas runtime.

Startup sequence (from Implementation Guide §3): 1. Read configuration 2. Create Serializer 3. Create Tracer 4. Create Transport 5. Create Registry 6. Create MessageBus 7. Create plugin instances 8. Instantiate / wire all AgentProcesses 9. Register all AgentProcesses in Registry 10. Start Transport 11. Walk supervision tree bottom-up, start each agent 12. Start all Supervisors 13. Runtime is ready

Source code in civitas/runtime.py

def __init__(
    self,
    supervisor: Supervisor | None = None,
    transport: str = "in_process",
    serializer: Serializer | None = None,
    model_provider: Any = None,
    tool_registry: Any = None,
    state_store: Any = None,
    metrics: Any = None,
    exporters: list[Any] | None = None,
    zmq_pub_addr: str = "tcp://127.0.0.1:5559",
    zmq_sub_addr: str = "tcp://127.0.0.1:5560",
    zmq_start_proxy: bool = True,
    nats_servers: str | list[str] = "nats://localhost:4222",
    nats_jetstream: bool = False,
    nats_stream_name: str = "AGENCY",
    components: ComponentSet | None = None,
) -> None:
    self._root_supervisor = supervisor
    self._transport_type = transport
    self._custom_serializer = serializer
    self._model_provider = model_provider
    self._tool_registry = tool_registry
    self._state_store = state_store
    self._metrics = metrics
    self._exporters = exporters or []
    self._components = components
    self._otel_agent_task: asyncio.Task[None] | None = None

    # ZMQ-specific config
    self._zmq_pub_addr = zmq_pub_addr
    self._zmq_sub_addr = zmq_sub_addr
    self._zmq_start_proxy = zmq_start_proxy

    # NATS-specific config
    self._nats_servers = nats_servers
    self._nats_jetstream = nats_jetstream
    self._nats_stream_name = nats_stream_name

    # MCP server configs parsed from topology YAML
    self._mcp_configs: list[Any] = []

    # Security config — populated by from_config() when a security: block is present
    self._security_config: Any = None
    self._topology_public_keys: dict[str, str] = {}

    # Signing infrastructure — built in start() when signing is enabled on a
    # distributed transport; used to verify cross-process announcements (R6 · D8).
    self._key_registry: Any = None
    self._message_signer: Any = None
    self._signing_on: bool = False

    # Per-agent credentials — populated by from_config() from credentials: blocks
    self._agent_credentials: dict[str, dict[str, str]] = {}

    # Per-agent capabilities — populated by from_config() from capabilities: blocks
    self._agent_capabilities: dict[str, tuple[list[str], dict[str, Any]]] = {}

    # Audit sink — populated by from_config() when an audit: block is present
    self._audit_sink: Any = None

    # Transport security (ZMQ CURVE / NATS TLS) — populated by from_config()
    self._transport_security: Any = None

    # Set during start() — exposed for stop(), ask()/send(), and get_agent()
    self._serializer: Serializer | None = None
    self._tracer: Any = None
    self._transport: Any = None
    self._registry: Any = None
    self._bus: Any = None
    self._agents_by_name: dict[str, AgentProcess] = {}  # F04-10: O(1) live process lookup
    self._started = False

from_config(path, agent_classes=None) classmethod

Build a Runtime from a YAML topology file.

The agent_classes dict maps type strings (e.g. "MyAgent") to the actual Python class. If not provided, types are resolved via importlib from dotted module paths (e.g. "myapp.agents.MyAgent").

Source code in civitas/runtime.py

@classmethod
def from_config(
    cls,
    path: str | Path,
    agent_classes: dict[str, type[AgentProcess]] | None = None,
) -> Runtime:
    """Build a Runtime from a YAML topology file.

    The ``agent_classes`` dict maps type strings (e.g. "MyAgent") to the
    actual Python class. If not provided, types are resolved via
    ``importlib`` from dotted module paths (e.g. "myapp.agents.MyAgent").
    """
    config = yaml.safe_load(Path(path).read_text())
    config = substitute_vars(config)
    return cls.from_config_dict(config, agent_classes=agent_classes)

start() async

Start the runtime following the canonical initialization sequence.

Source code in civitas/runtime.py

async def start(self) -> None:
    """Start the runtime following the canonical initialization sequence."""
    if self._started:
        return

    # Steps 2–6: build or use provided ComponentSet.
    # Note: if a pre-built ComponentSet is provided, its transport must support
    # being started by this call — transport.start() is always called below. (F04-11)
    if self._components is not None:
        cs = self._components
    else:
        ts = self._transport_security
        cs = build_component_set(
            transport_type=self._transport_type,
            serializer=self._custom_serializer,
            model_provider=self._model_provider,
            tool_registry=self._tool_registry,
            state_store=self._state_store,
            audit_sink=self._audit_sink,
            metrics=self._metrics,
            exporters=self._exporters,
            zmq_pub_addr=self._zmq_pub_addr,
            zmq_sub_addr=self._zmq_sub_addr,
            zmq_start_proxy=self._zmq_start_proxy,
            zmq_curve_config=ts.zmq if ts is not None and ts.zmq.enabled else None,
            nats_servers=self._nats_servers,
            nats_jetstream=self._nats_jetstream,
            nats_stream_name=self._nats_stream_name,
            nats_tls_config=ts.nats if ts is not None and ts.nats.enabled else None,
        )

    # Expose on self for stop(), ask(), send(), and get_agent()
    self._serializer = cs.serializer
    self._tracer = cs.tracer
    self._transport = cs.transport
    self._registry = cs.registry
    self._bus = cs.bus
    self._state_store = cs.store

    # Drain span_queue via OTELAgent when exporters are configured (FD-07/FD-09)
    if cs.span_queue is not None and cs.export_backend is not None:
        self._otel_agent_task = asyncio.create_task(
            run_otel_agent(cs.span_queue, cs.export_backend)
        )

    if self._root_supervisor is None:
        self._started = True
        return

    # 8. Inject dependencies into all AgentProcesses
    all_agents = self._root_supervisor.all_agents()

    # Security: build signing infrastructure if configured for non-InProcess transports.
    # InProcess transport skips signing entirely (D9 — same OS process, no wire to protect).
    if (
        self._security_config is not None
        and self._security_config.signing.enabled
        and self._transport_type != "in_process"
    ):
        key_dir = self._security_config.identity.key_dir
        identities: dict[str, AgentIdentity] = {}
        for agent in all_agents:
            # A DynamicSupervisor needs an identity to sign the cluster-wide
            # child announcements it publishes (R6 · D8); only TopologyServer
            # (read-only, never signs) is exempt.
            if isinstance(agent, TopologyServer):
                continue
            if self._security_config.identity.mode == "auto":
                identities[agent.name] = AgentIdentity.load_or_generate(agent.name, key_dir)
            else:
                identities[agent.name] = AgentIdentity.load(agent.name, key_dir)

        registry = KeyRegistry()
        for name, identity in identities.items():
            registry.register(name, identity.verify_key)
        for name, pub_b64 in self._topology_public_keys.items():
            if name not in registry:
                registry.register_b64(name, pub_b64)

        signer = MessageSigner(identities, registry, self._security_config.signing)
        signing_ser = SigningSerializer(signer, self._security_config.signing)
        self._serializer = signing_ser
        cs.bus._serializer = signing_ser
        self._key_registry = registry
        self._message_signer = signer
        self._signing_on = True

    for agent in all_agents:
        cs.inject(agent)
        # Wire per-agent credentials from topology credentials: blocks
        if self._agent_credentials:
            agent._credentials = self._agent_credentials.get(agent.name, {})

    # Inject into supervisors (supervisor-specific wiring, not via ComponentSet)
    for sup in self._root_supervisor.all_supervisors():
        sup._bus = cs.bus
        sup._registry = cs.registry
        sup._tracer = cs.tracer

    # Wire _dynamic_supervisor_name for all agents based on the static topology.
    # Each agent receives the name of the nearest DynamicSupervisor in its
    # ancestor-or-sibling subtree, enabling self.spawn() without explicit naming.
    def _wire_dyn_sup(
        node: Supervisor | AgentProcess,
        nearest_dyn: str | None,
    ) -> None:
        if isinstance(node, DynamicSupervisor):
            node._dynamic_supervisor_name = node.name  # spawns into itself
        elif isinstance(node, Supervisor):
            dyn_child = next(
                (c for c in node.children if isinstance(c, DynamicSupervisor)), None
            )
            new_nearest = dyn_child.name if dyn_child is not None else nearest_dyn
            for child in node.children:
                _wire_dyn_sup(child, new_nearest)
        else:
            node._dynamic_supervisor_name = nearest_dyn

    _wire_dyn_sup(self._root_supervisor, None)

    # 9. Register all AgentProcesses in Registry; build O(1) name→process map (F04-10)
    for agent in all_agents:
        yaml_caps = self._agent_capabilities.get(agent.name)
        if yaml_caps is not None:
            caps, meta = yaml_caps
        else:
            caps = list(agent.capabilities)
            meta = dict(agent.capability_metadata)
        if isinstance(agent, DynamicSupervisor) and DYNAMIC_SUPERVISOR_CAPABILITY not in caps:
            caps = [*caps, DYNAMIC_SUPERVISOR_CAPABILITY]
        self._registry.register(agent.name, capabilities=caps, capability_metadata=meta)
    self._agents_by_name = {a.name: a for a in all_agents}

    # Inject topology server references before supervision tree starts
    for agent in all_agents:
        if isinstance(agent, TopologyServer):
            agent._root_supervisor = self._root_supervisor
            agent._agents = self._agents_by_name

    # 10. Start Transport
    await self._transport.start()

    # Set up transport subscriptions for each agent
    for agent in all_agents:
        await self._bus.setup_agent(agent)

    # Subscribe to cross-process agent announcements from Worker processes.
    # Workers publish _agency.register on startup so this runtime's bus can
    # route messages to remote agents without a shared registry service.
    await self._transport.subscribe("_agency.register", self._on_remote_register)
    await self._transport.subscribe("_agency.deregister", self._on_remote_deregister)

    # Wait for subscriptions to propagate (ZMQ slow joiner mitigation)
    if hasattr(self._transport, "wait_ready"):
        await self._transport.wait_ready()

    # Connect MCP servers declared in topology YAML to all agents
    if self._mcp_configs:
        for agent in all_agents:
            for mcp_cfg in self._mcp_configs:
                try:
                    await agent.connect_mcp(mcp_cfg)
                except Exception as exc:
                    logger.warning(
                        "Failed to connect agent '%s' to MCP server '%s': %s",
                        agent.name,
                        mcp_cfg.name,
                        exc,
                    )

    # 11-12. Start supervision tree (supervisors start their children)
    await self._root_supervisor.start()

    # 13. Runtime is ready
    self._started = True

stop() async

Shutdown sequence: stop agents, transport, flush tracer.

Source code in civitas/runtime.py

async def stop(self) -> None:
    """Shutdown sequence: stop agents, transport, flush tracer."""
    if not self._started:
        return

    # 1. Stop supervision tree (sends shutdown, awaits on_stop for each agent)
    if self._root_supervisor is not None:
        await self._root_supervisor.stop()

    # 2. Stop Transport
    if self._transport is not None:
        await self._transport.stop()

    # 3. Close StateStore
    if self._state_store is not None and hasattr(self._state_store, "close"):
        await self._state_store.close()

    # 4. Flush Tracer
    if self._tracer is not None:
        self._tracer.flush()

    # 5. Stop OTELAgent — cancel triggers its own drain-remaining-spans logic
    if self._otel_agent_task is not None:
        self._otel_agent_task.cancel()
        await asyncio.gather(self._otel_agent_task, return_exceptions=True)
        self._otel_agent_task = None

    # 6. Close Audit sink
    if self._audit_sink is not None:
        await self._audit_sink.close()

    self._agents_by_name.clear()
    self._started = False

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

Send a message to an agent and await a reply.

Source code in civitas/runtime.py

async def ask(
    self,
    agent_name: str,
    payload: dict[str, Any],
    timeout: float = 30.0,
    message_type: str = "message",
) -> Message:
    """Send a message to an agent and await a reply."""
    if self._bus is None or self._tracer is None:
        raise RuntimeError("Runtime not started")

    trace_id = self._tracer.new_trace_id()
    message = Message(
        type=message_type,
        sender="_runtime",
        recipient=agent_name,
        payload=payload,
        correlation_id=_uuid7(),
        trace_id=trace_id,
        span_id=_new_span_id(),
    )
    return cast(Message, await self._bus.request(message, timeout=timeout))

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

Fire-and-forget: send a message to an agent.

Source code in civitas/runtime.py

async def send(
    self,
    agent_name: str,
    payload: dict[str, Any],
    message_type: str = "message",
) -> None:
    """Fire-and-forget: send a message to an agent."""
    if self._bus is None or self._tracer is None:
        raise RuntimeError("Runtime not started")

    trace_id = self._tracer.new_trace_id()
    message = Message(
        type=message_type,
        sender="_runtime",
        recipient=agent_name,
        payload=payload,
        trace_id=trace_id,
        span_id=_new_span_id(),
    )
    await self._bus.route(message)

get_agent(name)

Return the live AgentProcess instance by name, or None.

O(1) lookup via the agents-by-name dict built during start(). Use this when you need to inspect process state (e.g. status). For routing messages use runtime.send/ask instead.

Source code in civitas/runtime.py

def get_agent(self, name: str) -> AgentProcess | None:
    """Return the live AgentProcess instance by name, or None.

    O(1) lookup via the agents-by-name dict built during start().
    Use this when you need to inspect process state (e.g. status).
    For routing messages use runtime.send/ask instead.
    """
    return self._agents_by_name.get(name)

all_agents()

Return all AgentProcess instances in the supervision tree.

Source code in civitas/runtime.py

def all_agents(self) -> list[AgentProcess]:
    """Return all AgentProcess instances in the supervision tree."""
    if self._root_supervisor is None:
        return []
    return self._root_supervisor.all_agents()

print_tree()

Return an ASCII representation of the supervision tree.

Source code in civitas/runtime.py

def print_tree(self) -> str:
    """Return an ASCII representation of the supervision tree."""
    if self._root_supervisor is None:
        return "(no supervision tree)"

    lines: list[str] = []

    def _walk(node: Supervisor | AgentProcess, prefix: str, is_last: bool) -> None:
        connector = "└── " if is_last else "├── "
        if isinstance(node, Supervisor):
            label = f"[sup] {node.name} ({node.strategy.value})"
        else:
            status = node.status.value if hasattr(node, "status") else "?"
            if isinstance(node, DynamicSupervisor):
                prefix_tag = "[dyn]"
            elif isinstance(node, TopologyServer):
                prefix_tag = "[topo]"
            elif isinstance(node, EvalAgent):
                prefix_tag = "[eval]"
            elif isinstance(node, HTTPGateway):
                prefix_tag = "[http]"
            elif isinstance(node, GenServer):
                prefix_tag = "[srv]"
            else:
                prefix_tag = "[agent]"
            label = f"{prefix_tag} {node.name} ({status})"
        lines.append(f"{prefix}{connector}{label}")

        if isinstance(node, Supervisor):
            child_prefix = prefix + ("    " if is_last else "│   ")
            for i, child in enumerate(node.children):
                _walk(child, child_prefix, i == len(node.children) - 1)

    # Root
    root = self._root_supervisor
    lines.append(f"[sup] {root.name} ({root.strategy.value})")
    for i, child in enumerate(root.children):
        _walk(child, "", i == len(root.children) - 1)

    return "\n".join(lines)

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

Spawn a dynamic agent via the named DynamicSupervisor.

Returns the agent name on success. Raises SpawnError on failure.

With wait=False the call returns as soon as the child's task exists, before on_start() completes; a later start failure is delivered to the spawner via on_child_terminated (R1 · D2).

Source code in civitas/runtime.py

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

    Returns the agent name on success. Raises SpawnError on failure.

    With ``wait=False`` the call returns as soon as the child's task exists,
    before ``on_start()`` completes; a later start failure is delivered to the
    spawner via ``on_child_terminated`` (R1 · D2).
    """
    class_path = f"{agent_class.__module__}.{agent_class.__qualname__}"
    reply = await self.ask(
        supervisor_name,
        {
            "class_path": class_path,
            "name": name,
            "config": config or {},
            "spawner": "_runtime",
            "wait": wait,
            "spawn_id": _uuid7(),
        },
        message_type="civitas.dynamic.spawn",
    )
    if reply.payload.get("status") != "ok":
        reason = reply.payload.get("reason") or reply.payload.get("error") or "spawn failed"
        raise SpawnError(reason)
    return name

despawn(supervisor_name, name) async

Hard-stop a dynamic child via the named DynamicSupervisor.

Source code in civitas/runtime.py

async def despawn(self, supervisor_name: str, name: str) -> None:
    """Hard-stop a dynamic child via the named DynamicSupervisor."""
    reply = await self.ask(
        supervisor_name,
        {"name": name},
        message_type="civitas.dynamic.despawn",
    )
    if reply.payload.get("status") != "ok":
        raise SpawnError(reply.payload.get("reason", "despawn failed"))

stop_agent(supervisor_name, name, drain='current', timeout=30.0) async

Soft-stop a dynamic child via the named DynamicSupervisor.

Source code in civitas/runtime.py

async def stop_agent(
    self,
    supervisor_name: str,
    name: str,
    drain: str = "current",
    timeout: float = 30.0,
) -> None:
    """Soft-stop a dynamic child via the named DynamicSupervisor."""
    reply = await self.ask(
        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.components.ComponentSet(transport, registry, serializer, tracer, store=None, model_provider=None, tool_registry=None, audit_sink=None, metrics=None, span_queue=None, export_backend=None) dataclass

Assembled infrastructure wiring for a single Runtime or Worker.

MessageBus is derived automatically from the other four fields in post_init — callers should not construct it separately.

Attributes:

Name	Type	Description
transport	Any	Transport layer (InProcess, ZMQ, or NATS).
registry	Registry	LocalRegistry for agent name → address mapping.
serializer	Serializer	Serializer used by transport and bus.
tracer	Tracer	Tracer instance for span emission.
store	Any	StateStore for agent checkpoint/restore. None means no persistence; callers should default to InMemoryStateStore when appropriate.
model_provider	Any	Injected into agent.llm at startup.
tool_registry	Any	Injected into agent.tools at startup.
bus	MessageBus	MessageBus built from the other four fields.

inject(agent)

Inject bus and plugin references into an agent process.

Source code in civitas/components.py

def inject(self, agent: AgentProcess) -> None:
    """Inject bus and plugin references into an agent process."""
    agent._bus = self.bus
    agent._tracer = self.tracer
    agent._registry = self.registry
    agent.llm = self.model_provider
    agent.tools = self.tool_registry
    agent.store = self.store
    agent._audit_sink = self.audit_sink
    agent._metrics = self.metrics

---

civitas.components.build_component_set(transport_type='in_process', serializer=None, model_provider=None, tool_registry=None, state_store=None, audit_sink=None, metrics=None, exporters=None, zmq_pub_addr='tcp://127.0.0.1:5559', zmq_sub_addr='tcp://127.0.0.1:5560', zmq_start_proxy=True, zmq_curve_config=None, nats_servers='nats://localhost:4222', nats_jetstream=False, nats_stream_name='AGENCY', nats_tls_config=None)

Build a ComponentSet from primitive configuration values.

Called by Runtime.start() and Worker.start() when no pre-built ComponentSet is provided. Handles serializer selection, transport construction, and store defaulting.

Source code in civitas/components.py

def build_component_set(
    transport_type: str = "in_process",
    serializer: Serializer | None = None,
    model_provider: Any = None,
    tool_registry: Any = None,
    state_store: Any = None,
    audit_sink: AuditSink | None = None,
    metrics: MetricsSink | None = None,
    exporters: list[Any] | None = None,
    zmq_pub_addr: str = "tcp://127.0.0.1:5559",
    zmq_sub_addr: str = "tcp://127.0.0.1:5560",
    zmq_start_proxy: bool = True,
    zmq_curve_config: ZmqCurveConfig | None = None,
    nats_servers: str | list[str] = "nats://localhost:4222",
    nats_jetstream: bool = False,
    nats_stream_name: str = "AGENCY",
    nats_tls_config: NatsTlsConfig | None = None,
) -> ComponentSet:
    """Build a ComponentSet from primitive configuration values.

    Called by Runtime.start() and Worker.start() when no pre-built
    ComponentSet is provided. Handles serializer selection, transport
    construction, and store defaulting.
    """
    # Serializer
    if serializer is not None:
        built_serializer = serializer
    elif settings.serializer == "json":
        built_serializer = JsonSerializer()
    else:
        built_serializer = MsgpackSerializer()

    # Tracer — when exporters are configured (plugins.exporters in topology YAML),
    # spans flow through SpanQueue -> OTELAgent -> ExportBackend instead of the
    # Tracer's own direct OTLP path, so the two never run simultaneously (FD-09).
    # Runtime/Worker are responsible for starting/stopping the OTELAgent task.
    built_span_queue: Any = None
    built_export_backend: Any = None
    if exporters:
        from civitas.observability.export_backend import FanOutBackend
        from civitas.observability.span_queue import SpanQueue

        built_span_queue = SpanQueue()
        built_export_backend = FanOutBackend(exporters) if len(exporters) > 1 else exporters[0]
    built_tracer = Tracer(span_queue=built_span_queue)

    # Transport — imports are intentionally scoped here: ZMQ and NATS are optional
    # extras (pyzmq, nats-py) that may not be installed. Importing at module level
    # would cause ImportError on every civitas import for users without those extras.
    built_transport: Transport
    if transport_type == "zmq":
        from civitas.transport.zmq import ZMQTransport

        built_transport = ZMQTransport(
            built_serializer,
            pub_addr=zmq_pub_addr,
            sub_addr=zmq_sub_addr,
            start_proxy=zmq_start_proxy,
            curve_config=zmq_curve_config,
        )
    elif transport_type == "nats":
        from civitas.transport.nats import NATSTransport

        built_transport = NATSTransport(
            built_serializer,
            servers=nats_servers,
            jetstream=nats_jetstream,
            stream_name=nats_stream_name,
            tls_config=nats_tls_config,
        )
    else:
        from civitas.transport.inprocess import InProcessTransport

        built_transport = InProcessTransport(built_serializer)

    # Registry
    built_registry = LocalRegistry()

    # State store default
    if state_store is None:
        state_store = InMemoryStateStore()

    return ComponentSet(
        transport=built_transport,
        registry=built_registry,
        serializer=built_serializer,
        tracer=built_tracer,
        store=state_store,
        model_provider=model_provider,
        tool_registry=tool_registry,
        audit_sink=audit_sink,
        metrics=metrics,
        span_queue=built_span_queue,
        export_backend=built_export_backend,
    )