Skip to content

datarobot_genai.dragent.cli.render

render

Shared terminal rendering for AG-UI events.

remote.py (SSE JSON dicts) normalises dicts to fields and calls :func:datarobot_genai.core.agents.render.render_event. console.py passes typed AG-UI objects to :func:datarobot_genai.core.agents.render.render_ag_ui_event.

render_sse_event 

render_sse_event(ev: dict[str, object]) -> str | None

Render an SSE JSON dict event. Returns the run-error message if RUN_ERROR, else None.

Source code in datarobot_genai/dragent/cli/render.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def render_sse_event(ev: dict[str, object]) -> str | None:
    """Render an SSE JSON dict event. Returns the run-error message if RUN_ERROR, else None."""
    raw_type = str(ev.get("type", ""))
    if not raw_type:
        logger.debug("Unhandled SSE event type: %s", raw_type)
        return None
    try:
        event_type = EventType(raw_type)
    except ValueError:
        logger.debug("Unknown SSE event type: %s", raw_type)
        return None

    if event_type == EventType.RUN_ERROR:
        return str(ev.get("message", "Unknown error"))

    rendered = render_event(
        event_type,
        delta=str(ev.get("delta", "")),
        name=str(ev.get("tool_call_name", "") or ev.get("step_name", "") or ev.get("name", "")),
        content=str(ev.get("content", "")),
        message=str(ev.get("message", "Unknown error")),
    )
    if rendered is not None:
        click.echo(rendered, nl=False)
    return None

render_object_event 

render_object_event(event: Event) -> bool

Render a typed AG-UI event object.

Returns True only when a streaming assistant text or reasoning content/chunk delta was printed, so the console can tell whether to skip the final result dump (see :meth:DRAgentConsoleFrontEndPlugin.run_workflow).

Source code in datarobot_genai/dragent/cli/render.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def render_object_event(event: Event) -> bool:
    """Render a typed AG-UI event object.

    Returns True only when a streaming assistant text or reasoning content/chunk
    delta was printed, so the console can tell whether to skip the final result
    dump (see :meth:`DRAgentConsoleFrontEndPlugin.run_workflow`).
    """
    event_type = event.type

    # The console frontend prints its own "Run finished" message after the
    # workflow completes, so skip it here to avoid duplicate output.
    if event_type in (EventType.RUN_FINISHED, EventType.RUN_ERROR):
        return False

    rendered = render_ag_ui_event(event)
    if rendered is not None:
        click.echo(rendered, nl=False)

    delta = str(getattr(event, "delta", "") or "")
    return event_type in (
        EventType.TEXT_MESSAGE_CONTENT,
        EventType.TEXT_MESSAGE_CHUNK,
        EventType.REASONING_MESSAGE_CONTENT,
        EventType.REASONING_MESSAGE_CHUNK,
    ) and bool(delta)