aea.manager.manager
This module contains the implementation of AEA agents manager.
AgentRunAsyncTask Objects
class AgentRunAsyncTask()
Async task wrapper for agent.
__
init__
| __init__(agent: AEA, loop: asyncio.AbstractEventLoop) -> None
Init task with agent and loop.
create_
run_
loop
| create_run_loop() -> None
Create run loop.
start
| start() -> None
Start task.
wait
| wait() -> asyncio.Future
Return future to wait task completed.
stop
| stop() -> None
Stop task.
run
| async run() -> None
Run task body.
is_
running
| @property
| is_running() -> bool
Return is task running.
AgentRunThreadTask Objects
class AgentRunThreadTask(AgentRunAsyncTask)
Threaded wrapper to run agent.
__
init__
| __init__(agent: AEA, loop: asyncio.AbstractEventLoop) -> None
Init task with agent and loop.
create_
run_
loop
| create_run_loop() -> None
Create run loop.
start
| start() -> None
Run task in a dedicated thread.
MultiAgentManager Objects
class MultiAgentManager()
Multi agents manager.
__
init__
| __init__(working_dir: str, mode: str = "async", registry_path: str = DEFAULT_REGISTRY_NAME) -> None
Initialize manager.
Arguments:
working_dir
: directory to store base agents.
keys_
dir
| @property
| keys_dir() -> str
Get the keys directory.
certs_
dir
| @property
| certs_dir() -> str
Get the certs directory.
is_
running
| @property
| is_running() -> bool
Is manager running.
dict_
state
| @property
| dict_state() -> Dict[str, Any]
Create MultiAgentManager dist state.
add_
error_
callback
| add_error_callback(error_callback: Callable[[str, BaseException], None]) -> None
Add error callback to call on error raised.
start_
manager
| start_manager(local: bool = False, remote: bool = False) -> "MultiAgentManager"
Start manager.
If local = False and remote = False, then the packages are fetched in mixed mode (i.e. first try from local registry, and then from remote registry in case of failure).
Arguments:
local
: whether or not to fetch from local registry.remote
: whether or not to fetch from remote registry.
Returns:
the MultiAgentManager instance.
stop_
manager
| stop_manager(cleanup: bool = True, save: bool = False) -> "MultiAgentManager"
Stop manager.
Stops all running agents and stop agent.
Arguments:
cleanup
: bool is cleanup on stop.save
: bool is save state to file on stop.
Returns:
None
add_
project
| add_project(public_id: PublicId, local: bool = False, remote: bool = False, restore: bool = False) -> "MultiAgentManager"
Fetch agent project and all dependencies to working_dir.
If local = False and remote = False, then the packages are fetched in mixed mode (i.e. first try from local registry, and then from remote registry in case of failure).
Arguments:
public_id
: the public if of the agent project.local
: whether or not to fetch from local registry.remote
: whether or not to fetch from remote registry.restore
: bool flag for restoring already fetched agent.
remove_
project
| remove_project(public_id: PublicId, keep_files: bool = False) -> "MultiAgentManager"
Remove agent project.
list_
projects
| list_projects() -> List[PublicId]
List all agents projects added.
Returns:
list of public ids of projects
add_
agent
| add_agent(public_id: PublicId, agent_name: Optional[str] = None, agent_overrides: Optional[dict] = None, component_overrides: Optional[List[dict]] = None) -> "MultiAgentManager"
Create new agent configuration based on project with config overrides applied.
Alias is stored in memory only!
Arguments:
public_id
: base agent project public idagent_name
: unique name for the agentagent_overrides
: overrides for agent config.component_overrides
: overrides for component section.config
: agent config (used for agent re-creation).
Returns:
manager
add_
agent_
with_
config
| add_agent_with_config(public_id: PublicId, config: List[dict], agent_name: Optional[str] = None) -> "MultiAgentManager"
Create new agent configuration based on project with config provided.
Alias is stored in memory only!
Arguments:
public_id
: base agent project public idagent_name
: unique name for the agentconfig
: agent config (used for agent re-creation).
Returns:
manager
get_
agent_
overridables
| get_agent_overridables(agent_name: str) -> Tuple[Dict, List[Dict]]
Get agent config overridables.
Arguments:
agent_name
: str
Returns:
Tuple of agent overridables dict and and list of component overridables dict.
set_
agent_
overrides
| set_agent_overrides(agent_name: str, agent_overides: Optional[Dict], components_overrides: Optional[List[Dict]]) -> None
Set agent overrides.
Arguments:
agent_name
: stragent_overides
: optional dict of agent config overridescomponents_overrides
: optional list of dict of components overrides
Returns:
None
list_
agents_
info
| list_agents_info() -> List[Dict[str, Any]]
List agents detailed info.
Returns:
list of dicts that represents agent info: public_id, name, is_running.
list_
agents
| list_agents(running_only: bool = False) -> List[str]
List all agents.
Arguments:
running_only
: returns only running if set to True
Returns:
list of agents names
remove_
agent
| remove_agent(agent_name: str) -> "MultiAgentManager"
Remove agent alias definition from registry.
Arguments:
agent_name
: agent name to remove
Returns:
None
start_
agent
| start_agent(agent_name: str) -> "MultiAgentManager"
Start selected agent.
Arguments:
agent_name
: agent name to start
Returns:
None
start_
all_
agents
| start_all_agents() -> "MultiAgentManager"
Start all not started agents.
Returns:
None
stop_
agent
| stop_agent(agent_name: str) -> "MultiAgentManager"
Stop running agent.
Arguments:
agent_name
: agent name to stop
Returns:
None
stop_
all_
agents
| stop_all_agents() -> "MultiAgentManager"
Stop all agents running.
Returns:
None
stop_
agents
| stop_agents(agent_names: List[str]) -> "MultiAgentManager"
Stop specified agents.
Returns:
None
start_
agents
| start_agents(agent_names: List[str]) -> "MultiAgentManager"
Stop specified agents.
Returns:
None
get_
agent_
alias
| get_agent_alias(agent_name: str) -> AgentAlias
Return details about agent alias definition.
Returns:
AgentAlias