@@ -3,11 +3,13 @@
 
 from beeai_framework.adapters.acp_zed.serve.server import ACPZedServer, ACPZedServerConfig
 from beeai_framework.adapters.acp_zed.tools.read_file import ACPZedReadFileTool
+from beeai_framework.adapters.acp_zed.tools.terminal import ACPZedTerminalTool
 from beeai_framework.adapters.acp_zed.tools.write_file import ACPZedWriteFileTool
 
 __all__ = [
     "ACPZedReadFileTool",
     "ACPZedServer",
     "ACPZedServerConfig",
+    "ACPZedTerminalTool",
     "ACPZedWriteFileTool",
 ]
@@ -73,6 +73,10 @@ def can_write(self) -> bool:
         fs = getattr(self._capabilities, "fs", None)
         return bool(getattr(fs, "write_text_file", False))
 
+    @property
+    def can_terminal(self) -> bool:
+        return bool(getattr(self._capabilities, "terminal", False))
+
     def _active_session_id(self) -> str:
         sid = _active_session.get()
         if sid is None:
@@ -93,6 +97,39 @@ async def read_text_file(self, path: str, *, line: int | None = None, limit: int
     async def write_text_file(self, path: str, content: str) -> None:
         await self._require_conn().write_text_file(content=content, path=path, session_id=self._active_session_id())
 
+    async def create_terminal(
+        self,
+        command: str,
+        *,
+        args: list[str] | None = None,
+        cwd: str | None = None,
+        env: list[Any] | None = None,
+        output_byte_limit: int | None = None,
+    ) -> str:
+        response = await self._require_conn().create_terminal(
+            command=command,
+            session_id=self._active_session_id(),
+            args=args,
+            cwd=cwd,
+            env=env,
+            output_byte_limit=output_byte_limit,
+        )
+        return response.terminal_id
+
+    async def terminal_output(self, terminal_id: str) -> Any:
+        return await self._require_conn().terminal_output(session_id=self._active_session_id(), terminal_id=terminal_id)
+
+    async def wait_for_terminal_exit(self, terminal_id: str) -> Any:
+        return await self._require_conn().wait_for_terminal_exit(
+            session_id=self._active_session_id(), terminal_id=terminal_id
+        )
+
+    async def release_terminal(self, terminal_id: str) -> None:
+        await self._require_conn().release_terminal(session_id=self._active_session_id(), terminal_id=terminal_id)
+
+    async def kill_terminal(self, terminal_id: str) -> None:
+        await self._require_conn().kill_terminal(session_id=self._active_session_id(), terminal_id=terminal_id)
+
 
 class ACPZedServerAgent(ACPBaseAgent):
     """Wrapper that exposes a BeeAI agent over the Zed Agent Client Protocol (stdio).
 
@@ -2,6 +2,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 from beeai_framework.adapters.acp_zed.tools.read_file import ACPZedReadFileTool
+from beeai_framework.adapters.acp_zed.tools.terminal import ACPZedTerminalTool
 from beeai_framework.adapters.acp_zed.tools.write_file import ACPZedWriteFileTool
 
-__all__ = ["ACPZedReadFileTool", "ACPZedWriteFileTool"]
+__all__ = ["ACPZedReadFileTool", "ACPZedTerminalTool", "ACPZedWriteFileTool"]
@@ -0,0 +1,101 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Self
+
+from pydantic import BaseModel, Field
+
+from beeai_framework.context import RunContext
+from beeai_framework.emitter.emitter import Emitter
+from beeai_framework.tools import JSONToolOutput
+from beeai_framework.tools.errors import ToolError
+from beeai_framework.tools.tool import Tool
+from beeai_framework.tools.types import ToolRunOptions
+
+try:
+    from acp.schema import EnvVariable
+except ModuleNotFoundError as e:
+    raise ModuleNotFoundError(
+        "Optional module [acp_zed] not found.\nRun 'pip install \"beeai-framework[acp-zed]\"' to install."
+    ) from e
+
+if TYPE_CHECKING:
+    from beeai_framework.adapters.acp_zed.serve.server import ACPZedServer
+
+
+class ACPZedTerminalToolInput(BaseModel):
+    command: str = Field(description="Program to run (e.g. 'pytest', 'npm').")
+    args: list[str] = Field(default_factory=list, description="Arguments passed to the program.")
+    cwd: str | None = Field(default=None, description="Working directory (absolute path).")
+    env: dict[str, str] | None = Field(default=None, description="Extra environment variables.")
+    output_byte_limit: int | None = Field(default=None, description="Cap captured output in bytes. Editor-enforced.")
+    wait: bool = Field(
+        default=True,
+        description=(
+            "Wait for the process to exit and return its output. If false, returns the terminal id immediately."
+        ),
+    )
+
+
+class ACPZedTerminalTool(Tool[ACPZedTerminalToolInput, ToolRunOptions, JSONToolOutput[dict[str, Any]]]):
+    """Run a command in the editor's terminal via ACP `terminal/*`.
+
+    Live output shows in the Zed terminal widget while the process is running; the
+    agent blocks on exit (by default) and gets back `{exit_code, stdout, stderr,
+    terminal_id}`. Requires the client to advertise `terminal` capability.
+    """
+
+    name = "acp_zed_terminal"
+    description = (
+        "Run a command in the user's editor terminal. Returns exit code, stdout, and stderr after the "
+        "process exits. Prefer this over generic shell when running under Zed."
+    )
+    input_schema = ACPZedTerminalToolInput
+
+    def __init__(self, server: ACPZedServer[Any], options: dict[str, Any] | None = None) -> None:
+        super().__init__(options=options)
+        self._server = server
+
+    async def clone(self) -> Self:
+        return type(self)(server=self._server, options=self.options)
+
+    def _create_emitter(self) -> Emitter:
+        return Emitter.root().child(namespace=["tool", "acp_zed", "terminal"], creator=self)
+
+    async def _run(
+        self, input: ACPZedTerminalToolInput, options: ToolRunOptions | None, context: RunContext
+    ) -> JSONToolOutput[dict[str, Any]]:
+        bridge = self._server.bridge
+        if not bridge.can_terminal:
+            raise ToolError("ACP client does not advertise terminal capability")
+
+        env_list = [EnvVariable(name=k, value=v) for k, v in (input.env or {}).items()] or None
+        terminal_id = await bridge.create_terminal(
+            input.command,
+            args=input.args,
+            cwd=input.cwd,
+            env=env_list,
+            output_byte_limit=input.output_byte_limit,
+        )
+
+        if not input.wait:
+            return JSONToolOutput({"terminal_id": terminal_id, "waited": False})
+
+        try:
+            exit_info = await bridge.wait_for_terminal_exit(terminal_id)
+            output = await bridge.terminal_output(terminal_id)
+        finally:
+            await bridge.release_terminal(terminal_id)
+
+        return JSONToolOutput(
+            {
+                "terminal_id": terminal_id,
+                "exit_code": getattr(exit_info, "exit_code", None),
+                "signal": getattr(exit_info, "signal", None),
+                "stdout": getattr(output, "output", "") or "",
+                "truncated": getattr(output, "truncated", False),
+                "waited": True,
+            }
+        )
@@ -4,6 +4,7 @@
 from beeai_framework.tools.code.output import PythonToolOutput
 from beeai_framework.tools.code.python import PythonTool
 from beeai_framework.tools.code.sandbox import SandboxTool
+from beeai_framework.tools.code.shell import ShellTool, ShellToolInput
 from beeai_framework.tools.code.storage import LocalPythonStorage, PythonStorage
 
 __all__ = [
@@ -12,4 +13,6 @@
     "PythonTool",
     "PythonToolOutput",
     "SandboxTool",
+    "ShellTool",
+    "ShellToolInput",
 ]
@@ -0,0 +1,97 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+import asyncio
+import os
+from typing import Any, Self
+
+from pydantic import BaseModel, Field
+
+from beeai_framework.context import RunContext
+from beeai_framework.emitter.emitter import Emitter
+from beeai_framework.tools import JSONToolOutput
+from beeai_framework.tools.errors import ToolError, ToolInputValidationError
+from beeai_framework.tools.tool import Tool
+from beeai_framework.tools.types import ToolRunOptions
+
+
+class ShellToolInput(BaseModel):
+    command: list[str] = Field(
+        description=(
+            "Command and arguments as a list (no shell interpretation). "
+            "Example: ['git', 'status', '--porcelain']. Use shlex.split on a user string if needed."
+        )
+    )
+    cwd: str | None = Field(default=None, description="Working directory. Defaults to the current process cwd.")
+    env: dict[str, str] | None = Field(
+        default=None, description="Additional environment variables merged over the current environment."
+    )
+    timeout_seconds: float | None = Field(
+        default=60.0, description="Kill the process if it runs longer. Set to None to disable."
+    )
+    input_text: str | None = Field(default=None, description="Optional text piped to the process on stdin.")
+
+
+class ShellTool(Tool[ShellToolInput, ToolRunOptions, JSONToolOutput[dict[str, Any]]]):
+    """Run a local subprocess and collect its output.
+
+    Takes the command as a list of arguments (no shell interpretation), so callers
+    don't have to worry about quoting or injection. Captures stdout, stderr, exit
+    code, and a `timed_out` flag. Under Zed's ACP adapter, prefer
+    `ACPZedTerminalTool` — it routes through the editor's terminal widget and shows
+    live output; this tool is for harnesses without a client-side terminal.
+    """
+
+    name = "shell"
+    description = (
+        "Run a local command as a subprocess and return its exit code, stdout, and stderr. "
+        "The command must be a list of arguments; it is NOT run through a shell."
+    )
+    input_schema = ShellToolInput
+
+    async def clone(self) -> Self:
+        return type(self)(options=self.options)
+
+    def _create_emitter(self) -> Emitter:
+        return Emitter.root().child(namespace=["tool", "code", "shell"], creator=self)
+
+    async def _run(
+        self, input: ShellToolInput, options: ToolRunOptions | None, context: RunContext
+    ) -> JSONToolOutput[dict[str, Any]]:
+        if not input.command:
+            raise ToolInputValidationError("`command` must be a non-empty list")
+
+        merged_env = {**os.environ, **(input.env or {})} if input.env else None
+        try:
+            process = await asyncio.create_subprocess_exec(
+                *input.command,
+                cwd=input.cwd,
+                env=merged_env,
+                stdin=asyncio.subprocess.PIPE if input.input_text is not None else None,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+        except FileNotFoundError as e:
+            raise ToolError(f"Command not found: {input.command[0]!r}") from e
+
+        stdin_bytes = input.input_text.encode() if input.input_text is not None else None
+        timed_out = False
+        try:
+            stdout_bytes, stderr_bytes = await asyncio.wait_for(
+                process.communicate(input=stdin_bytes), timeout=input.timeout_seconds
+            )
+        except TimeoutError:
+            process.kill()
+            stdout_bytes, stderr_bytes = await process.communicate()
+            timed_out = True
+
+        return JSONToolOutput(
+            {
+                "exit_code": process.returncode if process.returncode is not None else -1,
+                "stdout": stdout_bytes.decode(errors="replace"),
+                "stderr": stderr_bytes.decode(errors="replace"),
+                "timed_out": timed_out,
+            }
+        )
@@ -0,0 +1,16 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from beeai_framework.tools.filesystem.glob_tool import GlobTool, GlobToolInput
+from beeai_framework.tools.filesystem.grep_tool import GrepTool, GrepToolInput
+
+# `FilePatchTool` lives in `.patch` and depends on the optional `patch-ng` package
+# (install via `pip install "beeai-framework[filesystem]"`). Import it directly
+# from `beeai_framework.tools.filesystem.patch` so package import doesn't fail.
+
+__all__ = [
+    "GlobTool",
+    "GlobToolInput",
+    "GrepTool",
+    "GrepToolInput",
+]
@@ -0,0 +1,55 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Self
+
+from pydantic import BaseModel, Field
+
+from beeai_framework.context import RunContext
+from beeai_framework.emitter.emitter import Emitter
+from beeai_framework.tools import JSONToolOutput
+from beeai_framework.tools.tool import Tool
+from beeai_framework.tools.types import ToolRunOptions
+
+
+class GlobToolInput(BaseModel):
+    pattern: str = Field(description="Glob pattern, e.g. '**/*.py' or 'src/**/*.ts'.")
+    root: str = Field(default=".", description="Directory to search from.")
+    include_hidden: bool = Field(default=False, description="Include entries whose name starts with a dot.")
+    limit: int = Field(default=1000, description="Maximum number of paths to return.")
+
+
+class GlobTool(Tool[GlobToolInput, ToolRunOptions, JSONToolOutput[dict[str, Any]]]):
+    """Find paths matching a glob pattern under a root directory.
+
+    Uses `pathlib.Path.glob` — patterns like `**/*.py` work out of the box. Hidden
+    files/directories (names starting with `.`) are excluded by default.
+    """
+
+    name = "glob"
+    description = "List files and directories matching a glob pattern (e.g. '**/*.py')."
+    input_schema = GlobToolInput
+
+    async def clone(self) -> Self:
+        return type(self)(options=self.options)
+
+    def _create_emitter(self) -> Emitter:
+        return Emitter.root().child(namespace=["tool", "filesystem", "glob"], creator=self)
+
+    async def _run(
+        self, input: GlobToolInput, options: ToolRunOptions | None, context: RunContext
+    ) -> JSONToolOutput[dict[str, Any]]:
+        root = Path(input.root).expanduser().resolve()
+        matches: list[str] = []
+        truncated = False
+        for path in root.glob(input.pattern):
+            if not input.include_hidden and any(part.startswith(".") for part in path.relative_to(root).parts):
+                continue
+            matches.append(str(path))
+            if len(matches) >= input.limit:
+                truncated = True
+                break
+        return JSONToolOutput({"matches": matches, "truncated": truncated})