Narrow bare dict/list annotations to parameterized types

aai_cli/agent/session.py

@@ -56,7 +56,7 @@ def should_send_audio(self) -> bool:
         with self._lock:
             return self.ready and not self.muted
 
-    def dispatch(self, event: dict) -> None:
+    def dispatch(self, event: dict[str, Any]) -> None:
         """Route one server event to its handler; unknown types are ignored.
 
         Handlers are registered in ``_EVENT_HANDLERS`` (below the class). Events with
@@ -66,41 +66,41 @@ def dispatch(self, event: dict) -> None:
         if handler is not None:
             handler(self, event)
 
-    def _on_session_ready(self, _event: dict) -> None:
+    def _on_session_ready(self, _event: dict[str, Any]) -> None:
         with self._lock:
             self.ready = True
         if self.ready_event is not None:
             self.ready_event.set()
         self.renderer.connected()
 
-    def _on_speech_started(self, _event: dict) -> None:
+    def _on_speech_started(self, _event: dict[str, Any]) -> None:
         if self.full_duplex:
             self.player.flush()
 
-    def _on_user_delta(self, event: dict) -> None:
+    def _on_user_delta(self, event: dict[str, Any]) -> None:
         self.renderer.user_partial(event.get("text", ""))
 
-    def _on_user_final(self, event: dict) -> None:
+    def _on_user_final(self, event: dict[str, Any]) -> None:
         self._saw_user = True
         self.renderer.user_final(event.get("text", ""))
 
-    def _on_reply_started(self, _event: dict) -> None:
+    def _on_reply_started(self, _event: dict[str, Any]) -> None:
         if not self.full_duplex:
             with self._lock:
                 self.muted = True
         self.renderer.reply_started()
 
-    def _on_reply_audio(self, event: dict) -> None:
+    def _on_reply_audio(self, event: dict[str, Any]) -> None:
         data = event.get("data")
         if data:
             self.player.enqueue(base64.b64decode(data))
 
-    def _on_agent_transcript(self, event: dict) -> None:
+    def _on_agent_transcript(self, event: dict[str, Any]) -> None:
         self.renderer.agent_transcript(
             event.get("text", ""), interrupted=bool(event.get("interrupted", False))
         )
 
-    def _on_reply_done(self, event: dict) -> None:
+    def _on_reply_done(self, event: dict[str, Any]) -> None:
         if not self.full_duplex:
             with self._lock:
                 self.muted = False
@@ -112,7 +112,7 @@ def _on_reply_done(self, event: dict) -> None:
         if self.exit_after_reply and self._saw_user and not interrupted:
             self.finished = True
 
-    def _raise_error(self, event: dict) -> None:
+    def _raise_error(self, event: dict[str, Any]) -> None:
         code = event.get("code", "")
         message = event.get("message") or code or "Voice agent error."
         if code in _AUTH_ERROR_CODES:
@@ -126,7 +126,7 @@ def _raise_error(self, event: dict) -> None:
 
 # Server event type -> the VoiceAgentSession method that handles it. Types absent
 # here (input.speech.stopped, tool.call, anything unrecognized) are ignored.
-_EVENT_HANDLERS: dict[str, Callable[[VoiceAgentSession, dict], None]] = {
+_EVENT_HANDLERS: dict[str, Callable[[VoiceAgentSession, dict[str, Any]], None]] = {
     "session.ready": VoiceAgentSession._on_session_ready,
     "input.speech.started": VoiceAgentSession._on_speech_started,
     "transcript.user.delta": VoiceAgentSession._on_user_delta,

aai_cli/commands/account.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from datetime import UTC, datetime, timedelta
+from typing import Any
 
 import typer
 from rich.markup import escape
@@ -64,7 +65,7 @@ def body(state: AppState, json_mode: bool) -> None:
         start_date = start or (today - timedelta(days=30)).isoformat()
         data = ams.get_usage(jwt, start_date, end_date, window)
 
-        def render(d: dict) -> Table:
+        def render(d: dict[str, Any]) -> Table:
             table = Table("window start", "window end", "total", header_style="aai.heading")
             for item in d.get("usage_items", []):
                 table.add_row(
@@ -96,7 +97,7 @@ def body(state: AppState, json_mode: bool) -> None:
         account_id, jwt = resolve_session(state)
         data = ams.get_rate_limits(account_id, jwt)
 
-        def render(d: dict) -> Table:
+        def render(d: dict[str, Any]) -> Table:
             table = Table("service", "limit", header_style="aai.heading")
             for limit in d.get("rate_limits", []):
                 table.add_row(escape(str(limit["service"])), f"{limit['magnitude']:,}")

aai_cli/commands/audit.py

@@ -34,7 +34,7 @@ def body(state: AppState, json_mode: bool) -> None:
         payload = ams.list_audit_logs(jwt, limit=limit, action_taken=action, resource_type=resource)
         rows = payload.get("data", [])
 
-        def render(data: list[dict]) -> Table:
+        def render(data: list[dict[str, object]]) -> Table:
             table = Table("time", "actor", "action", "resource", header_style="aai.heading")
             for entry in data:
                 actor = str(entry["actor_type"])

aai_cli/commands/sessions.py

@@ -49,7 +49,7 @@ def body(state: AppState, json_mode: bool) -> None:
         payload = ams.list_streaming(jwt, limit=limit, status=status)
         rows = payload.get("data", [])
 
-        def render(data: list[dict]) -> Table:
+        def render(data: list[dict[str, object]]) -> Table:
             table = Table(
                 "session id",
                 "status",
@@ -92,7 +92,7 @@ def body(state: AppState, json_mode: bool) -> None:
         _, jwt = resolve_session(state)
         data = ams.get_streaming(session_id, jwt)
 
-        def render(d: dict) -> Table:
+        def render(d: dict[str, object]) -> Table:
             table = Table(show_header=False)
             for field in _DETAIL_FIELDS:
                 value = d.get(field)

aai_cli/commands/transcribe.py

@@ -2,6 +2,7 @@
 
 import tempfile
 from pathlib import Path
+from typing import Any
 
 import typer
 
@@ -23,7 +24,7 @@
 app = typer.Typer()
 
 
-def _render_transform_steps(d: dict) -> str:
+def _render_transform_steps(d: dict[str, Any]) -> str:
     """Human view of chained LLM-Gateway steps: the lone output, or each step labeled."""
     steps = d["transform"]["steps"]
     if len(steps) == 1:

aai_cli/config.py

@@ -61,7 +61,7 @@ def _load() -> dict[str, Any]:
         return data
 
 
-def _dump(data: dict) -> None:
+def _dump(data: dict[str, Any]) -> None:
     path = _config_file()
     path.parent.mkdir(parents=True, exist_ok=True)
     # Write to a sibling temp file and atomically rename over the target, so a crash