Skip to content

datarobot_genai.crewai.agui_stream

agui_stream

AG-UI event emitter for CrewAI streaming.

Translates a CrewAI chunk/tool-event stream into a well-formed AG-UI event sequence. The emitter owns the open step and the open text/reasoning message, so callers just declare intent (step/text/reasoning/tool_call/finish) and every opened step and message is closed before the next one opens.

AGUIStreamEmitter

Stateful translator from CrewAI stream events to AG-UI events.

Owns the open step and the open text/reasoning message; every opened step/message is closed before the next opens. Methods yield bare AG-UI events -- the caller attaches usage metrics and the run-level RunStarted/RunFinished events.

Source code in datarobot_genai/crewai/agui_stream.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
class AGUIStreamEmitter:
    """Stateful translator from CrewAI stream events to AG-UI events.

    Owns the open step and the open text/reasoning message; every opened step/message is
    closed before the next opens. Methods yield bare AG-UI events -- the caller attaches
    usage metrics and the run-level ``RunStarted``/``RunFinished`` events.
    """

    def __init__(self) -> None:
        self._role = ""  # role of the open step ("" = none open)
        self._mid = str(uuid.uuid4())  # current message id
        self._text = False
        self._reasoning = False

    def step(self, role: str) -> Iterator[Any]:
        """Make ``role`` the open step, closing the previous one first. No-op if ``role`` is
        unchanged; an empty ``role`` only closes the current step.
        """
        if role == self._role:
            return
        if self._role:
            yield from self.close_messages()
            yield StepFinishedEvent(type=EventType.STEP_FINISHED, step_name=self._role)
            self._mid = str(uuid.uuid4())
        if role:
            yield StepStartedEvent(type=EventType.STEP_STARTED, step_name=role)
        self._role = role

    def _reasoning_id(self) -> str:
        """Reasoning message id, derived from the text id but distinct from it: a UI grouping by id
        renders reasoning as its own block instead of folding it into the assistant bubble. Matches
        the langgraph/llamaindex adapters (``uuid5(NAMESPACE_OID, f"{text_id}-reasoning")``).
        """
        return str(uuid.uuid5(uuid.NAMESPACE_OID, f"{self._mid}-reasoning"))

    def reasoning(self, on: bool) -> Iterator[Any]:
        """Open or close the reasoning message at a reasoning-mode boundary."""
        if on and not self._reasoning:
            yield ReasoningStartEvent(
                type=EventType.REASONING_START, message_id=self._reasoning_id()
            )
            yield ReasoningMessageStartEvent(
                type=EventType.REASONING_MESSAGE_START,
                message_id=self._reasoning_id(),
                role="reasoning",
            )
            self._reasoning = True
        elif not on and self._reasoning:
            yield from self._close_reasoning()

    def text(self, delta: str) -> Iterator[Any]:
        """Emit a content delta as reasoning content (in reasoning mode) or assistant text."""
        if self._reasoning:
            yield ReasoningMessageContentEvent(
                type=EventType.REASONING_MESSAGE_CONTENT,
                message_id=self._reasoning_id(),
                delta=delta,
            )
            return
        if not self._text:
            yield TextMessageStartEvent(type=EventType.TEXT_MESSAGE_START, message_id=self._mid)
            self._text = True
        yield TextMessageContentEvent(
            type=EventType.TEXT_MESSAGE_CONTENT, message_id=self._mid, delta=delta
        )

    def tool_call(self, record: ToolCallRecord) -> Iterator[Any]:
        """Emit a tool call/result as a clean sibling: open its owning agent's step, close any
        open message, and attach the call to that message if one was open.
        """
        if record.kind == "call" and record.agent_role:
            yield from self.step(record.agent_role)
        parent = self._mid if (self._text or self._reasoning) else ""
        yield from self.close_messages()
        yield from _tool_call_record_to_events(record, parent)
        self._mid = str(uuid.uuid4())

    def finish(self) -> Iterator[Any]:
        """Close any open message and the open step at end of stream."""
        yield from self.close_messages()
        if self._role:
            yield StepFinishedEvent(type=EventType.STEP_FINISHED, step_name=self._role)
            self._role = ""

    def close_messages(self) -> Iterator[Any]:
        """Close the open text/reasoning message, if any (at a step boundary or stream end)."""
        if self._text:
            yield TextMessageEndEvent(type=EventType.TEXT_MESSAGE_END, message_id=self._mid)
            self._text = False
        if self._reasoning:
            yield from self._close_reasoning()

    def _close_reasoning(self) -> Iterator[Any]:
        yield ReasoningMessageEndEvent(
            type=EventType.REASONING_MESSAGE_END, message_id=self._reasoning_id()
        )
        yield ReasoningEndEvent(type=EventType.REASONING_END, message_id=self._reasoning_id())
        self._reasoning = False

step 

step(role: str) -> Iterator[Any]

Make role the open step, closing the previous one first. No-op if role is unchanged; an empty role only closes the current step.

Source code in datarobot_genai/crewai/agui_stream.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def step(self, role: str) -> Iterator[Any]:
    """Make ``role`` the open step, closing the previous one first. No-op if ``role`` is
    unchanged; an empty ``role`` only closes the current step.
    """
    if role == self._role:
        return
    if self._role:
        yield from self.close_messages()
        yield StepFinishedEvent(type=EventType.STEP_FINISHED, step_name=self._role)
        self._mid = str(uuid.uuid4())
    if role:
        yield StepStartedEvent(type=EventType.STEP_STARTED, step_name=role)
    self._role = role

reasoning 

reasoning(on: bool) -> Iterator[Any]

Open or close the reasoning message at a reasoning-mode boundary.

Source code in datarobot_genai/crewai/agui_stream.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def reasoning(self, on: bool) -> Iterator[Any]:
    """Open or close the reasoning message at a reasoning-mode boundary."""
    if on and not self._reasoning:
        yield ReasoningStartEvent(
            type=EventType.REASONING_START, message_id=self._reasoning_id()
        )
        yield ReasoningMessageStartEvent(
            type=EventType.REASONING_MESSAGE_START,
            message_id=self._reasoning_id(),
            role="reasoning",
        )
        self._reasoning = True
    elif not on and self._reasoning:
        yield from self._close_reasoning()

text 

text(delta: str) -> Iterator[Any]

Emit a content delta as reasoning content (in reasoning mode) or assistant text.

Source code in datarobot_genai/crewai/agui_stream.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def text(self, delta: str) -> Iterator[Any]:
    """Emit a content delta as reasoning content (in reasoning mode) or assistant text."""
    if self._reasoning:
        yield ReasoningMessageContentEvent(
            type=EventType.REASONING_MESSAGE_CONTENT,
            message_id=self._reasoning_id(),
            delta=delta,
        )
        return
    if not self._text:
        yield TextMessageStartEvent(type=EventType.TEXT_MESSAGE_START, message_id=self._mid)
        self._text = True
    yield TextMessageContentEvent(
        type=EventType.TEXT_MESSAGE_CONTENT, message_id=self._mid, delta=delta
    )

tool_call 

tool_call(record: ToolCallRecord) -> Iterator[Any]

Emit a tool call/result as a clean sibling: open its owning agent's step, close any open message, and attach the call to that message if one was open.

Source code in datarobot_genai/crewai/agui_stream.py 
143
144
145
146
147
148
149
150
151
152
def tool_call(self, record: ToolCallRecord) -> Iterator[Any]:
    """Emit a tool call/result as a clean sibling: open its owning agent's step, close any
    open message, and attach the call to that message if one was open.
    """
    if record.kind == "call" and record.agent_role:
        yield from self.step(record.agent_role)
    parent = self._mid if (self._text or self._reasoning) else ""
    yield from self.close_messages()
    yield from _tool_call_record_to_events(record, parent)
    self._mid = str(uuid.uuid4())

finish 

finish() -> Iterator[Any]

Close any open message and the open step at end of stream.

Source code in datarobot_genai/crewai/agui_stream.py 
154
155
156
157
158
159
def finish(self) -> Iterator[Any]:
    """Close any open message and the open step at end of stream."""
    yield from self.close_messages()
    if self._role:
        yield StepFinishedEvent(type=EventType.STEP_FINISHED, step_name=self._role)
        self._role = ""

close_messages 

close_messages() -> Iterator[Any]

Close the open text/reasoning message, if any (at a step boundary or stream end).

Source code in datarobot_genai/crewai/agui_stream.py 
161
162
163
164
165
166
167
def close_messages(self) -> Iterator[Any]:
    """Close the open text/reasoning message, if any (at a step boundary or stream end)."""
    if self._text:
        yield TextMessageEndEvent(type=EventType.TEXT_MESSAGE_END, message_id=self._mid)
        self._text = False
    if self._reasoning:
        yield from self._close_reasoning()