feat(sdk): extend existing conversation and workspace APIs for automation

openhands-sdk/openhands/sdk/conversation/impl/remote_conversation.py

@@ -7,7 +7,7 @@
 import uuid
 from collections.abc import Callable, Mapping
 from queue import Empty, Queue
-from typing import TYPE_CHECKING, Final, SupportsIndex, overload
+from typing import TYPE_CHECKING, Final, Self, SupportsIndex, overload
 from urllib.parse import urlparse
 
 import httpx
@@ -19,6 +19,7 @@
 
 
 if TYPE_CHECKING:
+    from openhands.sdk.conversation.request import StartConversationRequest
     from openhands.sdk.tool.schema import Action, Observation
 from openhands.sdk.conversation.conversation_stats import ConversationStats
 from openhands.sdk.conversation.events_list_base import EventsListBase
@@ -765,18 +766,6 @@ def __init__(
             observability_span_name: Optional child span name for observability
                       backends. The root span remains named "conversation".
         """
-        super().__init__()  # Initialize base class with span tracking
-        self.agent = agent
-        self._callbacks = callbacks or []
-        self.max_iteration_per_run = max_iteration_per_run
-        self.workspace = workspace
-        self._client = workspace.client
-        self._conversation_info_base_path = LEGACY_CONVERSATIONS_PATH
-        self._conversation_action_base_path = LEGACY_CONVERSATIONS_PATH
-        self._cleanup_initiated = False
-        self._terminal_status_queue: Queue[str] = Queue()
-        self._run_armed = threading.Event()
-
         # Client tool specs the server already has persisted for this
         # conversation (populated when re-attaching to an existing one). These
         # must be registered locally before the initial event sync so that
@@ -787,9 +776,9 @@ def __init__(
         if conversation_id is not None:
             # Try to attach to existing conversation
             resp = _send_request(
-                self._client,
+                workspace.client,
                 "GET",
-                f"{self._conversation_info_base_path}/{conversation_id}",
+                f"{LEGACY_CONVERSATIONS_PATH}/{conversation_id}",
                 acceptable_status_codes={404},
             )
             if resp.status_code == 404:
@@ -808,8 +797,6 @@ def __init__(
                     attached_client_tools.append(
                         ClientToolSpec.model_validate(raw_spec)
                     )
-                # Conversation exists, use the provided ID
-                self._id = conversation_id
 
         if should_create:
             # Import here to avoid circular imports
@@ -832,7 +819,7 @@ def __init__(
                 "stuck_detection": stuck_detection,
                 # We need to convert RemoteWorkspace to LocalWorkspace for the server
                 "workspace": LocalWorkspace(
-                    working_dir=self.workspace.working_dir
+                    working_dir=workspace.working_dir
                 ).model_dump(),
                 # Include tool module qualnames for dynamic registration on server
                 "tool_module_qualnames": tool_qualnames,
@@ -874,9 +861,9 @@ def __init__(
             if conversation_id is not None:
                 payload["conversation_id"] = str(conversation_id)
             resp = _send_request(
-                self._client,
+                workspace.client,
                 "POST",
-                self._conversation_info_base_path,
+                LEGACY_CONVERSATIONS_PATH,
                 json=payload,
             )
             data = resp.json()
@@ -886,17 +873,155 @@ def __init__(
                 raise RuntimeError(
                     "Invalid response from server: missing conversation id"
                 )
-            self._id = uuid.UUID(cid)
+            conversation_id = uuid.UUID(cid)
+
+            workspace.register_conversation(str(conversation_id))
+
+        assert conversation_id is not None
+        self._initialize_connection(
+            agent=agent,
+            workspace=workspace,
+            conversation_id=conversation_id,
+            callbacks=callbacks,
+            max_iteration_per_run=max_iteration_per_run,
+            client_tools=[*(client_tools or []), *attached_client_tools],
+            visualizer=visualizer,
+        )
 
-            workspace.register_conversation(str(self._id))
+        # Initialize secrets if provided
+        if secrets:
+            # Convert dict[str, str] to dict[str, SecretValue]
+            secret_values: dict[str, SecretValue] = {k: v for k, v in secrets.items()}
+            self.update_secrets(secret_values)
+
+        self._start_observability_span(
+            str(self._id),
+            span_name=observability_span_name,
+            user_id=user_id,
+            metadata=observability_metadata,
+            tags=observability_tags,
+            conversation_tags=tags,
+        )
+        # All hooks (including SessionStart/SessionEnd) are executed server-side.
+        # hook_config is sent in the creation payload.
+        self.delete_on_close = delete_on_close
+
+    @classmethod
+    def create(
+        cls,
+        workspace: RemoteWorkspace,
+        request: "StartConversationRequest",
+        *,
+        callbacks: list[ConversationCallbackType] | None = None,
+        visualizer: (
+            type[ConversationVisualizerBase] | ConversationVisualizerBase | None
+        ) = DefaultConversationVisualizer,
+    ) -> Self:
+        """Submit a creation request and connect to the returned conversation.
+
+        The request selects the agent or saved server profile and all server
+        options. A supplied conversation ID follows the server's idempotency
+        contract; this method does not probe for an existing conversation.
+        """
+        response = _send_request(
+            workspace.client,
+            "POST",
+            LEGACY_CONVERSATIONS_PATH,
+            json=request.model_dump(mode="json", context={"expose_secrets": True}),
+        )
+        info = response.json()
+        workspace.register_conversation(info["id"])
+        conversation = cls._from_info(workspace, info, callbacks, visualizer)
+        conversation._start_observability_span(
+            str(conversation.id),
+            span_name=request.observability_span_name,
+            user_id=request.user_id,
+            metadata=request.observability_metadata,
+            tags=request.observability_tags,
+            conversation_tags=request.tags,
+        )
+        return conversation
+
+    @classmethod
+    def attach(
+        cls,
+        workspace: RemoteWorkspace,
+        conversation_id: ConversationID,
+        *,
+        callbacks: list[ConversationCallbackType] | None = None,
+        visualizer: (
+            type[ConversationVisualizerBase] | ConversationVisualizerBase | None
+        ) = DefaultConversationVisualizer,
+    ) -> Self:
+        """Connect using the saved agent; never create or update a conversation.
 
+        Missing or inaccessible conversations raise an HTTP error.
+        """
+        response = _send_request(
+            workspace.client, "GET", f"{LEGACY_CONVERSATIONS_PATH}/{conversation_id}"
+        )
+        conversation = cls._from_info(workspace, response.json(), callbacks, visualizer)
+        conversation._start_observability_span(str(conversation.id))
+        return conversation
+
+    @classmethod
+    def _from_info(
+        cls,
+        workspace: RemoteWorkspace,
+        info: dict,
+        callbacks: list[ConversationCallbackType] | None,
+        visualizer: (
+            type[ConversationVisualizerBase] | ConversationVisualizerBase | None
+        ),
+    ) -> Self:
+        conversation = cls.__new__(cls)
+        conversation._initialize_connection(
+            agent=_validate_remote_agent(info["agent"]),
+            workspace=workspace,
+            conversation_id=uuid.UUID(info["id"]),
+            callbacks=callbacks,
+            max_iteration_per_run=info["max_iterations"],
+            client_tools=[
+                ClientToolSpec.model_validate(spec)
+                for spec in info.get("client_tools") or []
+            ],
+            visualizer=visualizer,
+        )
+        return conversation
+
+    def _initialize_connection(
+        self,
+        *,
+        agent: AgentBase,
+        workspace: RemoteWorkspace,
+        conversation_id: ConversationID,
+        callbacks: list[ConversationCallbackType] | None,
+        max_iteration_per_run: int,
+        client_tools: list[ClientToolSpec],
+        visualizer: (
+            type[ConversationVisualizerBase] | ConversationVisualizerBase | None
+        ),
+    ) -> None:
+        super().__init__()  # Initialize base class with span tracking
+        self.agent = agent
+        self._callbacks = callbacks or []
+        self.max_iteration_per_run = max_iteration_per_run
+        self.workspace = workspace
+        self._client = workspace.client
+        self._conversation_info_base_path = LEGACY_CONVERSATIONS_PATH
+        self._conversation_action_base_path = LEGACY_CONVERSATIONS_PATH
+        self._cleanup_initiated = False
+        self._terminal_status_queue: Queue[str] = Queue()
+        self._run_armed = threading.Event()
+
+        self._id = conversation_id
         # Register client tool action types locally so WebSocket/persisted
         # events with ClientAction_* action_type can be deserialized by the
         # event loop. This must cover both the specs the caller passed in and
         # the specs the server already had persisted (when re-attaching), so a
         # plain reattach by conversation_id can still sync persisted events.
         seen_client_tool_names: set[str] = set()
-        for spec in [*(client_tools or []), *attached_client_tools]:
+        for spec in client_tools:
             if spec.name in seen_client_tool_names:
                 continue
             seen_client_tool_names.add(spec.name)
@@ -1034,24 +1159,6 @@ def run_complete_callback(event: Event) -> None:
         # This is the "reconciliation" part of the subscription handshake.
         self._state.events.reconcile()
 
-        # Initialize secrets if provided
-        if secrets:
-            # Convert dict[str, str] to dict[str, SecretValue]
-            secret_values: dict[str, SecretValue] = {k: v for k, v in secrets.items()}
-            self.update_secrets(secret_values)
-
-        self._start_observability_span(
-            str(self._id),
-            span_name=observability_span_name,
-            user_id=user_id,
-            metadata=observability_metadata,
-            tags=observability_tags,
-            conversation_tags=tags,
-        )
-        # All hooks (including SessionStart/SessionEnd) are executed server-side.
-        # hook_config is sent in the creation payload.
-        self.delete_on_close = delete_on_close
-
     def _create_llm_completion_log_callback(self) -> ConversationCallbackType:
         """Create a callback that writes LLM completion logs to client filesystem."""
 
@@ -1510,6 +1617,15 @@ def ask_agent(self, question: str) -> str:
         data = resp.json()
         return data["response"]
 
+    def set_title(self, title: str) -> None:
+        """Set the persisted display title without running the agent."""
+        _send_request(
+            self._client,
+            "PATCH",
+            f"{self._conversation_info_base_path}/{self._id}",
+            json={"title": title},
+        )
+
     @observe(
         name="conversation.generate_title",
         ignore_inputs=["llm"],

openhands-sdk/openhands/sdk/workspace/remote/async_remote_workspace.py

@@ -1,6 +1,7 @@
 from collections.abc import Generator
 from pathlib import Path
-from typing import Any
+from types import TracebackType
+from typing import Any, Self
 from urllib.request import urlopen
 
 import httpx
@@ -16,6 +17,25 @@ class AsyncRemoteWorkspace(RemoteWorkspaceMixin):
 
     _client: httpx.AsyncClient | None = PrivateAttr(default=None)
 
+    async def __aenter__(self) -> Self:
+        """Enter a workspace whose HTTP pool is closed on context exit."""
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None:
+        """Close the client without releasing the server's runtime."""
+        await self.reset_client()
+
+    async def get_server_info(self) -> dict[str, Any]:
+        """Return server metadata, matching RemoteWorkspace.get_server_info."""
+        response = await self.client.get("/server_info")
+        response.raise_for_status()
+        return response.json()
+
     async def reset_client(self) -> None:
         """Reset the HTTP client to force re-initialization.
 
@@ -54,6 +74,29 @@ async def _execute(self, generator: Generator[dict[str, Any], httpx.Response, An
         except StopIteration as e:
             return e.value
 
+    async def start_command(
+        self,
+        command: str,
+        cwd: str | Path | None = None,
+        timeout: float = 30,
+    ) -> str:
+        """Start a command and return its ID without waiting for completion."""
+        return await self._execute(self._start_command_generator(command, cwd, timeout))
+
+    async def get_command_output(
+        self, command_id: str | None = None
+    ) -> dict[str, Any] | None:
+        """Read the latest output; a missing exit code means it is still running."""
+        return await self._execute(self._get_command_output_generator(command_id))
+
+    async def get_runtime_session_key(self) -> str:
+        """Get the scoped worker credential for this conversation runtime."""
+        return await self._execute(self._runtime_lifecycle_generator(release=False))
+
+    async def release_runtime(self) -> None:
+        """Release execution resources while retaining conversation history."""
+        await self._execute(self._runtime_lifecycle_generator(release=True))
+
     async def execute_command(
         self,
         command: str,
@@ -79,15 +122,15 @@ async def execute_command(
 
     async def file_upload(
         self,
-        source_path: str | Path,
+        source_path: str | Path | bytes,
         destination_path: str | Path,
     ) -> FileOperationResult:
         """Upload a file to the remote system.
 
         Reads the local file and sends it to the remote system via HTTP API.
 
         Args:
-            source_path: Path to the local source file
+            source_path: Local file path or in-memory bytes
             destination_path: Path where the file should be uploaded on remote system
 
         Returns: