Skip to content

aea.agent

This module contains the implementation of a generic agent.

Agent Objects

class Agent(AbstractAgent, WithLogger)

This class provides an abstract base class for a generic agent.

__init__

def __init__(identity: Identity,
             connections: List[Connection],
             loop: Optional[AbstractEventLoop] = None,
             period: float = 1.0,
             loop_mode: Optional[str] = None,
             runtime_mode: Optional[str] = None,
             storage_uri: Optional[str] = None,
             logger: Logger = _default_logger,
             task_manager_mode: Optional[str] = None) -> None

Instantiate the agent.

Arguments:

  • identity: the identity of the agent.
  • connections: the list of connections of the agent.
  • loop: the event loop to run the connections.
  • period: period to call agent's act
  • loop_mode: loop_mode to choose agent run loop.
  • runtime_mode: runtime mode to up agent.
  • storage_uri: optional uri to set generic storage
  • task_manager_mode: task manager mode.
  • logger: the logger.
  • task_manager_mode: mode of the task manager.

storage_uri

@property
def storage_uri() -> Optional[str]

Return storage uri.

is_running

@property
def is_running() -> bool

Get running state of the runtime and agent.

is_stopped

@property
def is_stopped() -> bool

Get running state of the runtime and agent.

identity

@property
def identity() -> Identity

Get the identity.

inbox

@property
def inbox() -> InBox

Get the inbox.

The inbox contains Envelopes from the Multiplexer. The agent can pick these messages for processing.

Returns:

InBox instance

outbox

@property
def outbox() -> OutBox

Get the outbox.

The outbox contains Envelopes for the Multiplexer. Envelopes placed in the Outbox are processed by the Multiplexer.

Returns:

OutBox instance

name

@property
def name() -> str

Get the agent name.

tick

@property
def tick() -> int

Get the tick or agent loop count.

Each agent loop (one call to each one of act(), react(), update()) increments the tick.

Returns:

tick count

state

@property
def state() -> RuntimeStates

Get state of the agent's runtime.

Returns:

RuntimeStates

period

@property
def period() -> float

Get a period to call act.

runtime

@property
def runtime() -> BaseRuntime

Get the runtime.

setup

def setup() -> None

Set up the agent.

start

def start() -> None

Start the agent.

Performs the following:

  • calls start() on runtime.
  • waits for runtime to complete running (blocking)

handle_envelope

def handle_envelope(envelope: Envelope) -> None

Handle an envelope.

Arguments:

  • envelope: the envelope to handle.

act

def act() -> None

Perform actions on period.

stop

def stop() -> None

Stop the agent.

Performs the following:

  • calls stop() on runtime
  • waits for runtime to stop (blocking)

teardown

def teardown() -> None

Tear down the agent.

get_periodic_tasks

def get_periodic_tasks(
) -> Dict[Callable, Tuple[float, Optional[datetime.datetime]]]

Get all periodic tasks for agent.

Returns:

dict of callable with period specified

get_message_handlers

def get_message_handlers() -> List[Tuple[Callable[[Any], None], Callable]]

Get handlers with message getters.

Returns:

List of tuples of callables: handler and coroutine to get a message

exception_handler

def exception_handler(exception: Exception, function: Callable) -> bool

Handle exception raised during agent main loop execution.

Arguments:

  • exception: exception raised
  • function: a callable exception raised in.

Returns:

bool, propagate exception if True otherwise skip it.