fix: remove stream() method and emit_state events

Streaming events add unnecessary complexity for MCP servers.
The execute() path is sufficient.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: edis-uipath

src/uipath_mcp/_cli/_runtime/_factory.py

@@ -2,13 +2,15 @@
 
 import logging
 import uuid
+from typing import Any
 
 from uipath.runtime import (
     UiPathRuntimeContext,
     UiPathRuntimeFactorySettings,
     UiPathRuntimeProtocol,
 )
 from uipath.runtime.errors import UiPathErrorCategory
+from uipath.runtime.storage import UiPathRuntimeStorageProtocol
 
 from uipath_mcp._cli._runtime._exception import McpErrorCode, UiPathMcpRuntimeError
 from uipath_mcp._cli._runtime._runtime import UiPathMcpRuntime
@@ -66,7 +68,7 @@ async def discover_runtimes(self) -> list[UiPathRuntimeProtocol]:
         return runtimes
 
     async def new_runtime(
-        self, entrypoint: str, runtime_id: str
+        self, entrypoint: str, runtime_id: str, **kwargs: Any
     ) -> UiPathRuntimeProtocol:
         """Create a new MCP runtime instance.
 
@@ -115,6 +117,14 @@ async def new_runtime(
             server_slug=self.context.mcp_server_slug,
         )
 
+    async def get_storage(self) -> UiPathRuntimeStorageProtocol | None:
+        """Get factory storage.
+
+        MCP servers are long-running processes and don't need
+        cross-invocation state persistence.
+        """
+        return None
+
     async def get_settings(self) -> UiPathRuntimeFactorySettings | None:
         """Get factory settings."""
         return None

src/uipath_mcp/_cli/_runtime/_runtime.py

@@ -27,7 +27,6 @@
     UiPathErrorCategory,
     UiPathErrorCode,
 )
-from uipath.runtime.events import UiPathRuntimeStateEvent
 
 from .._utils._config import McpServer
 from ._context import UiPathServerType
@@ -153,85 +152,23 @@ async def stream(
         input: dict[str, Any] | None = None,
         options: UiPathStreamOptions | None = None,
     ) -> AsyncGenerator[UiPathRuntimeEvent, None]:
-        """Stream execution events from MCP server runtime in real-time.
+        """Stream execution for MCP server runtime.
 
-        Yields state events at key lifecycle points (initialization, registration,
-        connection, sessions) and finally yields the runtime result.
-
-        Args:
-            input: Optional input dictionary (unused for MCP servers).
-            options: Optional stream options.
-
-        Yields:
-            UiPathRuntimeStateEvent: State updates during server lifecycle.
-            UiPathRuntimeResult: Final result when server stops.
-
-        Raises:
-            UiPathMcpRuntimeError: If execution fails.
+        MCP servers don't emit intermediate events, so this just yields the final result.
         """
-        # Queue for real-time event streaming
-        # None is used as sentinel to signal completion
-        event_queue: asyncio.Queue[UiPathRuntimeEvent | None] = asyncio.Queue()
-        run_exception: list[Exception] = []
-
-        async def run_with_queue() -> None:
-            try:
-                result = await self._run_server(event_queue=event_queue)
-                await event_queue.put(result)
-            except Exception as e:
-                run_exception.append(e)
-            finally:
-                await event_queue.put(None)  # Signal completion
-
-        # Start the server task
-        run_task = asyncio.create_task(run_with_queue())
-
-        try:
-            # Consume events from queue in real-time
-            while True:
-                event = await event_queue.get()
-                if event is None:
-                    break
-                yield event
-        finally:
-            # Ensure the run task is properly awaited
-            if not run_task.done():
-                run_task.cancel()
-                try:
-                    await run_task
-                except asyncio.CancelledError:
-                    pass
+        result = await self._run_server()
+        yield result
 
-        # Re-raise any exception from the run task
-        if run_exception:
-            raise run_exception[0]
-
-    async def _run_server(
-        self,
-        event_queue: asyncio.Queue[UiPathRuntimeEvent | None] | None = None,
-    ) -> UiPathRuntimeResult:
+    async def _run_server(self) -> UiPathRuntimeResult:
         """Core server execution logic.
 
-        Args:
-            event_queue: Optional queue to push state events for real-time streaming.
-
         Returns:
             UiPathRuntimeResult with execution results.
 
         Raises:
             UiPathMcpRuntimeError: If execution fails.
         """
-
-        async def emit_state(node_name: str, payload: dict[str, Any]) -> None:
-            if event_queue:
-                event = UiPathRuntimeStateEvent(
-                    payload=payload,
-                    node_name=node_name,
-                )
-                await event_queue.put(event)
-
         try:
-            await emit_state("initializing", {"status": "validating_auth"})
 
             # Validate authentication configuration
             self._validate_auth()
@@ -262,11 +199,6 @@ async def emit_state(node_name: str, payload: dict[str, Any]) -> None:
 
             logger.info(f"Folder key: {self._folder_key}")
 
-            await emit_state(
-                "initializing",
-                {"status": "folder_resolved", "folder_key": self._folder_key},
-            )
-
             with tracer.start_as_current_span(self.slug) as root_span:
                 root_span.set_attribute("runtime_id", self._runtime_id)
                 root_span.set_attribute("command", str(self._server.command))
@@ -290,26 +222,13 @@ async def emit_state(node_name: str, payload: dict[str, Any]) -> None:
                 self._signalr_client.on_open(self._handle_signalr_open)
                 self._signalr_client.on_close(self._handle_signalr_close)
 
-                await emit_state("registering", {"status": "registering_server"})
-
                 # Register the local server with UiPath MCP Server
                 await self._register()
 
-                await emit_state(
-                    "registered",
-                    {
-                        "status": "server_registered",
-                        "slug": self.slug,
-                        "server_type": self.server_type.value,
-                    },
-                )
-
                 run_task = asyncio.create_task(self._signalr_client.run())
                 cancel_task = asyncio.create_task(self._cancel_event.wait())
                 self._keep_alive_task = asyncio.create_task(self._keep_alive())
 
-                await emit_state("running", {"status": "server_running"})
-
                 try:
                     # Wait for either the run to complete or cancellation
                     done, pending = await asyncio.wait(
@@ -334,8 +253,6 @@ async def emit_state(node_name: str, payload: dict[str, Any]) -> None:
                 if self._session_output:
                     output_result["content"] = self._session_output
 
-                await emit_state("completed", {"status": "server_stopped"})
-
                 return UiPathRuntimeResult(
                     output=output_result,
                     status=UiPathRuntimeStatus.SUCCESSFUL,