diff --git a/server/main.py b/server/main.py index 5c995dd..850b74d 100644 --- a/server/main.py +++ b/server/main.py @@ -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}")