Agent

Agent Objects

class Agent(Sink)

An agent that interacts within a communication environment.

Attributes:

Properties:

__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,
             version: Optional[str] = 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.
• version Optional[str] - The version of the agent.

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.

wallet

@property
def wallet() -> LocalWallet

Get the wallet of the agent.

Returns:

• LocalWallet - The agent's wallet.

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 (deprecated and replaced by agentverse).

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() -> MailboxClient

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

Returns:

• MailboxClient - The mailbox client instance.

mailbox

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

Set the mailbox configuration for the agent (deprecated and replaced by agentverse).

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() -> str

Sign the registration data for Almanac contract.

Returns:

• str - The signature of the registration data.

Raises:

• AssertionError - If the Almanac contract address is None.

update_endpoints

def update_endpoints(endpoints: List[Dict[str, Any]])

Update the list of endpoints.

Arguments:

• endpoints List[Dict[str, Any]] - 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.

register

async def register()

Register with the Almanac contract.

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[Model, Set[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.

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.

setup

def setup()

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

start_background_tasks

def start_background_tasks()

Start background tasks for the agent.

run

def run()

Run the agent.

Bureau Objects

class Bureau()

A class representing a Bureau of agents.

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

Arguments:

• port Optional[int] - The port number for the server.
• endpoint Optional[Union[str, List[str], Dict[str, dict]]] - Configuration for agent endpoints.

Attributes:

• _loop asyncio.AbstractEventLoop - The event loop.
• _agents List[Agent] - The list of agents contained in 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.
• _use_mailbox bool - A flag indicating whether mailbox functionality is enabled for any of the agents.

__init__

def __init__(port: Optional[int] = None,
             endpoint: Optional[Union[str, List[str], Dict[str,
                                                           dict]]] = None)

Initialize a Bureau instance.

Arguments:

• port Optional[int] - The port on which the bureau's server will run.
• endpoint Optional[Union[str, List[str], Dict[str, dict]]] - The endpoint configuration for the bureau.

add

def add(agent: Agent)

Add an agent to the bureau.

Arguments:

• agent Agent - The agent to be added.

run

def run()

Run the agents managed by the bureau.