Add TCP network transport to ACPAgent

ACPAgent currently requires spawning the ACP server as a local subprocess,
which needs Node.js installed in the container. When using APIRemoteWorkspace
with K8s pods that lack Node.js, this forces custom container images.

Add an alternative TCP transport mode where ACPAgent connects to an
already-running ACP server over the network via asyncio.open_connection().
The ClientSideConnection works unchanged since it accepts any
StreamReader/StreamWriter pair.

Transport is configured via mutually exclusive fields validated by a
model_validator: either acp_command (subprocess) or acp_host+acp_port (TCP).

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

Author: simonrosenberg

openhands-sdk/openhands/sdk/agent/acp_agent.py

@@ -14,7 +14,7 @@
 from collections.abc import Generator
 from typing import TYPE_CHECKING, Any
 
-from pydantic import Field, PrivateAttr
+from pydantic import Field, PrivateAttr, model_validator
 
 from openhands.sdk.agent.base import AgentBase
 from openhands.sdk.conversation.state import ConversationExecutionStatus
@@ -263,20 +263,21 @@ def on_connect(self, conn: Any) -> None:  # noqa: ARG002
 class ACPAgent(AgentBase):
     """Agent that delegates to an ACP (Agent Client Protocol) server.
 
-    Instead of calling an LLM directly, this agent spawns an ACP-compatible
-    server (e.g. ``claude-code-acp``) as a subprocess and communicates with
-    it via the ACP protocol.  The server manages its own LLM, tools, and
-    execution lifecycle.
+    Instead of calling an LLM directly, this agent communicates with an
+    ACP-compatible server (e.g. ``claude-code-acp``) via the ACP protocol.
+    The server manages its own LLM, tools, and execution lifecycle.
 
-    Example::
+    Two transport modes are supported (mutually exclusive):
 
-        from openhands.sdk.agent import ACPAgent
-        from openhands.sdk.conversation import Conversation
+    **Subprocess mode** — the agent spawns the ACP server as a child process
+    and communicates via stdin/stdout JSON-RPC::
 
         agent = ACPAgent(acp_command=["npx", "-y", "claude-code-acp"])
-        conversation = Conversation(agent=agent, workspace="./workspace")
-        conversation.send_message("Hello! What is 2+2?")
-        conversation.run()
+
+    **TCP mode** — the agent connects to an already-running ACP server over
+    the network::
+
+        agent = ACPAgent(acp_host="acp-server.internal", acp_port=4001)
     """
 
     # Override required fields with ACP-appropriate defaults
@@ -285,12 +286,39 @@ class ACPAgent(AgentBase):
     include_default_tools: list[str] = Field(default_factory=list)
 
     # ACP-specific configuration
-    acp_command: list[str] = Field(
-        ...,
+    acp_command: list[str] | None = Field(
+        default=None,
         description=(
-            "Command to start the ACP server, e.g. ['npx', '-y', 'claude-code-acp']"
+            "Command to start the ACP server, e.g. ['npx', '-y', 'claude-code-acp']. "
+            "Mutually exclusive with acp_host."
         ),
     )
+    acp_host: str | None = Field(
+        default=None,
+        description=(
+            "Hostname of a remote ACP server. Mutually exclusive with acp_command."
+        ),
+    )
+    acp_port: int | None = Field(
+        default=None,
+        description="Port of the remote ACP server. Required when acp_host is set.",
+    )
+
+    @model_validator(mode="before")
+    @classmethod
+    def _validate_transport(cls, data):
+        if not isinstance(data, dict):
+            return data
+        has_command = data.get("acp_command") is not None
+        has_host = data.get("acp_host") is not None
+        if has_command and has_host:
+            raise ValueError("acp_command and acp_host are mutually exclusive")
+        if not has_command and not has_host:
+            raise ValueError("Either acp_command or acp_host must be provided")
+        if has_host and data.get("acp_port") is None:
+            raise ValueError("acp_port is required when acp_host is set")
+        return data
+
     acp_args: list[str] = Field(
         default_factory=list,
         description="Additional arguments for the ACP server command",
@@ -307,6 +335,7 @@ class ACPAgent(AgentBase):
     _process: Any = PrivateAttr(default=None)  # asyncio subprocess
     _client: Any = PrivateAttr(default=None)  # _OpenHandsACPClient
     _filtered_reader: Any = PrivateAttr(default=None)  # StreamReader
+    _tcp_writer: Any = PrivateAttr(default=None)  # asyncio.StreamWriter (TCP mode)
     _closed: bool = PrivateAttr(default=False)
     _working_dir: str = PrivateAttr(default="")
 
@@ -375,70 +404,86 @@ def init_state(
         self._initialized = True
 
     def _start_acp_server(self, state: ConversationState) -> None:
-        """Start the ACP subprocess and initialize the session."""
+        """Start the ACP connection and initialize the session.
+
+        In subprocess mode (``acp_command``), spawns the server as a child
+        process and communicates via stdin/stdout.  In TCP mode
+        (``acp_host``/``acp_port``), connects to an already-running server
+        over the network.
+        """
         import asyncio
 
         from acp.client.connection import (
             ClientSideConnection,
         )
-        from acp.transports import (
-            default_environment,
-        )
 
         client = _OpenHandsACPClient()
         client._llm_ref = self.llm
         self._client = client
 
-        # Build environment: inherit current env + ACP extras
-        env = default_environment()
-        env.update(os.environ)
-        env.update(self.acp_env)
-
-        command = self.acp_command[0]
-        args = list(self.acp_command[1:]) + list(self.acp_args)
-
         working_dir = str(state.workspace.working_dir)
 
-        async def _init() -> tuple[Any, Any, Any, str]:
-            # Spawn the subprocess directly so we can install a
-            # filtering reader that skips non-JSON-RPC lines some
-            # ACP servers (e.g. claude-code-acp v0.1.x) write to
-            # stdout.
-            process = await asyncio.create_subprocess_exec(
-                command,
-                *args,
-                stdin=asyncio.subprocess.PIPE,
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.PIPE,
-                env=env,
-            )
-            assert process.stdin is not None
-            assert process.stdout is not None
-
-            # Wrap the subprocess stdout in a filtering reader that
-            # only passes lines starting with '{' (JSON-RPC messages).
-            filtered_reader = asyncio.StreamReader()
-            asyncio.get_event_loop().create_task(
-                _filter_jsonrpc_lines(process.stdout, filtered_reader)
-            )
+        if self.acp_host is not None:
+            # --- TCP mode ---
+            async def _init_tcp() -> tuple[Any, str, Any]:
+                reader, writer = await asyncio.open_connection(
+                    self.acp_host, self.acp_port
+                )
+                conn = ClientSideConnection(
+                    client,
+                    writer,  # input_stream (write to server)
+                    reader,  # output_stream (read from server)
+                )
+                await conn.initialize(protocol_version=1)
+                response = await conn.new_session(cwd=working_dir)
+                return conn, response.session_id, writer
 
-            conn = ClientSideConnection(
-                client,
-                process.stdin,  # write to subprocess
-                filtered_reader,  # read filtered output
+            result = self._executor.run_async(_init_tcp)
+            self._conn, self._session_id, self._tcp_writer = result
+        else:
+            # --- Subprocess mode ---
+            from acp.transports import (
+                default_environment,
             )
 
-            # Initialize the protocol
-            await conn.initialize(protocol_version=1)
+            env = default_environment()
+            env.update(os.environ)
+            env.update(self.acp_env)
+
+            assert self.acp_command is not None
+            command = self.acp_command[0]
+            args = list(self.acp_command[1:]) + list(self.acp_args)
+
+            async def _init_subprocess() -> tuple[Any, Any, Any, str]:
+                process = await asyncio.create_subprocess_exec(
+                    command,
+                    *args,
+                    stdin=asyncio.subprocess.PIPE,
+                    stdout=asyncio.subprocess.PIPE,
+                    stderr=asyncio.subprocess.PIPE,
+                    env=env,
+                )
+                assert process.stdin is not None
+                assert process.stdout is not None
 
-            # Create a new session
-            response = await conn.new_session(cwd=working_dir)
-            session_id = response.session_id
+                filtered_reader = asyncio.StreamReader()
+                asyncio.get_event_loop().create_task(
+                    _filter_jsonrpc_lines(process.stdout, filtered_reader)
+                )
 
-            return conn, process, filtered_reader, session_id
+                conn = ClientSideConnection(
+                    client,
+                    process.stdin,
+                    filtered_reader,
+                )
+
+                await conn.initialize(protocol_version=1)
+                response = await conn.new_session(cwd=working_dir)
+                return conn, process, filtered_reader, response.session_id
+
+            result = self._executor.run_async(_init_subprocess)
+            self._conn, self._process, self._filtered_reader, self._session_id = result
 
-        result = self._executor.run_async(_init)
-        self._conn, self._process, self._filtered_reader, self._session_id = result
         self._working_dir = working_dir
 
     def step(
@@ -640,6 +685,14 @@ def _cleanup(self) -> None:
                 logger.debug("Error closing ACP connection: %s", e)
             self._conn = None
 
+        # Close TCP writer if in network mode
+        if self._tcp_writer is not None:
+            try:
+                self._tcp_writer.close()
+            except Exception as e:
+                logger.debug("Error closing TCP writer: %s", e)
+            self._tcp_writer = None
+
         # Terminate the subprocess
         if self._process is not None:
             try:

tests/sdk/agent/test_acp_agent.py

@@ -26,7 +26,15 @@
 
 
 def _make_agent(**kwargs) -> ACPAgent:
-    return ACPAgent(acp_command=["echo", "test"], **kwargs)
+    if "acp_command" not in kwargs and "acp_host" not in kwargs:
+        kwargs["acp_command"] = ["echo", "test"]
+    return ACPAgent(**kwargs)
+
+
+def _make_tcp_agent(**kwargs) -> ACPAgent:
+    kwargs.setdefault("acp_host", "localhost")
+    kwargs.setdefault("acp_port", 4001)
+    return ACPAgent(**kwargs)
 
 
 def _make_state(tmp_path) -> ConversationState:
@@ -57,9 +65,23 @@ def test_creates_with_empty_default_tools(self):
         agent = _make_agent()
         assert agent.include_default_tools == []
 
-    def test_requires_acp_command(self):
-        with pytest.raises(Exception):
-            ACPAgent()  # type: ignore[call-arg]
+    def test_creates_with_tcp_transport(self):
+        agent = ACPAgent(acp_host="localhost", acp_port=4001)
+        assert agent.acp_host == "localhost"
+        assert agent.acp_port == 4001
+        assert agent.acp_command is None
+
+    def test_rejects_both_command_and_host(self):
+        with pytest.raises(Exception, match="mutually exclusive"):
+            ACPAgent(acp_command=["echo"], acp_host="localhost", acp_port=4001)
+
+    def test_rejects_neither_command_nor_host(self):
+        with pytest.raises(Exception, match="Either acp_command or acp_host"):
+            ACPAgent()
+
+    def test_rejects_host_without_port(self):
+        with pytest.raises(Exception, match="acp_port is required"):
+            ACPAgent(acp_host="localhost")
 
     def test_acp_command_stored(self):
         agent = ACPAgent(acp_command=["npx", "-y", "claude-code-acp"])
@@ -122,6 +144,26 @@ def test_deserialization_from_dict(self):
         assert isinstance(agent, ACPAgent)
         assert agent.acp_command == ["echo", "test"]
 
+    def test_roundtrip_serialization_tcp(self):
+        agent = ACPAgent(acp_host="acp.internal", acp_port=5000)
+        dumped = agent.model_dump_json()
+        restored = AgentBase.model_validate_json(dumped)
+        assert isinstance(restored, ACPAgent)
+        assert restored.acp_host == "acp.internal"
+        assert restored.acp_port == 5000
+        assert restored.acp_command is None
+
+    def test_deserialization_from_dict_tcp(self):
+        data = {
+            "kind": "ACPAgent",
+            "acp_host": "10.0.0.1",
+            "acp_port": 4001,
+        }
+        agent = AgentBase.model_validate(data)
+        assert isinstance(agent, ACPAgent)
+        assert agent.acp_host == "10.0.0.1"
+        assert agent.acp_port == 4001
+
 
 # ---------------------------------------------------------------------------
 # Feature validation (init_state guards)
@@ -175,6 +217,31 @@ def test_emits_system_prompt_event(self, tmp_path):
         assert events[0].system_prompt.text == "ACP-managed agent"
         assert events[0].tools == []
 
+    def test_tcp_transport_connects(self, tmp_path):
+        """TCP mode stores connection, session_id, and tcp_writer."""
+        agent = _make_tcp_agent()
+        state = _make_state(tmp_path)
+        events: list = []
+
+        mock_conn = MagicMock()
+        mock_writer = MagicMock()
+
+        # Mock AsyncExecutor so run_async returns the expected TCP tuple
+        mock_executor = MagicMock()
+        mock_executor.run_async.return_value = (mock_conn, "tcp-session-1", mock_writer)
+
+        with patch(
+            "openhands.sdk.utils.async_executor.AsyncExecutor",
+            return_value=mock_executor,
+        ):
+            agent.init_state(state, on_event=events.append)
+
+        assert agent._conn is mock_conn
+        assert agent._session_id == "tcp-session-1"
+        assert agent._tcp_writer is mock_writer
+        assert agent._process is None
+        assert agent._filtered_reader is None
+
 
 # ---------------------------------------------------------------------------
 # _OpenHandsACPClient
@@ -490,6 +557,45 @@ def test_close_handles_errors_gracefully(self):
         # Should not raise
         agent.close()
 
+    def test_close_closes_tcp_writer(self):
+        agent = _make_tcp_agent()
+        mock_writer = MagicMock()
+        agent._tcp_writer = mock_writer
+        agent._executor = MagicMock()
+        agent._conn = None
+        agent._process = None
+
+        agent.close()
+
+        mock_writer.close.assert_called_once()
+        assert agent._tcp_writer is None
+
+    def test_close_skips_process_in_tcp_mode(self):
+        agent = _make_tcp_agent()
+        mock_writer = MagicMock()
+        agent._tcp_writer = mock_writer
+        agent._executor = MagicMock()
+        agent._conn = None
+        agent._process = None  # No process in TCP mode
+
+        agent.close()
+
+        # No process terminate/kill should be attempted
+        mock_writer.close.assert_called_once()
+
+    def test_close_tcp_writer_handles_errors(self):
+        agent = _make_tcp_agent()
+        mock_writer = MagicMock()
+        mock_writer.close.side_effect = OSError("connection reset")
+        agent._tcp_writer = mock_writer
+        agent._executor = MagicMock()
+        agent._conn = None
+        agent._process = None
+
+        # Should not raise
+        agent.close()
+        assert agent._tcp_writer is None
+
 
 # ---------------------------------------------------------------------------
 # _filter_jsonrpc_lines