fz-mirrors/AstrBot
1
0
Fork 0
mirror of https://github.com/AstrBotDevs/AstrBot synced 2026-07-16 01:40:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: add python tool timeout param (#7953)

Browse Source 
* feat: add python tool timeout param

* Update python.py

---------

Co-authored-by: Weilong Liao <37870767+Soulter@users.noreply.github.com>
This commit is contained in:
Haoran Xu
2026-05-03 15:08:10 +08:00
committed by GitHub
parent 77fe2de2c1
commit 9e09299dcb
1 changed files with 35 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
astrbot/core/tools/computer_tools/python.py
View File

@@ -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)}"