add chat runtime

Author: GabrielVasilescu04

src/uipath/runtime/__init__.py

@@ -8,6 +8,7 @@
     UiPathStreamOptions,
 )
 from uipath.runtime.chat.bridge import UiPathChatBridgeProtocol
+from uipath.runtime.chat.runtime import UiPathChatRuntime
 from uipath.runtime.context import UiPathRuntimeContext
 from uipath.runtime.debug.breakpoint import UiPathBreakpointResult
 from uipath.runtime.debug.bridge import UiPathDebugBridgeProtocol
@@ -68,4 +69,5 @@
     "UiPathStreamNotSupportedError",
     "UiPathResumeTriggerName",
     "UiPathChatBridgeProtocol",
+    "UiPathChatRuntime",
 ]

src/uipath/runtime/chat/__init__.py

@@ -1,5 +1,6 @@
-"""Chat bridge protocol for conversational agents."""
+"""Chat bridge protocol and runtime for conversational agents."""
 
 from uipath.runtime.chat.bridge import UiPathChatBridgeProtocol
+from uipath.runtime.chat.runtime import UiPathChatRuntime
 
-__all__ = ["UiPathChatBridgeProtocol"]
+__all__ = ["UiPathChatBridgeProtocol", "UiPathChatRuntime"]

src/uipath/runtime/chat/runtime.py

@@ -0,0 +1,94 @@
+"""Chat runtime implementation."""
+
+import logging
+from typing import Any, AsyncGenerator, cast
+
+from uipath.runtime.base import (
+    UiPathExecuteOptions,
+    UiPathRuntimeProtocol,
+    UiPathStreamNotSupportedError,
+    UiPathStreamOptions,
+)
+from uipath.runtime.chat.bridge import UiPathChatBridgeProtocol
+from uipath.runtime.events import (
+    UiPathRuntimeEvent,
+    UiPathRuntimeMessageEvent,
+)
+from uipath.runtime.result import (
+    UiPathRuntimeResult,
+    UiPathRuntimeStatus,
+)
+from uipath.runtime.schema import UiPathRuntimeSchema
+
+logger = logging.getLogger(__name__)
+
+
+class UiPathChatRuntime:
+    """Specialized runtime for chat mode that streams message events to a chat bridge."""
+
+    def __init__(
+        self,
+        delegate: UiPathRuntimeProtocol,
+        chat_bridge: UiPathChatBridgeProtocol,
+    ):
+        """Initialize the UiPathChatRuntime.
+
+        Args:
+            delegate: The underlying runtime to wrap
+            chat_bridge: Bridge for chat event communication
+        """
+        super().__init__()
+        self.delegate = delegate
+        self.chat_bridge = chat_bridge
+
+    async def execute(
+        self,
+        input: dict[str, Any] | None = None,
+        options: UiPathExecuteOptions | None = None,
+    ) -> UiPathRuntimeResult:
+        """Execute the workflow with chat support."""
+        result: UiPathRuntimeResult | None = None
+        async for event in self.stream(input, cast(UiPathStreamOptions, options)):
+            if isinstance(event, UiPathRuntimeResult):
+                result = event
+
+        return (
+            result
+            if result
+            else UiPathRuntimeResult(status=UiPathRuntimeStatus.SUCCESSFUL)
+        )
+
+    async def stream(
+        self,
+        input: dict[str, Any] | None = None,
+        options: UiPathStreamOptions | None = None,
+    ) -> AsyncGenerator[UiPathRuntimeEvent, None]:
+        """Stream execution events with chat support."""
+        await self.chat_bridge.connect()
+
+        result: UiPathRuntimeResult | None = None
+
+        async for event in self.delegate.stream(input, options=options):
+            # Send message events to chat bridge
+            if isinstance(event, UiPathRuntimeMessageEvent):
+                if event.payload:
+                    await self.chat_bridge.emit_message_event(event.payload)
+
+            yield event
+
+            if isinstance(event, UiPathRuntimeResult):
+                result = event
+
+        if result:
+            logger.info(f"Chat execution completed with status: {result.status}")
+
+        # Always disconnect from chat bridge
+        await self.chat_bridge.disconnect("", "")
+
+    async def get_schema(self) -> UiPathRuntimeSchema:
+        """Get schema from the delegate runtime."""
+        return await self.delegate.get_schema()
+
+    async def dispose(self) -> None:
+        """Dispose the delegate runtime."""
+        await self.delegate.dispose()