src.uagents.agent

Agent

AgentRepresentation Objects


class AgentRepresentation()

Represents an agent in the context of a message.

Attributes:

_address str - The address of the agent.
_name Optional[str] - The name of the agent.
_signing_callback Callable - The callback for signing messages.

Properties:
name str - The name of the agent.
address str - The address of the agent.
identifier str - The agent’s address and network prefix.

Methods:

sign_digest(data - bytes) -> str: Sign the provided data with the agent’s identity.

init


def __init__(address: str, name: Optional[str], signing_callback: Callable)

Initialize the AgentRepresentation instance.

Arguments:

address str - The address of the context.
name Optional[str] - The optional name associated with the context.
signing_callback Callable - The callback for signing messages.

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.

identifier


@property
def identifier() -> str

Get the address of the agent used for communication including the network prefix.

Returns:

str - The agent’s address and network prefix.

sign_digest


def sign_digest(data: bytes) -> str

Sign the provided data with the callback of the agent’s identity.

Arguments:

data bytes - The data to sign.

Returns:

str - The signature of the data.

Agent Objects


class Agent(Sink)

An agent that interacts within a communication environment.

Attributes:

_name str - The name of the agent.
_port int - The port on which the agent’s server runs.
_background_tasks Set[asyncio.Task] - Set of background tasks associated with the agent.
_resolver Resolver - The resolver for agent communication.
_loop asyncio.AbstractEventLoop - The asyncio event loop used by the agent.
_logger - The logger instance for logging agent activities.
_endpoints List[AgentEndpoint] - List of endpoints at which the agent is reachable.
_use_mailbox bool - Indicates if the agent uses a mailbox for communication.
_agentverse dict - Agentverse configuration settings.
_mailbox_client MailboxClient - The client for interacting with the agentverse mailbox.
_ledger - The client for interacting with the blockchain ledger.
_almanac_contract - The almanac contract for registering agent addresses to endpoints.
_storage - Key-value store for agent data storage.
_interval_handlers List[Tuple[IntervalCallback, float]] - List of interval handlers and their periods.
_interval_messages Set[str] - Set of message digests that may be sent by interval tasks.
_signed_message_handlers Dict[str, MessageCallback] - Handlers for signed messages.
_unsigned_message_handlers Dict[str, MessageCallback] - Handlers for unsigned messages.
_message_cache EnvelopeHistory - History of messages received by the agent.
_models Dict[str, Type[Model]] - Dictionary mapping supported message digests to messages.
_replies Dict[str, Dict[str, Type[Model]]] - Dictionary of allowed replies for each type of incoming message.
_queries Dict[str, asyncio.Future] - Dictionary mapping query senders to their response Futures.
_dispatcher - The dispatcher for internal handling/sorting of messages.
_dispenser - The dispatcher for external message handling.
_message_queue - Asynchronous queue for incoming messages.
_on_startup List[Callable] - List of functions to run on agent startup.
_on_shutdown List[Callable] - List of functions to run on agent shutdown.
_version str - The version of the agent.
_protocol Protocol - The internal agent protocol consisting of all interval and message handlers assigned with agent decorators.
protocols Dict[str, Protocol] - Dictionary mapping all supported protocol digests to their corresponding protocols.
_ctx Context - The context for agent interactions.
_test bool - True if the agent will register and transact on the testnet.
_enable_agent_inspector bool - Enable the agent inspector REST endpoints.
_metadata Dict[str, Any] - Metadata associated with the agent.

Properties:
name str - The name of the agent.
address str - The address of the agent used for communication.
identifier str - The Agent Identifier, including network prefix and address.
wallet LocalWallet - The agent’s wallet for transacting on the ledger.
storage KeyValueStore - The key-value store for storage operations.
mailbox Dict[str, str] - The mailbox configuration for the agent.
agentverse Dict[str, str] - The agentverse configuration for the agent.
mailbox_client MailboxClient - The client for interacting with the agentverse mailbox.
protocols Dict[str, Protocol] - Dictionary mapping all supported protocol digests to their corresponding protocols.
metadata Optional[Dict[str, Any]] - Metadata associated with the agent.

init


def __init__(name: Optional[str] = None,
             port: Optional[int] = None,
             seed: Optional[str] = None,
             endpoint: Optional[Union[str, List[str], Dict[str, dict]]] = None,
             agentverse: Optional[Union[str, Dict[str, str]]] = None,
             mailbox: Optional[Union[str, Dict[str, str]]] = None,
             resolve: Optional[Resolver] = None,
             registration_policy: Optional[AgentRegistrationPolicy] = None,
             enable_wallet_messaging: Union[bool, Dict[str, str]] = False,
             wallet_key_derivation_index: Optional[int] = 0,
             max_resolver_endpoints: Optional[int] = None,
             version: Optional[str] = None,
             test: bool = True,
             loop: Optional[asyncio.AbstractEventLoop] = None,
             log_level: Union[int, str] = logging.INFO,
             enable_agent_inspector: bool = True,
             metadata: Optional[Dict[str, Any]] = None)

Initialize an Agent instance.

Arguments:

name Optional[str] - The name of the agent.
port Optional[int] - The port on which the agent’s server will run.
seed Optional[str] - The seed for generating keys.
endpoint Optional[Union[str, List[str], Dict[str, dict]]] - The endpoint configuration.
agentverse Optional[Union[str, Dict[str, str]]] - The agentverse configuration.
mailbox Optional[Union[str, Dict[str, str]]] - The mailbox configuration.
resolve Optional[Resolver] - The resolver to use for agent communication.
enable_wallet_messaging Optional[Union[bool, Dict[str, str]]] - Whether to enable wallet messaging. If ’{“chain_id”: CHAIN_ID}’ is provided, this sets the chain ID for the messaging server.
wallet_key_derivation_index Optional[int] - The index used for deriving the wallet key.
max_resolver_endpoints Optional[int] - The maximum number of endpoints to resolve.
version Optional[str] - The version of the agent.
test Optional[bool] - True if the agent will register and transact on the testnet.
loop Optional[asyncio.AbstractEventLoop] - The asyncio event loop to use.
log_level Union[int, str] - The logging level for the agent.
enable_agent_inspector bool - Enable the agent inspector for debugging.
metadata Optional[Dict[str, Any]] - Optional metadata to include in the agent object.

initialize_wallet_messaging


def initialize_wallet_messaging(enable_wallet_messaging: Union[bool,
                                                               Dict[str,
                                                                    str]])

Initialize wallet messaging for the agent.

Arguments:

enable_wallet_messaging Union[bool, Dict[str, str]] - Wallet messaging configuration.

name


@property
def name() -> str

Get the name of the agent.

Returns:

str - The name of the agent.

address


@property
def address() -> str

Get the address of the agent used for communication.

Returns:

str - The agent’s address.

identifier


@property
def identifier() -> str

Get the Agent Identifier, including network prefix and address.

Returns:

str - The agent’s identifier.

wallet


@property
def wallet() -> LocalWallet

Get the wallet of the agent.

Returns:

LocalWallet - The agent’s wallet.

ledger


@property
def ledger() -> LedgerClient

Get the ledger of the agent.

Returns:

LedgerClient - The agent’s ledger

storage


@property
def storage() -> KeyValueStore

Get the key-value store used by the agent for data storage.

Returns:

KeyValueStore - The key-value store instance.

mailbox


@property
def mailbox() -> Dict[str, str]

Get the mailbox configuration of the agent. Agentverse overrides it but mailbox is kept for backwards compatibility.

Returns:

Dict[str, str]: The mailbox configuration.

agentverse


@property
def agentverse() -> Dict[str, str]

Get the agentverse configuration of the agent.

Returns:

Dict[str, str]: The agentverse configuration.

mailbox_client


@property
def mailbox_client() -> Optional[MailboxClient]

Get the mailbox client used by the agent for mailbox communication.

Returns:

Optional[MailboxClient] - The mailbox client instance.

balance


@property
def balance() -> int

Get the balance of the agent.

Returns:

int - Bank balance.

info


@property
def info() -> AgentInfo

Get basic information about the agent.

Returns:

AgentInfo - The agent’s address, endpoints, protocols, and metadata.

metadata


@property
def metadata() -> Dict[str, Any]

Get the metadata associated with the agent.

Returns:

Dict[str, Any]: The metadata associated with the agent.

mailbox


@mailbox.setter
def mailbox(config: Union[str, Dict[str, str]])

Set the mailbox configuration for the agent. Agentverse overrides it but mailbox is kept for backwards compatibility.

Arguments:

config Union[str, Dict[str, str]] - The new mailbox configuration.

agentverse


@agentverse.setter
def agentverse(config: Union[str, Dict[str, str]])

Set the agentverse configuration for the agent.

Arguments:

config Union[str, Dict[str, str]] - The new agentverse configuration.

sign


def sign(data: bytes) -> str

Sign the provided data.

Arguments:

data bytes - The data to be signed.

Returns:

str - The signature of the data.

sign_digest


def sign_digest(digest: bytes) -> str

Sign the provided digest.

Arguments:

digest bytes - The digest to be signed.

Returns:

str - The signature of the digest.

sign_registration


def sign_registration(timestamp: int,
                      sender_wallet_address: Optional[str] = None) -> str

Sign the registration data for Almanac contract.

Arguments:

timestamp int - The timestamp for the registration.
sender_wallet_address Optional[str] - The wallet address of the transaction sender.

Returns:

str - The signature of the registration data.

Raises:

AssertionError - If the Almanac contract is None.

update_endpoints


def update_endpoints(endpoints: List[AgentEndpoint])

Update the list of endpoints.

Arguments:

endpoints List[AgentEndpoint] - List of endpoint dictionaries.

update_loop


def update_loop(loop)

Update the event loop.

Arguments:

loop - The event loop.

update_queries


def update_queries(queries)

Update the queries attribute.

Arguments:

queries - The queries attribute.

update_registration_policy


def update_registration_policy(policy: AgentRegistrationPolicy)

Update the registration policy.

Arguments:

policy - The registration policy.

register


async def register()

This method checks for registration conditions and performs registration if necessary.

on_interval


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

Decorator to register an interval handler for the provided period.

Arguments:

period float - The interval period.
messages Optional[Union[Type[Model], Set[Type[Model]]]] - Optional message types.

Returns:

Callable - The decorator function for registering interval handlers.

on_query


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

Set up a query event with a callback.

Arguments:

model Type[Model] - The query model.
replies Optional[Union[Model, Set[Model]]] - Optional reply models.

Returns:

Callable - The decorator function for registering query handlers.

on_message


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

Decorator to register an message handler for the provided message model.

Arguments:

model Type[Model] - The message model.
replies Optional[Union[Type[Model], Set[Type[Model]]]] - Optional reply models.
allow_unverified Optional[bool] - Allow unverified messages.

Returns:

Callable - The decorator function for registering message handlers.

on_event


def on_event(event_type: str)

Decorator to register an event handler for a specific event type.

Arguments:

event_type str - The type of event.

Returns:

Callable - The decorator function for registering event handlers.

on_wallet_message


def on_wallet_message()

Add a handler for wallet messages.

include


def include(protocol: Protocol, publish_manifest: Optional[bool] = False)

Include a protocol into the agent’s capabilities.

Arguments:

protocol Protocol - The protocol to include.
publish_manifest Optional[bool] - Flag to publish the protocol’s manifest.

Raises:

RuntimeError - If a duplicate model, signed message handler, or message handler is encountered.

publish_manifest


def publish_manifest(manifest: Dict[str, Any])

Publish a protocol manifest to the Almanac service.

Arguments:

manifest Dict[str, Any] - The protocol manifest.

handle_message


async def handle_message(sender, schema_digest: str, message: JsonStr,
                         session: uuid.UUID)

Handle an incoming message.

Arguments:

sender - The sender of the message.
schema_digest str - The digest of the message schema.
message JsonStr - The message content in JSON format.
session uuid.UUID - The session UUID.

handle_rest


async def handle_rest(
        method: RestMethod, endpoint: str,
        message: Optional[Model]) -> Optional[Union[Dict[str, Any], Model]]

Handle a REST request.

Arguments:

method RestMethod - The REST method.
endpoint str - The REST endpoint.
message Model - The message content.

setup


async def setup()

Include the internal agent protocol, run startup tasks, and start background tasks.

start_registration_loop


def start_registration_loop()

Start the registration loop.

start_message_dispenser


def start_message_dispenser()

Start the message dispenser.

start_interval_tasks


def start_interval_tasks()

Start interval tasks for the agent.

start_message_receivers


def start_message_receivers()

Start message receiving tasks for the agent.

start_server


async def start_server()

Start the agent’s server.

run_async


async def run_async()

Create all tasks for the agent.

run


def run()

Run the agent by itself. A fresh event loop is created for the agent and it is closed after the agent stops.

get_message_protocol


def get_message_protocol(
        message_schema_digest) -> Optional[Tuple[str, Protocol]]

Get the protocol for a given message schema digest.

Bureau Objects


class Bureau()

A class representing a Bureau of agents.

This class manages a collection of agents and orchestrates their execution.

Attributes:

_loop asyncio.AbstractEventLoop - The event loop.
_agents List[Agent] - The list of agents to be managed by the bureau.
_endpoints List[Dict[str, Any]] - The endpoint configuration for the bureau.
_port int - The port on which the bureau’s server runs.
_queries Dict[str, asyncio.Future] - Dictionary mapping query senders to their response Futures.
_logger Logger - The logger instance.
_server ASGIServer - The ASGI server instance for handling requests.
_agentverse Dict[str, str] - The agentverse configuration for the bureau.
_use_mailbox bool - A flag indicating whether mailbox functionality is enabled for any of the agents.
_registration_policy AgentRegistrationPolicy - The registration policy for the bureau.

init


def __init__(agents: Optional[List[Agent]] = None,
             port: Optional[int] = None,
             endpoint: Optional[Union[str, List[str], Dict[str, dict]]] = None,
             agentverse: Optional[Union[str, Dict[str, str]]] = None,
             registration_policy: Optional[BatchRegistrationPolicy] = None,
             ledger: Optional[LedgerClient] = None,
             wallet: Optional[LocalWallet] = None,
             seed: Optional[str] = None,
             test: bool = True,
             loop: Optional[asyncio.AbstractEventLoop] = None,
             log_level: Union[int, str] = logging.INFO)

Initialize a Bureau instance.

Arguments:

agents Optional[List[Agent]] - The list of agents to be managed by the bureau.
port Optional[int] - The port number for the server.
endpoint Optional[Union[str, List[str], Dict[str, dict]]] - The endpoint configuration.
agentverse Optional[Union[str, Dict[str, str]]] - The agentverse configuration.
registration_policy Optional[BatchRegistrationPolicy] - The registration policy.
ledger Optional[LedgerClient] - The ledger for the bureau.
wallet Optional[LocalWallet] - The wallet for the bureau (overrides ‘seed’).
seed Optional[str] - The seed phrase for the wallet (overridden by ‘wallet’).
test Optional[bool] - True if the bureau will register and transact on the testnet.
loop Optional[asyncio.AbstractEventLoop] - The event loop.
log_level Union[int, str] - The logging level for the bureau.

add


def add(agent: Agent)

Add an agent to the bureau.

Arguments:

agent Agent - The agent to be added.

run_async


async def run_async()

Run the agents managed by the bureau.

run


def run()

Run the bureau.