Skip to content

autonomy.chain.service

Helper functions to manage on-chain services

MultiSendOperation Objects

class MultiSendOperation(Enum)

Operation types.

get_deployment_payload

def get_deployment_payload(fallback_handler: Optional[str] = None) -> str

Calculates deployment payload.

get_deployment_with_recovery_payload

def get_deployment_with_recovery_payload(
        fallback_handler: Optional[str] = None) -> str

Calculates deployment payload.

get_poly_safe_deployment_payload

def get_poly_safe_deployment_payload(ledger_api: LedgerApi,
                                     chain_type: ChainType,
                                     crypto: Crypto) -> str

Calculates Poly Safe deployment payload.

get_agent_instances

def get_agent_instances(ledger_api: LedgerApi, chain_type: ChainType,
                        token_id: int) -> Dict

Get the list of agent instances.

Arguments:

  • ledger_api: aea.crypto.LedgerApi object for interacting with the chain
  • chain_type: Chain type
  • token_id: Token ID pointing to the on-chain service

Returns:

number of agent instances and the list of registered addressed

get_service_info

def get_service_info(ledger_api: LedgerApi, chain_type: ChainType,
                     token_id: int) -> ServiceInfo

Returns service info.

Arguments:

  • ledger_api: aea.crypto.LedgerApi object for interacting with the chain
  • chain_type: Chain type
  • token_id: Token ID pointing to the on-chain service

Returns:

security deposit, multisig address, IPFS hash for config, threshold, max number of agent instances, number of agent instances, service state, list of cannonical agents

get_token_deposit_amount

def get_token_deposit_amount(ledger_api: LedgerApi,
                             chain_type: ChainType,
                             service_id: int,
                             agent_id: Optional[int] = None) -> int

Returns service info.

get_activate_registration_amount

def get_activate_registration_amount(ledger_api: LedgerApi,
                                     chain_type: ChainType, service_id: int,
                                     agents: List[int]) -> int

Get activate registration amount.

is_service_token_secured

def is_service_token_secured(ledger_api: LedgerApi, chain_type: ChainType,
                             service_id: int) -> bool

Check if the service is token secured.

approve_erc20_usage

def approve_erc20_usage(ledger_api: LedgerApi,
                        crypto: Crypto,
                        chain_type: ChainType,
                        spender: str,
                        amount: int,
                        sender: str,
                        dry_run: bool = False,
                        timeout: Optional[float] = None,
                        retries: Optional[int] = None,
                        sleep: Optional[float] = None) -> None

Approve ERC20 token usage.

ServiceManager Objects

class ServiceManager()

Service manager.

__init__

def __init__(ledger_api: LedgerApi,
             crypto: Crypto,
             chain_type: ChainType,
             dry_run: bool = False,
             timeout: Optional[float] = None,
             retries: Optional[int] = None,
             sleep: Optional[float] = None) -> None

Initialize object.

get_service_info

def get_service_info(token_id: int) -> ServiceInfo

Returns service info.

Arguments:

  • token_id: Token ID pointing to the on-chain service

Returns:

security deposit, multisig address, IPFS hash for config, threshold, max number of agent instances, number of agent instances, service state, list of cannonical agents

activate

def activate(service_id: int) -> None

Activate service.

Once you have minted the service on-chain, you'll have to activate the service before you can proceed further.

Arguments:

  • service_id: Service ID retrieved after minting a service

register_instance

def register_instance(service_id: int, instances: List[str],
                      agent_ids: List[int]) -> None

Register instance.

Once you have a service with an active registration, you can register agent which will be a part of the service deployment. Using this method you can register maximum N amounts per agents, N being the number of slots for an agent with agent id being agent_id.

Make sure the instance address you provide is not already a part of any service and not as same as the service owner.

Arguments:

  • service_id: Service ID retrieved after minting a service
  • instances: Address of the agent instance
  • agent_ids: Agent ID of the agent that you want this instance to be a part of when deployed

deploy

def deploy(service_id: int,
           fallback_handler: Optional[str] = None,
           reuse_multisig: bool = False,
           use_recovery_module: bool = False) -> None

Deploy service.

Using this method you can deploy a service on-chain once you have activated the service and registered the required agent instances.

Arguments:

  • service_id: Service ID retrieved after minting a service
  • fallback_handler: Fallback handler address for gnosis safe multisig
  • reuse_multisig: Use multisig from the previous deployment
  • use_recovery_module: Use multisig with recovery module

terminate

def terminate(service_id: int) -> None

Terminate service.

Using this method you can terminate a service on-chain once you have activated the service and registered the required agent instances.

Arguments:

  • service_id: Service ID retrieved after minting a service

unbond

def unbond(service_id: int) -> None

Unbond service.

Using this method you can unbond a service on-chain once you have terminated the service.

Arguments:

  • service_id: Service ID retrieved after minting a service

recover_multisig

def recover_multisig(service_id: int) -> None

Recover the service multisig.

This method allows the service owner to reclaim the multisig wallet from the previous deployment if it was not properly transferred by the agents after service termination.

Service multisig recovery is only possible if: - The original deployment was performed with the --use-recovery-module flag. - The service is currently in the PRE_REGISTRATION state (i.e., all operators have unbonded).

Arguments:

  • service_id: Service ID retrieved after minting a service

get_reuse_multisig_payload

def get_reuse_multisig_payload(
        ledger_api: LedgerApi, crypto: Crypto, chain_type: ChainType,
        service_id: int) -> Tuple[Optional[str], Optional[str]]

Reuse multisig.

get_reuse_multisig_with_recovery_payload

def get_reuse_multisig_with_recovery_payload(
        ledger_api: LedgerApi, crypto: Crypto, chain_type: ChainType,
        service_id: int) -> Tuple[Optional[str], Optional[str]]

Reuse multisig.