src.uagents.context

Agent Context and Message Handling

Context Objects ↗ (opens in a new tab)


class Context(ABC)

Represents the context in which messages are handled and processed.

Properties: agent (AgentRepresentation): The agent representation associated with the context. storage (KeyValueStore): The key-value store for storage operations. ledger (LedgerClient): The client for interacting with the blockchain ledger. logger (logging.Logger): The logger instance. session (uuid.UUID): The session UUID associated with the context.

Methods:

get_agents_by_protocol(protocol_digest, limit, logger): Retrieve a list of agent addresses using a specific protocol digest. broadcast(destination_protocol, message, limit, timeout): Broadcast a message to agents with a specific protocol. 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.

agent ↗ (opens in a new tab)


@property
@abstractmethod
def agent() -> AgentRepresentation

Get the agent representation associated with the context.

Returns:

AgentRepresentation - The agent representation.

storage ↗ (opens in a new tab)


@property
@abstractmethod
def storage() -> KeyValueStore

Get the key-value store associated with the context.

Returns:

KeyValueStore - The key-value store.

ledger ↗ (opens in a new tab)


@property
@abstractmethod
def ledger() -> LedgerClient

Get the ledger client associated with the context.

Returns:

LedgerClient - The ledger client.

logger ↗ (opens in a new tab)


@property
@abstractmethod
def logger() -> logging.Logger

Get the logger instance associated with the context.

Returns:

logging.Logger - The logger instance.

session ↗ (opens in a new tab)


@property
@abstractmethod
def session() -> uuid.UUID

Get the session UUID associated with the context.

Returns:

uuid.UUID - The session UUID.

get_agents_by_protocol ↗ (opens in a new tab)


@abstractmethod
def get_agents_by_protocol(
        protocol_digest: str,
        limit: int = DEFAULT_SEARCH_LIMIT,
        logger: Optional[logging.Logger] = 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.

broadcast ↗ (opens in a new tab)


@abstractmethod
async def broadcast(
        destination_protocol: str,
        message: Model,
        limit: int = DEFAULT_SEARCH_LIMIT,
        timeout: int = DEFAULT_ENVELOPE_TIMEOUT_SECONDS) -> List[MsgStatus]

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:

List[MsgStatus] - A list of message delivery statuses.

send ↗ (opens in a new tab)


@abstractmethod
async def send(destination: str,
               message: Model,
               sync: bool = False,
               timeout: int = DEFAULT_ENVELOPE_TIMEOUT_SECONDS) -> MsgStatus

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.
sync bool - Whether to send the message synchronously or asynchronously.
timeout Optional[int] - The optional timeout for sending the message, in seconds.

Returns:

MsgStatus - The delivery status of the message.

send_raw ↗ (opens in a new tab)


@abstractmethod
async def send_raw(
        destination: str,
        message_schema_digest: str,
        message_body: JsonStr,
        sync: bool = False,
        timeout: int = DEFAULT_ENVELOPE_TIMEOUT_SECONDS,
        protocol_digest: Optional[str] = None,
        queries: Optional[Dict[str, asyncio.Future]] = None) -> MsgStatus

Send a message to the specified destination where the message body and message schema digest are sent separately.

Arguments:

destination str - The destination address to send the message to.
message_schema_digest str - The schema digest of the message to be sent.
message_body JsonStr - The JSON-encoded message body to be sent.
sync bool - Whether to send the message synchronously or asynchronously.
timeout Optional[int] - The optional timeout for sending the message, in seconds.
protocol_digest Optional[str] - The protocol digest of the message to be sent.
queries Optional[Dict[str, asyncio.Future]] - The dictionary of queries to resolve.

Returns:

MsgStatus - The delivery status of the message.

send_wallet_message ↗ (opens in a new tab)


@abstractmethod
async def send_wallet_message(destination: str, text: str, msg_type: int = 1)

Send a message to the wallet of the specified destination.

Arguments:

destination str - The destination address to send the message to.
text str - The text of the message to be sent.
msg_type int - The type of the message to be sent.

Returns:

None

InternalContext Objects ↗ (opens in a new tab)


class InternalContext(Context)

Represents the agent internal context for proactive behaviour.

session ↗ (opens in a new tab)


@property
def session() -> uuid.UUID

Get the session UUID associated with the context.

Returns:

uuid.UUID - The session UUID.

outbound_messages ↗ (opens in a new tab)


@property
def outbound_messages() -> Dict[str, Tuple[JsonStr, str]]

Get the dictionary of outbound messages associated with the context.

Returns:

Dict[str, Tuple[JsonStr, str]]: The dictionary of outbound messages.

address ↗ (opens in a new tab)


@property
@deprecated("Please use `ctx.agent.address` instead.")
def address() -> str

Get the agent address associated with the context. This is a deprecated property and will be removed in a future release. Please use the ctx.agent.address property instead.

Returns:

str - The agent address.

send ↗ (opens in a new tab)


async def send(destination: str,
               message: Model,
               sync: bool = False,
               timeout: int = DEFAULT_ENVELOPE_TIMEOUT_SECONDS) -> MsgStatus

This is the pro-active send method which is used in on_event and on_interval methods. In these methods, interval messages are set but we don't have access properties that are only necessary in re-active contexts, like 'replies', 'message_received', or 'protocol'.

ExternalContext Objects ↗ (opens in a new tab)


class ExternalContext(InternalContext)

Represents the reactive context in which messages are handled and processed.

Attributes:

_queries Dict[str, asyncio.Future] - Dictionary mapping query senders to their response Futures.
_replies Optional[Dict[str, Dict[str, Type[Model]]]] - Dictionary of allowed reply digests for each type of incoming message.
_message_received Optional[MsgDigest] - The message digest received.
_protocol Optional[Tuple[str, Protocol]] - The supported protocol digest and the corresponding protocol.

init ↗ (opens in a new tab) ↗ (opens in a new tab)


def __init__(message_received: MsgDigest,
             queries: Optional[Dict[str, asyncio.Future]] = None,
             replies: Optional[Dict[str, Dict[str, Type[Model]]]] = None,
             protocol: Optional[Tuple[str, Protocol]] = None,
             **kwargs)

Initialize the ExternalContext instance and attributes needed from the InternalContext.

Arguments:

message_received MsgDigest - The optional message digest received.
queries Dict[str, asyncio.Future] - Dictionary mapping query senders to their response Futures.
replies Optional[Dict[str, Dict[str, Type[Model]]]] - Dictionary of allowed replies for each type of incoming message.
protocol Optional[Tuple[str, Protocol]] - The optional Tuple of protocols.

send ↗ (opens in a new tab)


async def send(destination: str,
               message: Model,
               sync: bool = False,
               timeout: int = DEFAULT_ENVELOPE_TIMEOUT_SECONDS) -> MsgStatus

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.
sync bool - Whether to send the message synchronously or asynchronously.
timeout Optional[int] - The optional timeout for sending the message, in seconds.

Returns:

MsgStatus - The delivery status of the message.

Was this page helpful?

You can also leave detailed feedbackon Github

Config Dispatch