uAgent Context and Message Handling

MsgDigest Objects

@dataclass
class MsgDigest()

Represents a message digest containing a message and its schema digest.

Attributes:

Context Objects

class Context()

Represents the context in which messages are handled and processed.

Attributes:

Properties:

Methods:

• get_message_protocol(message_schema_digest): Get the protocol associated with a message schema digest.
• send(destination, message, timeout): Send a message to a destination.
• send_raw(destination, json_message, schema_digest, message_type, timeout): Send a message with the provided schema digest to a destination.
• experimental_broadcast(destination_protocol, message, limit, timeout): Broadcast a message to agents with a specific protocol.

__init__

def __init__(address: str,
             name: Optional[str],
             storage: KeyValueStore,
             resolve: Resolver,
             identity: Identity,
             wallet: LocalWallet,
             ledger: LedgerClient,
             queries: Dict[str, asyncio.Future],
             session: Optional[uuid.UUID] = None,
             replies: Optional[Dict[str, Set[Type[Model]]]] = None,
             interval_messages: Optional[Set[str]] = None,
             message_received: Optional[MsgDigest] = None,
             protocols: Optional[Dict[str, Protocol]] = None,
             logger: Optional[logging.Logger] = None)

Initialize the Context instance.

Arguments:

• address str - The address of the context.
• name Optional[str] - The optional name associated with the context.
• storage KeyValueStore - The key-value store for storage operations.
• resolve Resolver - The resolver for name-to-address resolution.
• identity Identity - The identity associated with the context.
• wallet LocalWallet - The local wallet instance for managing identities.
• ledger LedgerClient - The ledger client for interacting with distributed ledgers.
• queries Dict[str, asyncio.Future] - Dictionary mapping query senders to their response Futures.
• session Optional[uuid.UUID] - The optional session UUID.
• replies Optional[Dict[str, Set[Type[Model]]]] - Optional dictionary of reply models.
• interval_messages Optional[Set[str]] - The optional set of interval messages.
• message_received Optional[MsgDigest] - The optional message digest received.
• protocols Optional[Dict[str, Protocol]] - The optional dictionary of protocols.
• logger Optional[logging.Logger] - The optional logger instance.

name

@property
def name() -> str

Get the name associated with the context or a truncated address if name is None.

Returns:

• str - The name or truncated address.

address

@property
def address() -> str

Get the address of the context.

Returns:

• str - The address of the context.

logger

@property
def logger() -> logging.Logger

Get the logger instance associated with the context.

Returns:

• logging.Logger - The logger instance.

protocols

@property
def protocols() -> Optional[Dict[str, Protocol]]

Get the dictionary of protocols associated with the context.

Returns:

Optional[Dict[str, Protocol]]: The dictionary of protocols.

session

@property
def session() -> uuid.UUID

Get the session UUID associated with the context.

Returns:

• uuid.UUID - The session UUID.

get_message_protocol

def get_message_protocol(message_schema_digest) -> Optional[str]

Get the protocol associated with a given message schema digest.

Arguments:

• message_schema_digest str - The schema digest of the message.

Returns:

• Optional[str] - The protocol digest associated with the message schema digest, or None if not found.

get_agents_by_protocol

def get_agents_by_protocol(protocol_digest: str,
                           limit: Optional[int] = None) -> List[str]

Retrieve a list of agent addresses using a specific protocol digest.

This method queries the Almanac API to retrieve a list of agent addresses that are associated with a given protocol digest. The list can be optionally limited to a specified number of addresses.

Arguments:

• protocol_digest str - The protocol digest to search for, starting with "proto:".
• limit int, optional - The maximum number of agent addresses to return.

Returns:

• List[str] - A list of agent addresses using the specified protocol digest.

send

async def send(destination: str,
               message: Model,
               timeout: Optional[int] = DEFAULT_ENVELOPE_TIMEOUT_SECONDS)

Send a message to the specified destination.

Arguments:

• destination str - The destination address to send the message to.
• message Model - The message to be sent.
• timeout Optional[int] - The optional timeout for sending the message, in seconds.

experimental_broadcast

async def experimental_broadcast(
        destination_protocol: str,
        message: Model,
        limit: Optional[int] = DEFAULT_SEARCH_LIMIT,
        timeout: Optional[int] = DEFAULT_ENVELOPE_TIMEOUT_SECONDS)

Broadcast a message to agents with a specific protocol.

This asynchronous method broadcasts a given message to agents associated with a specific protocol. The message is sent to multiple agents concurrently. The schema digest of the message is used for verification.

Arguments:

• destination_protocol str - The protocol to filter agents by.
• message Model - The message to broadcast.
• limit int, optional - The maximum number of agents to send the message to.
• timeout int, optional - The timeout for sending each message.

Returns:

None

send_raw

async def send_raw(destination: str,
                   json_message: JsonStr,
                   schema_digest: str,
                   message_type: Optional[Type[Model]] = None,
                   timeout: Optional[int] = DEFAULT_ENVELOPE_TIMEOUT_SECONDS)

Send a raw message to the specified destination.

Arguments:

• destination str - The destination address to send the message to.
• json_message JsonStr - The JSON-encoded message to be sent.
• schema_digest str - The schema digest of the message.
• message_type Optional[Type[Model]] - The optional type of the message being sent.
• timeout Optional[int] - The optional timeout for sending the message, in seconds.