feat(tools): support shell execution in run_code (#709)

Co-authored-by: yuanfeng.dev <yuanfeng.dev@bytedance.com>

Author: dyike

tests/tools/builtin_tools/test_agentkit.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import importlib.util
+import json
 import os
 import sys
 import types
@@ -160,5 +161,47 @@ def test_resolve_raises_when_all_tool_ids_missing(self):
             self.agentkit_module.resolve_agentkit_tool_id("AGENTKIT_TOOL_ID_SCRIPT")
 
 
+class TestInvokeAgentkitExecBash(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.agentkit_module = _load_agentkit_module()
+
+    def test_builds_exec_bash_invoke_tool_request(self):
+        with patch.object(
+            self.agentkit_module,
+            "ve_request",
+            return_value={"Result": {"Result": "shell output"}},
+        ) as ve_request:
+            result = self.agentkit_module.invoke_agentkit_exec_bash(
+                tool_id="shell-tool",
+                tool_user_session_id="kk",
+                command="echo hello",
+                exec_dir="/tmp",
+                env={"DEMO_ENV": "from-invoke-tool"},
+                timeout=30,
+                hard_timeout=60,
+                max_output_length=30000,
+                ttl=1800,
+            )
+
+        self.assertEqual(result, {"Result": {"Result": "shell output"}})
+        request_body = ve_request.call_args.kwargs["request_body"]
+        self.assertEqual(request_body["ToolId"], "shell-tool")
+        self.assertEqual(request_body["OperationType"], "ExecBash")
+        self.assertEqual(request_body["UserSessionId"], "kk")
+        self.assertEqual(request_body["Ttl"], 1800)
+        self.assertEqual(
+            json.loads(request_body["OperationPayload"]),
+            {
+                "command": "echo hello",
+                "exec_dir": "/tmp",
+                "env": {"DEMO_ENV": "from-invoke-tool"},
+                "timeout": 30,
+                "hard_timeout": 60,
+                "max_output_length": 30000,
+            },
+        )
+
+
 if __name__ == "__main__":
     unittest.main()

veadk/tools/builtin_tools/_agentkit.py

@@ -153,3 +153,56 @@ def invoke_agentkit_run_code(
         header=header,
         scheme=scheme,
     )
+
+
+def invoke_agentkit_exec_bash(
+    *,
+    tool_id: str,
+    tool_user_session_id: str,
+    command: str,
+    exec_dir: Optional[str] = None,
+    env: Optional[dict[str, str]] = None,
+    timeout: int = 30,
+    hard_timeout: Optional[int] = None,
+    max_output_length: Optional[int] = None,
+    tool_state: Optional[dict[str, Any]] = None,
+    ttl: Optional[int] = None,
+) -> dict[str, Any]:
+    """Invoke AgentKit's Bash execution operation through InvokeTool."""
+    service, region, host, scheme = get_agentkit_endpoint_config()
+    ak, sk, header = get_agentkit_credentials(tool_state)
+
+    operation_payload: dict[str, Any] = {
+        "command": command,
+        "timeout": timeout,
+    }
+    if exec_dir is not None:
+        operation_payload["exec_dir"] = exec_dir
+    if env is not None:
+        operation_payload["env"] = env
+    if hard_timeout is not None:
+        operation_payload["hard_timeout"] = hard_timeout
+    if max_output_length is not None:
+        operation_payload["max_output_length"] = max_output_length
+
+    request_body: dict[str, Any] = {
+        "ToolId": tool_id,
+        "OperationType": "ExecBash",
+        "UserSessionId": tool_user_session_id,
+        "OperationPayload": json.dumps(operation_payload),
+    }
+    if ttl is not None:
+        request_body["Ttl"] = ttl
+
+    return ve_request(
+        request_body=request_body,
+        action="InvokeTool",
+        ak=ak,
+        sk=sk,
+        service=service,
+        version="2025-10-30",
+        region=region,
+        host=host,
+        header=header,
+        scheme=scheme,
+    )

veadk/tools/builtin_tools/run_code.py

@@ -18,6 +18,7 @@
 
 from veadk.tools.builtin_tools._agentkit import (
     get_agentkit_endpoint_config,
+    invoke_agentkit_exec_bash,
     invoke_agentkit_run_code,
     resolve_agentkit_tool_id,
 )
@@ -27,15 +28,26 @@
 
 
 def run_code(
-    code: str, language: str, tool_context: ToolContext, timeout: int = 30
+    code: str,
+    language: str,
+    tool_context: ToolContext,
+    timeout: int = 30,
+    exec_dir: str = "/tmp",
+    env: dict[str, str] | None = None,
+    hard_timeout: int = 300,
+    max_output_length: int = 30000,
 ) -> str:
     """Run code in a code sandbox and return the output.
     For C++ code, don't execute it directly, compile and execute via Python; write sources and object files to /tmp.
 
     Args:
         code (str): The code to run.
-        language (str): The programming language of the code. Language must be one of the supported languages: python3.
+        language (str): The execution language. Use ``python3`` for code or ``bash`` for shell scripts.
         timeout (int, optional): The timeout in seconds for the code execution. Defaults to 30.
+        exec_dir (str, optional): Working directory for Bash execution. Defaults to ``/tmp``.
+        env (dict[str, str], optional): Environment variables for Bash execution.
+        hard_timeout (int, optional): Hard timeout for Bash execution. Defaults to 300 seconds.
+        max_output_length (int, optional): Maximum Bash output length. Defaults to 30000.
 
     Returns:
         str: The output of the code execution.
@@ -55,19 +67,35 @@ def run_code(
         f"Running code in language: {language}, session_id={session_id}, code={code}, tool_id={tool_id}, host={host}, service={service}, region={region}, timeout={timeout}"
     )
 
-    res = invoke_agentkit_run_code(
-        tool_id=tool_id,
-        tool_user_session_id=tool_user_session_id,
-        code=code,
-        timeout=timeout,
-        kernel_name=language,
-        tool_state=tool_context.state if tool_context else None,
-        ttl=int(os.getenv("AGENTKIT_TOOL_TTL", "1800")),
-    )
+    tool_state = tool_context.state if tool_context else None
+    ttl = int(os.getenv("AGENTKIT_TOOL_TTL", "1800"))
+    if language.lower() in {"bash", "shell"}:
+        res = invoke_agentkit_exec_bash(
+            tool_id=tool_id,
+            tool_user_session_id=tool_user_session_id,
+            command=code,
+            exec_dir=exec_dir,
+            env=env,
+            timeout=timeout,
+            hard_timeout=hard_timeout,
+            max_output_length=max_output_length,
+            tool_state=tool_state,
+            ttl=ttl,
+        )
+    else:
+        res = invoke_agentkit_run_code(
+            tool_id=tool_id,
+            tool_user_session_id=tool_user_session_id,
+            code=code,
+            timeout=timeout,
+            kernel_name=language,
+            tool_state=tool_state,
+            ttl=ttl,
+        )
     logger.debug(f"Invoke run code response: {res}")
 
     try:
         return res["Result"]["Result"]
-    except KeyError as e:
+    except (KeyError, TypeError) as e:
         logger.error(f"Error occurred while running code: {e}, response is {res}")
         return res