Tool framework extension

ddev/src/ddev/ai/tools/core/base.py

@@ -101,7 +101,8 @@ async def run(self, raw: dict[str, object]) -> ToolResult:
         try:
             return await self(validated)
         except Exception as e:
-            return ToolResult(success=False, error=f"{type(e).__name__}: {str(e)}")
+            msg = str(e) or repr(e)
+            return ToolResult(success=False, error=f"{type(e).__name__}: {msg}")
 
     @abstractmethod
     async def __call__(self, tool_input: TInput) -> ToolResult:

ddev/src/ddev/ai/tools/fs/__init__.py

@@ -0,0 +1,3 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)

ddev/src/ddev/ai/tools/fs/append_file.py

@@ -0,0 +1,43 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from pathlib import Path
+from typing import Annotated
+
+from pydantic import Field
+
+from ddev.ai.tools.core.base import BaseToolInput
+from ddev.ai.tools.core.types import ToolResult
+
+from .base import FileRegistryTool
+
+
+class AppendFileInput(BaseToolInput):
+    path: Annotated[str, Field(description="Path of the file to append to")]
+    content: Annotated[str, Field(description="Content to append to the file")]
+
+
+class AppendFileTool(FileRegistryTool[AppendFileInput]):
+    """Appends content to the end of an existing file.
+    Can only append to files registered in the file registry.
+    Fails if the file was modified since the last read."""
+
+    @property
+    def name(self) -> str:
+        return "append_file"
+
+    async def __call__(self, tool_input: AppendFileInput) -> ToolResult:
+        path = Path(tool_input.path)
+
+        async with self._registry.lock_for(str(path)):
+            current_content, fail = self._read_verified(str(path))
+            if fail:
+                return fail
+
+            content_to_append = tool_input.content.replace("\r\n", "\n")
+            separator = "" if not current_content or current_content.endswith("\n") else "\n"
+            new_content = current_content + separator + content_to_append
+
+            path.write_text(new_content, encoding="utf-8")
+            self._record(str(path), new_content)
+        return ToolResult(success=True, data=f"Content appended to: {path}")

ddev/src/ddev/ai/tools/fs/base.py

@@ -0,0 +1,35 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from pathlib import Path
+
+from ddev.ai.tools.core.base import BaseTool, BaseToolInput
+from ddev.ai.tools.core.types import ToolResult
+
+from .file_registry import FileRegistry
+
+
+class FileRegistryTool[TInput: BaseToolInput](BaseTool[TInput]):
+    """Abstract base for file system tools with hash-based consistency checks."""
+
+    def __init__(self, file_registry: FileRegistry) -> None:
+        self._registry = file_registry
+
+    def _refresh_if_known(self, path: str, content: str) -> None:
+        if self._registry.is_known(path):
+            self._registry.record(path, content)
+
+    def _record(self, path: str, content: str) -> None:
+        self._registry.record(path, content)
+
+    def _read_verified(self, path: str) -> tuple[str, ToolResult | None]:
+        """Read file content and verify it matches the last recorded hash."""
+        if not self._registry.is_known(path):
+            return "", ToolResult(success=False, error=f"Not authorized to modify '{path}'.")
+        try:
+            content = Path(path).read_text(encoding="utf-8")
+        except OSError as e:
+            return "", ToolResult(success=False, error=str(e))
+        if not self._registry.verify(path, content):
+            return "", ToolResult(success=False, error=f"File '{path}' has changed since last read. Re-read and retry.")
+        return content, None

ddev/src/ddev/ai/tools/fs/create_file.py

@@ -0,0 +1,41 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from pathlib import Path
+from typing import Annotated
+
+from pydantic import Field
+
+from ddev.ai.tools.core.base import BaseToolInput
+from ddev.ai.tools.core.types import ToolResult
+
+from .base import FileRegistryTool
+
+
+class CreateFileInput(BaseToolInput):
+    path: Annotated[str, Field(description="Path of the file to create")]
+    content: Annotated[str, Field(description="Content of the file to create")] = ""
+
+
+class CreateFileTool(FileRegistryTool[CreateFileInput]):
+    """Creates a new file and writes content into it (default: empty content).
+    Parent directories are created automatically if they do not exist (no need to call mkdir first).
+    Registers the file in the file registry.
+    Fails if the file already exists.
+    Use edit_file to modify existing files."""
+
+    @property
+    def name(self) -> str:
+        return "create_file"
+
+    async def __call__(self, tool_input: CreateFileInput) -> ToolResult:
+        path = Path(tool_input.path)
+
+        async with self._registry.lock_for(str(path)):
+            if path.exists():
+                return ToolResult(success=False, error=f"File already exists: {path}")
+
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(tool_input.content, encoding="utf-8")
+            self._record(str(path), tool_input.content)
+        return ToolResult(success=True, data=f"File created: {path}")

ddev/src/ddev/ai/tools/fs/edit_file.py

@@ -0,0 +1,65 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from pathlib import Path
+from typing import Annotated
+
+from pydantic import Field
+
+from ddev.ai.tools.core.base import BaseToolInput
+from ddev.ai.tools.core.types import ToolResult
+
+from .base import FileRegistryTool
+
+
+class EditFileInput(BaseToolInput):
+    path: Annotated[str, Field(description="Path of the file to edit")]
+    old_string: Annotated[
+        str,
+        Field(
+            description="Exact non-empty text to replace. Must appear exactly once in the file \
+        (hint: include surrounding context if needed)."
+        ),
+    ]
+    new_string: Annotated[str, Field(description="Text to replace old_string with")]
+
+
+class EditFileTool(FileRegistryTool[EditFileInput]):
+    """Edits a file by replacing an exact string with a new one.
+    Can only edit files registered in the file registry.
+    Fails if the file was modified since the last read.
+    old_string must appear exactly once in the file — if it appears multiple times, the call fails."""
+
+    @property
+    def name(self) -> str:
+        return "edit_file"
+
+    async def __call__(self, tool_input: EditFileInput) -> ToolResult:
+        path = Path(tool_input.path)
+
+        async with self._registry.lock_for(str(path)):
+            content, fail = self._read_verified(str(path))
+            if fail:
+                return fail
+
+            # Normalize line endings to avoid issues with different OSs
+            old_string = tool_input.old_string.replace("\r\n", "\n")
+            new_string = tool_input.new_string.replace("\r\n", "\n")
+
+            if not old_string:
+                return ToolResult(success=False, error="old_string must not be empty")
+
+            count = content.count(old_string)
+            if count == 0:
+                return ToolResult(success=False, error="old_string not found in file")
+            if count > 1:
+                return ToolResult(
+                    success=False,
+                    error=f"old_string appears {count} times in the file",
+                    hint="Include more surrounding context to make it unique",
+                )
+
+            new_content = content.replace(old_string, new_string, 1)
+            path.write_text(new_content, encoding="utf-8")
+            self._record(str(path), new_content)
+        return ToolResult(success=True, data=f"File edited: {path}")

ddev/src/ddev/ai/tools/fs/file_registry.py

@@ -0,0 +1,35 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+import asyncio
+import hashlib
+from pathlib import Path
+
+
+class FileRegistry:
+    """Tracks files created by the agent and their last-seen content hash."""
+
+    def __init__(self) -> None:
+        self._hashes: dict[str, str] = {}
+        self._locks: dict[str, asyncio.Lock] = {}
+
+    def _normalize(self, path: str) -> str:
+        return Path(path).resolve().as_posix()
+
+    def _hash(self, content: str) -> str:
+        return hashlib.sha256(content.encode()).hexdigest()
+
+    def record(self, path: str, content: str) -> None:
+        self._hashes[self._normalize(path)] = self._hash(content)
+
+    def is_known(self, path: str) -> bool:
+        return self._normalize(path) in self._hashes
+
+    def lock_for(self, path: str) -> asyncio.Lock:
+        return self._locks.setdefault(self._normalize(path), asyncio.Lock())
+
+    def verify(self, path: str, content: str) -> bool:
+        """Check whether content matches what was last recorded for path."""
+        normalized = self._normalize(path)
+        stored = self._hashes.get(normalized)
+        return stored is not None and self._hash(content) == stored

ddev/src/ddev/ai/tools/fs/read_file.py

@@ -0,0 +1,51 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from pathlib import Path
+from typing import Annotated
+
+from pydantic import Field
+
+from ddev.ai.tools.core.base import BaseToolInput
+from ddev.ai.tools.core.types import ToolResult
+
+from .base import FileRegistryTool
+
+
+class ReadFileInput(BaseToolInput):
+    path: Annotated[str, Field(description="Absolute or relative path to the file to read")]
+    offset: Annotated[
+        int, Field(description="Line number to start reading from (0-indexed, default: 0). Must be >= 0.", ge=0)
+    ] = 0
+    limit: Annotated[
+        int | None, Field(description="Number of lines to read (default: all remaining lines). Must be >= 1.", ge=1)
+    ] = None
+
+
+class ReadFileTool(FileRegistryTool[ReadFileInput]):
+    """Reads contents of a text file from the host filesystem.
+    Use to inspect config files, logs, source code. Do not use for binary files.
+    The output is a numbered list of lines starting from 0.
+    Supports offset/limit for paging through large files.
+    File does not need to be registered in the file registry.
+    Note: data="" is a valid result meaning no lines in range."""
+
+    @property
+    def name(self) -> str:
+        return "read_file"
+
+    async def __call__(self, tool_input: ReadFileInput) -> ToolResult:
+        try:
+            content = Path(tool_input.path).read_text(encoding="utf-8")
+        except (OSError, UnicodeDecodeError) as e:
+            return ToolResult(success=False, error=f"{tool_input.path}: {e}")
+
+        self._refresh_if_known(tool_input.path, content)
+
+        offset = tool_input.offset
+        limit = tool_input.limit
+        lines = content.splitlines(keepends=True)
+        slice_ = lines[offset : offset + limit] if limit is not None else lines[offset:]
+        width = len(str(offset + len(slice_)))
+        numbered = "".join(f"{offset + i:{width}}: {line}" for i, line in enumerate(slice_))
+        return ToolResult(success=True, data=numbered)

ddev/src/ddev/ai/tools/http/__init__.py

@@ -0,0 +1,3 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)

ddev/src/ddev/ai/tools/http/http_get.py

@@ -0,0 +1,59 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from typing import Annotated
+
+import httpx
+from pydantic import Field
+
+from ddev.ai.tools.core.base import BaseTool, BaseToolInput
+from ddev.ai.tools.core.truncation import TruncateResult, truncate
+from ddev.ai.tools.core.types import ToolResult
+
+
+class HttpGetInput(BaseToolInput):
+    url: Annotated[str, Field(description="Full URL to probe (must start with http:// or https://)")]
+    timeout: Annotated[float, Field(description="Request timeout in seconds (default: 10)", gt=0)] = 10.0
+
+
+class HttpGetTool(BaseTool[HttpGetInput]):
+    """Performs an HTTP GET request to check if an endpoint is reachable.
+    Use to validate that a metrics endpoint is accessible and inspect its response.
+    Returns the HTTP status code and response body (truncated if large)."""
+
+    @property
+    def name(self) -> str:
+        return "http_get"
+
+    async def __call__(self, tool_input: HttpGetInput) -> ToolResult:
+        url: str = tool_input.url
+        timeout: float = tool_input.timeout
+
+        if not url.startswith(("http://", "https://")):
+            return ToolResult(success=False, error="URL must start with http:// or https://")
+
+        try:
+            async with httpx.AsyncClient(timeout=timeout) as client:
+                response = await client.get(url)
+        except httpx.TimeoutException:
+            return ToolResult(success=False, error=f"Request timed out after {timeout}s")
+        except httpx.RequestError as e:
+            return ToolResult(success=False, error=f"Request failed for {url}: {e}")
+
+        body = response.text
+        result: TruncateResult = truncate(body)
+
+        status_line = f"Status: {response.status_code}"
+        output = f"{status_line}\n\n{result.output}"
+
+        if result.truncated and result.meta is not None:
+            return ToolResult(
+                success=response.is_success,
+                data=output,
+                truncated=True,
+                total_size=result.meta.total_size,
+                shown_size=result.meta.shown_size,
+                hint=result.meta.hint,
+            )
+
+        return ToolResult(success=response.is_success, data=output)

ddev/tests/ai/tools/fs/__init__.py

@@ -0,0 +1,3 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)