aea.context.base

This module contains the agent context class.

AgentContext Objects

class AgentContext()

Provide read access to relevant objects of the agent for the skills.

``init``

def __init__(identity: Identity,
             connection_status: MultiplexerStatus,
             outbox: OutBox,
             decision_maker_message_queue: Queue,
             decision_maker_handler_context: SimpleNamespace,
             task_manager: TaskManager,
             default_ledger_id: str,
             currency_denominations: Dict[str, str],
             default_connection: Optional[PublicId],
             default_routing: Dict[PublicId, PublicId],
             search_service_address: Address,
             decision_maker_address: Address,
             data_dir: str,
             storage_callable: Callable[[], Optional[Storage]] = lambda: None,
             send_to_skill: Optional[Callable] = None,
             **kwargs: Any) -> None

Initialize an agent context.

Arguments:

identity: the identity object
connection_status: the connection status of the multiplexer
outbox: the outbox
decision_maker_message_queue: the (in) queue of the decision maker
decision_maker_handler_context: the decision maker's name space
task_manager: the task manager
default_ledger_id: the default ledger id
currency_denominations: mapping from ledger ids to currency denominations
default_connection: the default connection
default_routing: the default routing
search_service_address: the address of the search service
decision_maker_address: the address of the decision maker
data_dir: directory where to put local files.
storage_callable: function that returns optional storage attached to agent.
send_to_skill: callable for sending envelopes to skills.
kwargs: keyword arguments to be attached in the agent context namespace.

send`_`to`_`skill

def send_to_skill(message_or_envelope: Union[Message, Envelope],
                  context: Optional[EnvelopeContext] = None) -> None

Send message or envelope to another skill.

If message passed it will be wrapped into envelope with optional envelope context.

Arguments:

message_or_envelope: envelope to send to another skill.
context: the optional envelope context

storage

@property
def storage() -> Optional[Storage]

Return storage instance if enabled in AEA.

data`_`dir

@property
def data_dir() -> str

Return assets directory.

shared`_`state

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

Get the shared state dictionary.

The shared state is the only object which skills can use to exchange state directly. It is accessible (read and write) from all skills.

Returns:

dictionary of the shared state.

identity

@property
def identity() -> Identity

Get the identity.

agent`_`name

@property
def agent_name() -> str

Get agent name.

addresses

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

Get addresses.

public`_`keys

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

Get public keys.

address

@property
def address() -> Address

Get the default address.

public`_`key

@property
def public_key() -> str

Get the default public key.

connection`_`status

@property
def connection_status() -> MultiplexerStatus

Get connection status of the multiplexer.

outbox

@property
def outbox() -> OutBox

Get outbox.

decision`_`maker`_`message`_`queue

@property
def decision_maker_message_queue() -> Queue

Get decision maker queue.

decision`_`maker`_`handler`_`context

@property
def decision_maker_handler_context() -> SimpleNamespace

Get the decision maker handler context.

task`_`manager

@property
def task_manager() -> TaskManager

Get the task manager.

search`_`service`_`address

@property
def search_service_address() -> Address

Get the address of the search service.

decision`_`maker`_`address

@property
def decision_maker_address() -> Address

Get the address of the decision maker.

default`_`ledger`_`id

@property
def default_ledger_id() -> str

Get the default ledger id.

currency`_`denominations

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

Get a dictionary mapping ledger ids to currency denominations.

default`_`connection

@property
def default_connection() -> Optional[PublicId]

Get the default connection.

default`_`routing

@property
def default_routing() -> Dict[PublicId, PublicId]

Get the default routing.

namespace

@property
def namespace() -> SimpleNamespace

Get the agent context namespace.