feat: conversational agent support

Author: mike-deem-uipath

src/uipath/_cli/_chat/_bridge.py

@@ -1,6 +1,7 @@
 """Chat bridge implementations for conversational agents."""
 
 import asyncio
+import json
 import logging
 import os
 import uuid
@@ -57,6 +58,10 @@ def __init__(
         self._client: AsyncClient | None = None
         self._connected_event = asyncio.Event()
 
+        # Set CAS_WEBSOCKET_DISABLED when using the debugger to prevent websocket errors from 
+        # interrupting the debugging session. Events will be logged instead of being sent.
+        self._debug_mode = os.environ.get("CAS_WEBSOCKET_DISABLED") == "true"
+
     async def connect(self, timeout: float = 10.0) -> None:
         """Establish WebSocket connection to the server.
 
@@ -89,34 +94,37 @@ async def connect(self, timeout: float = 10.0) -> None:
 
         self._connected_event.clear()
 
-        try:
-            # Attempt to connect with timeout
-            await asyncio.wait_for(
-                self._client.connect(
-                    url=self.websocket_url,
-                    socketio_path=self.websocket_path,
-                    headers=self.headers,
-                    auth=self.auth,
-                    transports=["websocket"],
-                ),
-                timeout=timeout,
-            )
+        if self._debug_mode:
+            logger.warning("SocketIOChatBridge is in debug mode. Not connecting websocket.")
+        else:
+            try:
+                # Attempt to connect with timeout
+                await asyncio.wait_for(
+                    self._client.connect(
+                        url=self.websocket_url,
+                        socketio_path=self.websocket_path,
+                        headers=self.headers,
+                        auth=self.auth,
+                        transports=["websocket"],
+                    ),
+                    timeout=timeout,
+                )
 
-            await asyncio.wait_for(self._connected_event.wait(), timeout=timeout)
+                await asyncio.wait_for(self._connected_event.wait(), timeout=timeout)
 
-        except asyncio.TimeoutError as e:
-            error_message = (
-                f"Failed to connect to WebSocket server within {timeout}s timeout"
-            )
-            logger.error(error_message)
-            await self._cleanup_client()
-            raise RuntimeError(error_message) from e
+            except asyncio.TimeoutError as e:
+                error_message = (
+                    f"Failed to connect to WebSocket server within {timeout}s timeout"
+                )
+                logger.error(error_message)
+                await self._cleanup_client()
+                raise RuntimeError(error_message) from e
 
-        except Exception as e:
-            error_message = f"Failed to connect to WebSocket server: {e}"
-            logger.error(error_message)
-            await self._cleanup_client()
-            raise RuntimeError(error_message) from e
+            except Exception as e:
+                error_message = f"Failed to connect to WebSocket server: {e}"
+                logger.error(error_message)
+                await self._cleanup_client()
+                raise RuntimeError(error_message) from e
 
     async def disconnect(self) -> None:
         """Close the WebSocket connection gracefully.
@@ -149,7 +157,7 @@ async def emit_message_event(
         if self._client is None:
             raise RuntimeError("WebSocket client not connected. Call connect() first.")
 
-        if not self._connected_event.is_set():
+        if not self._connected_event.is_set() and not self._debug_mode:
             raise RuntimeError("WebSocket client not in connected state")
 
         try:
@@ -166,7 +174,10 @@ async def emit_message_event(
                 mode="json", exclude_none=True, by_alias=True
             )
 
-            await self._client.emit("ConversationEvent", event_data)
+            if self._debug_mode:
+                logger.info(f"SocketIOChatBridge is in debug mode. Not sending event: {json.dumps(event_data)}")
+            else:
+                await self._client.emit("ConversationEvent", event_data)
 
             # Store the current message ID, used for emitting interrupt events.
             self._current_message_id = message_event.message_id
@@ -184,7 +195,7 @@ async def emit_exchange_end_event(self) -> None:
         if self._client is None:
             raise RuntimeError("WebSocket client not connected. Call connect() first.")
 
-        if not self._connected_event.is_set():
+        if not self._connected_event.is_set() and not self._debug_mode:
             raise RuntimeError("WebSocket client not in connected state")
 
         try:
@@ -200,7 +211,10 @@ async def emit_exchange_end_event(self) -> None:
                 mode="json", exclude_none=True, by_alias=True
             )
 
-            await self._client.emit("ConversationEvent", event_data)
+            if self._debug_mode:
+                logger.info(f"SocketIOChatBridge is in debug mode. Not sending event: {json.dumps(event_data)}")
+            else:
+                await self._client.emit("ConversationEvent", event_data)
 
         except Exception as e:
             logger.error(f"Error sending conversation event to WebSocket: {e}")
@@ -230,7 +244,10 @@ async def emit_interrupt_event(self, runtime_result: UiPathRuntimeResult):
                 event_data = interrupt_event.model_dump(
                     mode="json", exclude_none=True, by_alias=True
                 )
-                await self._client.emit("ConversationEvent", event_data)
+                if self._debug_mode:
+                    logger.info(f"SocketIOChatBridge is in debug mode. Not sending event: {json.dumps(event_data)}")
+                else:
+                    await self._client.emit("ConversationEvent", event_data)
             except Exception as e:
                 logger.warning(f"Error sending interrupt event: {e}")
 
@@ -315,6 +332,11 @@ def get_chat_bridge(
     websocket_url = f"wss://{host}?conversationId={context.conversation_id}"
     websocket_path = "autopilotforeveryone_/websocket_/socket.io"
 
+    if os.environ.get("CAS_WEBSOCKET_HOST"):
+        websocket_url = f"ws://{os.environ.get('CAS_WEBSOCKET_HOST')}?conversationId={context.conversation_id}"
+        websocket_path = "/socket.io"
+        logger.warning(f"CAS_WEBSOCKET_HOST is set. Using websocket_url '{websocket_url}{websocket_path}'.")
+
     # Build headers from context
     headers = {
         "Authorization": f"Bearer {os.environ.get('UIPATH_ACCESS_TOKEN', '')}",

src/uipath/agent/models/agent.py

@@ -795,6 +795,11 @@ class AgentDefinition(BaseModel):
         validate_by_name=True, validate_by_alias=True, extra="allow"
     )
 
+    @property
+    def is_conversational(self) -> bool:
+        """Checks the settings.engine property to determine if the agent is conversational."""
+        return self.settings.engine == "conversational-v1"
+
     @staticmethod
     def _normalize_guardrails(v: Dict[str, Any]) -> None:
         guards = v.get("guardrails")

src/uipath/agent/react/prompts.py

@@ -8,7 +8,7 @@
 
 {{systemPrompt}}
 
-Your adhere strictly to the following rules to ensure accuracy and data validity:
+You adhere strictly to the following rules to ensure accuracy and data validity:
 
 <rules>
 Data Verification and Tool Analysis:

src/uipath/platform/common/interrupt_models.py

@@ -134,3 +134,7 @@ class WaitDocumentExtraction(BaseModel):
     """Model representing a wait document extraction task creation."""
 
     extraction: StartExtractionResponse
+
+class UserMessageWait(BaseModel):
+    """Model representing a wait for a new user input message."""
+    

src/uipath/platform/resume_triggers/_protocol.py

@@ -35,6 +35,7 @@
     WaitJob,
     WaitTask,
 )
+from uipath.platform.common.interrupt_models import UserMessageWait
 from uipath.platform.context_grounding import DeepRagStatus
 from uipath.platform.errors import (
     BatchTransformNotCompleteException,
@@ -362,6 +363,10 @@ async def create_trigger(self, suspend_value: Any) -> UiPathResumeTrigger:
                     await self._handle_ixp_extraction_trigger(
                         suspend_value, resume_trigger, uipath
                     )
+                case UiPathResumeTriggerType.USER_MESSAGE_WAIT:
+                    self._handle_user_message_wait_trigger(
+                        suspend_value, resume_trigger, uipath
+                    )                    
                 case _:
                     raise UiPathFaultedTriggerError(
                         ErrorCategory.SYSTEM,
@@ -395,6 +400,8 @@ def _determine_trigger_type(self, value: Any) -> UiPathResumeTriggerType:
             return UiPathResumeTriggerType.BATCH_RAG
         if isinstance(value, (DocumentExtraction, WaitDocumentExtraction)):
             return UiPathResumeTriggerType.IXP_EXTRACTION
+        if isinstance(value, UserMessageWait):
+            return UiPathResumeTriggerType.USER_MESSAGE_WAIT
         # default to API trigger
         return UiPathResumeTriggerType.API
 
@@ -419,6 +426,8 @@ def _determine_trigger_name(self, value: Any) -> UiPathResumeTriggerName:
             return UiPathResumeTriggerName.BATCH_RAG
         if isinstance(value, (DocumentExtraction, WaitDocumentExtraction)):
             return UiPathResumeTriggerName.EXTRACTION
+        if isinstance(value, UserMessageWait):
+            return UiPathResumeTriggerName.USER_MESSAGE_WAIT
         # default to API trigger
         return UiPathResumeTriggerName.API
 
@@ -579,7 +588,17 @@ def _handle_api_trigger(
             inbox_id=str(uuid.uuid4()), request=serialize_object(value)
         )
 
+    def _handle_user_message_wait_trigger(
+        self, value: Any, resume_trigger: UiPathResumeTrigger, uipath: UiPath
+    ) -> None:
+        """Handle user message wait resume triggers.
 
+        Args:
+            value: The suspend value
+            resume_trigger: The resume trigger to populate
+        """
+        # Nothing to do?
+        
 class UiPathResumeTriggerHandler:
     """Combined handler for creating and reading resume triggers.