src.uagents.network
Network and Contracts.
InsufficientFundsError Objects - Github
class InsufficientFundsError(Exception)
Raised when an agent has insufficient funds for a transaction.
get_ledger - Github
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 - Github
def get_faucet() -> FaucetApi
Get the Faucet API instance.
Returns:
FaucetApi
- The Faucet API instance.
add_testnet_funds - Github
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 - Github
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 - Github
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 - Github
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 - Github
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 - Github
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 - Github
def get_contract_version() -> str
Get the version of the contract.
Returns:
str
- The version of the contract.
is_registered - Github
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 - Github
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 - Github
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 - Github
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 - Github
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 - Github
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 - Github
async def register(ledger: LedgerClient, wallet: LocalWallet,
agent_address: str, protocols: List[str],
endpoints: List[AgentEndpoint], signature: str,
current_time: int)
Register an agent with the Almanac contract.
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 - Github
async def register_batch(ledger: LedgerClient, wallet: LocalWallet,
agent_records: List[AlmanacContractRecord])
Register multiple agents with the Almanac contract.
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 - Github
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 - Github
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 - Github
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 - Github
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 - Github
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 - Github
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 - Github
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 - Github
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 - Github
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 - Github
async def register(ledger: LedgerClient,
wallet: LocalWallet,
agent_records: Optional[Union[str, List[str], Dict[str,
dict]]],
name: str,
domain: str,
overwrite: bool = True)
Register a name within a domain using the NameService contract.
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 - Github
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 - Github
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.