src.uagents.protocol

Exchange Protocol

Protocol Objects ↗ (opens in a new tab)

class Protocol()

The Protocol class encapsulates a particular set of functionalities for an agent. It typically relates to the exchange of messages between agents for executing some task. It includes the message (model) types it supports, the allowed replies, and the interval message handlers that define the logic of the protocol.

init ↗ (opens in a new tab)

def __init__(name: Optional[str] = None, version: Optional[str] = None)

Initialize a Protocol instance.

Arguments:

• name Optional[str], optional - The name of the protocol. Defaults to None.
• version Optional[str], optional - The version of the protocol. Defaults to None.

intervals ↗ (opens in a new tab)

@property
def intervals()

Property to access the interval handlers.

Returns:

List[Tuple[IntervalCallback, float]]: List of interval handlers and their periods.

models ↗ (opens in a new tab)

@property
def models()

Property to access the registered models.

Returns:

Dict[str, Type[Model]]: Dictionary of registered models with schema digests as keys.

replies ↗ (opens in a new tab)

@property
def replies()

Property to access the registered replies.

Returns:

Dict[str, Dict[str, Type[Model]]]: Dictionary mapping message schema digests to their allowed replies.

interval_messages ↗ (opens in a new tab)

@property
def interval_messages()

Property to access the interval message digests.

Returns:

• Set[str] - Set of message digests that may be sent by interval handlers.

signed_message_handlers ↗ (opens in a new tab)

@property
def signed_message_handlers()

Property to access the signed message handlers.

Returns:

Dict[str, MessageCallback]: Dictionary mapping message schema digests to their handlers.

unsigned_message_handlers ↗ (opens in a new tab)

@property
def unsigned_message_handlers()

Property to access the unsigned message handlers.

Returns:

Dict[str, MessageCallback]: Dictionary mapping message schema digests to their handlers.

name ↗ (opens in a new tab)

@property
def name()

Property to access the protocol name.

Returns:

• str - The protocol name.

version ↗ (opens in a new tab)

@property
def version()

Property to access the protocol version.

Returns:

• str - The protocol version.

canonical_name ↗ (opens in a new tab)

@property
def canonical_name()

Property to access the canonical name of the protocol ('name:version').

Returns:

• str - The canonical name of the protocol.

digest ↗ (opens in a new tab)

@property
def digest()

Property to access the digest of the protocol's manifest.

Returns:

• str - The digest of the protocol's manifest.

on_interval ↗ (opens in a new tab)

def on_interval(period: float,
                messages: Optional[Union[Type[Model],
                                         Set[Type[Model]]]] = None)

Decorator to register an interval handler for the protocol.

Arguments:

• period float - The interval period in seconds.
• messages Optional[Union[Type[Model], Set[Type[Model]]]], optional - The associated message types. Defaults to None.

Returns:

• Callable - The decorator to register the interval handler.

on_query ↗ (opens in a new tab)

def on_query(model: Type[Model],
             replies: Optional[Union[Type[Model], Set[Type[Model]]]] = None)

Decorator to register a query handler for the protocol.

Arguments:

• model Type[Model] - The message model type.
• replies Optional[Union[Type[Model], Set[Type[Model]]]], optional - The associated reply types. Defaults to None.

Returns:

• Callable - The decorator to register the query handler.

on_message ↗ (opens in a new tab)

def on_message(model: Type[Model],
               replies: Optional[Union[Type[Model], Set[Type[Model]]]] = None,
               allow_unverified: Optional[bool] = False)

Decorator to register a message handler for the protocol.

Arguments:

• model Type[Model] - The message model type.
• replies Optional[Union[Type[Model], Set[Type[Model]]]], optional - The associated reply types. Defaults to None.
• allow_unverified Optional[bool], optional - Whether to allow unverified messages. Defaults to False.

Returns:

• Callable - The decorator to register the message handler.

manifest ↗ (opens in a new tab)

def manifest() -> Dict[str, Any]

Generate the protocol's manifest, a long-form machine readable description of the protocol details and interface.

Returns:

Dict[str, Any]: The protocol's manifest.

compute_digest ↗ (opens in a new tab)

@staticmethod
def compute_digest(manifest: Dict[str, Any]) -> str

Compute the digest of a given manifest.

Arguments:

• manifest Dict[str, Any] - The manifest to compute the digest for.

Returns:

• str - The computed digest.