feat: add ThreadListSidebar component for managing conversations

feat: implement tool render components for AG-UI with various widgets

feat: create useRunManager hook for managing agent run lifecycle

feat: add CopilotKitProvider for AG-UI context integration

test: implement e2e tests for conversation functionality and thread API
Signed-off-by: Andre Bossard <anbossar@microsoft.com>

Author: abossard

backend/agent_builder/engine/ag_ui_events.py

@@ -0,0 +1,196 @@
+"""
+Agent Builder — AG-UI Event Conversion
+
+Calculations: pure functions that convert ReAct execution data
+into AG-UI protocol events. No I/O, no database, no side effects.
+Each function takes data in, returns AG-UI event(s) out.
+"""
+
+import json
+import uuid
+from typing import Any, Optional
+
+from ag_ui.core import (
+    CustomEvent,
+    EventType,
+    RunErrorEvent,
+    RunFinishedEvent,
+    RunStartedEvent,
+    StepFinishedEvent,
+    StepStartedEvent,
+    TextMessageContentEvent,
+    TextMessageEndEvent,
+    TextMessageStartEvent,
+    ToolCallArgsEvent,
+    ToolCallEndEvent,
+    ToolCallResultEvent,
+    ToolCallStartEvent,
+    StateSnapshotEvent,
+)
+from ag_ui.encoder import EventEncoder
+
+
+# Shared encoder — stateless, safe to reuse
+_encoder = EventEncoder()
+
+
+# ---------------------------------------------------------------------------
+# Pure conversion functions
+# ---------------------------------------------------------------------------
+
+def run_started_event(thread_id: str, run_id: str) -> RunStartedEvent:
+    return RunStartedEvent(
+        type=EventType.RUN_STARTED,
+        thread_id=thread_id,
+        run_id=run_id,
+    )
+
+
+def run_finished_event(thread_id: str, run_id: str) -> RunFinishedEvent:
+    return RunFinishedEvent(
+        type=EventType.RUN_FINISHED,
+        thread_id=thread_id,
+        run_id=run_id,
+    )
+
+
+def run_error_event(message: str) -> RunErrorEvent:
+    return RunErrorEvent(
+        type=EventType.RUN_ERROR,
+        message=message,
+    )
+
+
+def text_message_start(message_id: str, role: str = "assistant") -> TextMessageStartEvent:
+    return TextMessageStartEvent(
+        type=EventType.TEXT_MESSAGE_START,
+        message_id=message_id,
+        role=role,
+    )
+
+
+def text_message_content(message_id: str, delta: str) -> TextMessageContentEvent:
+    return TextMessageContentEvent(
+        type=EventType.TEXT_MESSAGE_CONTENT,
+        message_id=message_id,
+        delta=delta,
+    )
+
+
+def text_message_end(message_id: str) -> TextMessageEndEvent:
+    return TextMessageEndEvent(
+        type=EventType.TEXT_MESSAGE_END,
+        message_id=message_id,
+    )
+
+
+def text_message_events(content: str, message_id: Optional[str] = None) -> list:
+    """Convert a complete text into the three-event TEXT_MESSAGE sequence."""
+    msg_id = message_id or str(uuid.uuid4())
+    return [
+        text_message_start(msg_id, role="assistant"),
+        text_message_content(msg_id, content),
+        text_message_end(msg_id),
+    ]
+
+
+def tool_call_start(tool_call_id: str, tool_name: str, parent_message_id: Optional[str] = None) -> ToolCallStartEvent:
+    return ToolCallStartEvent(
+        type=EventType.TOOL_CALL_START,
+        tool_call_id=tool_call_id,
+        tool_call_name=tool_name,
+        parent_message_id=parent_message_id,
+    )
+
+
+def tool_call_args(tool_call_id: str, args: Any) -> ToolCallArgsEvent:
+    args_str = json.dumps(args) if not isinstance(args, str) else args
+    return ToolCallArgsEvent(
+        type=EventType.TOOL_CALL_ARGS,
+        tool_call_id=tool_call_id,
+        delta=args_str,
+    )
+
+
+def tool_call_end(tool_call_id: str) -> ToolCallEndEvent:
+    return ToolCallEndEvent(
+        type=EventType.TOOL_CALL_END,
+        tool_call_id=tool_call_id,
+    )
+
+
+def tool_call_result(tool_call_id: str, content: str, message_id: Optional[str] = None) -> ToolCallResultEvent:
+    msg_id = message_id or str(uuid.uuid4())
+    return ToolCallResultEvent(
+        type=EventType.TOOL_CALL_RESULT,
+        message_id=msg_id,
+        tool_call_id=tool_call_id,
+        content=content,
+        role="tool",
+    )
+
+
+def tool_call_events(
+    tool_name: str,
+    args: Any,
+    result: str,
+    tool_call_id: Optional[str] = None,
+    parent_message_id: Optional[str] = None,
+) -> list:
+    """Convert a complete tool invocation into the four-event TOOL_CALL sequence."""
+    tc_id = tool_call_id or str(uuid.uuid4())
+    return [
+        tool_call_start(tc_id, tool_name, parent_message_id),
+        tool_call_args(tc_id, args),
+        tool_call_end(tc_id),
+        tool_call_result(tc_id, result),
+    ]
+
+
+def step_started_event(step_name: str) -> StepStartedEvent:
+    return StepStartedEvent(
+        type=EventType.STEP_STARTED,
+        step_name=step_name,
+    )
+
+
+def step_finished_event(step_name: str) -> StepFinishedEvent:
+    return StepFinishedEvent(
+        type=EventType.STEP_FINISHED,
+        step_name=step_name,
+    )
+
+
+def state_snapshot_event(snapshot: dict[str, Any]) -> StateSnapshotEvent:
+    return StateSnapshotEvent(
+        type=EventType.STATE_SNAPSHOT,
+        snapshot=snapshot,
+    )
+
+
+def structured_output_event(
+    output: Any,
+    schema: Optional[dict[str, Any]] = None,
+) -> CustomEvent:
+    """Emit structured agent output as a CUSTOM event with widget metadata.
+
+    The frontend uses the schema's x-ui annotations to pick the right
+    CopilotKit tool render for each field.
+    """
+    return CustomEvent(
+        type=EventType.CUSTOM,
+        name="structured_output",
+        value={
+            "output": output,
+            "schema": schema or {},
+        },
+    )
+
+
+# ---------------------------------------------------------------------------
+# SSE encoding — thin wrapper, still a pure calculation
+# ---------------------------------------------------------------------------
+
+def encode_event(event: Any) -> str:
+    """Serialize an AG-UI event to an SSE data line."""
+    return _encoder.encode(event)

backend/agent_builder/engine/callbacks.py

@@ -157,13 +157,14 @@ def _extract_llm_call_metadata(
 # ---------------------------------------------------------------------------
 
 def make_streaming_callback(run_id: str, event_bus: AgentEventBus) -> Any:
-    """Create a callback handler that publishes agent events to the SSE event bus."""
+    """Create a callback handler that publishes AG-UI events to the SSE event bus."""
     from langchain_core.callbacks import BaseCallbackHandler
 
     class StreamingCallbackHandler(BaseCallbackHandler):
         def __init__(self) -> None:
             super().__init__()
             self._start_times: dict[Any, float] = {}
+            self._tool_names: dict[Any, str] = {}
 
         def on_tool_start(
             self,
@@ -175,30 +176,43 @@ def on_tool_start(
         ) -> None:
             self._start_times[run_id] = perf_counter()
             name = serialized.get("name", "unknown")
+            self._tool_names[run_id] = name
             preview = input_str[:500] if isinstance(input_str, str) else str(input_str)[:500]
             event_bus.publish(AgentEvent(
                 run_id=run_id_outer,
-                event_type="tool_start",
-                data={"tool_name": name, "input": preview},
+                event_type="TOOL_CALL_START",
+                data={"toolCallId": str(run_id), "toolCallName": name, "args": preview},
             ))
 
         def on_tool_end(self, output: str, *, run_id: Any = None, **kwargs: Any) -> None:
             started = self._start_times.pop(run_id, None)
             duration_ms = int((perf_counter() - started) * 1000) if started is not None else None
+            name = self._tool_names.pop(run_id, "")
             preview = output[:500] if isinstance(output, str) else str(output)[:500]
             event_bus.publish(AgentEvent(
                 run_id=run_id_outer,
-                event_type="tool_end",
-                data={"tool_name": kwargs.get("name", ""), "output": preview, "duration_ms": duration_ms},
+                event_type="TOOL_CALL_END",
+                data={"toolCallId": str(run_id), "toolCallName": name},
+            ))
+            event_bus.publish(AgentEvent(
+                run_id=run_id_outer,
+                event_type="TOOL_CALL_RESULT",
+                data={
+                    "toolCallId": str(run_id),
+                    "toolCallName": name,
+                    "content": preview,
+                    "durationMs": duration_ms,
+                },
             ))
 
         def on_tool_error(self, error: BaseException, *, run_id: Any = None, **kwargs: Any) -> None:
             started = self._start_times.pop(run_id, None)
             duration_ms = int((perf_counter() - started) * 1000) if started is not None else None
+            name = self._tool_names.pop(run_id, "")
             event_bus.publish(AgentEvent(
                 run_id=run_id_outer,
-                event_type="tool_error",
-                data={"error": str(error), "duration_ms": duration_ms},
+                event_type="TOOL_CALL_END",
+                data={"toolCallId": str(run_id), "toolCallName": name, "error": str(error), "durationMs": duration_ms},
             ))
 
         def on_llm_start(
@@ -219,8 +233,8 @@ def on_llm_start(
                 )
             event_bus.publish(AgentEvent(
                 run_id=run_id_outer,
-                event_type="llm_start",
-                data={"model": model_name or ""},
+                event_type="STEP_STARTED",
+                data={"stepName": "llm_call", "model": model_name or ""},
             ))
 
         def on_llm_end(self, response: Any, *, run_id: UUID = None, **kwargs: Any) -> None:
@@ -229,12 +243,13 @@ def on_llm_end(self, response: Any, *, run_id: UUID = None, **kwargs: Any) -> No
             token_usage, model_name, finish_reason = _extract_llm_call_metadata(response)
             event_bus.publish(AgentEvent(
                 run_id=run_id_outer,
-                event_type="llm_end",
+                event_type="STEP_FINISHED",
                 data={
+                    "stepName": "llm_call",
                     "model": model_name or "",
-                    "duration_ms": duration_ms,
-                    "token_usage": token_usage or {},
-                    "finish_reason": finish_reason or "",
+                    "durationMs": duration_ms,
+                    "tokenUsage": token_usage or {},
+                    "finishReason": finish_reason or "",
                 },
             ))
 
@@ -243,8 +258,8 @@ def on_llm_error(self, error: BaseException, *, run_id: UUID = None, **kwargs: A
             duration_ms = int((perf_counter() - started_at) * 1000) if started_at is not None else None
             event_bus.publish(AgentEvent(
                 run_id=run_id_outer,
-                event_type="llm_error",
-                data={"error": str(error), "duration_ms": duration_ms},
+                event_type="STEP_FINISHED",
+                data={"stepName": "llm_call", "error": str(error), "durationMs": duration_ms},
             ))
 
     # Closure captures run_id as run_id_outer to avoid shadowing with LangChain's run_id param

backend/agent_builder/engine/event_bus.py

@@ -1,18 +1,18 @@
 """
 Agent Builder — Event Bus
 
-Pub/sub for real-time agent activity events.
+Pub/sub for real-time agent activity events using AG-UI protocol format.
 Subscribers receive events via asyncio.Queue; a history buffer
 provides catch-up for new SSE connections.
 
-Data: AgentEvent dataclass
+Data: AgentEvent dataclass (carries AG-UI typed event dicts)
 Action: publish / subscribe / unsubscribe (I/O via queues)
 """
 
 import asyncio
 import logging
 import time
-from dataclasses import asdict, dataclass, field
+from dataclasses import dataclass, field
 from typing import Any
 
 logger = logging.getLogger(__name__)
@@ -22,14 +22,20 @@
 
 @dataclass
 class AgentEvent:
-    """A single event from an agent run."""
+    """A single AG-UI event from an agent run."""
     run_id: str
     event_type: str
     data: dict[str, Any] = field(default_factory=dict)
     timestamp: float = field(default_factory=time.time)
 
     def to_sse_dict(self) -> dict[str, Any]:
-        return asdict(self)
+        """Return the AG-UI event dict for SSE serialization."""
+        return {
+            "type": self.event_type,
+            "runId": self.run_id,
+            "timestamp": int(self.timestamp * 1000),
+            **self.data,
+        }
 
 
 class AgentEventBus:

backend/agent_builder/models/__init__.py

@@ -17,6 +17,14 @@
     SuccessCriteria,
 )
 from .run import AgentRun, AgentRunCreate, RunStatus
+from .thread import (
+    ConversationThread,
+    MessageRole,
+    ThreadCreate,
+    ThreadMessage,
+    ThreadMessageCreate,
+    ThreadStatus,
+)
 
 __all__ = [
     "AgentDefinition",
@@ -27,8 +35,14 @@
     "AgentResponse",
     "AgentRun",
     "AgentRunCreate",
+    "ConversationThread",
     "CriteriaResult",
     "CriteriaType",
+    "MessageRole",
     "RunStatus",
     "SuccessCriteria",
+    "ThreadCreate",
+    "ThreadMessage",
+    "ThreadMessageCreate",
+    "ThreadStatus",
 ]