Skip to Content
References

src.uagents.resolver

Endpoint Resolver.

weighted_random_sample 

def weighted_random_sample(items: List[Any],
                           weights: Optional[List[float]] = None,
                           k: int = 1,
                           rng=random) -> List[Any]

Weighted random sample from a list of items without replacement.

Ref: Efraimidis, Pavlos S. “Weighted random sampling over data streams.”

Arguments:

  • items List[Any] - The list of items to sample from.
  • weights Optional[List[float]] - The optional list of weights for each item.
  • k int - The number of items to sample.
  • rng random - The random number generator.

Returns:

  • List[Any] - The sampled items.

is_valid_address 

def is_valid_address(address: str) -> bool

Check if the given string is a valid address.

Arguments:

  • address str - The address to be checked.

Returns:

  • bool - True if the address is valid; False otherwise.

is_valid_prefix 

def is_valid_prefix(prefix: str) -> bool

Check if the given string is a valid prefix.

Arguments:

  • prefix str - The prefix to be checked.

Returns:

  • bool - True if the prefix is valid; False otherwise.

parse_identifier 

def parse_identifier(identifier: str) -> Tuple[str, str, str]

Parse an agent identifier string into prefix, name, and address.

Arguments:

  • identifier str - The identifier string to be parsed.

Returns:

Tuple[str, str, str]: A tuple containing the prefix, name, and address as strings.

query_record 

def query_record(agent_address: str, service: str, test: bool) -> dict

Query a record from the Almanac contract.

Arguments:

  • agent_address str - The address of the agent.
  • service str - The type of service to query.

Returns:

  • dict - The query result.

get_agent_address 

def get_agent_address(name: str, test: bool) -> Optional[str]

Get the agent address associated with the provided name from the name service contract.

Arguments:

  • name str - The name to query.
  • test bool - Whether to use the testnet or mainnet contract.

Returns:

  • Optional[str] - The associated agent address if found.

Resolver Objects 

class Resolver(ABC)

resolve 

@abstractmethod
async def resolve(destination: str) -> Tuple[Optional[str], List[str]]

Resolve the destination to an address and endpoint.

Arguments:

  • destination str - The destination name or address to resolve.

Returns:

Tuple[Optional[str], List[str]]: The address (if available) and resolved endpoints.

GlobalResolver Objects 

class GlobalResolver(Resolver)

init 

def __init__(max_endpoints: Optional[int] = None,
             almanac_api_url: Optional[str] = None)

Initialize the GlobalResolver.

Arguments:

  • max_endpoints Optional[int] - The maximum number of endpoints to return.
  • almanac_api_url Optional[str] - The url for almanac api

resolve 

async def resolve(destination: str) -> Tuple[Optional[str], List[str]]

Resolve the destination using the appropriate resolver.

Arguments:

  • destination str - The destination name or address to resolve.

Returns:

Tuple[Optional[str], List[str]]: The address (if available) and resolved endpoints.

AlmanacContractResolver Objects 

class AlmanacContractResolver(Resolver)

init 

def __init__(max_endpoints: Optional[int] = None)

Initialize the AlmanacContractResolver.

Arguments:

  • max_endpoints Optional[int] - The maximum number of endpoints to return.

resolve 

async def resolve(destination: str) -> Tuple[Optional[str], List[str]]

Resolve the destination using the Almanac contract.

Arguments:

  • destination str - The destination address to resolve.

Returns:

Tuple[str, List[str]]: The address and resolved endpoints.

AlmanacApiResolver Objects 

class AlmanacApiResolver(Resolver)

init 

def __init__(max_endpoints: Optional[int] = None,
             almanac_api_url: Optional[str] = None)

Initialize the AlmanacApiResolver.

Arguments:

  • max_endpoints Optional[int] - The maximum number of endpoints to return.
  • almanac_api_url Optional[str] - The url for almanac api

resolve 

async def resolve(destination: str) -> Tuple[Optional[str], List[str]]

Resolve the destination using the Almanac API. If the resolution using API fails, it retries using the Almanac Contract.

Arguments:

  • destination str - The destination address to resolve.

Returns:

Tuple[Optional[str], List[str]]: The address and resolved endpoints.

NameServiceResolver Objects 

class NameServiceResolver(Resolver)

init 

def __init__(max_endpoints: Optional[int] = None)

Initialize the NameServiceResolver.

Arguments:

  • max_endpoints Optional[int] - The maximum number of endpoints to return.

resolve 

async def resolve(destination: str) -> Tuple[Optional[str], List[str]]

Resolve the destination using the NameService contract.

Arguments:

  • destination str - The destination name to resolve.

Returns:

Tuple[Optional[str], List[str]]: The address (if available) and resolved endpoints.

RulesBasedResolver Objects 

class RulesBasedResolver(Resolver)

init 

def __init__(rules: Dict[str, str], max_endpoints: Optional[int] = None)

Initialize the RulesBasedResolver with the provided rules.

Arguments:

  • rules Dict[str, str] - A dictionary of rules mapping destinations to endpoints.
  • max_endpoints Optional[int] - The maximum number of endpoints to return.

resolve 

async def resolve(destination: str) -> Tuple[Optional[str], List[str]]

Resolve the destination using the provided rules.

Arguments:

  • destination str - The destination to resolve.

Returns:

Tuple[str, List[str]]: The address and resolved endpoints.

Last updated on