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.