simplify: dedupe webhook-header, code-gen gateway, proc-detail, and nested-attr helpers

aai_cli/agent/session.py

@@ -159,9 +159,8 @@ def run_session(
     agent's first reply to the spoken input and the capture thread waits for
     session.ready before streaming the source.
     """
-    _connect = connect
-    if _connect is None:
-        from websockets.sync.client import connect as _connect
+    if connect is None:
+        from websockets.sync.client import connect
 
     ready_event = threading.Event() if exit_after_reply else None
     session = VoiceAgentSession(
@@ -173,7 +172,7 @@ def run_session(
     )
 
     try:
-        ws = _connect(WS_URL, additional_headers={"Authorization": f"Bearer {api_key}"})
+        ws = connect(WS_URL, additional_headers={"Authorization": f"Bearer {api_key}"})
     except Exception as exc:
         if _is_auth_rejection(exc):
             raise auth_failure() from exc

aai_cli/code_gen/__init__.py

@@ -5,6 +5,13 @@
 from aai_cli.code_gen import transcribe as _transcribe
 
 
+def gateway_options(prompts: list[str], model: str, max_tokens: int) -> dict[str, object] | None:
+    """The LLM-gateway options dict consumed by `transcribe`/`stream`, or None if no prompts."""
+    if not prompts:
+        return None
+    return {"prompts": list(prompts), "model": model, "max_tokens": max_tokens}
+
+
 def agent(voice: str, system_prompt: str, greeting: str) -> str:
     """Generate runnable Python that reproduces this voice-agent session."""
     return _agent.render(voice, system_prompt, greeting)

aai_cli/commands/claude.py

@@ -46,6 +46,11 @@ def _run(cmd: list[str], *, timeout: float = 120) -> subprocess.CompletedProcess
         )
 
 
+def _proc_detail(proc: subprocess.CompletedProcess[str]) -> str:
+    """The error text from a finished process: stderr if present, else stdout."""
+    return (proc.stderr or proc.stdout).strip()
+
+
 def _mcp_present() -> bool:
     return _run(["claude", "mcp", "get", MCP_NAME]).returncode == 0
 
@@ -69,14 +74,13 @@ def _install_mcp(scope: str, force: bool) -> Step:
             return {
                 "name": "mcp",
                 "status": "failed",
-                "detail": f"could not remove existing {MCP_NAME}: "
-                + (removed.stderr or removed.stdout).strip(),
+                "detail": f"could not remove existing {MCP_NAME}: " + _proc_detail(removed),
             }
     proc = _run(
         ["claude", "mcp", "add", "--transport", "http", "--scope", scope, MCP_NAME, MCP_URL]
     )
     if proc.returncode != 0:
-        return {"name": "mcp", "status": "failed", "detail": (proc.stderr or proc.stdout).strip()}
+        return {"name": "mcp", "status": "failed", "detail": _proc_detail(proc)}
     return {"name": "mcp", "status": "installed", "detail": f"{MCP_NAME} @ {scope} scope"}
 
 
@@ -106,7 +110,7 @@ def _install_skill(force: bool) -> Step:
     # looks. npx -y skips its install prompt; the longer timeout covers the download.
     proc = _run(_SKILL_ADD, timeout=300)
     if proc.returncode != 0:
-        return {"name": "skill", "status": "failed", "detail": (proc.stderr or proc.stdout).strip()}
+        return {"name": "skill", "status": "failed", "detail": _proc_detail(proc)}
     # Trust the filesystem, not the exit code: confirm the skill actually landed
     # where `status` looks, so the two commands can never disagree.
     if not _skill_installed():
@@ -162,7 +166,7 @@ def _remove_mcp(scope: str | None) -> Step:
         cmd += ["--scope", scope]
     proc = _run(cmd)
     if proc.returncode != 0:
-        return {"name": "mcp", "status": "failed", "detail": (proc.stderr or proc.stdout).strip()}
+        return {"name": "mcp", "status": "failed", "detail": _proc_detail(proc)}
     return {"name": "mcp", "status": "removed", "detail": MCP_NAME}
 
 
@@ -179,7 +183,7 @@ def _remove_skill() -> Step:
     # do the removal (a plain rmtree would choke on the symlink and orphan the store).
     proc = _run(_SKILL_REMOVE, timeout=120)
     if proc.returncode != 0 or _skill_installed():
-        detail = (proc.stderr or proc.stdout).strip() or "skill still present after removal"
+        detail = _proc_detail(proc) or "skill still present after removal"
         return {"name": "skill", "status": "failed", "detail": detail}
     return {"name": "skill", "status": "removed", "detail": str(_skill_dir())}
 

aai_cli/commands/stream.py

@@ -155,10 +155,7 @@ def make_flags(rate: int) -> dict[str, object]:
                 "webhook_url": webhook_url,
                 "prompt": prompt,
             }
-            header = config_builder.parse_auth_header(webhook_auth_header)
-            if header is not None:
-                flags["webhook_auth_header_name"] = header[0]
-                flags["webhook_auth_header_value"] = header[1]
+            flags.update(config_builder.auth_header_flags(webhook_auth_header))
             return flags
 
         if show_code:
@@ -170,11 +167,7 @@ def make_flags(rate: int) -> dict[str, object]:
                 overrides=list(config_kv or []),
                 config_file=config_file,
             )
-            gateway = (
-                {"prompts": list(llm_prompt), "model": model, "max_tokens": max_tokens}
-                if llm_prompt
-                else None
-            )
+            gateway = code_gen.gateway_options(llm_prompt, model, max_tokens)
             output.print_code(code_gen.stream(merged, llm=gateway))
             return
 

aai_cli/commands/transcribe.py

@@ -196,10 +196,7 @@ def body(state: AppState, json_mode: bool) -> None:
                 config_builder.translation_request(list(translate_to)) if translate_to else None
             ),
         }
-        header = config_builder.parse_auth_header(webhook_auth_header)
-        if header is not None:
-            flags["webhook_auth_header_name"] = header[0]
-            flags["webhook_auth_header_value"] = header[1]
+        flags.update(config_builder.auth_header_flags(webhook_auth_header))
 
         merged = config_builder.merge_transcribe_config(
             flags=flags, overrides=list(config_kv or []), config_file=config_file
@@ -210,11 +207,7 @@ def body(state: AppState, json_mode: bool) -> None:
             # transcribing or authenticating. Raw stdout so `--show-code > script.py`
             # yields a runnable file.
             audio = client.resolve_audio_source(source, sample=sample)
-            gateway = (
-                {"prompts": list(llm_prompt), "model": model, "max_tokens": max_tokens}
-                if llm_prompt
-                else None
-            )
+            gateway = code_gen.gateway_options(llm_prompt, model, max_tokens)
             output.print_code(code_gen.transcribe(merged, audio, llm_gateway=gateway))
             return
 

aai_cli/config_builder.py

@@ -317,6 +317,14 @@ def parse_auth_header(value: str | None) -> tuple[str, str] | None:
     return name.strip(), header_value.strip()
 
 
+def auth_header_flags(value: str | None) -> dict[str, object]:
+    """Webhook auth-header config flags for a `NAME:VALUE` value, or `{}` when unset."""
+    header = parse_auth_header(value)
+    if header is None:
+        return {}
+    return {"webhook_auth_header_name": header[0], "webhook_auth_header_value": header[1]}
+
+
 def load_custom_spelling(path: str) -> dict[str, object]:
     """Load a custom-spelling JSON map (e.g. {"AssemblyAI": ["assembly ai"]})."""
     try:

aai_cli/render.py

@@ -44,8 +44,8 @@ def _status(self, message: str) -> None:
 
     # --- JSON output (plain text; preserves BrokenPipe for `| head`) -------
     def _emit(self, obj: object) -> None:
-        """Write one NDJSON event."""
-        self._write(json.dumps(obj) + "\n")
+        """Write one NDJSON event (default=str matches output.emit_ndjson's safety)."""
+        self._write(json.dumps(obj, default=str) + "\n")
 
     def _write(self, text: str) -> None:
         try:

aai_cli/transcribe_render.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from collections import Counter
+from typing import Any
 
 from rich.console import Console
 from rich.text import Text
@@ -17,6 +18,12 @@ def _enum_value(obj: object) -> str:
     return str(getattr(obj, "value", obj))
 
 
+def _nested(transcript: object, outer: str, inner: str) -> Any:
+    """Return ``transcript.<outer>.<inner>``, or None if either link is missing/falsy."""
+    mid = getattr(transcript, outer, None)
+    return getattr(mid, inner, None) if mid else None
+
+
 def render_transcript_result(transcript: object, console: Console) -> None:
     """Print the transcript text, then a section per analysis feature present."""
     _render_text(transcript, console)
@@ -63,8 +70,7 @@ def _render_chapters(transcript: object, console: Console) -> None:
 
 
 def _render_highlights(transcript: object, console: Console) -> None:
-    highlights = getattr(transcript, "auto_highlights", None)
-    results = getattr(highlights, "results", None) if highlights else None
+    results = _nested(transcript, "auto_highlights", "results")
     if not results:
         return
     console.print("\n[bold]Highlights:[/bold]")
@@ -92,8 +98,7 @@ def _render_entities(transcript: object, console: Console) -> None:
 
 
 def _render_topics(transcript: object, console: Console) -> None:
-    iab = getattr(transcript, "iab_categories", None)
-    summary = getattr(iab, "summary", None) if iab else None
+    summary = _nested(transcript, "iab_categories", "summary")
     if not summary:
         return
     console.print("\n[bold]Topics:[/bold]")
@@ -102,8 +107,7 @@ def _render_topics(transcript: object, console: Console) -> None:
 
 
 def _render_content_safety(transcript: object, console: Console) -> None:
-    safety = getattr(transcript, "content_safety", None)
-    summary = getattr(safety, "summary", None) if safety else None
+    summary = _nested(transcript, "content_safety", "summary")
     if not summary:
         return
     console.print("\n[bold]Content Safety:[/bold]")