academy.message

academy/message.py

ActionRequest pydantic-model

Bases: BaseModel

Agent action request message.

Warning

The positional and keywords arguments for the invoked action are pickled and base64-encoded when serialized to JSON. This can have non-trivial time and space overheads for large arguments.

Config:

• default: DEFAULT_MUTABLE_CONFIG

Fields:

• action (str)
• pargs (SkipValidation[tuple[Any, ...]])
• kargs (SkipValidation[dict[str, Any]])
• kind (Literal['action-request'])

action pydantic-field

action: str

Name of the requested action.

pargs pydantic-field

pargs: SkipValidation[tuple[Any, ...]]

Positional arguments to the action method.

kargs pydantic-field

kargs: SkipValidation[dict[str, Any]]

Keyword arguments to the action method.

get_args

get_args() -> tuple[Any, ...]

Get the positional arguments.

Lazily deserializes and returns the positional arguments. Caches the result to avoid redundant decoding.

Returns:

• tuple[Any, ...] –

  The deserialized tuple of positional arguments.

Source code in academy/message.py

def get_args(self) -> tuple[Any, ...]:
    """Get the positional arguments.

    Lazily deserializes and returns the positional arguments.
    Caches the result to avoid redundant decoding.

    Returns:
        The deserialized tuple of positional arguments.
    """
    if isinstance(self.pargs, str):
        self.pargs = pickle.loads(base64.b64decode(self.pargs))
    return self.pargs

get_kwargs

get_kwargs() -> dict[str, Any]

Get the keyword arguments.

Lazily deserializes and returns the keyword arguments. Caches the result to avoid redundant decoding.

Returns:

• dict[str, Any] –

  The deserialized dictionary of keyword arguments.

Source code in academy/message.py

def get_kwargs(self) -> dict[str, Any]:
    """Get the keyword arguments.

    Lazily deserializes and returns the keyword arguments.
    Caches the result to avoid redundant decoding.

    Returns:
        The deserialized dictionary of keyword arguments.
    """
    if isinstance(self.kargs, str):
        self.kargs = pickle.loads(base64.b64decode(self.kargs))
    return self.kargs

PingRequest pydantic-model

Bases: BaseModel

Agent ping request message.

Config:

• default: DEFAULT_FROZEN_CONFIG

Fields:

• kind (Literal['ping-request'])

ShutdownRequest pydantic-model

Bases: BaseModel

Agent shutdown request message.

Config:

• default: DEFAULT_FROZEN_CONFIG

Fields:

• terminate (Optional[bool])
• kind (Literal['shutdown-request'])

terminate pydantic-field

terminate: Optional[bool] = None

Override the termination behavior of the agent.

ActionResponse pydantic-model

Bases: BaseModel

Agent action response message.

Warning

The result is pickled and base64-encoded when serialized to JSON. This can have non-trivial time and space overheads for large results.

Config:

• default: DEFAULT_MUTABLE_CONFIG

Fields:

• action (str)
• result (SkipValidation[Any])
• kind (Literal['action-response'])

action pydantic-field

action: str

Name of the requested action.

result pydantic-field

result: SkipValidation[Any]

Result of the action, if successful.

get_result

get_result() -> Any

Get the result.

Lazily deserializes and returns the result of the action. Caches the result to avoid redundant decoding.

Returns:

• Any –

  The deserialized result of the action.

Source code in academy/message.py

def get_result(self) -> Any:
    """Get the result.

    Lazily deserializes and returns the result of the action.
    Caches the result to avoid redundant decoding.

    Returns:
        The deserialized result of the action.
    """
    if (
        isinstance(self.result, list)
        and len(self.result) == 2  # noqa PLR2004
        and self.result[0] == '__pickled__'
    ):
        self.result = pickle.loads(base64.b64decode(self.result[1]))
    return self.result

ErrorResponse pydantic-model

Bases: BaseModel

Error response message.

Contains the exception raised by a failed request.

Config:

• default: DEFAULT_MUTABLE_CONFIG

Fields:

• exception (SkipValidation[Exception])
• kind (Literal['error-response'])

exception pydantic-field

exception: SkipValidation[Exception]

Exception of the failed request.

get_exception

get_exception() -> Exception

Get the exception.

Lazily deserializes and returns the exception object. Caches the result to avoid redundant decoding.

Returns:

• Exception –

  The deserialized exception.

Source code in academy/message.py

def get_exception(self) -> Exception:
    """Get the exception.

    Lazily deserializes and returns the exception object.
    Caches the result to avoid redundant decoding.

    Returns:
        The deserialized exception.
    """
    if isinstance(self.exception, str):
        self.exception = pickle.loads(base64.b64decode(self.exception))
    return self.exception

SuccessResponse pydantic-model

Bases: BaseModel

Success response message.

Config:

• default: DEFAULT_FROZEN_CONFIG

Fields:

• kind (Literal['success-response'])

Header pydantic-model

Bases: BaseModel

Message metadata header.

Contains information about the sender, receiver, and message context.

Config:

• default: DEFAULT_FROZEN_CONFIG

Fields:

• src (EntityId)
• dest (EntityId)
• tag (UUID)
• label (Optional[UUID])
• kind (Literal['request', 'response'])

src pydantic-field

src: EntityId

Message source ID.

dest pydantic-field

dest: EntityId

Message destination ID.

tag pydantic-field

tag: UUID

Unique message tag used to match requests and responses.

label pydantic-field

label: Optional[UUID] = None

Optional label used to disambiguate response messages when multiple objects (i.e., handles) share the same mailbox. This is a different usage from the tag.

create_response_header

create_response_header() -> Self

Create a response header based on the current request header.

Swaps the source and destination, retains the tag and label, and sets the kind to 'response'.

Returns:

• Self –

  A new header instance with reversed roles.

Raises:

• ValueError –

  If the current header is already a response.

Source code in academy/message.py

def create_response_header(self) -> Self:
    """Create a response header based on the current request header.

    Swaps the source and destination, retains the tag and label,
    and sets the kind to 'response'.

    Returns:
        A new header instance with reversed roles.

    Raises:
        ValueError: If the current header is already a response.
    """
    if self.kind == 'response':
        raise ValueError(
            'Cannot create response header from another response header',
        )
    return type(self)(
        tag=self.tag,
        src=self.dest,
        dest=self.src,
        label=self.label,
        kind='response',
    )

Message pydantic-model

Bases: BaseModel, Generic[BodyT]

A complete message with header and body.

Wraps a header and a typed request/response body. Supports lazy deserialization of message bodies, metadata access, and convenient construction.

Note

The body value is ignored when testing equality or hashing an instance because the body value could be in either a serialized or deserialized state until get_body() is called.

Config:

• default: DEFAULT_MUTABLE_CONFIG

Fields:

• header (Header)
• body (SkipValidation[BodyT])

src property

src: EntityId

Message source ID.

dest property

dest: EntityId

Message destination ID.

tag property

tag: UUID

Message tag.

label property

label: UUID | None

Message label.

create classmethod

create(
    src: EntityId,
    dest: EntityId,
    body: BodyT,
    *,
    label: UUID | None = None
) -> Message[BodyT]

Create a new message with the specified header and body.

Parameters:

• src (EntityId) –

  Source entity ID.

• dest (EntityId) –

  Destination entity ID.

• body (BodyT) –

  Message body.

• label (UUID | None, default: None ) –

  Optional label for disambiguation.

Returns:

• Message[BodyT] –

  A new message instance.

Source code in academy/message.py

@classmethod
def create(
    cls,
    src: EntityId,
    dest: EntityId,
    body: BodyT,
    *,
    label: uuid.UUID | None = None,
) -> Message[BodyT]:
    """Create a new message with the specified header and body.

    Args:
        src: Source entity ID.
        dest: Destination entity ID.
        body: Message body.
        label: Optional label for disambiguation.

    Returns:
        A new message instance.
    """
    if isinstance(body, get_args(Request)):
        kind = 'request'
    elif isinstance(body, get_args(Response)):
        kind = 'response'
    else:
        raise AssertionError('Unreachable.')
    header = Header(src=src, dest=dest, label=label, kind=kind)
    request: Message[BodyT] = Message(header=header, body=body)
    return request

create_response

create_response(body: ResponseT) -> Message[ResponseT]

Create a response message from this request message.

Parameters:

• body (ResponseT) –

  Response message body.

Returns:

• Message[ResponseT] –

  A new response message instance.

Raises:

• ValueError –

  If this message is already a response.

Source code in academy/message.py

def create_response(self, body: ResponseT) -> Message[ResponseT]:
    """Create a response message from this request message.

    Args:
        body: Response message body.

    Returns:
        A new response message instance.

    Raises:
        ValueError: If this message is already a response.
    """
    header = self.header.create_response_header()
    response: Message[ResponseT] = Message(header=header, body=body)
    return response

get_body

get_body() -> BodyT

Return the message body, deserializing if needed.

Lazily deserializes and returns the body object. Caches the body to avoid redundant decoding.

Returns:

• BodyT –

  The deserialized body.

Source code in academy/message.py

def get_body(self) -> BodyT:
    """Return the message body, deserializing if needed.

    Lazily deserializes and returns the body object.
    Caches the body to avoid redundant decoding.

    Returns:
        The deserialized body.
    """
    if isinstance(self.body, get_args(Body)):
        return self.body

    adapter: TypeAdapter[BodyT] = TypeAdapter(Body)
    body = (
        adapter.validate_json(self.body)
        if isinstance(self.body, str)
        else adapter.validate_python(self.body)
    )
    self.body = body
    return self.body

is_request

is_request() -> bool

Check if the message is a request.

Source code in academy/message.py

def is_request(self) -> bool:
    """Check if the message is a request."""
    return self.header.kind == 'request'

is_response

is_response() -> bool

Check if the message is a response.

Source code in academy/message.py

def is_response(self) -> bool:
    """Check if the message is a response."""
    return self.header.kind == 'response'

model_deserialize classmethod

model_deserialize(data: bytes) -> Message[BodyT]

Deserialize a message from bytes using pickle.

Warning

This uses pickle and is therefore susceptible to all the typical pickle warnings about code injection.

Parameters:

• data (bytes) –

  The serialized message as bytes.

Returns:

• Message[BodyT] –

  The deserialized message instance.

Raises:

• TypeError –

  If the deserialized object is not a Message.

Source code in academy/message.py

@classmethod
def model_deserialize(cls, data: bytes) -> Message[BodyT]:
    """Deserialize a message from bytes using pickle.

    Warning:
        This uses pickle and is therefore susceptible to all the
        typical pickle warnings about code injection.

    Args:
        data: The serialized message as bytes.

    Returns:
        The deserialized message instance.

    Raises:
        TypeError: If the deserialized object is not a Message.
    """
    message = pickle.loads(data)
    if not isinstance(message, cls):
        raise TypeError(
            'Deserialized message is not of type Message.',
        )
    return message

model_serialize

model_serialize() -> bytes

Serialize the message to bytes using pickle.

Warning

This uses pickle and is therefore susceptible to all the typical pickle warnings about code injection.

Returns:

• bytes –

  The serialized message as bytes.

Source code in academy/message.py

def model_serialize(self) -> bytes:
    """Serialize the message to bytes using pickle.

    Warning:
        This uses pickle and is therefore susceptible to all the
        typical pickle warnings about code injection.

    Returns:
        The serialized message as bytes.
    """
    return pickle.dumps(self)