src.uagents.communication

Agent dispatch of exchange envelopes and synchronous messages.

Dispenser Objects ↗ (opens in a new tab)

class Dispenser()

Dispenses messages externally.

add_envelope ↗ (opens in a new tab)

def add_envelope(envelope: Envelope,
                 endpoints: List[str],
                 response_future: asyncio.Future,
                 sync: bool = False)

Add an envelope to the dispenser.

Arguments:

• envelope Envelope - The envelope to send.
• endpoints List[str] - The endpoints to send the envelope to.
• response_future asyncio.Future - The future to set the response on.
• sync bool, optional - True if the message is synchronous. Defaults to False.

run ↗ (opens in a new tab)

async def run()

Run the dispenser routine.

dispatch_local_message ↗ (opens in a new tab)

async def dispatch_local_message(sender: str, destination: str,
                                 schema_digest: str, message: JsonStr,
                                 session_id: uuid.UUID) -> MsgStatus

Process a message locally.

send_exchange_envelope ↗ (opens in a new tab)

async def send_exchange_envelope(
        envelope: Envelope,
        endpoints: List[str],
        sync: bool = False) -> Union[MsgStatus, Envelope]

Method to send an exchange envelope.

Arguments:

• envelope Envelope - The envelope to send.
• resolver Optional[Resolver], optional - The resolver to use. Defaults to None.
• sync bool, optional - True if the message is synchronous. Defaults to False.

Returns:

Union[MsgStatus, Envelope]: Either the status of the message or the response envelope.

dispatch_sync_response_envelope ↗ (opens in a new tab)

async def dispatch_sync_response_envelope(
        env: Envelope) -> Union[MsgStatus, Envelope]

Dispatch a synchronous response envelope locally.

send_message_raw ↗ (opens in a new tab)

async def send_message_raw(
        destination: str,
        message_schema_digest: str,
        message_body: JsonStr,
        response_type: Optional[Type[Model]] = None,
        sender: Optional[Union[Identity, str]] = None,
        resolver: Optional[Resolver] = None,
        timeout: int = DEFAULT_ENVELOPE_TIMEOUT_SECONDS,
        sync: bool = False) -> Union[Model, JsonStr, MsgStatus, Envelope]

Standalone function to send a message to an agent.

Arguments:

• destination str - The destination address to send the message to.
• message_schema_digest str - The schema digest of the message.
• message_body JsonStr - The JSON-formatted message to be sent.
• response_type Optional[Type[Model]] - The optional type of the response message.
• sender Optional[Union[Identity, str]] - The optional sender identity or user address.
• resolver Optional[Resolver] - The optional resolver for address-to-endpoint resolution.
• timeout int - The timeout for the message response in seconds. Defaults to 30.
• sync bool - True if the message is synchronous.

Returns:

Union[Model, JsonStr, MsgStatus, Envelope]: On success, if the response type is provided, the response message is returned with that type. Otherwise, the JSON message is returned. If the sender is a user address, the response envelope is returned. On failure, a message status is returned.

send_message ↗ (opens in a new tab)

async def send_message(
        destination: str,
        message: Model,
        response_type: Optional[Type[Model]] = None,
        sender: Optional[Union[Identity, str]] = None,
        resolver: Optional[Resolver] = None,
        timeout: int = DEFAULT_ENVELOPE_TIMEOUT_SECONDS,
        sync: bool = False) -> Union[Model, JsonStr, MsgStatus, Envelope]

Standalone function to send a message to an agent.

Arguments:

• destination str - The destination address to send the message to.
• message Model - The message to be sent.
• response_type Optional[Type[Model]] - The optional type of the response message.
• sender Optional[Union[Identity, str]] - The optional sender identity or user address.
• resolver Optional[Resolver] - The optional resolver for address-to-endpoint resolution.
• timeout int - The timeout for the message response in seconds. Defaults to 30.
• sync bool - True if the message is synchronous.

Returns:

Union[Model, JsonStr, MsgStatus, Envelope]: On success, if the response type is provided, the response message is returned with that type. Otherwise, the JSON message is returned. If the sender is a user address, the response envelope is returned. On failure, a message status is returned.

send_sync_message ↗ (opens in a new tab)

async def send_sync_message(
    destination: str,
    message: Model,
    response_type: Optional[Type[Model]] = None,
    sender: Optional[Union[Identity, str]] = None,
    resolver: Optional[Resolver] = None,
    timeout: int = DEFAULT_ENVELOPE_TIMEOUT_SECONDS
) -> Union[Model, JsonStr, MsgStatus, Envelope]

Standalone function to send a synchronous message to an agent.

Arguments:

• destination str - The destination address to send the message to.
• message Model - The message to be sent.
• response_type Optional[Type[Model]] - The optional type of the response message.
• sender Optional[Union[Identity, str]] - The optional sender identity or user address.
• resolver Optional[Resolver] - The optional resolver for address-to-endpoint resolution.
• timeout int - The timeout for the message response in seconds. Defaults to 30.
• sync bool - True if the message is synchronous.

Returns:

Union[Model, JsonStr, MsgStatus, Envelope]: On success, if the response type is provided, the response message is returned with that type. Otherwise, the JSON message is returned. If the sender is a user address, the response envelope is returned. On failure, a message status is returned.

enclose_response ↗ (opens in a new tab)

def enclose_response(message: Model,
                     sender: str,
                     session: UUID4,
                     target: str = "") -> JsonStr

Enclose a response message within an envelope.

Arguments:

• message Model - The response message to enclose.
• sender str - The sender's address.
• session str - The session identifier.
• target str - The target address.

Returns:

• str - The JSON representation of the response envelope.

enclose_response_raw ↗ (opens in a new tab)

def enclose_response_raw(json_message: JsonStr,
                         schema_digest: str,
                         sender: str,
                         session: UUID4,
                         target: str = "") -> JsonStr

Enclose a raw response message within an envelope.

Arguments:

• json_message JsonStr - The JSON-formatted response message to enclose.
• schema_digest str - The schema digest of the message.
• sender str - The sender's address.
• session UUID4 - The session identifier.
• target str - The target address.

Returns:

• str - The JSON representation of the response envelope.