"""Read Python tool source code from the tools/ directory."""
import asyncio
import json
import logging
import os
from pathlib import Path
import aiofiles
logger = logging.getLogger(__name__)
TOOL_NAME = "read_tool_code"
TOOL_DESCRIPTION = (
"Read the source code of a Python tool file from the tools/ directory. "
"Useful for examining existing tool implementations or debugging."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"filename": {
"type": "string",
"description": "Name of the tool file to read (e.g. 'shell_tool.py').",
},
"max_lines": {
"type": "integer",
"description": "Max number of lines to return (default: all).",
},
"start_line": {
"type": "integer",
"description": "Starting line number (1-indexed).",
},
"end_line": {
"type": "integer",
"description": "Ending line number (1-indexed).",
},
},
"required": ["filename"],
}
TOOLS_ROOT = Path(__file__).resolve().parent
[docs]
async def run(
filename: str,
max_lines: int = None,
start_line: int = None,
end_line: int = None,
**_kwargs,
) -> str:
"""Execute this tool and return the result.
Args:
filename (str): The filename value.
max_lines (int): The max lines value.
start_line (int): The start line value.
end_line (int): The end line value.
Returns:
str: Result string.
"""
if not filename or not filename.strip():
return json.dumps({"success": False, "error": "Filename is required"})
filename = filename.strip()
if os.path.basename(filename) != filename:
return json.dumps({"success": False, "error": "Invalid filename: subdirectories not allowed"})
filepath = (TOOLS_ROOT / filename).resolve()
if not filepath.is_relative_to(TOOLS_ROOT):
return json.dumps({"success": False, "error": "Invalid filename: path traversal not allowed"})
if not await asyncio.to_thread(filepath.exists):
return json.dumps({"success": False, "error": f"File not found: {filename}"})
if not await asyncio.to_thread(filepath.is_file):
return json.dumps({"success": False, "error": f"Path is not a file: {filename}"})
if not filename.endswith(".py"):
return json.dumps({"success": False, "error": f"Only Python files (.py) are allowed: {filename}"})
try:
async with aiofiles.open(str(filepath), "r", encoding="utf-8") as f:
content = await f.read()
lines = content.splitlines(keepends=True)
total_lines = len(lines)
if start_line is not None or end_line is not None:
start_idx = (start_line - 1) if start_line is not None else 0
end_idx = end_line if end_line is not None else total_lines
if start_idx < 0 or start_idx >= total_lines:
return json.dumps({"success": False, "error": f"Invalid start_line: {start_line}. File has {total_lines} lines."})
if end_idx < 1 or end_idx > total_lines:
return json.dumps({"success": False, "error": f"Invalid end_line: {end_line}. File has {total_lines} lines."})
if start_idx >= end_idx:
return json.dumps({"success": False, "error": f"start_line ({start_line}) must be less than end_line ({end_line})"})
lines = lines[start_idx:end_idx]
elif max_lines is not None:
lines = lines[:max_lines]
return json.dumps({
"success": True,
"filename": filename,
"content": "".join(lines),
"total_lines": total_lines,
"returned_lines": len(lines),
})
except PermissionError:
return json.dumps({"success": False, "error": f"Permission denied reading file: {filename}"})
except UnicodeDecodeError:
return json.dumps({"success": False, "error": f"File encoding error (not UTF-8): {filename}"})
except Exception as e:
logger.error("Error reading tool file %s: %s", filepath, e)
return json.dumps({"success": False, "error": f"Unexpected error reading file: {e}"})