feat(beta): add LocalShellTool and LocalShellEnvironment for local sh…

autogen/beta/__init__.py

@@ -8,14 +8,16 @@
 from .annotations import Context, Inject, Variable
 from .response import PromptedSchema, ResponseSchema, response_schema
 from .stream import MemoryStream
-from .tools import ToolResult, tool
+from .tools import LocalShellEnvironment, LocalShellTool, ToolResult, tool
 
 __all__ = (
     "Agent",
     "AgentReply",
     "Context",
     "Depends",
     "Inject",
+    "LocalShellEnvironment",
+    "LocalShellTool",
     "MemoryStream",
     "PromptedSchema",
     "ResponseSchema",

autogen/beta/tools/__init__.py

@@ -18,14 +18,16 @@
     WebFetchTool,
     WebSearchTool,
 )
-from .final import Toolkit, tool
+from .final import LocalShellEnvironment, LocalShellTool, Toolkit, tool
 
 __all__ = (
     "CodeExecutionTool",
     "ContainerAutoEnvironment",
     "ContainerReferenceEnvironment",
     "ImageGenerationTool",
     "LocalEnvironment",
+    "LocalShellEnvironment",
+    "LocalShellTool",
     "MemoryTool",
     "NetworkPolicy",
     "ShellTool",

autogen/beta/tools/final/__init__.py

@@ -4,6 +4,7 @@
 
 from .client_tool import ClientTool
 from .function_tool import FunctionDefinition, FunctionParameters, FunctionTool, FunctionToolSchema, tool
+from .local_shell import LocalShellEnvironment, LocalShellTool
 from .toolkit import Toolkit
 
 __all__ = (
@@ -12,6 +13,8 @@
     "FunctionParameters",
     "FunctionTool",
     "FunctionToolSchema",
+    "LocalShellEnvironment",
+    "LocalShellTool",
     "Toolkit",
     "tool",
 )

autogen/beta/tools/final/local_shell.py

@@ -0,0 +1,105 @@
+# Copyright (c) 2026, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+import atexit
+import shutil
+import subprocess
+import tempfile
+from collections.abc import Iterable
+from contextlib import ExitStack
+from dataclasses import dataclass
+from pathlib import Path
+
+from autogen.beta.annotations import Context
+from autogen.beta.middleware import BaseMiddleware
+from autogen.beta.tools.tool import Tool
+
+from .function_tool import FunctionTool, tool
+
+
+@dataclass(slots=True)
+class LocalShellEnvironment:
+    """Working directory configuration for :class:`LocalShellTool`.
+
+    Args:
+        path: Directory to use as the shell working directory. If ``None``,
+              a temporary directory is created automatically.
+        cleanup: Whether to delete the directory when the tool is garbage-
+                 collected. Defaults to ``True`` when ``path`` is ``None``
+                 (auto temp dir) and ``False`` when ``path`` is provided.
+    """
+
+    path: str | Path | None = None
+    cleanup: bool | None = None
+
+
+class LocalShellTool(Tool):
+    """Client-side shell execution tool. Works with any LLM provider.
+
+    Unlike :class:`~autogen.beta.tools.ShellTool` (provider schema), this tool
+    executes shell commands locally using ``subprocess``. The working directory
+    lifecycle is controlled by the ``environment`` parameter.
+
+    Args:
+        environment: Directory configuration. If ``None``, defaults to
+                     :class:`LocalShellEnvironment` with a temporary directory
+                     that is cleaned up on exit.
+
+    """
+
+    __slots__ = ("_environment", "_workdir", "_tempdir", "_tool")
+
+    def __init__(self, *, environment: LocalShellEnvironment | None = None) -> None:
+        if environment is None:
+            environment = LocalShellEnvironment()
+
+        self._environment = environment
+        self._tempdir: tempfile.TemporaryDirectory[str] | None = None
+
+        # cleanup default: True for auto-tempdir, False for user-specified path
+        cleanup = environment.cleanup if environment.cleanup is not None else (environment.path is None)
+
+        if environment.path is None:
+            self._tempdir = tempfile.TemporaryDirectory(prefix="ag2_shell_", delete=cleanup)
+            self._workdir = Path(self._tempdir.name)
+        else:
+            self._workdir = Path(environment.path).resolve()
+            self._workdir.mkdir(parents=True, exist_ok=True)
+            if cleanup:
+                workdir = self._workdir
+                atexit.register(lambda: shutil.rmtree(workdir, ignore_errors=True))
+
+        workdir = self._workdir
+
+        @tool
+        def shell(command: str) -> str:
+            """Execute a shell command in the working directory."""
+            result = subprocess.run(
+                command,
+                shell=True,
+                cwd=workdir,
+                capture_output=True,
+                text=True,
+                timeout=60,
+            )
+            return (result.stdout + result.stderr).strip()
+
+        self._tool: FunctionTool = shell
+
+    @property
+    def workdir(self) -> Path:
+        """The working directory used for command execution."""
+        return self._workdir
+
+    async def schemas(self, context: "Context") -> list:
+        return await self._tool.schemas(context)
+
+    def register(
+        self,
+        stack: "ExitStack",
+        context: "Context",
+        *,
+        middleware: Iterable["BaseMiddleware"] = (),
+    ) -> None:
+        self._tool.register(stack, context, middleware=middleware)