From 9e09299dcb67bdb56d4db770f6b01eb7b7dcebbc Mon Sep 17 00:00:00 2001
From: Haoran Xu <3230105281@zju.edu.cn>
Date: Sun, 3 May 2026 15:08:10 +0800
Subject: [PATCH] feat: add python tool timeout param (#7953)

* feat: add python tool timeout param

* Update python.py

---------

Co-authored-by: Weilong Liao <37870767+Soulter@users.noreply.github.com>
---
 astrbot/core/tools/computer_tools/python.py | 39 ++++++++++++++++++---
 1 file changed, 35 insertions(+), 4 deletions(-)

diff --git a/astrbot/core/tools/computer_tools/python.py b/astrbot/core/tools/computer_tools/python.py
index e0bb6c9de..be909f6d2 100644
--- a/astrbot/core/tools/computer_tools/python.py
+++ b/astrbot/core/tools/computer_tools/python.py
@@ -33,6 +33,11 @@ param_schema = {
             "description": "Whether to suppress the output of the code execution.",
             "default": False,
         },
+        "timeout": {
+            "type": "integer",
+            "description": "Optional timeout in seconds for code execution.",
+            "default": 30,
+        },
     },
     "required": ["code"],
 }
@@ -77,7 +82,11 @@ 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 = 30,
     ) -> ToolExecResult:
         if permission_error := check_admin_permission(context, "Python execution"):
             return permission_error
@@ -85,8 +94,17 @@ class PythonTool(FunctionTool):
             context.context.context,
             context.context.event.unified_msg_origin,
         )
+        effective_timeout = (
+            min(timeout, context.tool_call_timeout)
+            if timeout > 0
+            else 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 +122,26 @@ 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 = 30,
     ) -> ToolExecResult:
         if permission_error := check_admin_permission(context, "Python execution"):
             return permission_error
         sb = get_local_booter()
+        effective_timeout = (
+            min(timeout, context.tool_call_timeout)
+            if timeout > 0
+            else 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)}"