Skip to content

Message

The standard message envelope passed between agents and through the runtime.

See Messaging for routing semantics and payload rules.

civitas.messages.Message(id=_uuid7(), type='message', sender='', recipient='', correlation_id=None, reply_to=None, payload=dict(), timestamp=_now(), trace_id='', span_id='', parent_span_id=None, attempt=0, priority=0) dataclass

Standard message envelope for all inter-agent communication.

Every message in Civitas is wrapped in this envelope. The envelope carries routing and observability metadata. The payload carries application data.

to_dict()

Convert to a plain dict for serialization.

Includes schema_version so receivers can detect and handle schema evolution. Unknown keys in received dicts are filtered by from_dict.

Source code in civitas/messages.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def to_dict(self) -> dict[str, Any]:
    """Convert to a plain dict for serialization.

    Includes ``schema_version`` so receivers can detect and handle schema
    evolution. Unknown keys in received dicts are filtered by ``from_dict``.
    """
    return {
        "schema_version": 1,
        "id": self.id,
        "type": self.type,
        "sender": self.sender,
        "recipient": self.recipient,
        "correlation_id": self.correlation_id,
        "reply_to": self.reply_to,
        "payload": self.payload,
        "timestamp": self.timestamp,
        "trace_id": self.trace_id,
        "span_id": self.span_id,
        "parent_span_id": self.parent_span_id,
        "attempt": self.attempt,
        "priority": self.priority,
    }

from_dict(data) classmethod

Reconstruct a Message from a plain dict.

Source code in civitas/messages.py 
111
112
113
114
@classmethod
def from_dict(cls, data: dict[str, Any]) -> Message:
    """Reconstruct a Message from a plain dict."""
    return cls(**{k: v for k, v in data.items() if k in _MESSAGE_FIELDS})

System message types

Messages with types prefixed _agency. are reserved for runtime internals. Application code must not use this prefix.

civitas.messages.SYSTEM_MESSAGE_TYPES = frozenset({'_agency.heartbeat', '_agency.heartbeat_ack', '_agency.shutdown', '_agency.restart', '_agency.register', '_agency.deregister'}) module-attribute