fix: remove stdin timeout for hooks and SDK MCP servers

src/claude_agent_sdk/_internal/query.py

@@ -112,9 +112,6 @@ def __init__(
 
         # Track first result for proper stream closure with SDK MCP servers
         self._first_result_event = anyio.Event()
-        self._stream_close_timeout = (
-            float(os.environ.get("CLAUDE_CODE_STREAM_CLOSE_TIMEOUT", "60000")) / 1000.0
-        )  # Convert ms to seconds
 
     async def initialize(self) -> dict[str, Any] | None:
         """Initialize control protocol if in streaming mode.
@@ -615,17 +612,19 @@ async def wait_for_result_and_end_input(self) -> None:
         """Wait for the first result (if needed) then close stdin.
 
         If SDK MCP servers or hooks require bidirectional communication,
-        keeps stdin open until the first result arrives (or timeout).
-        Otherwise closes stdin immediately.
+        keeps stdin open until the first result arrives. The control protocol
+        requires stdin to remain open for the entire conversation, so no
+        timeout is applied. The event is guaranteed to fire: either when the
+        result message arrives, or in _read_messages' finally block if the
+        process exits early.
         """
         if self.sdk_mcp_servers or self.hooks:
             logger.debug(
                 "Waiting for first result before closing stdin "
                 f"(sdk_mcp_servers={len(self.sdk_mcp_servers)}, "
                 f"has_hooks={bool(self.hooks)})"
             )
-            with anyio.move_on_after(self._stream_close_timeout):
-                await self._first_result_event.wait()
+            await self._first_result_event.wait()
 
         await self.transport.end_input()
 

tests/test_query.py

@@ -21,6 +21,7 @@
     query,
     tool,
 )
+from claude_agent_sdk._internal.query import Query
 from claude_agent_sdk.types import HookMatcher
 
 
@@ -436,3 +437,58 @@ async def prompt_stream():
             assert len(control_responses) == 2
 
         anyio.run(_test)
+
+
+class TestNoTimeoutForHooksAndMcpServers:
+    """Regression test for #730: stdin must not be closed by a timeout when
+    hooks or SDK MCP servers are active."""
+
+    def test_hooks_wait_without_timeout(self):
+        """wait_for_result_and_end_input() should wait indefinitely for the
+        result event when hooks are configured, not cut off after 60s."""
+
+        async def _test():
+            mock_transport = _make_mock_transport(messages=[])
+            end_input_called = anyio.Event()
+
+            async def tracking_end_input():
+                end_input_called.set()
+
+            mock_transport.end_input = tracking_end_input
+
+            q = Query(
+                transport=mock_transport,
+                is_streaming_mode=True,
+                hooks={"PreToolUse": [{"matcher": "Bash", "hooks": ["hook_0"]}]},
+            )
+
+            async with anyio.create_task_group() as tg:
+
+                async def wait_then_check():
+                    await anyio.sleep(0.05)
+                    assert not end_input_called.is_set()
+                    q._first_result_event.set()
+                    await anyio.sleep(0.05)
+                    assert end_input_called.is_set()
+
+                tg.start_soon(q.wait_for_result_and_end_input)
+                tg.start_soon(wait_then_check)
+
+        anyio.run(_test)
+
+    def test_no_hooks_closes_immediately(self):
+        """Without hooks or SDK MCP servers, end_input should be called
+        immediately without waiting for any event."""
+
+        async def _test():
+            mock_transport = _make_mock_transport(messages=[])
+
+            q = Query(
+                transport=mock_transport,
+                is_streaming_mode=True,
+            )
+
+            await q.wait_for_result_and_end_input()
+            mock_transport.end_input.assert_called_once()
+
+        anyio.run(_test)