feat: codex can now call tools. implemented functions.shell, functions.read_text_file and functions.write_text_file. Use A2A_TOOLS_CONFIG=tools.codex.yaml A2A_AGENT_COMMAND='./path/to/codex-acp/target/debug/codex-acp' make run to use codex-acp with tools

Author: mrorigo

src/a2a_acp/sandbox.py

@@ -321,7 +321,7 @@ async def validate_script_security(self, script: str, allowed_commands: Optional
             (r';\s*curl\s+', "Command chaining with curl (potential injection)"),
             (r'>\s*/dev/null', "Output redirection hiding (potential injection)"),
             (r'2>\s*/dev/null', "Error redirection hiding (potential injection)"),
-            (r'\$\(', "Command substitution (potential injection)"),
+            (r'\$\((?!\()', "Command substitution (potential injection)"),
             (r'`.*`', "Backtick command execution (potential injection)"),
             (r';\s*cat\s+', "Command chaining with cat (potential injection)"),
             (r'\|\s*cat\s+', "Pipe to cat (potential injection)"),
@@ -746,4 +746,4 @@ async def managed_sandbox(
     try:
         yield env, working_dir
     finally:
-        await sandbox.cleanup_sandbox(working_dir, context)
+        await sandbox.cleanup_sandbox(working_dir, context)

src/a2a_acp/tool_config.py

@@ -215,13 +215,24 @@ def __init__(self, config_paths: Optional[List[str]] = None):
             config_paths: List of paths to search for tool configuration files.
                          Defaults to standard locations.
         """
-        self.config_paths = config_paths or [
+        default_paths = [
             "tools.yaml",
             "tools.yml",
             "config/tools.yaml",
             "config/tools.yml",
             "/etc/a2a-acp/tools.yaml"
         ]
+        base_paths = config_paths if config_paths is not None else default_paths
+
+        env_value = os.environ.get("A2A_TOOLS_CONFIG", "")
+        env_paths = [p.strip() for p in env_value.split(os.pathsep) if p.strip()] if env_value else []
+
+        combined_paths: List[str] = []
+        for path in env_paths + base_paths:
+            if path not in combined_paths:
+                combined_paths.append(path)
+
+        self.config_paths = combined_paths or default_paths
         self._tools: Dict[str, BashTool] = {}
         self._config_mtimes: Dict[str, float] = {}
         self._hot_reload_enabled = True
@@ -448,4 +459,4 @@ async def get_tool(tool_id: str) -> Optional[BashTool]:
 async def list_tools() -> List[BashTool]:
     """Convenience function to list all available tools."""
     manager = get_tool_configuration_manager()
-    return await manager.list_tools()
+    return await manager.list_tools()

src/a2a_acp/zed_agent.py

@@ -5,6 +5,7 @@
 import logging
 from asyncio import StreamReader, StreamWriter
 from typing import Any, Awaitable, Callable, Coroutine, Optional, Sequence
+from uuid import uuid4
 
 from .bash_executor import get_bash_executor, ToolExecutionResult
 from .sandbox import ExecutionContext
@@ -379,6 +380,12 @@ async def handler(payload: dict[str, Any]) -> None:
                 "payload_keys": list(payload.keys())
             })
 
+            # Handle agent-initiated requests (e.g., fs/read_text_file)
+            if payload.get("id") is not None and payload.get("method"):
+                handled = await self._handle_agent_request(payload, session_id)
+                if handled:
+                    return
+
             if payload.get("method") == "session/update":
                 params = payload.get("params", {})
                 update_data = params.get("update", {})
@@ -580,6 +587,131 @@ def stderr(self) -> str:
         """Return aggregated stderr output."""
         return "\n".join(self._stderr_buffer)
 
+    async def _handle_agent_request(self, payload: dict[str, Any], session_id: str) -> bool:
+        """Handle JSON-RPC requests initiated by the agent."""
+        method = payload.get("method")
+
+        if method == "fs/read_text_file":
+            await self._handle_fs_read_text_file(payload, session_id)
+            return True
+
+        return False
+
+    async def _handle_fs_read_text_file(self, payload: dict[str, Any], session_id: str) -> None:
+        """Execute the Codex filesystem read request via the bash tool."""
+        request_id = payload.get("id")
+        params = payload.get("params") or {}
+
+        if request_id is None:
+            self._logger.warning("fs/read_text_file request missing id", extra={"payload": payload})
+            return
+
+        self._logger.info("Handling fs/read_text_file request", extra={
+            "request_id": request_id,
+            "path": params.get("path"),
+            "line": params.get("line"),
+            "limit": params.get("limit")
+        })
+
+        # Resolve tool configuration
+        tool = await get_tool("functions.acp_fs__read_text_file")
+        if not tool:
+            await self._write_json({
+                "jsonrpc": "2.0",
+                "id": request_id,
+                "error": {
+                    "code": -32001,
+                    "message": "Tool functions.acp_fs__read_text_file is not available"
+                }
+            })
+            return
+
+        tool_params: dict[str, Any] = {}
+        path = params.get("path")
+        if not path:
+            await self._write_json({
+                "jsonrpc": "2.0",
+                "id": request_id,
+                "error": {
+                    "code": -32602,
+                    "message": "Missing required parameter: path"
+                }
+            })
+            return
+        tool_params["path"] = path
+
+        # Optional parameters with validation
+        if "line" in params and params["line"] is not None:
+            try:
+                tool_params["line"] = int(params["line"])
+            except (TypeError, ValueError):
+                await self._write_json({
+                    "jsonrpc": "2.0",
+                    "id": request_id,
+                    "error": {
+                        "code": -32602,
+                        "message": "Invalid line parameter; must be an integer"
+                    }
+                })
+                return
+
+        if "limit" in params and params["limit"] is not None:
+            try:
+                tool_params["limit"] = int(params["limit"])
+            except (TypeError, ValueError):
+                await self._write_json({
+                    "jsonrpc": "2.0",
+                    "id": request_id,
+                    "error": {
+                        "code": -32602,
+                        "message": "Invalid limit parameter; must be an integer"
+                    }
+                })
+                return
+
+        # Execute the tool
+        executor = get_bash_executor()
+        exec_session_id = params.get("sessionId") or session_id
+        context = ExecutionContext(
+            tool_id=tool.id,
+            session_id=exec_session_id,
+            task_id=f"fs_read_text_file_{uuid4().hex}",
+            user_id="zedacp_user"
+        )
+
+        result = await executor.execute_tool(tool, tool_params, context)
+
+        if result.success:
+            await self._write_json({
+                "jsonrpc": "2.0",
+                "id": request_id,
+                "result": {
+                    "content": result.output
+                }
+            })
+            self._logger.info("fs/read_text_file completed", extra={
+                "request_id": request_id,
+                "path": path,
+                "line": tool_params.get("line"),
+                "limit": tool_params.get("limit"),
+                "output_length": len(result.output) if result.output else 0
+            })
+        else:
+            await self._write_json({
+                "jsonrpc": "2.0",
+                "id": request_id,
+                "error": {
+                    "code": -32002,
+                    "message": result.error or "Tool execution failed"
+                }
+            })
+            self._logger.error("fs/read_text_file failed", extra={
+                "request_id": request_id,
+                "path": path,
+                "error": result.error,
+                "return_code": result.return_code
+            })
+
     async def _process_zedacp_tool_calls(self, response: Dict[str, Any], session_id: str) -> Dict[str, Any]:
         """Process ZedACP tool calls in agent response.