Skip to content

aea.launcher

This module contains the implementation of multiple AEA configs launcher.

load_agent

def load_agent(agent_dir: Union[PathLike, str],
               password: Optional[str] = None) -> AEA

Load AEA from directory.

Arguments:

  • agent_dir: agent configuration directory
  • password: the password to encrypt/decrypt the private key.

Returns:

AEA instance

AEADirTask Objects

class AEADirTask(AbstractExecutorTask)

Task to run agent from agent configuration directory.

__init__

def __init__(agent_dir: Union[PathLike, str],
             password: Optional[str] = None) -> None

Init AEA config dir task.

Arguments:

  • agent_dir: directory with AEA config.
  • password: the password to encrypt/decrypt the private key.

id

@property
def id() -> Union[PathLike, str]

Return agent_dir.

start

def start() -> None

Start task.

stop

def stop() -> None

Stop task.

create_async_task

def create_async_task(loop: AbstractEventLoop) -> TaskAwaitable

Return asyncio Task for task run in asyncio loop.

AEADirMultiprocessTask Objects

class AEADirMultiprocessTask(AbstractMultiprocessExecutorTask)

Task to run agent from agent configuration directory.

Version for multiprocess executor mode.

__init__

def __init__(agent_dir: Union[PathLike, str],
             log_level: Optional[str] = None,
             password: Optional[str] = None) -> None

Init AEA config dir task.

Arguments:

  • agent_dir: directory with AEA config.
  • log_level: debug level applied for AEA in subprocess
  • password: the password to encrypt/decrypt the private key.

id

@property
def id() -> Union[PathLike, str]

Return agent_dir.

failed

@property
def failed() -> bool

Return was exception failed or not.

If it's running it's not failed.

Returns:

bool

start

def start() -> Tuple[Callable, Sequence[Any]]

Return function and arguments to call within subprocess.

stop

def stop() -> None

Stop task.

AEALauncher Objects

class AEALauncher(AbstractMultipleRunner)

Run multiple AEA instances.

__init__

def __init__(agent_dirs: Sequence[Union[PathLike, str]],
             mode: str,
             fail_policy: ExecutorExceptionPolicies = ExecutorExceptionPolicies
             .propagate,
             log_level: Optional[str] = None,
             password: Optional[str] = None) -> None

Init AEALauncher.

Arguments:

  • agent_dirs: sequence of AEA config directories.
  • mode: executor name to use.
  • fail_policy: one of ExecutorExceptionPolicies to be used with Executor
  • log_level: debug level applied for AEA in subprocesses
  • password: the password to encrypt/decrypt the private key.