CosmPy

Some messages, such as MsgExec from the authz module, require nesting one message inside another. Because protobuf is strongly typed, you must pack the inner message into a google.protobuf.any_pb2.Any object before including it in the outer message.

Workflow:

1. Create an instance of any_pb2.Any().
2. Use msg.Pack(inner_message, "") to wrap your specific protobuf message.
3. Pass the Any object into the field of the outer message (e.g., the msgs list in MsgExec).

from cosmpy.aerial.client import LedgerClient, NetworkConfig
from cosmpy.aerial.wallet import LocalWallet
from cosmpy.aerial.tx import Transaction
from cosmpy.aerial.client.utils import prepare_and_broadcast_basic_transaction
from cosmpy.crypto.keypairs import PrivateKey
from cosmpy.protos.cosmos.distribution.v1beta1.tx_pb2 import MsgWithdrawValidatorCommission
from cosmpy.protos.cosmos.authz.v1beta1.tx_pb2 import MsgExec
from google.protobuf import any_pb2

client = LedgerClient(NetworkConfig.fetchai_dorado_testnet())
wallet = LocalWallet(PrivateKey("rBDA3Q0vK5T+JVQmXSoooqUY/mSO4mmhMHQJI31+h1o="))

# Pack the nested message into Any
msg = any_pb2.Any()
msg.Pack(
    MsgWithdrawValidatorCommission(
        validator_address="fetchvaloper1rsane988vksrgp2mlqzclmt8wucxv0ej4hrn2k"
    ),
    "",
)

tx = Transaction()
tx.add_message(MsgExec(grantee=str(wallet.address()), msgs=[msg]))

tx = prepare_and_broadcast_basic_transaction(client, tx, wallet)
tx.wait_to_complete()

The Transaction class manages the lifecycle of a Cosmos transaction through several states defined by TxState (Draft, Sealed, Final).

Workflow:

1. Draft: Start with a Transaction object and add messages using .add_message(msg). This method raises a RuntimeError if the transaction is not in the Draft state.
2. Seal: Use .seal(signing_cfgs, fee, memo, timeout_height) to transition the transaction to the Sealed state. This requires signing_cfgs (a SigningCfg or a list of them) and a TxFee object.
3. Sign: Use .sign(signer, chain_id, account_number, deterministic) to sign the sealed transaction. This raises a RuntimeError if the transaction is not yet Sealed.
4. Complete: Use .complete() to update the transaction state to Final.

Accessing Data:

• .state: Returns the current TxState.
• .msgs: Returns the transaction messages.
• .fee: Returns the transaction fee.
• .tx: Returns the transaction (raises RuntimeError if not completed).

from cosmpy.aerial.tx import Transaction, TxFee, SigningCfg

# 1. Draft
tx = Transaction().add_message(my_msg)

# 2. Seal
fee = TxFee(amount="1000afet", gas_limit=200000)
sealed_tx = tx.seal(signing_cfgs=my_signing_cfg, fee=fee, memo="my memo")

# 3. Sign
signed_tx = sealed_tx.sign(signer=my_signer, chain_id="fetchai", account_number=1)

# 4. Complete
final_tx = signed_tx.complete()

You can authorize a grantee (the authz_wallet) to perform specific actions on behalf of a granter (the wallet) using MsgGrant. This requires defining a Grant object containing a SendAuthorization with a spend_limit and an expiration timestamp.

To implement this:

1. Create a SendAuthorization with the allowed spend_limit (a list of Coin objects).
2. Pack the authorization into a google.protobuf.any_pb2.Any message.
3. Define the Grant with an expiration time.
4. Construct a MsgGrant specifying the granter, grantee, and the grant.
5. Broadcast the transaction using the granter's wallet.

from cosmpy.protos.cosmos.authz.v1beta1.authz_pb2 import Grant
from cosmpy.protos.cosmos.authz.v1beta1.tx_pb2 import MsgGrant
from cosmpy.protos.cosmos.bank.v1beta1.authz_pb2 import SendAuthorization
from cosmpy.aerial.client.utils import prepare_and_broadcast_basic_transaction
from cosmpy.aerial.tx import Transaction
from google.protobuf import any_pb2, timestamp_pb2
from datetime import datetime, timedelta

# Configuration
total_authz_time = 10000
amount = 1000000000000000000
spend_amount = Coin(amount=str(amount), denom="atestfet")

# 1. Prepare SendAuthorization
authz_any = any_pb2.Any()
authz_any.Pack(
    SendAuthorization(spend_limit=[spend_amount]),
    "",
)

# 2. Set Expiration
expiry = timestamp_pb2.Timestamp()
expiry.FromDatetime(datetime.now() + timedelta(seconds=total_authz_time * 60))
grant = Grant(authorization=authz_any, expiration=expiry)

# 3. Create MsgGrant
msg = MsgGrant(
    granter=str(wallet.address()),
    grantee=str(authz_wallet.address()),
    grant=grant,
)

# 4. Broadcast
tx = Transaction()
tx.add_message(msg)
tx = prepare_and_broadcast_basic_transaction(ledger, tx, wallet)
tx.wait_to_complete()

An Oracle Client is a contract that allows other users to request the oracle's data in exchange for a fee. To use an Oracle Client:

1. Deploy Client: Create a LedgerContract using oracle_client.wasm. The instantiation message must include the oracle_contract_address of the deployed oracle.
2. Request Value: Execute the query_oracle_value message on the client contract. This requires sending funds to cover the fee.
3. Retrieve Data: Use contract.query() with the key {"oracle_value": {}} to read the current value from the contract state.

Example workflow:

# Deploying the client
contract = LedgerContract("oracle_client.wasm", ledger)
instantiation_message = {"oracle_contract_address": str(ORACLE_CONTRACT_ADDRESS)}
contract.deploy(instantiation_message, wallet)

# Requesting and querying
request_message = {"query_oracle_value": {}}
contract.execute(request_message, wallet, funds="100atestfet").wait_to_complete()
result = contract.query({"oracle_value": {}})

# Requesting and querying
request_message = {"query_oracle_value": {}}
contract.execute(request_message, wallet, funds="100atestfet").wait_to_complete()
result = contract.query({"oracle_value": {}})
print(f"Oracle value successfully retrieved: {result}")

To allow an account to execute messages on behalf of another (using authz), you must first establish a grant.

Using FetchD CLI

Use the fetchd tx authz grant command to authorize a specific message type:

fetchd tx authz grant $(fetchd keys show grantee --output json | jq -r .address) generic --msg-type "/cosmos.distribution.v1beta1.MsgWithdrawValidatorCommission" --from=$(fetchd keys show grantor --output json | .address) --gas auto --gas-adjustment 1.5 --gas-prices 5000000000atestfet

Using Python

If you have access to the keys in Python, use MsgGrant with a Grant object containing a GenericAuthorization and an expiration Timestamp.

from cosmpy.aerial.client import LedgerClient, NetworkConfig
from cosmpy.aerial.wallet import LocalWallet
from cosmpy.aerial.tx import Transaction
from cosmpy.aerial.client.utils import prepare_and_broadcast_basic_transaction
from cosmpy.crypto.keypairs import PrivateKey
from cosmpy.protos.cosmos.distribution.v1beta1.tx_pb2 import MsgWithdrawValidatorCommission
from cosmpy.protos.cosmos.authz.v1beta1.tx_pb2 import MsgGrant
from cosmpy.protos.cosmos.authz.v1beta1.authz_pb2 import GenericAuthorization, Grant

from google.protobuf import any_pb2, timestamp_pb2
from datetime import datetime, timedelta

client = LedgerClient(NetworkConfig.fetchai_dorado_testnet())
wallet = LocalWallet(PrivateKey("rBDA3Q0vK5T+JVQmXSoooqUY/mSO4mmhMHQJI31+h1o="))
validator = LocalWallet(PrivateKey("<redacted>private key of dorado validator0"))

# Prepare authorization
authz_any = any_pb2.Any()
authz_any.Pack(
    GenericAuthorization(
        msg="/cosmos.distribution.v1beta1.MsgWithdrawValidatorCommission"
    ),
    "",
)

expiry = timestamp_pb2.Timestamp()
expiry.FromDatetime(datetime.now() + timedelta(seconds=60))
grant = Grant(authorization=authz_any, expiration=expiry)

# Create the grant message
msg = MsgGrant(
    granter=str(validator.address()),
    grantee=str(wallet.address()),
    grant=grant,
)

tx = Transaction()
tx.add_message(msg)

tx = prepare_and_broadcast_basic_transaction(client, tx, validator)
tx.wait_to_complete()