feat: add reconnect_mcp_server and toggle_mcp_server client methods

Author: qing-ant

src/claude_agent_sdk/_internal/query.py

@@ -567,6 +567,34 @@ async def rewind_files(self, user_message_id: str) -> None:
             }
         )
 
+    async def reconnect_mcp_server(self, server_name: str) -> None:
+        """Reconnect a disconnected or failed MCP server.
+
+        Args:
+            server_name: The name of the MCP server to reconnect
+        """
+        await self._send_control_request(
+            {
+                "subtype": "mcp_reconnect",
+                "serverName": server_name,
+            }
+        )
+
+    async def toggle_mcp_server(self, server_name: str, enabled: bool) -> None:
+        """Enable or disable an MCP server.
+
+        Args:
+            server_name: The name of the MCP server to toggle
+            enabled: Whether the server should be enabled
+        """
+        await self._send_control_request(
+            {
+                "subtype": "mcp_toggle",
+                "serverName": server_name,
+                "enabled": enabled,
+            }
+        )
+
     async def stream_input(self, stream: AsyncIterable[dict[str, Any]]) -> None:
         """Stream input messages to transport.
 

src/claude_agent_sdk/client.py

@@ -304,6 +304,54 @@ async def rewind_files(self, user_message_id: str) -> None:
             raise CLIConnectionError("Not connected. Call connect() first.")
         await self._query.rewind_files(user_message_id)
 
+    async def reconnect_mcp_server(self, server_name: str) -> None:
+        """Reconnect a disconnected or failed MCP server (only works with streaming mode).
+
+        Use this to retry connecting to an MCP server that failed to connect
+        or was disconnected. Raises an exception if the reconnection fails.
+
+        Args:
+            server_name: The name of the MCP server to reconnect
+
+        Example:
+            ```python
+            async with ClaudeSDKClient(options) as client:
+                status = await client.get_mcp_status()
+                for server in status.get("mcpServers", []):
+                    if server["status"] == "failed":
+                        await client.reconnect_mcp_server(server["name"])
+            ```
+        """
+        if not self._query:
+            raise CLIConnectionError("Not connected. Call connect() first.")
+        await self._query.reconnect_mcp_server(server_name)
+
+    async def toggle_mcp_server(self, server_name: str, enabled: bool) -> None:
+        """Enable or disable an MCP server (only works with streaming mode).
+
+        Disabling a server disconnects it and removes its tools from the
+        available tool set. Enabling a server reconnects it and makes its
+        tools available again. Raises an exception on failure.
+
+        Args:
+            server_name: The name of the MCP server to toggle
+            enabled: True to enable the server, False to disable it
+
+        Example:
+            ```python
+            async with ClaudeSDKClient(options) as client:
+                # Temporarily disable a server
+                await client.toggle_mcp_server("my-server", enabled=False)
+                await client.query("Do something without my-server tools")
+
+                # Re-enable it later
+                await client.toggle_mcp_server("my-server", enabled=True)
+            ```
+        """
+        if not self._query:
+            raise CLIConnectionError("Not connected. Call connect() first.")
+        await self._query.toggle_mcp_server(server_name, enabled)
+
     async def get_mcp_status(self) -> dict[str, Any]:
         """Get current MCP server connection status (only works with streaming mode).
 

src/claude_agent_sdk/types.py

@@ -828,6 +828,23 @@ class SDKControlRewindFilesRequest(TypedDict):
     user_message_id: str
 
 
+class SDKControlMcpReconnectRequest(TypedDict):
+    """Reconnects a disconnected or failed MCP server."""
+
+    subtype: Literal["mcp_reconnect"]
+    # Note: wire protocol uses camelCase for this field
+    serverName: str
+
+
+class SDKControlMcpToggleRequest(TypedDict):
+    """Enables or disables an MCP server."""
+
+    subtype: Literal["mcp_toggle"]
+    # Note: wire protocol uses camelCase for this field
+    serverName: str
+    enabled: bool
+
+
 class SDKControlRequest(TypedDict):
     type: Literal["control_request"]
     request_id: str
@@ -839,6 +856,8 @@ class SDKControlRequest(TypedDict):
         | SDKHookCallbackRequest
         | SDKControlMcpMessageRequest
         | SDKControlRewindFilesRequest
+        | SDKControlMcpReconnectRequest
+        | SDKControlMcpToggleRequest
     )
 
 

tests/test_streaming_client.py

@@ -107,6 +107,54 @@ async def control_protocol_generator():
     return mock_transport
 
 
+def _create_mock_transport_with_control_responses():
+    """Create a mock transport that responds with success to all control requests.
+
+    Useful for testing client methods that send control requests (e.g.
+    reconnect_mcp_server, toggle_mcp_server) without needing to special-case
+    each subtype in the mock.
+    """
+    mock_transport = AsyncMock()
+    mock_transport.connect = AsyncMock()
+    mock_transport.close = AsyncMock()
+    mock_transport.end_input = AsyncMock()
+    mock_transport.is_ready = Mock(return_value=True)
+
+    written_messages: list[str] = []
+
+    async def mock_write(data):
+        written_messages.append(data)
+
+    mock_transport.write = AsyncMock(side_effect=mock_write)
+
+    async def control_protocol_generator():
+        # Poll for control requests and respond with success to each one.
+        last_check = 0
+        timeout_counter = 0
+        while timeout_counter < 200:  # Avoid infinite loop
+            await asyncio.sleep(0.01)
+            timeout_counter += 1
+
+            for msg_str in written_messages[last_check:]:
+                try:
+                    msg = json.loads(msg_str.strip())
+                    if msg.get("type") == "control_request":
+                        yield {
+                            "type": "control_response",
+                            "response": {
+                                "request_id": msg.get("request_id"),
+                                "subtype": "success",
+                                "response": {},
+                            },
+                        }
+                except (json.JSONDecodeError, KeyError, AttributeError):
+                    pass
+            last_check = len(written_messages)
+
+    mock_transport.read_messages = control_protocol_generator
+    return mock_transport
+
+
 class TestClaudeSDKClientStreaming:
     """Test ClaudeSDKClient streaming functionality."""
 
@@ -467,6 +515,128 @@ async def _test():
 
         anyio.run(_test)
 
+    def test_reconnect_mcp_server(self):
+        """Test reconnect_mcp_server sends correct control request."""
+
+        async def _test():
+            with patch(
+                "claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
+            ) as mock_transport_class:
+                mock_transport = _create_mock_transport_with_control_responses()
+                mock_transport_class.return_value = mock_transport
+
+                async with ClaudeSDKClient() as client:
+                    await client.reconnect_mcp_server("my-server")
+                    # Check that a control request was sent via write
+                    write_calls = mock_transport.write.call_args_list
+                    request_found = False
+                    for call in write_calls:
+                        data = call[0][0]
+                        try:
+                            msg = json.loads(data.strip())
+                            req = msg.get("request", {})
+                            if (
+                                msg.get("type") == "control_request"
+                                and req.get("subtype") == "mcp_reconnect"
+                            ):
+                                # Verify wire format uses camelCase serverName
+                                assert req.get("serverName") == "my-server"
+                                request_found = True
+                                break
+                        except (json.JSONDecodeError, KeyError, AttributeError):
+                            pass
+                    assert request_found, "mcp_reconnect control request not found"
+
+        anyio.run(_test)
+
+    def test_reconnect_mcp_server_not_connected(self):
+        """Test reconnect_mcp_server when not connected raises error."""
+
+        async def _test():
+            client = ClaudeSDKClient()
+            with pytest.raises(CLIConnectionError, match="Not connected"):
+                await client.reconnect_mcp_server("my-server")
+
+        anyio.run(_test)
+
+    def test_toggle_mcp_server(self):
+        """Test toggle_mcp_server sends correct control request."""
+
+        async def _test():
+            with patch(
+                "claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
+            ) as mock_transport_class:
+                mock_transport = _create_mock_transport_with_control_responses()
+                mock_transport_class.return_value = mock_transport
+
+                async with ClaudeSDKClient() as client:
+                    await client.toggle_mcp_server("my-server", False)
+                    # Check that a control request was sent via write
+                    write_calls = mock_transport.write.call_args_list
+                    request_found = False
+                    for call in write_calls:
+                        data = call[0][0]
+                        try:
+                            msg = json.loads(data.strip())
+                            req = msg.get("request", {})
+                            if (
+                                msg.get("type") == "control_request"
+                                and req.get("subtype") == "mcp_toggle"
+                            ):
+                                # Verify wire format uses camelCase serverName
+                                assert req.get("serverName") == "my-server"
+                                assert req.get("enabled") is False
+                                request_found = True
+                                break
+                        except (json.JSONDecodeError, KeyError, AttributeError):
+                            pass
+                    assert request_found, "mcp_toggle control request not found"
+
+        anyio.run(_test)
+
+    def test_toggle_mcp_server_enabled_true(self):
+        """Test toggle_mcp_server with enabled=True."""
+
+        async def _test():
+            with patch(
+                "claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
+            ) as mock_transport_class:
+                mock_transport = _create_mock_transport_with_control_responses()
+                mock_transport_class.return_value = mock_transport
+
+                async with ClaudeSDKClient() as client:
+                    await client.toggle_mcp_server("other-server", True)
+                    write_calls = mock_transport.write.call_args_list
+                    request_found = False
+                    for call in write_calls:
+                        data = call[0][0]
+                        try:
+                            msg = json.loads(data.strip())
+                            req = msg.get("request", {})
+                            if (
+                                msg.get("type") == "control_request"
+                                and req.get("subtype") == "mcp_toggle"
+                            ):
+                                assert req.get("serverName") == "other-server"
+                                assert req.get("enabled") is True
+                                request_found = True
+                                break
+                        except (json.JSONDecodeError, KeyError, AttributeError):
+                            pass
+                    assert request_found, "mcp_toggle control request not found"
+
+        anyio.run(_test)
+
+    def test_toggle_mcp_server_not_connected(self):
+        """Test toggle_mcp_server when not connected raises error."""
+
+        async def _test():
+            client = ClaudeSDKClient()
+            with pytest.raises(CLIConnectionError, match="Not connected"):
+                await client.toggle_mcp_server("my-server", True)
+
+        anyio.run(_test)
+
     def test_client_with_options(self):
         """Test client initialization with options."""