Skip to content

ANTA as a Python Library

ANTA is a Python library that can be used in user applications. This section describes how you can leverage ANTA Python modules to help you create your own NRFU solution.

Tip

If you are unfamiliar with asyncio, refer to the Python documentation relevant to your Python version - https://docs.python.org/3/library/asyncio.html

AntaDevice Abstract Class

A device is represented in ANTA as an instance of a subclass of the AntaDevice abstract class. There are a few abstract methods that need to be implemented by child classes:

  • The collect() coroutine is in charge of collecting outputs of AntaCommand instances.
  • The refresh() coroutine is in charge of updating attributes of the AntaDevice instance. These attributes are used by AntaInventory to filter out unreachable devices or by AntaTest to skip devices based on their hardware models.

The copy() coroutine is used to copy files to and from the device. It does not need to be implemented if tests are not using it.

AsyncEOSDevice Class

The AsyncEOSDevice class is an implementation of AntaDevice for Arista EOS. It uses asynceapi for eAPI and the AsyncSSH library for SCP. asynceapi is ANTA’s fork and packaged client derived from aio-eapi.

  • The _collect() coroutine collects AntaCommand outputs using eAPI.
  • The refresh() coroutine tries to open a TCP connection on the eAPI port and update the is_online attribute accordingly. If the TCP connection succeeds, it sends a show version command to gather the hardware model of the device and updates the established and hw_model attributes.
  • The copy() coroutine copies files to and from the device using the SCP protocol.

ANTA uses the term use_session_auth as a device-agnostic concept: it signals that the device should maintain an authenticated session across multiple requests rather than re-authenticating on every call. AsyncEOSDevice implements this via eAPI cookie-session authentication — when use_session_auth is set to True in the constructor (default: False), asynceapi establishes an HTTP session and reuses the session cookie across eAPI calls, reducing authentication overhead.

AntaInventory Class

The AntaInventory class is a subclass of the standard Python type dict. The keys of this dictionary are the device names, the values are AntaDevice instances.

AntaInventory provides methods to interact with the ANTA inventory:

Warning

When using ANTA as a Python library, inventory and device objects are owned by the caller. If you reuse the same AntaInventory or AntaDevice objects across multiple runs or runners, do not enable disconnect=True on a run that overlaps with other work using those objects. Instead, call disconnect_inventory() once all runs using that inventory are complete. The disconnect=True runner option is intended for cases where a single run owns the inventory lifecycle.

Examples

Parse an ANTA inventory file

# Copyright (c) 2024 Arista Networks, Inc.
# Use of this source code is governed by the Apache License 2.0
# that can be found in the LICENSE file.
"""Script that parses an ANTA inventory file, connects to devices and print their status."""

import asyncio

from anta.inventory import AntaInventory


async def main(inv: AntaInventory) -> None:
    """Read an AntaInventory and try to connect to every device in the inventory.

    Print a message for every device connection status
    """
    await inv.connect_inventory()

    for device in inv.values():
        if device.established:
            print(f"Device {device.name} is online")
        else:
            print(f"Could not connect to device {device.name}")


if __name__ == "__main__":
    # Create the AntaInventory instance
    inventory = AntaInventory.parse(
        filename="inventory.yaml",
        username="arista",
        password="@rista123",
    )

    # Run the main coroutine
    res = asyncio.run(main(inventory))

Note

How to create your inventory file

Please visit this dedicated section for how to use inventory and catalog files.

Run EOS commands

# Copyright (c) 2024 Arista Networks, Inc.
# Use of this source code is governed by the Apache License 2.0
# that can be found in the LICENSE file.
"""Script that runs a list of EOS commands on reachable devices."""

from __future__ import annotations

import asyncio
from pprint import pprint

from anta.inventory import AntaInventory
from anta.models import AntaCommand


async def main(inv: AntaInventory, commands: list[str]) -> dict[str, list[AntaCommand]]:
    """Run a list of commands against each valid device in the inventory.

    Take an AntaInventory and a list of commands as string
    1. try to connect to every device in the inventory
    2. collect the results of the commands from each device

    Returns
    -------
    dict[str, list[AntaCommand]]
        a dictionary where key is the device name and the value is the list of AntaCommand ran towards the device
    """
    await inv.connect_inventory()

    # Make a list of coroutine to run commands towards each connected device
    coros = []
    # dict to keep track of the commands per device
    result_dict = {}
    for name, device in inv.get_inventory(established_only=True).items():
        anta_commands = [AntaCommand(command=command, ofmt="json") for command in commands]
        result_dict[name] = anta_commands
        coros.append(device.collect_commands(anta_commands))

    # Run the coroutines
    await asyncio.gather(*coros)

    return result_dict


if __name__ == "__main__":
    # Create the AntaInventory instance
    inventory = AntaInventory.parse(
        filename="inventory.yaml",
        username="arista",
        password="@rista123",
    )

    # Create a list of commands with json output
    command_list = ["show version", "show ip bgp summary"]

    # Run the main asyncio  entry point
    res = asyncio.run(main(inventory, command_list))

    pprint(res)