From 67145cad93dbcd8b7c78ef59ea5e96f7e9c28517 Mon Sep 17 00:00:00 2001 From: Abhishek Mishra Date: Tue, 18 Aug 2026 18:03:39 +0530 Subject: [PATCH 1/2] feat(cli): stream live call events + build logs from the crew runtime MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Surfaces observability data that the platform already emits but the CLI did not expose: - smallestai calls events — stream a live call's events over SSE (GET /events): transcript turns, per-turn latency, node transitions, tool calls, and errors, as they happen. 400 (call finished) and 404 map to clear messages. - smallestai calls transcript --follow — same stream, filtered to the conversation turns. - smallestai agent-crew logs [buildId] — stream a build's compile + deploy logs (GET .../builds/:id/stream); defaults to the latest build. Revives the previously-disabled logs command. Adds two SSE helpers to the CLI HTTP client (stream_call_events, stream_agent_build) sharing one data-frame parser. All use the session token the CLI already holds. Tests cover SSE frame parsing and the event renderers. --- src/smallestai/cli/agent_crew.py | 121 +++++++++++------------ src/smallestai/cli/calls.py | 109 +++++++++++++++++++- src/smallestai/cli/lib/atoms.py | 69 ++++++++----- tests/custom/test_cli_event_streaming.py | 87 ++++++++++++++++ 4 files changed, 296 insertions(+), 90 deletions(-) create mode 100644 tests/custom/test_cli_event_streaming.py diff --git a/src/smallestai/cli/agent_crew.py b/src/smallestai/cli/agent_crew.py index f2967545..822f12cc 100644 --- a/src/smallestai/cli/agent_crew.py +++ b/src/smallestai/cli/agent_crew.py @@ -476,70 +476,63 @@ async def _manage_build(agent_id: str, build, access_token: str): ) console.print(f"[bold green]✓ Build {build.id[:12]}... has been taken down.[/bold green]") - # @app.command("logs") - # def stream_build( - # build_id: str = typer.Argument(..., help="The build ID to stream logs for"), - # ): - # """ - # Stream build logs in real-time using Server-Sent Events. - # """ - # asyncio.run(async_stream_build(build_id)) - - # async def async_stream_build(build_id: str): - # """Async implementation of stream build command.""" - # agent_id = project_config.get_agent_id() - - # if not agent_id: - # console.print( - # "[red]Agent not initialized. Run 'smallestai agent init' first.[/red]" - # ) - # return - - # credentials = auth_client.get_credentials() - # if not credentials or not credentials.get("access_token"): - # console.print( - # "[red]Error: You must be logged in first. Run 'smallestai auth login'[/red]" - # ) - # raise typer.Exit(1) - - # access_token = credentials["access_token"] - - # console.print(f"[bold cyan]Streaming logs for build: {build_id}[/bold cyan]") - # console.print("[dim]Press Ctrl+C to stop streaming[/dim]\n") - - # try: - # async for event in atoms_client.stream_agent_build( - # agent_id=agent_id, - # build_id=build_id, - # api_key=access_token, - # ): - # event_type = event.get("type") - - # if event_type == "log": - # console.print(f"[dim]LOG:[/dim] {event.get('message', '')}") - # elif event_type == "status": - # status = event.get("status", "") - # status_style = { - # "SUCCEEDED": "[green]SUCCEEDED[/green]", - # "BUILD_FAILED": "[red]BUILD_FAILED[/red]", - # "DEPLOY_FAILED": "[red]DEPLOY_FAILED[/red]", - # "PENDING": "[yellow]PENDING[/yellow]", - # "BUILDING": "[yellow]BUILDING[/yellow]", - # "DEPLOYING": "[yellow]DEPLOYING[/yellow]", - # }.get(status, status) - # console.print(f"[bold]STATUS:[/bold] {status_style}") - - # if status in ["SUCCEEDED", "BUILD_FAILED", "DEPLOY_FAILED"]: - # console.print("\n[bold]Build stream ended.[/bold]") - # break - # elif event_type == "error": - # console.print(f"[red]ERROR:[/red] {event.get('message', '')}") - # break - - # except KeyboardInterrupt: - # console.print("\n[yellow]Streaming stopped by user.[/yellow]") - # except Exception as e: - # console.print(f"[red]Error streaming build logs: {e}[/red]") + @app.command("logs") + def build_logs( + build_id: str = typer.Argument(None, help="Build ID to stream logs for (defaults to the latest build)"), + ): + """Stream a build's logs (compile + deploy) in real time. + + With no build ID, streams the most recent build for the current agent. + Use this to debug a deploy that failed or to watch one in progress. + """ + asyncio.run(async_build_logs(build_id)) + + async def async_build_logs(build_id: str | None): + agent_id = project_config.get_agent_id() + if not agent_id: + console.print("[red]Agent not initialized. Run 'smallestai agent-crew init' first.[/red]") + raise typer.Exit(1) + + credentials = auth_client.get_credentials() + if not credentials or not credentials.get("access_token"): + console.print("[red]Error: You must be logged in first. Run 'smallestai auth login'[/red]") + raise typer.Exit(1) + access_token = credentials["access_token"] + + if not build_id: + result = await atoms_client.list_agent_builds(agent_id=agent_id, api_key=access_token, limit=1, offset=0) + if not result.builds: + console.print("[yellow]No builds found. Run 'smallestai agent-crew deploy' first.[/yellow]") + raise typer.Exit(1) + build_id = result.builds[0].id + console.print(f"[dim]Latest build: {build_id}[/dim]") + + console.print( + f"[bold cyan]Streaming logs for build {build_id[:12]}...[/bold cyan] [dim](Ctrl+C to stop)[/dim]\n" + ) + + terminal = {"SUCCEEDED", "BUILD_FAILED", "DEPLOY_FAILED"} + try: + async for event in atoms_client.stream_agent_build( + agent_id=agent_id, build_id=build_id, access_token=access_token + ): + etype = event.get("type") + if etype == "log": + console.print(event.get("message", ""), highlight=False) + elif etype == "status": + status = str(event.get("status", "")) + color = {"SUCCEEDED": "green", "BUILD_FAILED": "red", "DEPLOY_FAILED": "red"}.get(status, "yellow") + console.print(f"[bold {color}]● {status}[/bold {color}]") + if status in terminal: + break + elif etype == "error": + console.print(f"[red]error:[/red] {event.get('message', '')}") + break + except KeyboardInterrupt: + console.print("\n[yellow]Stopped.[/yellow]") + except Exception as e: + console.print(f"[red]Error streaming build logs: {e}[/red]") + raise typer.Exit(1) @app.command() def doctor( diff --git a/src/smallestai/cli/calls.py b/src/smallestai/cli/calls.py index 4066c6b8..b4d8becf 100644 --- a/src/smallestai/cli/calls.py +++ b/src/smallestai/cli/calls.py @@ -4,18 +4,101 @@ `smallestai agents call`. """ +import asyncio import json as _json +import httpx import typer from rich.console import Console from rich.table import Table +from smallestai.cli.lib.atoms import AtomsAPIClient from smallestai.cli.lib.auth import AuthClient from smallestai.cli.lib.client import make_client console = Console() +def _session_token(auth_client: AuthClient) -> str: + """Session access token for live endpoints (events SSE) that need it.""" + creds = auth_client.get_credentials() + token = (creds or {}).get("access_token") + if not token: + console.print("[red]You must be logged in. Run 'smallestai auth login'.[/red]") + raise typer.Exit(1) + return token + + +def _render_event(ev: dict) -> None: + """Pretty-print one live call event.""" + et = ev.get("event_type") + if et == "call_start": + console.print("[dim]● call started[/dim]") + elif et == "user_transcription": + console.print(f"[cyan]user[/cyan]: {ev.get('user_transcription_text', '')}") + elif et == "tts_completed": + console.print(f"[green]agent[/green]: {ev.get('tts_text', '')}") + elif et == "turn_latency": + console.print(f"[dim] latency: {ev.get('turn_latency')}[/dim]") + elif et == "agent_node_state": + console.print(f"[dim] node: {ev.get('node_name')} ({ev.get('node_type')})[/dim]") + elif et == "hopping": + console.print(f"[dim] hop: {ev.get('from_node_id')} -> {ev.get('to_node_id')}[/dim]") + elif et in ("tool_call_start", "tool_call_end", "tool_call_error"): + console.print( + f"[magenta] {et}[/magenta]: {_json.dumps({k: v for k, v in ev.items() if k != 'event_type'}, default=str)}" + ) + elif et in ("agent_error", "generic_call_error"): + console.print(f"[red] error: {ev.get('error') or ev.get('message') or ev}[/red]") + else: + rest = {k: v for k, v in ev.items() if k != "event_type"} + console.print(f"[dim] {et}{': ' + _json.dumps(rest, default=str) if rest else ''}[/dim]") + + +def _render_transcript_event(ev: dict) -> None: + """Print only the conversation turns from a live event stream.""" + et = ev.get("event_type") + if et == "user_transcription" and ev.get("user_transcription_text"): + console.print(f"[cyan]user[/cyan]: {ev['user_transcription_text']}") + elif et == "tts_completed" and ev.get("tts_text"): + console.print(f"[green]agent[/green]: {ev['tts_text']}") + + +async def _stream_call_events(call_id: str, token: str, as_json: bool, transcript_only: bool) -> None: + client = AtomsAPIClient() + try: + async for ev in client.stream_call_events(call_id, token): + et = ev.get("event_type") + if et == "sse_init": + if not as_json and not transcript_only: + console.print("[dim]connected — waiting for events…[/dim]") + continue + if as_json: + console.print_json(_json.dumps(ev, default=str)) + elif transcript_only: + _render_transcript_event(ev) + else: + _render_event(ev) + if et in ("sse_close", "call_end"): + if not as_json and not transcript_only: + console.print("[dim]— call ended —[/dim]") + break + except httpx.HTTPStatusError as e: + code = e.response.status_code + if code == 400: + console.print( + f"[yellow]This call is not live (already finished). " + f"Run 'smallestai calls transcript {call_id}' for the final transcript.[/yellow]" + ) + elif code == 404: + console.print("[yellow]Call not found.[/yellow]") + else: + console.print(f"[red]Error: {e}[/red]") + raise typer.Exit(1) + except KeyboardInterrupt: + console.print("\n[yellow]Stopped.[/yellow]") + + def _data(resp): """Unwrap `.data` from an SDK response (or return the object itself).""" return getattr(resp, "data", resp) @@ -93,12 +176,36 @@ def get_call( console.print(f" recording : {rec or '[dim]none[/dim]'}") console.print(" [dim]transcript: smallestai calls transcript " + call_id + "[/dim]") + @calls_app.command("events") + def events( + call_id: str, + as_json: bool = typer.Option(False, "--json", help="Emit raw JSON, one event per line"), + ): + """Stream a live call's events as they happen. + + Shows transcript turns, per-turn latency, node transitions, tool calls, + and errors while the call is in progress. For a finished call use + `smallestai calls transcript`. + """ + token = _session_token(auth_client) + asyncio.run(_stream_call_events(call_id, token, as_json, transcript_only=False)) + @calls_app.command("transcript") def transcript( call_id: str, + follow: bool = typer.Option( + False, "--follow", "-f", help="Stream the transcript live (call must be in progress)" + ), as_json: bool = typer.Option(False, "--json", help="Emit raw JSON"), ): - """Print a call's transcript, one turn per line.""" + """Print a call's transcript, one turn per line. + + With --follow, streams turns live while the call is in progress. + """ + if follow: + token = _session_token(auth_client) + asyncio.run(_stream_call_events(call_id, token, as_json, transcript_only=True)) + return d = _data(make_client(auth_client).atoms.calls.get(id=call_id)) turns = getattr(d, "transcript", None) or [] if as_json: diff --git a/src/smallestai/cli/lib/atoms.py b/src/smallestai/cli/lib/atoms.py index abbec48f..32bc0866 100644 --- a/src/smallestai/cli/lib/atoms.py +++ b/src/smallestai/cli/lib/atoms.py @@ -1,3 +1,4 @@ +import json as _json from enum import Enum from typing import List, Optional @@ -305,28 +306,46 @@ async def update_agent_build( return update_build_response.data - # async def stream_agent_build( - # self, - # agent_id: str, - # build_id: str, - # api_key: str, - # ): - # """ - # Stream build logs using Server-Sent Events. - # Yields tuples of (event_type, data) where event_type is 'log', 'status', or 'error'. - # """ - # async with httpx.AsyncClient(timeout=None) as client: - # async with client.stream( - # "GET", - # f"{self.base_url}/atoms/v1/sdk/agents/{agent_id}/builds/{build_id}/stream", - # headers={ - # "Authorization": f"Bearer {api_key}", - # }, - # ) as response: - # response.raise_for_status() - # async for line in response.aiter_lines(): - # if line.startswith("data: "): - # import json - - # data = json.loads(line[6:]) - # yield data + async def _stream_sse(self, url: str, access_token: str): + """Open an SSE stream and yield each `data:` frame as a parsed dict. + + Shared by build-log and call-event streaming. Blank keep-alive lines and + non-JSON frames are skipped. The stream ends when the server closes it. + """ + async with httpx.AsyncClient(timeout=None) as client: + async with client.stream( + "GET", + url, + headers={"Authorization": f"Bearer {access_token}"}, + ) as response: + response.raise_for_status() + async for line in response.aiter_lines(): + if not line.startswith("data:"): + continue + payload = line[5:].lstrip() + if not payload: + continue + try: + yield _json.loads(payload) + except _json.JSONDecodeError: + continue + + async def stream_agent_build(self, agent_id: str, build_id: str, access_token: str): + """Stream a build's logs (SSE). Yields dicts of the form + `{"type": "log"|"status"|"error", "message"|"status": ...}`. + """ + url = f"{self.base_url}/atoms/v1/sdk/agents/{agent_id}/builds/{build_id}/stream" + async for event in self._stream_sse(url, access_token): + yield event + + async def stream_call_events(self, call_id: str, access_token: str): + """Stream a live call's events (SSE). Yields dicts with an `event_type` + field (e.g. `user_transcription`, `tts_completed`, `turn_latency`, + `agent_node_state`, `tool_call_start`, `agent_error`, `call_end`). + + The call must be in progress; the platform returns 400 for a completed + call (use `calls transcript` for finished calls). + """ + url = f"{self.base_url}/atoms/v1/events?callId={call_id}" + async for event in self._stream_sse(url, access_token): + yield event diff --git a/tests/custom/test_cli_event_streaming.py b/tests/custom/test_cli_event_streaming.py new file mode 100644 index 00000000..e61270ed --- /dev/null +++ b/tests/custom/test_cli_event_streaming.py @@ -0,0 +1,87 @@ +"""SSE parsing for build logs / call events, and the call-event renderers.""" + +import asyncio + +import pytest + +import smallestai.cli.calls as calls +from smallestai.cli.lib.atoms import AtomsAPIClient + + +class _FakeStream: + def __init__(self, lines): + self._lines = lines + + def raise_for_status(self): + return None + + async def aiter_lines(self): + for ln in self._lines: + yield ln + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + +class _FakeClient: + def __init__(self, lines): + self._lines = lines + + def stream(self, method, url, headers=None): + return _FakeStream(self._lines) + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + +def test_stream_sse_parses_data_frames(monkeypatch): + lines = [ + 'data: {"event_type": "sse_init"}', + "", + ": keep-alive", + 'data: {"event_type": "user_transcription", "user_transcription_text": "hi"}', + "data: not-json", + 'data:{"event_type":"call_end"}', + ] + monkeypatch.setattr(calls.httpx, "AsyncClient", lambda *a, **k: _FakeClient(lines)) + client = AtomsAPIClient() + + async def _collect(): + return [ev async for ev in client._stream_sse("http://x", "tok")] + + got = asyncio.run(_collect()) + assert [e["event_type"] for e in got] == ["sse_init", "user_transcription", "call_end"] + assert got[1]["user_transcription_text"] == "hi" + + +def test_render_event_shapes(capsys): + calls._render_event({"event_type": "user_transcription", "user_transcription_text": "hello"}) + calls._render_event({"event_type": "tts_completed", "tts_text": "hi there"}) + calls._render_event({"event_type": "turn_latency", "turn_latency": 820}) + calls._render_event({"event_type": "agent_error", "error": "boom"}) + calls._render_event({"event_type": "some_new_type", "foo": 1}) + out = capsys.readouterr().out + assert "user" in out and "hello" in out + assert "agent" in out and "hi there" in out + assert "820" in out + assert "boom" in out + assert "some_new_type" in out # unknown types still surface + + +def test_render_transcript_event_only_turns(capsys): + calls._render_transcript_event({"event_type": "turn_latency", "turn_latency": 1}) + calls._render_transcript_event({"event_type": "user_transcription", "user_transcription_text": "q"}) + calls._render_transcript_event({"event_type": "tts_completed", "tts_text": "a"}) + out = capsys.readouterr().out + assert "q" in out and "a" in out + assert "latency" not in out # non-turn events are dropped + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) From d846b4bc35e0bd58f4a930ff1a5c9b7ba47ced99 Mon Sep 17 00:00:00 2001 From: Abhishek Mishra Date: Wed, 19 Aug 2026 13:10:42 +0530 Subject: [PATCH 2/2] =?UTF-8?q?docs(cli):=20document=20calls=20events,=20t?= =?UTF-8?q?ranscript=20--follow,=20agent-crew=20logs=20+=20waves=E2=86=92m?= =?UTF-8?q?odels?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Keep the README CLI section and the mkdocs calls/crew guides in sync with the new commands this PR adds, and note the waves→models group rename (hidden back-compat alias). --- README.md | 16 ++++++++++++---- docs/guides/calls.md | 13 +++++++++++++ docs/guides/crew.md | 1 + 3 files changed, 26 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 4eff40e0..b7c5cb4e 100644 --- a/README.md +++ b/README.md @@ -116,12 +116,20 @@ deps in the pyproject). ## CLI ```sh -smallestai auth login # store your API key -smallestai agents list # list, get, call, and manage agents -smallestai agent-crew deploy ... # package and deploy crew code -smallestai agent-crew chat # talk to a running crew locally +smallestai auth login # store your API key +smallestai agents list # list, get, call, and manage agents +smallestai calls list # inspect call logs, transcripts, recordings +smallestai calls events # stream a live call's events (transcript, latency, tools) +smallestai calls transcript -f # stream the transcript live +smallestai models # text-to-speech, speech-to-text, voices +smallestai agent-crew deploy ... # package and deploy crew code +smallestai agent-crew logs [build-id] # stream a build's compile + deploy logs +smallestai agent-crew chat # talk to a running crew locally ``` +> The speech command group is now `models` (text-to-speech, speech-to-text, +> voices); `waves` still works as a hidden, back-compatible alias. + ## Async client The SDK exports an `async` client with the same surface: diff --git a/docs/guides/calls.md b/docs/guides/calls.md index ae12730e..55433036 100644 --- a/docs/guides/calls.md +++ b/docs/guides/calls.md @@ -33,6 +33,19 @@ smallestai calls transcript CALL-... smallestai calls recording CALL-... ``` +## Live events (while a call is in progress) + +Stream a call's events as they happen — transcript turns, per-turn latency, node +transitions, tool calls, and errors: + +```bash +smallestai calls events CALL-... # full live event stream +smallestai calls transcript CALL-... --follow # just the conversation, live +``` + +The call must be in progress; for a finished call use `calls transcript` (no +`--follow`) or `calls get`. Add `--json` to emit one raw event per line. + ## Reading a transfer from the logs A transfer produces a second **transfer leg**. If that leg shows diff --git a/docs/guides/crew.md b/docs/guides/crew.md index 9e1f6eaf..0cf3e1ae 100644 --- a/docs/guides/crew.md +++ b/docs/guides/crew.md @@ -44,6 +44,7 @@ crew, and a `requirements.txt`. Deploy with: ```bash smallestai agent-crew deploy --entry-point server.py smallestai agent-crew builds # watch build status; Make Live when SUCCEEDED +smallestai agent-crew logs [build-id] # stream a build's compile + deploy logs (latest by default) ``` Pin the SDK in `requirements.txt`: