feat: add chat bridge protocol

Author: GabrielVasilescu04

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath-runtime"
-version = "0.2.0"
+version = "0.2.1"
 description = "Runtime abstractions and interfaces for building agents and automation scripts in the UiPath ecosystem"
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"

src/uipath/runtime/__init__.py

@@ -7,6 +7,8 @@
     UiPathStreamNotSupportedError,
     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
@@ -66,4 +68,6 @@
     "UiPathBreakpointResult",
     "UiPathStreamNotSupportedError",
     "UiPathResumeTriggerName",
+    "UiPathChatBridgeProtocol",
+    "UiPathChatRuntime",
 ]

src/uipath/runtime/chat/__init__.py

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

src/uipath/runtime/chat/bridge.py

@@ -0,0 +1,28 @@
+"""Abstract conversation bridge interface."""
+
+from typing import Any, Protocol
+
+from uipath.core.chat import UiPathConversationMessageEvent
+
+
+class UiPathChatBridgeProtocol(Protocol):
+    """Abstract interface for chat communication.
+
+    Implementations: WebSocket, etc.
+    """
+
+    async def connect(self) -> None:
+        """Establish connection to chat service."""
+        ...
+
+    async def disconnect(self) -> None:
+        """Close connection and send exchange end event."""
+        ...
+
+    async def emit_message_event(self, message_event: UiPathConversationMessageEvent) -> None:
+        """Wrap and send a message event.
+
+        Args:
+            message_event: UiPathConversationMessageEvent to wrap and send
+        """
+        ...

src/uipath/runtime/chat/runtime.py

@@ -0,0 +1,93 @@
+"""Chat runtime implementation."""
+
+import logging
+from typing import Any, AsyncGenerator, cast
+
+from uipath.runtime.base import (
+    UiPathExecuteOptions,
+    UiPathRuntimeProtocol,
+    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()