src.uagents.network

Network and Contracts.

InsufficientFundsError Objects ↗ (opens in a new tab)


class InsufficientFundsError(Exception)

Raised when an agent has insufficient funds for a transaction.

get_ledger ↗ (opens in a new tab)


def get_ledger(test: bool = True) -> LedgerClient

Get the Ledger client.

Arguments:

test bool - Whether to use the testnet or mainnet. Defaults to True.

Returns:

LedgerClient - The Ledger client instance.

get_faucet ↗ (opens in a new tab)


def get_faucet() -> FaucetApi

Get the Faucet API instance.

Returns:

FaucetApi - The Faucet API instance.

add_testnet_funds ↗ (opens in a new tab)


def add_testnet_funds(wallet_address: str)

Add testnet funds to the provided wallet address.

Arguments:

wallet_address str - The wallet address to add funds to.

parse_record_config ↗ (opens in a new tab)


def parse_record_config(
    record: Optional[Union[str, List[str], Dict[str, dict]]]
) -> Optional[List[Dict[str, Any]]]

Parse the user-provided record configuration.

Returns:

Optional[List[Dict[str, Any]]]: The parsed record configuration in correct format.

wait_for_tx_to_complete ↗ (opens in a new tab)


async def wait_for_tx_to_complete(
        tx_hash: str,
        ledger: LedgerClient,
        timeout: Optional[timedelta] = None,
        poll_period: Optional[timedelta] = None) -> TxResponse

Wait for a transaction to complete on the Ledger.

Arguments:

tx_hash str - The hash of the transaction to monitor.
ledger LedgerClient - The Ledger client to poll.
timeout Optional[timedelta], optional - The maximum time to wait. the transaction to complete. Defaults to None.
poll_period Optional[timedelta], optional - The time interval to poll

Returns:

TxResponse - The response object containing the transaction details.

AlmanacContract Objects ↗ (opens in a new tab)


class AlmanacContract(LedgerContract)

A class representing the Almanac contract for agent registration.

This class provides methods to interact with the Almanac contract, including checking if an agent is registered, retrieving the expiry height of an agent's registration, and getting the endpoints associated with an agent's registration.

check_version ↗ (opens in a new tab)


def check_version() -> bool

Check if the contract version supported by this version of uAgents matches the deployed version.

Returns:

bool - True if the contract version is supported, False otherwise.

query_contract ↗ (opens in a new tab)


def query_contract(query_msg: Dict[str, Any]) -> Any

Execute a query with additional checks and error handling.

Arguments:

query_msg Dict[str, Any] - The query message.

Returns:

Any - The query response.

Raises:

RuntimeError - If the contract address is not set or the query fails.

get_contract_version ↗ (opens in a new tab)


def get_contract_version() -> str

Get the version of the contract.

Returns:

str - The version of the contract.

is_registered ↗ (opens in a new tab)


def is_registered(address: str) -> bool

Check if an agent is registered in the Almanac contract.

Arguments:

address str - The agent's address.

Returns:

bool - True if the agent is registered, False otherwise.

registration_needs_update ↗ (opens in a new tab)


def registration_needs_update(address: str, endpoints: List[AgentEndpoint],
                              protocols: List[str],
                              min_seconds_left: int) -> bool

Check if an agent's registration needs to be updated.

Arguments:

address str - The agent's address.
endpoints List[AgentEndpoint] - The agent's endpoints.
protocols List[str] - The agent's protocols.
min_time_left int - The minimum time left before the agent's registration expires

Returns:

bool - True if the agent's registration needs to be updated or will expire sooner than the specified minimum time, False otherwise.

query_agent_record ↗ (opens in a new tab)


def query_agent_record(
        address: str) -> Tuple[int, List[AgentEndpoint], List[str]]

Get the records associated with an agent's registration.

Arguments:

address str - The agent's address.

Returns:

Tuple[int, List[AgentEndpoint], List[str]]: The expiry height of the agent's registration, the agent's endpoints, and the agent's protocols.

get_expiry ↗ (opens in a new tab)


def get_expiry(address: str) -> int

Get the approximate seconds to expiry of an agent's registration.

Arguments:

address str - The agent's address.

Returns:

int - The approximate seconds to expiry of the agent's registration.

get_endpoints ↗ (opens in a new tab)


def get_endpoints(address: str) -> List[AgentEndpoint]

Get the endpoints associated with an agent's registration.

Arguments:

address str - The agent's address.

Returns:

List[AgentEndpoint] - The agent's registered endpoints.

get_protocols ↗ (opens in a new tab)


def get_protocols(address: str) -> List[str]

Get the protocols associated with an agent's registration.

Arguments:

address str - The agent's address.

Returns:

List[str] - The agent's registered protocols.

register ↗ (opens in a new tab)


async def register(ledger: LedgerClient, wallet: LocalWallet,
                   agent_address: str, protocols: List[str],
                   endpoints: List[AgentEndpoint], signature: str,
                   current_time: int)

Arguments:

ledger LedgerClient - The Ledger client.
wallet LocalWallet - The agent's wallet.
agent_address str - The agent's address.
protocols List[str] - List of protocols.
endpoints List[Dict[str, Any]] - List of endpoint dictionaries.
signature str - The agent's signature.

register_batch ↗ (opens in a new tab)


async def register_batch(ledger: LedgerClient, wallet: LocalWallet,
                         agent_records: List[AlmanacContractRecord])

Arguments:

ledger LedgerClient - The Ledger client.
wallet LocalWallet - The wallet of the registration sender.
agents List[ALmanacContractRecord] - The list of signed agent records to register.

get_sequence ↗ (opens in a new tab)


def get_sequence(address: str) -> int

Get the agent's sequence number for Almanac registration.

Arguments:

address str - The agent's address.

Returns:

int - The agent's sequence number.

get_almanac_contract ↗ (opens in a new tab)


def get_almanac_contract(test: bool = True) -> Optional[AlmanacContract]

Get the AlmanacContract instance.

Arguments:

test bool - Whether to use the testnet or mainnet. Defaults to True.

Returns:

AlmanacContract - The AlmanacContract instance if version is supported.

NameServiceContract Objects ↗ (opens in a new tab)


class NameServiceContract(LedgerContract)

A class representing the NameService contract for managing domain names and ownership.

This class provides methods to interact with the NameService contract, including checking name availability, checking ownership, querying domain public status, obtaining registration transaction details, and registering a name within a domain.

query_contract ↗ (opens in a new tab)


def query_contract(query_msg: Dict[str, Any]) -> Any

Execute a query with additional checks and error handling.

Arguments:

query_msg Dict[str, Any] - The query message.

Returns:

Any - The query response.

Raises:

RuntimeError - If the contract address is not set or the query fails.

is_name_available ↗ (opens in a new tab)


def is_name_available(name: str, domain: str) -> bool

Check if a name is available within a domain.

Arguments:

name str - The name to check.
domain str - The domain to check within.

Returns:

bool - True if the name is available, False otherwise.

is_owner ↗ (opens in a new tab)


def is_owner(name: str, domain: str, wallet_address: str) -> bool

Check if the provided wallet address is the owner of a name within a domain.

Arguments:

name str - The name to check ownership for.
domain str - The domain to check within.
wallet_address str - The wallet address to check ownership against.

Returns:

bool - True if the wallet address is the owner, False otherwise.

is_domain_public ↗ (opens in a new tab)


def is_domain_public(domain: str) -> bool

Check if a domain is public.

Arguments:

domain str - The domain to check.

Returns:

bool - True if the domain is public, False otherwise.

get_previous_records ↗ (opens in a new tab)


def get_previous_records(name: str, domain: str)

Retrieve the previous records for a given name within a specified domain.

Arguments:

name str - The name whose records are to be retrieved.
domain str - The domain within which the name is registered.

Returns:

A list of dictionaries, where each dictionary contains details of a record associated with the given name.

get_registration_tx ↗ (opens in a new tab)


def get_registration_tx(name: str, wallet_address: Address,
                        agent_records: Union[List[Dict[str, Any]],
                                             str], domain: str, test: bool)

Get the registration transaction for registering a name within a domain.

Arguments:

name str - The name to be registered.
wallet_address str - The wallet address initiating the registration.
agent_address str - The address of the agent.
domain str - The domain in which the name is registered.
test bool - The agent type

Returns:

Optional[Transaction] - The registration transaction, or None if the name is not available or not owned by the wallet address.

register ↗ (opens in a new tab)


async def register(ledger: LedgerClient,
                   wallet: LocalWallet,
                   agent_records: Optional[Union[str, List[str], Dict[str,
                                                                      dict]]],
                   name: str,
                   domain: str,
                   overwrite: bool = True)

Arguments:

ledger LedgerClient - The Ledger client.
wallet LocalWallet - The wallet of the agent.
agent_address str - The address of the agent.
name str - The name to be registered.
domain str - The domain in which the name is registered.
overwrite bool, optional - Specifies whether to overwrite any existing addresses registered to the domain. If False, the address will be appended to the previous records. Defaults to True.

unregister ↗ (opens in a new tab)


async def unregister(name: str, domain: str, wallet: LocalWallet)

Unregister a name within a domain using the NameService contract.

Arguments:

name str - The name to be unregistered.
domain str - The domain in which the name is registered.
wallet LocalWallet - The wallet of the agent.

get_name_service_contract ↗ (opens in a new tab)


def get_name_service_contract(test: bool = True) -> NameServiceContract

Get the NameServiceContract instance.

Arguments:

test bool - Whether to use the testnet or mainnet. Defaults to True.

Returns:

NameServiceContract - The NameServiceContract instance.

Was this page helpful?

You can also leave detailed feedbackon Github

Models Protocol