360 lines
13 KiB
Python
360 lines
13 KiB
Python
"""Tool definitions and implementations for nano claude."""
|
|
import os
|
|
import re
|
|
import glob as _glob
|
|
import subprocess
|
|
from pathlib import Path
|
|
from typing import Callable, Optional
|
|
|
|
# ── Tool JSON schemas (sent to Claude API) ─────────────────────────────────
|
|
|
|
TOOL_SCHEMAS = [
|
|
{
|
|
"name": "Read",
|
|
"description": (
|
|
"Read a file's contents. Returns content with line numbers "
|
|
"(format: 'N\\tline'). Use limit/offset to read large files in chunks."
|
|
),
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"file_path": {"type": "string", "description": "Absolute file path"},
|
|
"limit": {"type": "integer", "description": "Max lines to read"},
|
|
"offset": {"type": "integer", "description": "Start line (0-indexed)"},
|
|
},
|
|
"required": ["file_path"],
|
|
},
|
|
},
|
|
{
|
|
"name": "Write",
|
|
"description": "Write content to a file, creating parent directories as needed.",
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"file_path": {"type": "string"},
|
|
"content": {"type": "string"},
|
|
},
|
|
"required": ["file_path", "content"],
|
|
},
|
|
},
|
|
{
|
|
"name": "Edit",
|
|
"description": (
|
|
"Replace exact text in a file. old_string must match exactly (including whitespace). "
|
|
"If old_string appears multiple times, use replace_all=true or add more context."
|
|
),
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"file_path": {"type": "string"},
|
|
"old_string": {"type": "string", "description": "Exact text to replace"},
|
|
"new_string": {"type": "string", "description": "Replacement text"},
|
|
"replace_all": {"type": "boolean", "description": "Replace all occurrences"},
|
|
},
|
|
"required": ["file_path", "old_string", "new_string"],
|
|
},
|
|
},
|
|
{
|
|
"name": "Bash",
|
|
"description": "Execute a shell command. Returns stdout+stderr. Stateless (no cd persistence).",
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"command": {"type": "string"},
|
|
"timeout": {"type": "integer", "description": "Seconds before timeout (default 30)"},
|
|
},
|
|
"required": ["command"],
|
|
},
|
|
},
|
|
{
|
|
"name": "Glob",
|
|
"description": "Find files matching a glob pattern. Returns sorted list of matching paths.",
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"pattern": {"type": "string", "description": "Glob pattern e.g. **/*.py"},
|
|
"path": {"type": "string", "description": "Base directory (default: cwd)"},
|
|
},
|
|
"required": ["pattern"],
|
|
},
|
|
},
|
|
{
|
|
"name": "Grep",
|
|
"description": "Search file contents with regex using ripgrep (falls back to grep).",
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"pattern": {"type": "string", "description": "Regex pattern"},
|
|
"path": {"type": "string", "description": "File or directory to search"},
|
|
"glob": {"type": "string", "description": "File filter e.g. *.py"},
|
|
"output_mode": {
|
|
"type": "string",
|
|
"enum": ["content", "files_with_matches", "count"],
|
|
"description": "content=matching lines, files_with_matches=file paths, count=match counts",
|
|
},
|
|
"case_insensitive": {"type": "boolean"},
|
|
"context": {"type": "integer", "description": "Lines of context around matches"},
|
|
},
|
|
"required": ["pattern"],
|
|
},
|
|
},
|
|
{
|
|
"name": "WebFetch",
|
|
"description": "Fetch a URL and return its text content (HTML stripped).",
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"url": {"type": "string"},
|
|
"prompt": {"type": "string", "description": "Hint for what to extract"},
|
|
},
|
|
"required": ["url"],
|
|
},
|
|
},
|
|
{
|
|
"name": "WebSearch",
|
|
"description": "Search the web via DuckDuckGo and return top results.",
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {"type": "string"},
|
|
},
|
|
"required": ["query"],
|
|
},
|
|
},
|
|
]
|
|
|
|
# ── Safe bash commands (never ask permission) ───────────────────────────────
|
|
|
|
_SAFE_PREFIXES = (
|
|
"ls", "cat", "head", "tail", "wc", "pwd", "echo", "printf", "date",
|
|
"which", "type", "env", "printenv", "uname", "whoami", "id",
|
|
"git log", "git status", "git diff", "git show", "git branch",
|
|
"git remote", "git stash list", "git tag",
|
|
"find ", "grep ", "rg ", "ag ", "fd ",
|
|
"python ", "python3 ", "node ", "ruby ", "perl ",
|
|
"pip show", "pip list", "npm list", "cargo metadata",
|
|
"df ", "du ", "free ", "top -bn", "ps ",
|
|
"curl -I", "curl --head",
|
|
)
|
|
|
|
def _is_safe_bash(cmd: str) -> bool:
|
|
c = cmd.strip()
|
|
return any(c.startswith(p) for p in _SAFE_PREFIXES)
|
|
|
|
|
|
# ── Tool implementations ───────────────────────────────────────────────────
|
|
|
|
def _read(file_path: str, limit: int = None, offset: int = None) -> str:
|
|
p = Path(file_path)
|
|
if not p.exists():
|
|
return f"Error: file not found: {file_path}"
|
|
if p.is_dir():
|
|
return f"Error: {file_path} is a directory"
|
|
try:
|
|
lines = p.read_text(errors="replace").splitlines(keepends=True)
|
|
start = offset or 0
|
|
chunk = lines[start:start + limit] if limit else lines[start:]
|
|
if not chunk:
|
|
return "(empty file)"
|
|
return "".join(f"{start + i + 1}\t{l}" for i, l in enumerate(chunk))
|
|
except Exception as e:
|
|
return f"Error: {e}"
|
|
|
|
|
|
def _write(file_path: str, content: str) -> str:
|
|
p = Path(file_path)
|
|
try:
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
p.write_text(content)
|
|
lc = content.count("\n") + (1 if content and not content.endswith("\n") else 0)
|
|
return f"Wrote {lc} lines to {file_path}"
|
|
except Exception as e:
|
|
return f"Error: {e}"
|
|
|
|
|
|
def _edit(file_path: str, old_string: str, new_string: str, replace_all: bool = False) -> str:
|
|
p = Path(file_path)
|
|
if not p.exists():
|
|
return f"Error: file not found: {file_path}"
|
|
try:
|
|
content = p.read_text()
|
|
count = content.count(old_string)
|
|
if count == 0:
|
|
return "Error: old_string not found in file"
|
|
if count > 1 and not replace_all:
|
|
return (f"Error: old_string appears {count} times. "
|
|
"Provide more context to make it unique, or use replace_all=true.")
|
|
new_content = content.replace(old_string, new_string) if replace_all else \
|
|
content.replace(old_string, new_string, 1)
|
|
p.write_text(new_content)
|
|
return f"Replaced {'all ' + str(count) if replace_all else '1'} occurrence(s) in {file_path}"
|
|
except Exception as e:
|
|
return f"Error: {e}"
|
|
|
|
|
|
def _bash(command: str, timeout: int = 30) -> str:
|
|
try:
|
|
r = subprocess.run(
|
|
command, shell=True, capture_output=True, text=True,
|
|
timeout=timeout, cwd=os.getcwd(),
|
|
)
|
|
out = r.stdout
|
|
if r.stderr:
|
|
out += ("\n" if out else "") + "[stderr]\n" + r.stderr
|
|
return out.strip() or "(no output)"
|
|
except subprocess.TimeoutExpired:
|
|
return f"Error: timed out after {timeout}s"
|
|
except Exception as e:
|
|
return f"Error: {e}"
|
|
|
|
|
|
def _glob(pattern: str, path: str = None) -> str:
|
|
base = Path(path) if path else Path.cwd()
|
|
try:
|
|
matches = sorted(base.glob(pattern))
|
|
if not matches:
|
|
return "No files matched"
|
|
return "\n".join(str(m) for m in matches[:500])
|
|
except Exception as e:
|
|
return f"Error: {e}"
|
|
|
|
|
|
def _has_rg() -> bool:
|
|
try:
|
|
subprocess.run(["rg", "--version"], capture_output=True, check=True)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _grep(pattern: str, path: str = None, glob: str = None,
|
|
output_mode: str = "files_with_matches",
|
|
case_insensitive: bool = False, context: int = 0) -> str:
|
|
use_rg = _has_rg()
|
|
cmd = ["rg" if use_rg else "grep", "--no-heading"]
|
|
if case_insensitive:
|
|
cmd.append("-i")
|
|
if output_mode == "files_with_matches":
|
|
cmd.append("-l")
|
|
elif output_mode == "count":
|
|
cmd.append("-c")
|
|
else:
|
|
cmd.append("-n")
|
|
if context:
|
|
cmd += ["-C", str(context)]
|
|
if glob:
|
|
cmd += (["--glob", glob] if use_rg else ["--include", glob])
|
|
cmd.append(pattern)
|
|
cmd.append(path or str(Path.cwd()))
|
|
try:
|
|
r = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
|
|
out = r.stdout.strip()
|
|
return out[:20000] if out else "No matches found"
|
|
except Exception as e:
|
|
return f"Error: {e}"
|
|
|
|
|
|
def _webfetch(url: str, prompt: str = None) -> str:
|
|
try:
|
|
import httpx
|
|
r = httpx.get(url, headers={"User-Agent": "NanoClaude/1.0"},
|
|
timeout=30, follow_redirects=True)
|
|
r.raise_for_status()
|
|
ct = r.headers.get("content-type", "")
|
|
if "html" in ct:
|
|
text = re.sub(r"<script[^>]*>.*?</script>", "", r.text,
|
|
flags=re.DOTALL | re.IGNORECASE)
|
|
text = re.sub(r"<style[^>]*>.*?</style>", "", text,
|
|
flags=re.DOTALL | re.IGNORECASE)
|
|
text = re.sub(r"<[^>]+>", " ", text)
|
|
text = re.sub(r"\s+", " ", text).strip()
|
|
else:
|
|
text = r.text
|
|
return text[:25000]
|
|
except ImportError:
|
|
return "Error: httpx not installed — run: pip install httpx"
|
|
except Exception as e:
|
|
return f"Error: {e}"
|
|
|
|
|
|
def _websearch(query: str) -> str:
|
|
try:
|
|
import httpx
|
|
url = "https://html.duckduckgo.com/html/"
|
|
r = httpx.get(url, params={"q": query},
|
|
headers={"User-Agent": "Mozilla/5.0 (compatible)"},
|
|
timeout=30, follow_redirects=True)
|
|
titles = re.findall(r'class="result__title"[^>]*>.*?<a[^>]*href="([^"]+)"[^>]*>(.*?)</a>',
|
|
r.text, re.DOTALL)
|
|
snippets = re.findall(r'class="result__snippet"[^>]*>(.*?)</div>', r.text, re.DOTALL)
|
|
results = []
|
|
for i, (link, title) in enumerate(titles[:8]):
|
|
t = re.sub(r"<[^>]+>", "", title).strip()
|
|
s = re.sub(r"<[^>]+>", "", snippets[i]).strip() if i < len(snippets) else ""
|
|
results.append(f"**{t}**\n{link}\n{s}")
|
|
return "\n\n".join(results) if results else "No results found"
|
|
except ImportError:
|
|
return "Error: httpx not installed — run: pip install httpx"
|
|
except Exception as e:
|
|
return f"Error: {e}"
|
|
|
|
|
|
# ── Dispatcher ─────────────────────────────────────────────────────────────
|
|
|
|
def execute_tool(
|
|
name: str,
|
|
inputs: dict,
|
|
permission_mode: str = "auto",
|
|
ask_permission: Optional[Callable[[str], bool]] = None,
|
|
) -> str:
|
|
"""Dispatch tool execution; ask permission for write/destructive ops."""
|
|
|
|
def _check(desc: str) -> bool:
|
|
"""Return True if action is allowed."""
|
|
if permission_mode == "accept-all":
|
|
return True
|
|
if ask_permission:
|
|
return ask_permission(desc)
|
|
return True # headless: allow everything
|
|
|
|
if name == "Read":
|
|
return _read(inputs["file_path"], inputs.get("limit"), inputs.get("offset"))
|
|
|
|
elif name == "Write":
|
|
if not _check(f"Write to {inputs['file_path']}"):
|
|
return "Denied: user rejected write operation"
|
|
return _write(inputs["file_path"], inputs["content"])
|
|
|
|
elif name == "Edit":
|
|
if not _check(f"Edit {inputs['file_path']}"):
|
|
return "Denied: user rejected edit operation"
|
|
return _edit(inputs["file_path"], inputs["old_string"],
|
|
inputs["new_string"], inputs.get("replace_all", False))
|
|
|
|
elif name == "Bash":
|
|
cmd = inputs["command"]
|
|
if permission_mode != "accept-all" and not _is_safe_bash(cmd):
|
|
if not _check(f"Bash: {cmd}"):
|
|
return "Denied: user rejected bash command"
|
|
return _bash(cmd, inputs.get("timeout", 30))
|
|
|
|
elif name == "Glob":
|
|
return _glob(inputs["pattern"], inputs.get("path"))
|
|
|
|
elif name == "Grep":
|
|
return _grep(
|
|
inputs["pattern"], inputs.get("path"), inputs.get("glob"),
|
|
inputs.get("output_mode", "files_with_matches"),
|
|
inputs.get("case_insensitive", False),
|
|
inputs.get("context", 0),
|
|
)
|
|
|
|
elif name == "WebFetch":
|
|
return _webfetch(inputs["url"], inputs.get("prompt"))
|
|
|
|
elif name == "WebSearch":
|
|
return _websearch(inputs["query"])
|
|
|
|
else:
|
|
return f"Unknown tool: {name}"
|