[agent] Optimize the agent structure by moving the confirmation logic and risk level assessment down to the security service.

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

@@ -2,7 +2,6 @@
 
 from pydantic import ValidationError, model_validator
 
-import openhands.sdk.security.analyzer as analyzer
 import openhands.sdk.security.risk as risk
 from openhands.sdk.agent.base import AgentBase
 from openhands.sdk.agent.utils import fix_malformed_tool_arguments
@@ -42,15 +41,12 @@
     should_enable_observability,
 )
 from openhands.sdk.observability.utils import extract_action_name
-from openhands.sdk.security.llm_analyzer import LLMSecurityAnalyzer
 from openhands.sdk.tool import (
     Action,
     Observation,
 )
 from openhands.sdk.tool.builtins import (
-    FinishAction,
     FinishTool,
-    ThinkAction,
 )
 
 
@@ -232,7 +228,6 @@ def step(
                     tool_call,
                     llm_response_id=llm_response.id,
                     on_event=on_event,
-                    security_analyzer=state.security_analyzer,
                     thought=thought_content
                     if i == 0
                     else [],  # Only first gets thought
@@ -249,7 +244,10 @@ def step(
                 action_events.append(action_event)
 
             # Handle confirmation mode - exit early if actions need confirmation
-            if self._requires_user_confirmation(state, action_events):
+            if self._security_service.requires_confirmation(action_events):
+                state.execution_status = (
+                    ConversationExecutionStatus.WAITING_FOR_CONFIRMATION
+                )
                 return
 
             if action_events:
@@ -279,87 +277,11 @@ def step(
             )
             on_event(token_event)
 
-    def _requires_user_confirmation(
-        self, state: ConversationState, action_events: list[ActionEvent]
-    ) -> bool:
-        """
-        Decide whether user confirmation is needed to proceed.
-
-        Rules:
-            1. Confirmation mode is enabled
-            2. Every action requires confirmation
-            3. A single `FinishAction` never requires confirmation
-            4. A single `ThinkAction` never requires confirmation
-        """
-        # A single `FinishAction` or `ThinkAction` never requires confirmation
-        if len(action_events) == 1 and isinstance(
-            action_events[0].action, (FinishAction, ThinkAction)
-        ):
-            return False
-
-        # If there are no actions there is nothing to confirm
-        if len(action_events) == 0:
-            return False
-
-        # If a security analyzer is registered, use it to grab the risks of the actions
-        # involved. If not, we'll set the risks to UNKNOWN.
-        if state.security_analyzer is not None:
-            risks = [
-                risk
-                for _, risk in state.security_analyzer.analyze_pending_actions(
-                    action_events
-                )
-            ]
-        else:
-            risks = [risk.SecurityRisk.UNKNOWN] * len(action_events)
-
-        # Grab the confirmation policy from the state and pass in the risks.
-        if any(state.confirmation_policy.should_confirm(risk) for risk in risks):
-            state.execution_status = (
-                ConversationExecutionStatus.WAITING_FOR_CONFIRMATION
-            )
-            return True
-
-        return False
-
-    def _extract_security_risk(
-        self,
-        arguments: dict,
-        tool_name: str,
-        read_only_tool: bool,
-        security_analyzer: analyzer.SecurityAnalyzerBase | None = None,
-    ) -> risk.SecurityRisk:
-        requires_sr = isinstance(security_analyzer, LLMSecurityAnalyzer)
-        raw = arguments.pop("security_risk", None)
-
-        # Default risk value for action event
-        # Tool is marked as read-only so security risk can be ignored
-        if read_only_tool:
-            return risk.SecurityRisk.UNKNOWN
-
-        # Raises exception if failed to pass risk field when expected
-        # Exception will be sent back to agent as error event
-        # Strong models like GPT-5 can correct itself by retrying
-        if requires_sr and raw is None:
-            raise ValueError(
-                f"Failed to provide security_risk field in tool '{tool_name}'"
-            )
-
-        # When using weaker models without security analyzer
-        # safely ignore missing security risk fields
-        if not requires_sr and raw is None:
-            return risk.SecurityRisk.UNKNOWN
-
-        # Raises exception if invalid risk enum passed by LLM
-        security_risk = risk.SecurityRisk(raw)
-        return security_risk
-
     def _get_action_event(
         self,
         tool_call: MessageToolCall,
         llm_response_id: str,
         on_event: ConversationCallbackType,
-        security_analyzer: analyzer.SecurityAnalyzerBase | None = None,
         thought: list[TextContent] | None = None,
         reasoning_content: str | None = None,
         thinking_blocks: list[ThinkingBlock | RedactedThinkingBlock] | None = None,
@@ -405,11 +327,10 @@ def _get_action_event(
 
             # Fix malformed arguments (e.g., JSON strings for list/dict fields)
             arguments = fix_malformed_tool_arguments(arguments, tool.action_type)
-            security_risk = self._extract_security_risk(
+            security_risk = self._security_service.extract_security_risk(
                 arguments,
                 tool.name,
                 tool.annotations.readOnlyHint if tool.annotations else False,
-                security_analyzer,
             )
             assert "security_risk" not in arguments, (
                 "Unexpected 'security_risk' key found in tool arguments"

openhands-sdk/openhands/sdk/agent/base.py

@@ -15,6 +15,7 @@
 from openhands.sdk.logger import get_logger
 from openhands.sdk.mcp import create_mcp_tools
 from openhands.sdk.security import analyzer
+from openhands.sdk.security.security_service import SecurityService
 from openhands.sdk.tool import BUILT_IN_TOOLS, Tool, ToolDefinition, resolve_tool
 from openhands.sdk.utils.models import DiscriminatedUnionMixin
 from openhands.sdk.utils.pydantic_diff import pretty_pydantic_diff
@@ -272,6 +273,8 @@ def _initialize(self, state: "ConversationState"):
 
         # Store tools in a dict for easy access
         self._tools = {tool.name: tool for tool in tools}
+        # Build the security service based on the state.
+        self._security_service = SecurityService(state)
 
     @abstractmethod
     def step(

openhands-sdk/openhands/sdk/security/security_service.py

@@ -0,0 +1,86 @@
+from typing import TYPE_CHECKING
+
+
+if TYPE_CHECKING:
+    from openhands.sdk.conversation.state import ConversationState
+
+from openhands.sdk.event.llm_convertible.action import ActionEvent
+from openhands.sdk.security import risk
+from openhands.sdk.security.llm_analyzer import LLMSecurityAnalyzer
+from openhands.sdk.tool.builtins.finish import FinishAction
+from openhands.sdk.tool.builtins.think import ThinkAction
+
+
+class SecurityService:
+    def __init__(
+        self,
+        state: "ConversationState",
+    ):
+        self._state = state
+
+    def requires_confirmation(
+        self,
+        action_events: list[ActionEvent],
+    ) -> bool:
+        """
+        Decide whether user confirmation is needed to proceed.
+
+        Rules:
+            1. Confirmation mode is enabled
+            2. Every action requires confirmation
+            3. A single `FinishAction` never requires confirmation
+            4. A single `ThinkAction` never requires confirmation
+        """
+        # If there are no actions there is nothing to confirm
+        if len(action_events) == 0:
+            return False
+
+        if all(
+            isinstance(action_event.action, (FinishAction, ThinkAction))
+            for action_event in action_events
+        ):
+            return False
+        # If a security analyzer is registered, use it to grab the risks of the actions
+        # involved. If not, we'll set the risks to UNKNOWN.
+        if self._state.security_analyzer is not None:
+            risks = (
+                r
+                for _, r in self._state.security_analyzer.analyze_pending_actions(
+                    action_events
+                )
+            )
+        else:
+            risks = (risk.SecurityRisk.UNKNOWN for _ in action_events)
+
+        return any(self._state.confirmation_policy.should_confirm(r) for r in risks)
+
+    def extract_security_risk(
+        self,
+        arguments: dict,
+        tool_name: str,
+        read_only_tool: bool,
+    ) -> risk.SecurityRisk:
+        requires_sr = isinstance(self._state.security_analyzer, LLMSecurityAnalyzer)
+        raw = arguments.pop("security_risk", None)
+
+        # Default risk value for action event
+        # Tool is marked as read-only so security risk can be ignored
+        if read_only_tool:
+            return risk.SecurityRisk.UNKNOWN
+
+        # Raises exception if failed to pass risk field when expected
+        # Exception will be sent back to agent as error event
+        # Strong models like GPT-5 can correct itself by retrying
+        if requires_sr and raw is None:
+            raise ValueError(
+                f"Failed to provide security_risk field in tool '{tool_name}'"
+            )
+
+        # When using weaker models without security analyzer
+        # safely ignore missing security risk fields
+        if not requires_sr and raw is None:
+            return risk.SecurityRisk.UNKNOWN
+
+        # Raises exception if invalid risk enum passed by LLM
+        security_risk = risk.SecurityRisk(raw)
+        return security_risk