feat: add python tool timeout param

Author: Fronut

astrbot/core/tools/computer_tools/python.py

@@ -33,6 +33,11 @@
             "description": "Whether to suppress the output of the code execution.",
             "default": False,
         },
+        "timeout": {
+            "type": "integer",
+            "description": "Optional timeout in seconds for code execution. Omit or set to 0 to use tool_call_timeout.",
+            "default": 0,
+        },
     },
     "required": ["code"],
 }
@@ -77,16 +82,32 @@ class PythonTool(FunctionTool):
     parameters: dict = field(default_factory=lambda: param_schema)
 
     async def call(
-        self, context: ContextWrapper[AstrAgentContext], code: str, silent: bool = False
+        self,
+        context: ContextWrapper[AstrAgentContext],
+        code: str,
+        silent: bool = False,
+        timeout: int | None = None,
     ) -> ToolExecResult:
         if permission_error := check_admin_permission(context, "Python execution"):
             return permission_error
         sb = await get_booter(
             context.context.context,
             context.context.event.unified_msg_origin,
         )
+        effective_timeout = context.tool_call_timeout
+        if timeout is not None:
+            try:
+                timeout_value = int(timeout)
+            except (TypeError, ValueError):
+                timeout_value = None
+            if timeout_value and timeout_value > 0:
+                effective_timeout = min(timeout_value, context.tool_call_timeout)
         try:
-            result = await sb.python.exec(code, silent=silent)
+            result = await sb.python.exec(
+                code,
+                timeout=effective_timeout,
+                silent=silent,
+            )
             return await handle_result(result, context.context.event)
         except Exception as e:
             return f"Error executing code: {str(e)}"
@@ -104,13 +125,29 @@ class LocalPythonTool(FunctionTool):
     parameters: dict = field(default_factory=lambda: param_schema)
 
     async def call(
-        self, context: ContextWrapper[AstrAgentContext], code: str, silent: bool = False
+        self,
+        context: ContextWrapper[AstrAgentContext],
+        code: str,
+        silent: bool = False,
+        timeout: int | None = None,
     ) -> ToolExecResult:
         if permission_error := check_admin_permission(context, "Python execution"):
             return permission_error
         sb = get_local_booter()
+        effective_timeout = context.tool_call_timeout
+        if timeout is not None:
+            try:
+                timeout_value = int(timeout)
+            except (TypeError, ValueError):
+                timeout_value = None
+            if timeout_value and timeout_value > 0:
+                effective_timeout = min(timeout_value, context.tool_call_timeout)
         try:
-            result = await sb.python.exec(code, silent=silent)
+            result = await sb.python.exec(
+                code,
+                timeout=effective_timeout,
+                silent=silent,
+            )
             return await handle_result(result, context.context.event)
         except Exception as e:
             return f"Error executing code: {str(e)}"