Skip to content

Module: router

Module containing the RouterClient class, which is a client for the Infernet Router's REST API.

Example Usage

You can use the Router client as follows:

from infernet_client import RouterClient

client = RouterClient()

# Get a list of all containers running in the network
containers = await client.get_containers()

# Get a list of nodes that support the given container IDs
nodes = await client.get_nodes_by_container_ids(
    [containers[0]["id"], containers[1]["id"]]
)

RouterClient

Source code in src/infernet_client/router.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class RouterClient:
    def __init__(self, base_url: str = "infernet-router.ritual.net"):
        """Initializes the client

        Args:
            base_url (str): The base URL of the REST server

        """
        self.base_url = base_url

    async def get_nodes_by_container_ids(
        self, container_ids: list[str], n: int = 3, offset: int = 0
    ) -> list[str]:
        """Get a list of node IPs that support the given container IDs

        Args:
            container_ids (list[str]): The container IDs to get the nodes for
            n (int): The number of nodes to return
            offset (int): The offset to start at, for pagination

        Returns:
            list[dict[str, Any]]: A list of node IPs

        """

        # add container(s) as repeated query params
        url = (
            f"{self.base_url}/api/v1/ips?"
            f"{'&'.join([f'container={id}' for id in container_ids])}"
            f"&n={n}&offset={offset}"
        )

        async with ClientSession() as session:
            async with session.get(url, timeout=ClientTimeout(total=3)) as response:
                response.raise_for_status()
                return cast(list[str], await response.json())

    async def get_containers(self) -> list[NetworkContainer]:
        """Get a list of all containers running in the network

        Returns:
            list[NetworkContainer]: A list of containers, with counts and descriptions
        """

        url = f"{self.base_url}/api/v1/containers"
        async with ClientSession() as session:
            async with session.get(url, timeout=ClientTimeout(total=3)) as response:
                response.raise_for_status()
                return cast(list[NetworkContainer], await response.json())

    async def get_resources(self) -> dict[str, dict[str, ServiceResources]]:
        """Collect resources available on the network

        Returns:
            dict[str, int]: A mapping of hostname / IP address to a node's resources,
                including hardware resources and supported models.

        Raises:
            aiohttp.ClientResponseError: If the request returns an error code
            aiohttp.TimeoutError: If the request times out
        """

        url = f"{self.base_url}/api/v1/resources"
        async with ClientSession() as session:
            async with session.get(url, timeout=ClientTimeout(total=3)) as response:
                response.raise_for_status()
                return cast(
                    dict[str, dict[str, ServiceResources]], await response.json()
                )

    async def check_model_support(
        self, model_id: str
    ) -> dict[str, dict[str, ModelSupport]]:
        """Check model support on the containers of each node

        Args:
            model_id (str): The model ID to check support for

        Returns:
            dict[str, int]: A mapping of hostname / IP address to a model support status
                for each container on the node

        Raises:
            aiohttp.ClientResponseError: If the request returns an error code
            aiohttp.TimeoutError: If the request times out
        """

        url = f"{self.base_url}/api/v1/resources?model_id={model_id}"
        async with ClientSession() as session:
            async with session.get(url, timeout=ClientTimeout(total=3)) as response:
                response.raise_for_status()
                return cast(dict[str, dict[str, ModelSupport]], await response.json())

__init__(base_url='infernet-router.ritual.net')

Initializes the client

Parameters:

Name Type Description Default
base_url str

The base URL of the REST server

 'infernet-router.ritual.net'
Source code in src/infernet_client/router.py 
32
33
34
35
36
37
38
39
def __init__(self, base_url: str = "infernet-router.ritual.net"):
    """Initializes the client

    Args:
        base_url (str): The base URL of the REST server

    """
    self.base_url = base_url

check_model_support(model_id) async

Check model support on the containers of each node

Parameters:

Name Type Description Default
model_id str

The model ID to check support for

 required

Returns:

Type Description
dict[str, dict[str, ModelSupport]]

dict[str, int]: A mapping of hostname / IP address to a model support status for each container on the node

Raises:

Type Description
ClientResponseError

If the request returns an error code
TimeoutError

If the request times out
Source code in src/infernet_client/router.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
async def check_model_support(
    self, model_id: str
) -> dict[str, dict[str, ModelSupport]]:
    """Check model support on the containers of each node

    Args:
        model_id (str): The model ID to check support for

    Returns:
        dict[str, int]: A mapping of hostname / IP address to a model support status
            for each container on the node

    Raises:
        aiohttp.ClientResponseError: If the request returns an error code
        aiohttp.TimeoutError: If the request times out
    """

    url = f"{self.base_url}/api/v1/resources?model_id={model_id}"
    async with ClientSession() as session:
        async with session.get(url, timeout=ClientTimeout(total=3)) as response:
            response.raise_for_status()
            return cast(dict[str, dict[str, ModelSupport]], await response.json())

get_containers() async

Get a list of all containers running in the network

Returns:

Type Description
list[NetworkContainer]

list[NetworkContainer]: A list of containers, with counts and descriptions
Source code in src/infernet_client/router.py 
68
69
70
71
72
73
74
75
76
77
78
79
async def get_containers(self) -> list[NetworkContainer]:
    """Get a list of all containers running in the network

    Returns:
        list[NetworkContainer]: A list of containers, with counts and descriptions
    """

    url = f"{self.base_url}/api/v1/containers"
    async with ClientSession() as session:
        async with session.get(url, timeout=ClientTimeout(total=3)) as response:
            response.raise_for_status()
            return cast(list[NetworkContainer], await response.json())

get_nodes_by_container_ids(container_ids, n=3, offset=0) async

Get a list of node IPs that support the given container IDs

Parameters:

Name Type Description Default
container_ids list[str]

The container IDs to get the nodes for

 required
n int

The number of nodes to return

 3
offset int

The offset to start at, for pagination

 0

Returns:

Type Description
list[str]

list[dict[str, Any]]: A list of node IPs
Source code in src/infernet_client/router.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
async def get_nodes_by_container_ids(
    self, container_ids: list[str], n: int = 3, offset: int = 0
) -> list[str]:
    """Get a list of node IPs that support the given container IDs

    Args:
        container_ids (list[str]): The container IDs to get the nodes for
        n (int): The number of nodes to return
        offset (int): The offset to start at, for pagination

    Returns:
        list[dict[str, Any]]: A list of node IPs

    """

    # add container(s) as repeated query params
    url = (
        f"{self.base_url}/api/v1/ips?"
        f"{'&'.join([f'container={id}' for id in container_ids])}"
        f"&n={n}&offset={offset}"
    )

    async with ClientSession() as session:
        async with session.get(url, timeout=ClientTimeout(total=3)) as response:
            response.raise_for_status()
            return cast(list[str], await response.json())

get_resources() async

Collect resources available on the network

Returns:

Type Description
dict[str, dict[str, ServiceResources]]

dict[str, int]: A mapping of hostname / IP address to a node's resources, including hardware resources and supported models.

Raises:

Type Description
ClientResponseError

If the request returns an error code
TimeoutError

If the request times out
Source code in src/infernet_client/router.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
async def get_resources(self) -> dict[str, dict[str, ServiceResources]]:
    """Collect resources available on the network

    Returns:
        dict[str, int]: A mapping of hostname / IP address to a node's resources,
            including hardware resources and supported models.

    Raises:
        aiohttp.ClientResponseError: If the request returns an error code
        aiohttp.TimeoutError: If the request times out
    """

    url = f"{self.base_url}/api/v1/resources"
    async with ClientSession() as session:
        async with session.get(url, timeout=ClientTimeout(total=3)) as response:
            response.raise_for_status()
            return cast(
                dict[str, dict[str, ServiceResources]], await response.json()
            )