add file_edit, file_move, file_glob, file_grep tools and improve existing file tools
- file_read: numbered output (cat -n), offset/limit for chunked reading - file_write: updated description to prefer file_edit for partial changes - file_edit: partial string replacement (old_string → new_string), fails on ambiguous matches - file_move: rename/move files and directories within /workspace - file_glob: file discovery via glob patterns (e.g. **/*.py) - file_grep: regex search in file contents with context lines, glob filter, ignore_case - exec: new cwd parameter to run commands in subdirectories of /workspace
This commit is contained in:
Sidney Marvin Fricke
+145
-7
@@ -4,6 +4,7 @@ Implements MCP Streamable HTTP transport (JSON-RPC 2.0)
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import fnmatch
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
@@ -143,11 +144,12 @@ _jobs: dict[str, BackgroundJob] = {}
|
||||
TOOLS = [
|
||||
{
|
||||
"name": "exec",
|
||||
"description": "Run a shell command and return stdout+stderr synchronously (max 30 s).",
|
||||
"description": "Run a shell command inside /workspace and return stdout+stderr synchronously (max 30 s). Use cwd to run in a subdirectory.",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"cmd": {"type": "string", "description": "Shell command to execute"},
|
||||
"cwd": {"type": "string", "description": "Working directory relative to /workspace (default: /workspace)"},
|
||||
"timeout": {"type": "integer", "description": "Timeout in seconds (default 30)", "default": 30},
|
||||
},
|
||||
"required": ["cmd"],
|
||||
@@ -250,27 +252,43 @@ TOOLS = [
|
||||
},
|
||||
{
|
||||
"name": "file_read",
|
||||
"description": "Read the content of a file inside /workspace.",
|
||||
"description": "Read a file inside /workspace. Returns content with 1-based line numbers (cat -n format). Use offset and limit to read large files in chunks.",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"path": {"type": "string", "description": "Path relative to /workspace"},
|
||||
"offset": {"type": "integer", "description": "Line number to start reading from (1-based, default: 1)"},
|
||||
"limit": {"type": "integer", "description": "Maximum number of lines to return"},
|
||||
},
|
||||
"required": ["path"],
|
||||
},
|
||||
},
|
||||
{
|
||||
"name": "file_write",
|
||||
"description": "Write (create or overwrite) a file inside /workspace.",
|
||||
"description": "Write (create or overwrite) a file inside /workspace with the given content. Prefer file_edit for modifying existing files.",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"path": {"type": "string", "description": "Path relative to /workspace"},
|
||||
"content": {"type": "string", "description": "File content to write"},
|
||||
"content": {"type": "string", "description": "Full file content to write"},
|
||||
},
|
||||
"required": ["path", "content"],
|
||||
},
|
||||
},
|
||||
{
|
||||
"name": "file_edit",
|
||||
"description": "Perform an exact string replacement in a file inside /workspace. old_string must appear exactly once unless replace_all is true. Fails if old_string is not found or is ambiguous.",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"path": {"type": "string", "description": "Path relative to /workspace"},
|
||||
"old_string": {"type": "string", "description": "Exact text to find and replace"},
|
||||
"new_string": {"type": "string", "description": "Replacement text"},
|
||||
"replace_all": {"type": "boolean", "description": "Replace all occurrences (default: false)", "default": False},
|
||||
},
|
||||
"required": ["path", "old_string", "new_string"],
|
||||
},
|
||||
},
|
||||
{
|
||||
"name": "file_list",
|
||||
"description": "List files and directories inside /workspace (optionally a subdirectory).",
|
||||
@@ -292,6 +310,46 @@ TOOLS = [
|
||||
"required": ["path"],
|
||||
},
|
||||
},
|
||||
{
|
||||
"name": "file_move",
|
||||
"description": "Move or rename a file or directory inside /workspace.",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"src": {"type": "string", "description": "Source path relative to /workspace"},
|
||||
"dst": {"type": "string", "description": "Destination path relative to /workspace"},
|
||||
},
|
||||
"required": ["src", "dst"],
|
||||
},
|
||||
},
|
||||
{
|
||||
"name": "file_glob",
|
||||
"description": "Find files inside /workspace matching a glob pattern (e.g. '**/*.py'). Returns matching paths relative to /workspace.",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"pattern": {"type": "string", "description": "Glob pattern (e.g. '**/*.py', 'src/**/*.ts')"},
|
||||
"path": {"type": "string", "description": "Subdirectory to search in (default: workspace root)", "default": ""},
|
||||
},
|
||||
"required": ["pattern"],
|
||||
},
|
||||
},
|
||||
{
|
||||
"name": "file_grep",
|
||||
"description": "Search file contents inside /workspace using a regex pattern. Returns matching lines with file path and line number.",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"pattern": {"type": "string", "description": "Regular expression to search for"},
|
||||
"path": {"type": "string", "description": "File or directory to search in (default: workspace root)", "default": ""},
|
||||
"glob": {"type": "string", "description": "Only search files matching this glob (e.g. '*.py')"},
|
||||
"context": {"type": "integer", "description": "Lines of context before and after each match (default: 0)", "default": 0},
|
||||
"ignore_case": {"type": "boolean", "description": "Case-insensitive search (default: false)", "default": False},
|
||||
"max_matches": {"type": "integer", "description": "Maximum number of matches to return (default: 100)", "default": 100},
|
||||
},
|
||||
"required": ["pattern"],
|
||||
},
|
||||
},
|
||||
]
|
||||
|
||||
# ─── Tool dispatch ────────────────────────────────────────────────────────────
|
||||
@@ -300,12 +358,13 @@ async def _dispatch(name: str, args: dict) -> dict:
|
||||
if name == "exec":
|
||||
cmd = args.get("cmd", "")
|
||||
timeout = int(args.get("timeout", 30))
|
||||
log.info("exec: %s", cmd)
|
||||
cwd = _workspace_path(args["cwd"]) if args.get("cwd") else pathlib.Path(WORKSPACE)
|
||||
log.info("exec (cwd=%s): %s", cwd, cmd)
|
||||
try:
|
||||
r = subprocess.run(
|
||||
cmd, shell=True, capture_output=True, text=True,
|
||||
timeout=timeout, executable="/bin/bash",
|
||||
cwd=WORKSPACE,
|
||||
cwd=str(cwd),
|
||||
)
|
||||
return {"stdout": r.stdout, "stderr": r.stderr, "exit_code": r.returncode}
|
||||
except subprocess.TimeoutExpired:
|
||||
@@ -385,7 +444,12 @@ async def _dispatch(name: str, args: dict) -> dict:
|
||||
elif name == "file_read":
|
||||
p = _workspace_path(args["path"])
|
||||
log.info("file_read: %s", p)
|
||||
return {"path": args["path"], "content": p.read_text(encoding="utf-8")}
|
||||
all_lines = p.read_text(encoding="utf-8").splitlines()
|
||||
offset = max(1, int(args.get("offset", 1))) - 1 # convert to 0-based
|
||||
limit = args.get("limit")
|
||||
sliced = all_lines[offset:offset + int(limit)] if limit is not None else all_lines[offset:]
|
||||
numbered = "\n".join(f"{offset + i + 1}\t{line}" for i, line in enumerate(sliced))
|
||||
return {"path": args["path"], "content": numbered, "total_lines": len(all_lines)}
|
||||
|
||||
elif name == "file_write":
|
||||
p = _workspace_path(args["path"])
|
||||
@@ -394,6 +458,28 @@ async def _dispatch(name: str, args: dict) -> dict:
|
||||
p.write_text(args.get("content", ""), encoding="utf-8")
|
||||
return {"path": args["path"], "written": len(args.get("content", ""))}
|
||||
|
||||
elif name == "file_edit":
|
||||
p = _workspace_path(args["path"])
|
||||
old_string = args["old_string"]
|
||||
new_string = args["new_string"]
|
||||
replace_all = bool(args.get("replace_all", False))
|
||||
log.info("file_edit: %s replace_all=%s", p, replace_all)
|
||||
content = p.read_text(encoding="utf-8")
|
||||
count = content.count(old_string)
|
||||
if count == 0:
|
||||
raise ValueError(f"old_string not found in {args['path']}")
|
||||
if not replace_all and count > 1:
|
||||
raise ValueError(
|
||||
f"old_string is not unique in {args['path']} ({count} occurrences). "
|
||||
"Use replace_all=true to replace all, or provide more context to make it unique."
|
||||
)
|
||||
if replace_all:
|
||||
new_content = content.replace(old_string, new_string)
|
||||
else:
|
||||
new_content = content.replace(old_string, new_string, 1)
|
||||
p.write_text(new_content, encoding="utf-8")
|
||||
return {"path": args["path"], "replacements": count if replace_all else 1}
|
||||
|
||||
elif name == "file_list":
|
||||
p = _workspace_path(args.get("path", ""))
|
||||
log.info("file_list: %s", p)
|
||||
@@ -409,6 +495,58 @@ async def _dispatch(name: str, args: dict) -> dict:
|
||||
p.unlink()
|
||||
return {"deleted": args["path"]}
|
||||
|
||||
elif name == "file_move":
|
||||
src = _workspace_path(args["src"])
|
||||
dst = _workspace_path(args["dst"])
|
||||
log.info("file_move: %s -> %s", src, dst)
|
||||
dst.parent.mkdir(parents=True, exist_ok=True)
|
||||
src.rename(dst)
|
||||
return {"src": args["src"], "dst": args["dst"]}
|
||||
|
||||
elif name == "file_glob":
|
||||
base = _workspace_path(args.get("path", ""))
|
||||
pattern = args["pattern"]
|
||||
log.info("file_glob: %s in %s", pattern, base)
|
||||
matches = sorted(
|
||||
str(p.relative_to(pathlib.Path(WORKSPACE)))
|
||||
for p in base.glob(pattern)
|
||||
if p.is_file()
|
||||
)
|
||||
return {"pattern": pattern, "matches": matches, "count": len(matches)}
|
||||
|
||||
elif name == "file_grep":
|
||||
base = _workspace_path(args.get("path", ""))
|
||||
pattern = args["pattern"]
|
||||
glob_filter = args.get("glob")
|
||||
context_lines = int(args.get("context", 0))
|
||||
ignore_case = bool(args.get("ignore_case", False))
|
||||
max_matches = int(args.get("max_matches", 100))
|
||||
log.info("file_grep: %r in %s", pattern, base)
|
||||
flags = re.IGNORECASE if ignore_case else 0
|
||||
regex = re.compile(pattern, flags)
|
||||
results = []
|
||||
files = base.rglob("*") if base.is_dir() else [base]
|
||||
for filepath in sorted(files):
|
||||
if not filepath.is_file():
|
||||
continue
|
||||
if glob_filter and not fnmatch.fnmatch(filepath.name, glob_filter):
|
||||
continue
|
||||
try:
|
||||
lines = filepath.read_text(encoding="utf-8", errors="replace").splitlines()
|
||||
except OSError:
|
||||
continue
|
||||
for i, line in enumerate(lines):
|
||||
if regex.search(line):
|
||||
rel = str(filepath.relative_to(pathlib.Path(WORKSPACE)))
|
||||
entry = {"file": rel, "line": i + 1, "match": line}
|
||||
if context_lines:
|
||||
entry["before"] = lines[max(0, i - context_lines):i]
|
||||
entry["after"] = lines[i + 1:i + 1 + context_lines]
|
||||
results.append(entry)
|
||||
if len(results) >= max_matches:
|
||||
return {"pattern": pattern, "matches": results, "truncated": True}
|
||||
return {"pattern": pattern, "matches": results, "truncated": False}
|
||||
|
||||
raise ValueError(f"Unknown tool: {name}")
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user