src.uagents.resolver

Endpoint Resolver.

weighted_random_sample ↗ (opens in a new tab)

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 ↗ (opens in a new tab)

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 ↗ (opens in a new tab)

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 ↗ (opens in a new tab)

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 ↗ (opens in a new tab)

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 ↗ (opens in a new tab)

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 ↗ (opens in a new tab)

class Resolver(ABC)

resolve ↗ (opens in a new tab)

@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 ↗ (opens in a new tab)

class GlobalResolver(Resolver)

init ↗ (opens in a new tab)

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 ↗ (opens in a new tab)

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 ↗ (opens in a new tab)

class AlmanacContractResolver(Resolver)

init ↗ (opens in a new tab)

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

Initialize the AlmanacContractResolver.

Arguments:

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

resolve ↗ (opens in a new tab)

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 ↗ (opens in a new tab)

class AlmanacApiResolver(Resolver)

init ↗ (opens in a new tab)

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 ↗ (opens in a new tab)

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 ↗ (opens in a new tab)

class NameServiceResolver(Resolver)

init ↗ (opens in a new tab)

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

Initialize the NameServiceResolver.

Arguments:

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

resolve ↗ (opens in a new tab)

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 ↗ (opens in a new tab)

class RulesBasedResolver(Resolver)

init ↗ (opens in a new tab)

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 ↗ (opens in a new tab)

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.