Repository: aristanetworks/anta
Pull request #1035 (2025-02-11 04:08:22)

feat(anta): Add timestamps option to asynceapi client and update typing
# Description

Add timestamps option to asynceapi for eAPI requests. Also updated typing for asynceapi.

# Checklist:

<!-- Delete not relevant items !-->

- [X] My code follows the style guidelines of this project
- [X] I have performed a self-review of my code
- [X] I have run pre-commit for code linting and typing (`pre-commit run`)
- [X] I have made corresponding changes to the documentation
- [x] I have added tests that prove my fix is effective or that my feature works
- [x] New and existing unit tests pass locally with my changes (`tox -e testenv`)


sonarqubecloud[bot]: ## [![Quality Gate Failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-failed-20px.png 'Quality Gate Failed')](https://sonarcloud.io/dashboard?id=aristanetworks_anta&pullRequest=1035) **Quality Gate failed**  
Failed conditions  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/failed-16px.png '') [75.6% Coverage on New Code](https://sonarcloud.io/component_measures?id=aristanetworks_anta&pullRequest=1035&metric=new_coverage&view=list) (required ≥ 80%)  
  
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=aristanetworks_anta&pullRequest=1035)


sonarqubecloud[bot]: ## [![Quality Gate Failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-failed-20px.png 'Quality Gate Failed')](https://sonarcloud.io/dashboard?id=aristanetworks_anta&pullRequest=1035) **Quality Gate failed**  
Failed conditions  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/failed-16px.png '') [75.6% Coverage on New Code](https://sonarcloud.io/component_measures?id=aristanetworks_anta&pullRequest=1035&metric=new_coverage&view=list) (required ≥ 80%)  
  
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=aristanetworks_anta&pullRequest=1035)


github-actions[bot]: This pull request has conflicts, please resolve those before we can evaluate the pull request.
github-actions[bot]: Conflicts have been resolved. A maintainer will review the pull request shortly.
sonarqubecloud[bot]: ## [![Quality Gate Passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-passed-20px.png 'Quality Gate Passed')](https://sonarcloud.io/dashboard?id=aristanetworks_anta&pullRequest=1035) **Quality Gate passed**  
Issues  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [0 New issues](https://sonarcloud.io/project/issues?id=aristanetworks_anta&pullRequest=1035&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/accepted-16px.png '') [0 Accepted issues](https://sonarcloud.io/project/issues?id=aristanetworks_anta&pullRequest=1035&issueStatuses=ACCEPTED)

Measures  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=aristanetworks_anta&pullRequest=1035&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [81.8% Coverage on New Code](https://sonarcloud.io/component_measures?id=aristanetworks_anta&pullRequest=1035&metric=new_coverage&view=list)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=aristanetworks_anta&pullRequest=1035&metric=new_duplicated_lines_density&view=list)  
  
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=aristanetworks_anta&pullRequest=1035)


sonarqubecloud[bot]: ## [![Quality Gate Passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-passed-20px.png 'Quality Gate Passed')](https://sonarcloud.io/dashboard?id=aristanetworks_anta&pullRequest=1035) **Quality Gate passed**  
Issues  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [0 New issues](https://sonarcloud.io/project/issues?id=aristanetworks_anta&pullRequest=1035&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/accepted-16px.png '') [0 Accepted issues](https://sonarcloud.io/project/issues?id=aristanetworks_anta&pullRequest=1035&issueStatuses=ACCEPTED)

Measures  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=aristanetworks_anta&pullRequest=1035&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [85.3% Coverage on New Code](https://sonarcloud.io/component_measures?id=aristanetworks_anta&pullRequest=1035&metric=new_coverage&view=list)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=aristanetworks_anta&pullRequest=1035&metric=new_duplicated_lines_density&view=list)  
  
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=aristanetworks_anta&pullRequest=1035)


codspeed-hq[bot]: <!-- __CODSPEED_PERFORMANCE_REPORT_COMMENT__ -->
<!-- __CODSPEED_INSTRUMENTATION_PERFORMANCE_REPORT_COMMENT__ -->

## [CodSpeed Performance Report](https://codspeed.io/aristanetworks/anta/branches/carl-baillargeon%3Afeat%2Fasynceapi_typing_and_timestamps)

### Merging #1035 will **not alter performance**

<sub>Comparing <code>carl-baillargeon:feat/asynceapi_typing_and_timestamps</code> (d460f16) with <code>main</code> (9ba0284)</sub>



### Summary

`✅ 22` untouched benchmarks  




sonarqubecloud[bot]: ## [![Quality Gate Passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-passed-20px.png 'Quality Gate Passed')](https://sonarcloud.io/dashboard?id=aristanetworks_anta&pullRequest=1035) **Quality Gate passed**  
Issues  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [1 New issue](https://sonarcloud.io/project/issues?id=aristanetworks_anta&pullRequest=1035&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/accepted-16px.png '') [0 Accepted issues](https://sonarcloud.io/project/issues?id=aristanetworks_anta&pullRequest=1035&issueStatuses=ACCEPTED)

Measures  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=aristanetworks_anta&pullRequest=1035&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [85.3% Coverage on New Code](https://sonarcloud.io/component_measures?id=aristanetworks_anta&pullRequest=1035&metric=new_coverage&view=list)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=aristanetworks_anta&pullRequest=1035&metric=new_duplicated_lines_density&view=list)  
  
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=aristanetworks_anta&pullRequest=1035)



Patch:
diff --git a/anta/cli/exec/utils.py b/anta/cli/exec/utils.py
index 1f874b0c..3df4b7a5 100644
--- a/anta/cli/exec/utils.py
+++ b/anta/cli/exec/utils.py
@@ -23,6 +23,7 @@ from asynceapi import EapiCommandError
 
 if TYPE_CHECKING:
     from anta.inventory import AntaInventory
+    from asynceapi._types import EapiComplexCommand, EapiSimpleCommand
 
 EOS_SCHEDULED_TECH_SUPPORT = "/mnt/flash/schedule/tech-support"
 INVALID_CHAR = "`~!@#$/"
@@ -135,13 +136,13 @@ async def collect_show_tech(inv: AntaInventory, root_dir: Path, *, configure: bo
                 )
                 logger.warning(msg)
 
-                commands = []
                 # TODO: @mtache - add `config` field to `AntaCommand` object to handle this use case.
                 # Otherwise mypy complains about enable as it is only implemented for AsyncEOSDevice
                 # TODO: Should enable be also included in AntaDevice?
                 if not isinstance(device, AsyncEOSDevice):
                     msg = "anta exec collect-tech-support is only supported with AsyncEOSDevice for now."
                     raise UsageError(msg)
+                commands: list[EapiSimpleCommand | EapiComplexCommand] = []
                 if device.enable and device._enable_password is not None:
                     commands.append({"cmd": "enable", "input": device._enable_password})
                 elif device.enable:
diff --git a/anta/device.py b/anta/device.py
index f685357c..7c1e6f64 100644
--- a/anta/device.py
+++ b/anta/device.py
@@ -26,6 +26,8 @@ if TYPE_CHECKING:
     from collections.abc import Iterator
     from pathlib import Path
 
+    from asynceapi._types import EapiComplexCommand, EapiSimpleCommand
+
 logger = logging.getLogger(__name__)
 
 # Do not load the default keypairs multiple times due to a performance issue introduced in cryptography 37.0
@@ -466,7 +468,7 @@ class AsyncEOSDevice(AntaDevice):
         semaphore = await self._get_semaphore()
 
         async with semaphore:
-            commands: list[dict[str, str | int]] = []
+            commands: list[EapiComplexCommand | EapiSimpleCommand] = []
             if self.enable and self._enable_password is not None:
                 commands.append(
                     {
@@ -479,12 +481,12 @@ class AsyncEOSDevice(AntaDevice):
                 commands.append({"cmd": "enable"})
             commands += [{"cmd": command.command, "revision": command.revision}] if command.revision else [{"cmd": command.command}]
             try:
-                response: list[dict[str, Any] | str] = await self._session.cli(
+                response = await self._session.cli(
                     commands=commands,
                     ofmt=command.ofmt,
                     version=command.version,
                     req_id=f"ANTA-{collection_id}-{id(command)}" if collection_id else f"ANTA-{id(command)}",
-                )  # type: ignore[assignment] # multiple commands returns a list
+                )
                 # Do not keep response of 'enable' command
                 command.output = response[-1]
             except asynceapi.EapiCommandError as e:
diff --git a/asynceapi/_types.py b/asynceapi/_types.py
new file mode 100644
index 00000000..675b7715
--- /dev/null
+++ b/asynceapi/_types.py
@@ -0,0 +1,49 @@
+# Copyright (c) 2024-2025 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""Type definitions used for the asynceapi package."""
+
+from __future__ import annotations
+
+import sys
+from typing import Any, Literal
+
+if sys.version_info >= (3, 11):
+    from typing import NotRequired, TypedDict
+else:
+    from typing_extensions import NotRequired, TypedDict
+
+EapiJsonOutput = dict[str, Any]
+"""Type definition of an eAPI JSON output response."""
+EapiTextOutput = str
+"""Type definition of an eAPI text output response."""
+EapiSimpleCommand = str
+"""Type definition of an eAPI simple command."""
+
+
+class EapiComplexCommand(TypedDict):
+    """Type definition of an eAPI complex command."""
+
+    cmd: str
+    input: NotRequired[str]
+    revision: NotRequired[int]
+
+
+class JsonRpc(TypedDict):
+    """Type definition of a JSON-RPC payload."""
+
+    jsonrpc: Literal["2.0"]
+    method: Literal["runCmds"]
+    params: JsonRpcParams
+    id: NotRequired[int | str]
+
+
+class JsonRpcParams(TypedDict):
+    """Type definition of JSON-RPC parameters."""
+
+    version: NotRequired[int | Literal["latest"]]
+    cmds: list[EapiSimpleCommand | EapiComplexCommand]
+    format: NotRequired[Literal["json", "text"]]
+    autoComplete: NotRequired[bool]
+    expandAliases: NotRequired[bool]
+    timestamps: NotRequired[bool]
diff --git a/asynceapi/config_session.py b/asynceapi/config_session.py
index 470a0d69..e5e1d085 100644
--- a/asynceapi/config_session.py
+++ b/asynceapi/config_session.py
@@ -10,9 +10,10 @@
 from __future__ import annotations
 
 import re
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
+    from ._types import EapiComplexCommand, EapiJsonOutput, EapiSimpleCommand
     from .device import Device
 
 # -----------------------------------------------------------------------------
@@ -78,7 +79,7 @@ class SessionConfig:
     #                               Public Methods
     # -------------------------------------------------------------------------
 
-    async def status_all(self) -> dict[str, Any]:
+    async def status_all(self) -> EapiJsonOutput:
         """Get the status of all the session config on the device.
 
         Run the following command on the device:
@@ -86,7 +87,7 @@ class SessionConfig:
 
         Returns
         -------
-        dict[str, Any]
+        EapiJsonOutput
             Dictionary of native EOS eAPI response; see `status` method for
             details.
 
@@ -116,9 +117,9 @@ class SessionConfig:
         }
         ```
         """
-        return await self._cli("show configuration sessions detail")  # type: ignore[return-value] # json outformat returns dict[str, Any]
+        return await self._cli(command="show configuration sessions detail")
 
-    async def status(self) -> dict[str, Any] | None:
+    async def status(self) -> EapiJsonOutput | None:
         """Get the status of a session config on the device.
 
         Run the following command on the device:
@@ -129,7 +130,7 @@ class SessionConfig:
 
         Returns
         -------
-        dict[str, Any] | None
+        EapiJsonOutput | None
             Dictionary instance of the session status. If the session does not exist,
             then this method will return None.
 
@@ -201,7 +202,7 @@ class SessionConfig:
         # prepare the initial set of command to enter the config session and
         # rollback clean if the `replace` argument is True.
 
-        commands: list[str | dict[str, Any]] = [self._cli_config_session]
+        commands: list[EapiSimpleCommand | EapiComplexCommand] = [self._cli_config_session]
         if replace:
             commands.append(self.CLI_CFG_FACTORY_RESET)
 
@@ -232,7 +233,7 @@ class SessionConfig:
         if timer:
             command += f" timer {timer}"
 
-        await self._cli(command)
+        await self._cli(command=command)
 
     async def abort(self) -> None:
         """Abort the configuration session.
@@ -240,7 +241,7 @@ class SessionConfig:
         Run the following command on the device:
             # configure session <name> abort
         """
-        await self._cli(f"{self._cli_config_session} abort")
+        await self._cli(command=f"{self._cli_config_session} abort")
 
     async def diff(self) -> str:
         """Return the "diff" of the session config relative to the running config.
@@ -257,7 +258,7 @@ class SessionConfig:
         ----------
             * https://www.gnu.org/software/diffutils/manual/diffutils.txt
         """
-        return await self._cli(f"show session-config named {self.name} diffs", ofmt="text")  # type: ignore[return-value] # text outformat returns str
+        return await self._cli(command=f"show session-config named {self.name} diffs", ofmt="text")
 
     async def load_file(self, filename: str, *, replace: bool = False) -> None:
         """Load the configuration from <filename> into the session configuration.
@@ -281,12 +282,12 @@ class SessionConfig:
             If there are any issues with loading the configuration file then a
             RuntimeError is raised with the error messages content.
         """
-        commands: list[str | dict[str, Any]] = [self._cli_config_session]
+        commands: list[EapiSimpleCommand | EapiComplexCommand] = [self._cli_config_session]
         if replace:
             commands.append(self.CLI_CFG_FACTORY_RESET)
 
         commands.append(f"copy {filename} session-config")
-        res: list[dict[str, Any]] = await self._cli(commands=commands)  # type: ignore[assignment] # JSON outformat of multiple commands returns list[dict[str, Any]]
+        res = await self._cli(commands=commands)
         checks_re = re.compile(r"error|abort|invalid", flags=re.IGNORECASE)
         messages = res[-1]["messages"]
 
@@ -295,4 +296,4 @@ class SessionConfig:
 
     async def write(self) -> None:
         """Save the running config to the startup config by issuing the command "write" to the device."""
-        await self._cli("write")
+        await self._cli(command="write")
diff --git a/asynceapi/device.py b/asynceapi/device.py
index 9b68dc58..907ecb33 100644
--- a/asynceapi/device.py
+++ b/asynceapi/device.py
@@ -10,7 +10,7 @@
 from __future__ import annotations
 
 from socket import getservbyname
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Literal, overload
 
 # -----------------------------------------------------------------------------
 # Public Imports
@@ -25,7 +25,7 @@ from .config_session import SessionConfig
 from .errors import EapiCommandError
 
 if TYPE_CHECKING:
-    from collections.abc import Sequence
+    from ._types import EapiComplexCommand, EapiJsonOutput, EapiSimpleCommand, EapiTextOutput, JsonRpc
 
 # -----------------------------------------------------------------------------
 # Exports
@@ -121,18 +121,148 @@ class Device(httpx.AsyncClient):
         """
         return await port_check_url(self.base_url)
 
+    # Single command, JSON output, no suppression
+    @overload
     async def cli(
         self,
-        command: str | dict[str, Any] | None = None,
-        commands: Sequence[str | dict[str, Any]] | None = None,
-        ofmt: str | None = None,
-        version: int | str | None = "latest",
+        *,
+        command: EapiSimpleCommand | EapiComplexCommand,
+        commands: None = None,
+        ofmt: Literal["json"] = "json",
+        version: int | Literal["latest"] = "latest",
+        suppress_error: Literal[False] = False,
+        auto_complete: bool = False,
+        expand_aliases: bool = False,
+        timestamps: bool = False,
+        req_id: int | str | None = None,
+    ) -> EapiJsonOutput: ...
+
+    # Multiple commands, JSON output, no suppression
+    @overload
+    async def cli(
+        self,
+        *,
+        command: None = None,
+        commands: list[EapiSimpleCommand | EapiComplexCommand],
+        ofmt: Literal["json"] = "json",
+        version: int | Literal["latest"] = "latest",
+        suppress_error: Literal[False] = False,
+        auto_complete: bool = False,
+        expand_aliases: bool = False,
+        timestamps: bool = False,
+        req_id: int | str | None = None,
+    ) -> list[EapiJsonOutput]: ...
+
+    # Single command, TEXT output, no suppression
+    @overload
+    async def cli(
+        self,
+        *,
+        command: EapiSimpleCommand | EapiComplexCommand,
+        commands: None = None,
+        ofmt: Literal["text"],
+        version: int | Literal["latest"] = "latest",
+        suppress_error: Literal[False] = False,
+        auto_complete: bool = False,
+        expand_aliases: bool = False,
+        timestamps: bool = False,
+        req_id: int | str | None = None,
+    ) -> EapiTextOutput: ...
+
+    # Multiple commands, TEXT output, no suppression
+    @overload
+    async def cli(
+        self,
+        *,
+        command: None = None,
+        commands: list[EapiSimpleCommand | EapiComplexCommand],
+        ofmt: Literal["text"],
+        version: int | Literal["latest"] = "latest",
+        suppress_error: Literal[False] = False,
+        auto_complete: bool = False,
+        expand_aliases: bool = False,
+        timestamps: bool = False,
+        req_id: int | str | None = None,
+    ) -> list[EapiTextOutput]: ...
+
+    # Single command, JSON output, with suppression
+    @overload
+    async def cli(
+        self,
+        *,
+        command: EapiSimpleCommand | EapiComplexCommand,
+        commands: None = None,
+        ofmt: Literal["json"] = "json",
+        version: int | Literal["latest"] = "latest",
+        suppress_error: Literal[True],
+        auto_complete: bool = False,
+        expand_aliases: bool = False,
+        timestamps: bool = False,
+        req_id: int | str | None = None,
+    ) -> EapiJsonOutput | None: ...
+
+    # Multiple commands, JSON output, with suppression
+    @overload
+    async def cli(
+        self,
+        *,
+        command: None = None,
+        commands: list[EapiSimpleCommand | EapiComplexCommand],
+        ofmt: Literal["json"] = "json",
+        version: int | Literal["latest"] = "latest",
+        suppress_error: Literal[True],
+        auto_complete: bool = False,
+        expand_aliases: bool = False,
+        timestamps: bool = False,
+        req_id: int | str | None = None,
+    ) -> list[EapiJsonOutput] | None: ...
+
+    # Single command, TEXT output, with suppression
+    @overload
+    async def cli(
+        self,
+        *,
+        command: EapiSimpleCommand | EapiComplexCommand,
+        commands: None = None,
+        ofmt: Literal["text"],
+        version: int | Literal["latest"] = "latest",
+        suppress_error: Literal[True],
+        auto_complete: bool = False,
+        expand_aliases: bool = False,
+        timestamps: bool = False,
+        req_id: int | str | None = None,
+    ) -> EapiTextOutput | None: ...
+
+    # Multiple commands, TEXT output, with suppression
+    @overload
+    async def cli(
+        self,
+        *,
+        command: None = None,
+        commands: list[EapiSimpleCommand | EapiComplexCommand],
+        ofmt: Literal["text"],
+        version: int | Literal["latest"] = "latest",
+        suppress_error: Literal[True],
+        auto_complete: bool = False,
+        expand_aliases: bool = False,
+        timestamps: bool = False,
+        req_id: int | str | None = None,
+    ) -> list[EapiTextOutput] | None: ...
+
+    # Actual implementation
+    async def cli(
+        self,
+        command: EapiSimpleCommand | EapiComplexCommand | None = None,
+        commands: list[EapiSimpleCommand | EapiComplexCommand] | None = None,
+        ofmt: Literal["json", "text"] = "json",
+        version: int | Literal["latest"] = "latest",
         *,
         suppress_error: bool = False,
         auto_complete: bool = False,
         expand_aliases: bool = False,
+        timestamps: bool = False,
         req_id: int | str | None = None,
-    ) -> list[dict[str, Any] | str] | dict[str, Any] | str | None:
+    ) -> EapiJsonOutput | EapiTextOutput | list[EapiJsonOutput] | list[EapiTextOutput] | None:
         """Execute one or more CLI commands.
 
         Parameters
@@ -143,6 +273,7 @@ class Device(httpx.AsyncClient):
             A list of commands to execute; results in a list of output responses.
         ofmt
             Either 'json' or 'text'; indicates the output format for the CLI commands.
+            eAPI defaults to 'json'.
         version
             By default the eAPI will use "version 1" for all API object models.
             This driver will, by default, always set version to "latest" so
@@ -158,33 +289,55 @@ class Device(httpx.AsyncClient):
 
                 response = dev.cli(..., suppress_error=True)
         auto_complete
-            Enabled/disables the command auto-compelete feature of the EAPI.  Per the
+            Enabled/disables the command auto-compelete feature of the eAPI. Per the
             documentation:
                 Allows users to use shorthand commands in eAPI calls. With this
                 parameter included a user can send 'sh ver' via eAPI to get the
                 output of 'show version'.
         expand_aliases
-            Enables/disables the command use of User defined alias.  Per the
+            Enables/disables the command use of user-defined alias. Per the
             documentation:
                 Allowed users to provide the expandAliases parameter to eAPI
                 calls. This allows users to use aliased commands via the API.
                 For example if an alias is configured as 'sv' for 'show version'
                 then an API call with sv and the expandAliases parameter will
                 return the output of show version.
+        timestamps
+            If True, return the per-command execution time.
         req_id
             A unique identifier that will be echoed back by the switch. May be a string or number.
 
         Returns
         -------
-        list[dict[str, Any] | str] | dict[str, Any] | str | None
-            One or List of output responses, per the description above.
+        dict[str, Any]
+            Single command, JSON output, suppress_error=False
+        list[dict[str, Any]]
+            Multiple commands, JSON output, suppress_error=False
+        str
+            Single command, TEXT output, suppress_error=False
+        list[str]
+            Multiple commands, TEXT output, suppress_error=False
+        dict[str, Any] | None
+            Single command, JSON output, suppress_error=True
+        list[dict[str, Any]] | None
+            Multiple commands, JSON output, suppress_error=True
+        str | None
+            Single command, TEXT output, suppress_error=True
+        list[str] | None
+            Multiple commands, TEXT output, suppress_error=True
         """
         if not any((command, commands)):
             msg = "Required 'command' or 'commands'"
             raise RuntimeError(msg)
 
         jsonrpc = self._jsonrpc_command(
-            commands=[command] if command else commands, ofmt=ofmt, version=version, auto_complete=auto_complete, expand_aliases=expand_aliases, req_id=req_id
+            commands=[command] if command else commands if commands else [],
+            ofmt=ofmt,
+            version=version,
+            auto_complete=auto_complete,
+            expand_aliases=expand_aliases,
+            timestamps=timestamps,
+            req_id=req_id,
         )
 
         try:
@@ -197,14 +350,15 @@ class Device(httpx.AsyncClient):
 
     def _jsonrpc_command(
         self,
-        commands: Sequence[str | dict[str, Any]] | None = None,
-        ofmt: str | None = None,
-        version: int | str | None = "latest",
+        commands: list[EapiSimpleCommand | EapiComplexCommand],
+        ofmt: Literal["json", "text"] = "json",
+        version: int | Literal["latest"] = "latest",
         *,
         auto_complete: bool = False,
         expand_aliases: bool = False,
+        timestamps: bool = False,
         req_id: int | str | None = None,
-    ) -> dict[str, Any]:
+    ) -> JsonRpc:
         """Create the JSON-RPC command dictionary object.
 
         Parameters
@@ -213,6 +367,7 @@ class Device(httpx.AsyncClient):
             A list of commands to execute; results in a list of output responses.
         ofmt
             Either 'json' or 'text'; indicates the output format for the CLI commands.
+            eAPI defaults to 'json'.
         version
             By default the eAPI will use "version 1" for all API object models.
             This driver will, by default, always set version to "latest" so
@@ -232,6 +387,8 @@ class Device(httpx.AsyncClient):
                 For example if an alias is configured as 'sv' for 'show version'
                 then an API call with sv and the expandAliases parameter will
                 return the output of show version.
+        timestamps
+            If True, return the per-command execution time.
         req_id
             A unique identifier that will be echoed back by the switch. May be a string or number.
 
@@ -241,25 +398,21 @@ class Device(httpx.AsyncClient):
             dict containing the JSON payload to run the command.
 
         """
-        cmd: dict[str, Any] = {
+        return {
             "jsonrpc": "2.0",
             "method": "runCmds",
             "params": {
                 "version": version,
                 "cmds": commands,
-                "format": ofmt or self.EAPI_DEFAULT_OFMT,
+                "format": ofmt,
+                "autoComplete": auto_complete,
+                "expandAliases": expand_aliases,
+                "timestamps": timestamps,
             },
             "id": req_id or id(self),
         }
-        if auto_complete is not None:
-            cmd["params"].update({"autoComplete": auto_complete})
-
-        if expand_aliases is not None:
-            cmd["params"].update({"expandAliases": expand_aliases})
-
-        return cmd
 
-    async def jsonrpc_exec(self, jsonrpc: dict[str, Any]) -> list[dict[str, Any] | str]:
+    async def jsonrpc_exec(self, jsonrpc: JsonRpc) -> list[EapiJsonOutput] | list[EapiTextOutput]:
         """Execute the JSON-RPC dictionary object.
 
         Parameters
@@ -315,7 +468,7 @@ class Device(httpx.AsyncClient):
         failed_cmd = commands[err_at]
 
         raise EapiCommandError(
-            passed=[get_output(cmd_data[cmd_i]) for cmd_i, cmd in enumerate(commands[:err_at])],
+            passed=[get_output(cmd_data[i]) for i in range(err_at)],
             failed=failed_cmd["cmd"] if isinstance(failed_cmd, dict) else failed_cmd,
             errors=cmd_data[err_at]["errors"],
             errmsg=err_msg,
diff --git a/asynceapi/errors.py b/asynceapi/errors.py
index ee1faf2e..50b02c6f 100644
--- a/asynceapi/errors.py
+++ b/asynceapi/errors.py
@@ -6,13 +6,16 @@
 
 from __future__ import annotations
 
-from typing import Any
+from typing import TYPE_CHECKING
 
 import httpx
 
+if TYPE_CHECKING:
+    from ._types import EapiComplexCommand, EapiJsonOutput, EapiSimpleCommand, EapiTextOutput
+
 
 class EapiCommandError(RuntimeError):
-    """Exception class for EAPI command errors.
+    """Exception class for eAPI command errors.
 
     Attributes
     ----------
@@ -23,7 +26,14 @@ class EapiCommandError(RuntimeError):
         not_exec: a list of commands that were not executed
     """
 
-    def __init__(self, failed: str, errors: list[str], errmsg: str, passed: list[str | dict[str, Any]], not_exec: list[dict[str, Any]]) -> None:
+    def __init__(
+        self,
+        failed: str,
+        errors: list[str],
+        errmsg: str,
+        passed: list[EapiJsonOutput] | list[EapiTextOutput],
+        not_exec: list[EapiSimpleCommand | EapiComplexCommand],
+    ) -> None:
         """Initialize for the EapiCommandError exception."""
         self.failed = failed
         self.errmsg = errmsg
diff --git a/sonar-project.properties b/sonar-project.properties
index 0abd967f..0cc0d7cc 100644
--- a/sonar-project.properties
+++ b/sonar-project.properties
@@ -3,7 +3,9 @@ sonar.organization=aristanetworks-1
 
 # Path to sources
 sonar.sources=anta/,asynceapi/
-#sonar.exclusions=
+
+# Exclude asynceapi unused modules for now
+sonar.exclusions=asynceapi/config_session.py
 #sonar.inclusions=
 
 # Path to tests


Test patch:
diff --git a/tests/units/asynceapi/test_data.py b/tests/units/asynceapi/test_data.py
index f164c117..a1400fe8 100644
--- a/tests/units/asynceapi/test_data.py
+++ b/tests/units/asynceapi/test_data.py
@@ -3,7 +3,9 @@
 # that can be found in the LICENSE file.
 """Unit tests data for the asynceapi client package."""
 
-SUCCESS_EAPI_RESPONSE = {
+from asynceapi._types import EapiJsonOutput, JsonRpc
+
+SUCCESS_EAPI_RESPONSE: EapiJsonOutput = {
     "jsonrpc": "2.0",
     "id": "EapiExplorer-1",
     "result": [
@@ -49,7 +51,7 @@ SUCCESS_EAPI_RESPONSE = {
 }
 """Successful eAPI JSON response."""
 
-ERROR_EAPI_RESPONSE = {
+ERROR_EAPI_RESPONSE: EapiJsonOutput = {
     "jsonrpc": "2.0",
     "id": "EapiExplorer-1",
     "error": {
@@ -84,5 +86,5 @@ ERROR_EAPI_RESPONSE = {
 }
 """Error eAPI JSON response."""
 
-JSONRPC_REQUEST_TEMPLATE = {"jsonrpc": "2.0", "method": "runCmds", "params": {"version": 1, "cmds": [], "format": "json"}, "id": "EapiExplorer-1"}
+JSONRPC_REQUEST_TEMPLATE: JsonRpc = {"jsonrpc": "2.0", "method": "runCmds", "params": {"version": 1, "cmds": [], "format": "json"}, "id": "EapiExplorer-1"}
 """Template for JSON-RPC eAPI request. `cmds` must be filled by the parametrize decorator."""
diff --git a/tests/units/asynceapi/test_device.py b/tests/units/asynceapi/test_device.py
index da7af123..b2293d91 100644
--- a/tests/units/asynceapi/test_device.py
+++ b/tests/units/asynceapi/test_device.py
@@ -5,7 +5,7 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 
 import pytest
 from httpx import HTTPStatusError
@@ -17,6 +17,8 @@ from .test_data import ERROR_EAPI_RESPONSE, JSONRPC_REQUEST_TEMPLATE, SUCCESS_EA
 if TYPE_CHECKING:
     from pytest_httpx import HTTPXMock
 
+    from asynceapi._types import EapiComplexCommand, EapiSimpleCommand
+
 
 @pytest.mark.parametrize(
     "cmds",
@@ -30,10 +32,10 @@ if TYPE_CHECKING:
 async def test_jsonrpc_exec_success(
     asynceapi_device: Device,
     httpx_mock: HTTPXMock,
-    cmds: list[str | dict[str, Any]],
+    cmds: list[EapiSimpleCommand | EapiComplexCommand],
 ) -> None:
     """Test the Device.jsonrpc_exec method with a successful response. Simple and complex commands are tested."""
-    jsonrpc_request: dict[str, Any] = JSONRPC_REQUEST_TEMPLATE.copy()
+    jsonrpc_request = JSONRPC_REQUEST_TEMPLATE.copy()
     jsonrpc_request["params"]["cmds"] = cmds
 
     httpx_mock.add_response(json=SUCCESS_EAPI_RESPONSE)
@@ -55,13 +57,13 @@ async def test_jsonrpc_exec_success(
 async def test_jsonrpc_exec_eapi_command_error(
     asynceapi_device: Device,
     httpx_mock: HTTPXMock,
-    cmds: list[str | dict[str, Any]],
+    cmds: list[EapiSimpleCommand | EapiComplexCommand],
 ) -> None:
     """Test the Device.jsonrpc_exec method with an error response. Simple and complex commands are tested."""
-    jsonrpc_request: dict[str, Any] = JSONRPC_REQUEST_TEMPLATE.copy()
+    jsonrpc_request = JSONRPC_REQUEST_TEMPLATE.copy()
     jsonrpc_request["params"]["cmds"] = cmds
 
-    error_eapi_response: dict[str, Any] = ERROR_EAPI_RESPONSE.copy()
+    error_eapi_response = ERROR_EAPI_RESPONSE.copy()
     httpx_mock.add_response(json=error_eapi_response)
 
     with pytest.raises(EapiCommandError) as exc_info:
@@ -76,7 +78,7 @@ async def test_jsonrpc_exec_eapi_command_error(
 
 async def test_jsonrpc_exec_http_status_error(asynceapi_device: Device, httpx_mock: HTTPXMock) -> None:
     """Test the Device.jsonrpc_exec method with an HTTPStatusError."""
-    jsonrpc_request: dict[str, Any] = JSONRPC_REQUEST_TEMPLATE.copy()
+    jsonrpc_request = JSONRPC_REQUEST_TEMPLATE.copy()
     jsonrpc_request["params"]["cmds"] = ["show version"]
 
     httpx_mock.add_response(status_code=500, text="Internal Server Error")