← All code samples

Stream agent chat

You want a chat UI, desktop widget, or editor panel that shows Aiden’s reply as it arrives, including tool calls, not a spinner that waits for the whole turn. Start a turn with an agent (or a workflow), then walk streamExecutionEvents / stream_execution_events until the run finishes.

Each yielded event is a normalized AG-UI type such as TEXT_MESSAGE_CONTENT, TOOL_CALL_START, or RUN_FINISHED. Control frames (connected, replay_complete, subscribe_end) stay inside the helper. The generator reconnects on drop so the server can replay missed frames.

Step What the SDK does
1. Pick a target listAgents / listWorkflows, or a known system agent name
2. Start startAgentChat / runWorkflow returns trace_id + session_id
3. Stream streamExecutionEvents yields text and tool events until RUN_FINISHED

When you need this

  • Chat UI or plugin. Render deltas as they arrive instead of polling waitForExecution.
  • Editor or desktop host. Pass an AbortSignal (TypeScript) so Stop cancels the SSE read without tearing down the whole process.
  • Target picker. Call listAgents / listWorkflows to fill a dropdown, then start chat against the chosen name.
  • Workflow smoke. Pass --workflow when you want a pinned workflow run rather than an agent chat turn.

For a fire-and-forget CI wait that only cares about completion, prefer Ask → execution or waitForSubscribeComplete. For a Go Grafana host with tool approvals, see Aiden chat from a Go backend.

How to run

export STACKGEN_URL=https://app.stackgen.com
export STACKGEN_TOKEN=stackgen_…
export STACKGEN_WORKSPACE_ID=<workspace-uuid>

pip install stackgen-sdk
# List chat targets, then stream a short reply.
python examples/ask/stream-agent-chat.py --list
python examples/ask/stream-agent-chat.py \
  --agent wfref-demo-responder \
  --message "Reply with exactly: pong"

Pass --workflow <name> instead of --agent to run a workflow. System agents such as aiden-router accept chat even when they do not appear in listAgents. For a fast live smoke, prefer a small demo agent (for example wfref-demo-responder) over a full router.

Minimal

from stackgen import StackgenClient, StackgenConfig

client = StackgenClient(
    StackgenConfig(
        base_url="https://app.stackgen.com",
        api_token="stackgen_…",
        workspace_id="<workspace-uuid>",
    )
)

# Optional: fill a picker. System agents may be absent from this list.
agents = client.aiden.list_agents()
workflows = client.aiden.list_workflows(latestOnly=True, limit=50)

started = client.aiden.start_agent_chat(
    "wfref-demo-responder",
    {"message": "Reply with exactly: pong"},
)
trace_id = started["trace_id"]

for ev in client.aiden.stream_execution_events(trace_id, max_wall_clock_ms=180_000):
    if ev["type"] == "TEXT_MESSAGE_CONTENT":
        print(ev["data"].get("Content") or ev["data"].get("content") or "", end="")
    elif ev["type"] == "RUN_FINISHED":
        print("\n[run finished]")

Complete scripts

#!/usr/bin/env python3
"""Stream an agent chat turn: print reply text and tool calls as they happen.

Defaults to wfref-demo-responder (fast smoke). Pass --agent <name> for a
specialist, or --workflow <name> to run a workflow instead. System agents such
as aiden-router accept chat even when they are absent from --list.

Usage:
  export STACKGEN_URL=https://app.stackgen.com
  export STACKGEN_TOKEN=stackgen_…
  export STACKGEN_WORKSPACE_ID=<workspace-uuid>

  # Discover agents and workflows in the workspace
  python examples/ask/stream-agent-chat.py --list

  # Stream a short reply (agent chat)
  python examples/ask/stream-agent-chat.py \
    --agent wfref-demo-responder \
    --message "Reply with exactly: pong"

  # Or start a workflow run and stream the same AG-UI events
  python examples/ask/stream-agent-chat.py \
    --workflow incident-triage \
    --message "Summarize checkout latency"

The generator stops on RUN_FINISHED or when max_wall_clock_ms elapses.
"""

from __future__ import annotations

import argparse
import os
import sys

from stackgen import StackgenClient, StackgenConfig


def require_env(name: str, *fallbacks: str) -> str:
    for key in (name, *fallbacks):
        value = (os.environ.get(key) or "").strip()
        if value:
            return value
    raise SystemExit(f"missing env {name}")


def text_field(data: dict, *keys: str) -> str:
    for key in keys:
        value = data.get(key)
        if isinstance(value, str) and value:
            return value
    return ""


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Stream AG-UI events for an agent chat or workflow run.",
    )
    parser.add_argument("--message", default="Reply with exactly: pong")
    parser.add_argument(
        "--agent",
        default="wfref-demo-responder",
        help="Agent name for start_agent_chat (ignored when --workflow is set)",
    )
    parser.add_argument(
        "--workflow",
        default="",
        help="Workflow name for run_workflow instead of agent chat",
    )
    parser.add_argument(
        "--list",
        action="store_true",
        help="Print list_agents and list_workflows, then exit",
    )
    parser.add_argument("--timeout-seconds", type=int, default=300)
    args = parser.parse_args()

    client = StackgenClient(
        StackgenConfig(
            base_url=require_env("STACKGEN_URL"),
            api_token=require_env("STACKGEN_TOKEN"),
            workspace_id=require_env("STACKGEN_WORKSPACE_ID", "STACKGEN_PROJECT"),
        )
    )

    if args.list:
        agents = client.aiden.list_agents()
        workflows = client.aiden.list_workflows(latestOnly=True, limit=50)
        print("agents:")
        for a in agents[:30]:
            print(f"  - {text_field(a, 'name') or a}")
        print("workflows:")
        for w in workflows[:30]:
            print(f"  - {text_field(w, 'name') or w}")
        return

    if args.workflow:
        started = client.aiden.run_workflow(
            args.workflow, {"input": {"message": args.message}}
        )
        label = f"workflow={args.workflow}"
    else:
        started = client.aiden.start_agent_chat(
            args.agent, {"message": args.message}
        )
        label = f"agent={args.agent}"

    session_id = text_field(started, "session_id", "sessionId")
    trace_id = text_field(started, "trace_id", "traceId")
    print(f"{label} session_id={session_id} trace_id={trace_id}")
    if not trace_id:
        raise SystemExit("start did not return trace_id")

    for ev in client.aiden.stream_execution_events(
        trace_id,
        max_wall_clock_ms=args.timeout_seconds * 1000,
    ):
        etype = ev.get("type")
        data = ev.get("data") if isinstance(ev.get("data"), dict) else {}
        if etype == "TEXT_MESSAGE_CONTENT":
            chunk = text_field(data, "Content", "content", "delta", "Delta")
            if chunk:
                sys.stdout.write(chunk)
                sys.stdout.flush()
        elif etype == "TOOL_CALL_START":
            name = text_field(data, "ToolCallName", "toolCallName", "name")
            tid = text_field(data, "ToolCallID", "toolCallId", "id")
            print(f"\n[tool start] {name or 'tool'} ({tid or '?'})")
        elif etype == "TOOL_CALL_ARGS":
            delta = text_field(data, "delta", "Delta", "Content", "content")
            if delta:
                sys.stdout.write(delta)
                sys.stdout.flush()
        elif etype in ("TOOL_CALL_RESULT", "TOOL_CALL_END"):
            print(f"\n[tool {'result' if etype == 'TOOL_CALL_RESULT' else 'end'}]")
        elif etype == "RUN_ERROR":
            print(f"\n[run error] {data}", file=sys.stderr)
        elif etype == "RUN_FINISHED":
            print("\n[run finished]")


if __name__ == "__main__":
    main()