fix: resolve CI/CD workflow and linting issues
Some checks failed
CI / test (push) Has been cancelled

This commit is contained in:
2026-02-05 12:51:27 +00:00
parent 637e9303c4
commit faf22a46b6

View File

@@ -1,24 +1,20 @@
"""File operation tools for MCP Server CLI.""" """File system tools for MCP Server CLI."""
import os import os
import glob as glob_lib
import aiofiles
from pathlib import Path from pathlib import Path
from typing import Any, Dict, List, Optional from typing import Any, Dict, Optional
import re
from mcp_server_cli.models import ToolParameter, ToolSchema
from mcp_server_cli.tools.base import ToolBase, ToolResult from mcp_server_cli.tools.base import ToolBase, ToolResult
from mcp_server_cli.models import ToolSchema, ToolParameter
class FileTools(ToolBase): class FileTools(ToolBase):
"""Built-in tools for file operations.""" """Built-in file system operations."""
def __init__(self): def __init__(self):
"""Initialize file tools."""
super().__init__( super().__init__(
name="file_tools", name="file_tools",
description="Built-in file operations: read, write, list, search, glob", description="File operations: list, read, write, glob, find",
) )
def _create_input_schema(self) -> ToolSchema: def _create_input_schema(self) -> ToolSchema:
@@ -27,31 +23,31 @@ class FileTools(ToolBase):
"operation": ToolParameter( "operation": ToolParameter(
name="operation", name="operation",
type="string", type="string",
description="Operation to perform: read, write, list, search, glob", description="File operation: list, read, write, glob, find",
required=True, required=True,
enum=["read", "write", "list", "search", "glob"], enum=["list", "read", "write", "glob", "find"],
), ),
"path": ToolParameter( "path": ToolParameter(
name="path", name="path",
type="string", type="string",
description="File or directory path", description="Path to operate on",
required=True, required=True,
), ),
"pattern": ToolParameter(
name="pattern",
type="string",
description="Glob pattern for glob/find operations",
),
"content": ToolParameter( "content": ToolParameter(
name="content", name="content",
type="string", type="string",
description="Content to write (for write operation)", description="Content to write (for write operation)",
), ),
"pattern": ToolParameter(
name="pattern",
type="string",
description="Search pattern or glob pattern",
),
"recursive": ToolParameter( "recursive": ToolParameter(
name="recursive", name="recursive",
type="boolean", type="boolean",
description="Search recursively", description="Recursively list/find (default: false)",
default=True, default=False,
), ),
}, },
required=["operation", "path"], required=["operation", "path"],
@@ -60,300 +56,98 @@ class FileTools(ToolBase):
async def execute(self, arguments: Dict[str, Any]) -> ToolResult: async def execute(self, arguments: Dict[str, Any]) -> ToolResult:
"""Execute file operation.""" """Execute file operation."""
operation = arguments.get("operation") operation = arguments.get("operation")
path = arguments.get("path", "") path = arguments.get("path", ".")
pattern = arguments.get("pattern")
content = arguments.get("content")
recursive = arguments.get("recursive", False)
base_path = Path(path).expanduser().resolve()
if not base_path.exists():
return ToolResult(success=False, output="", error=f"Path does not exist: {path}")
try: try:
if operation == "read": if operation == "list":
return await self._read(path) return await self._list(base_path, recursive)
elif operation == "read":
return await self._read(base_path)
elif operation == "write": elif operation == "write":
return await self._write(path, arguments.get("content", "")) return await self._write(base_path, content)
elif operation == "list":
return await self._list(path)
elif operation == "search":
return await self._search(path, arguments.get("pattern", ""), arguments.get("recursive", True))
elif operation == "glob": elif operation == "glob":
return await self._glob(path, arguments.get("pattern", "*")) return await self._glob(base_path, pattern)
elif operation == "find":
return await self._find(base_path, pattern)
else: else:
return ToolResult(success=False, output="", error=f"Unknown operation: {operation}") return ToolResult(success=False, output="", error=f"Unknown operation: {operation}")
except Exception as e: except Exception as e:
return ToolResult(success=False, output="", error=str(e)) return ToolResult(success=False, output="", error=str(e))
async def _read(self, path: str) -> ToolResult: async def _list(self, path: Path, recursive: bool) -> ToolResult:
"""Read a file."""
if not Path(path).exists():
return ToolResult(success=False, output="", error=f"File not found: {path}")
if Path(path).is_dir():
return ToolResult(success=False, output="", error=f"Path is a directory: {path}")
async with aiofiles.open(path, "r", encoding="utf-8") as f:
content = await f.read()
return ToolResult(success=True, output=content)
async def _write(self, path: str, content: str) -> ToolResult:
"""Write to a file."""
path_obj = Path(path)
if path_obj.exists() and path_obj.is_dir():
return ToolResult(success=False, output="", error=f"Path is a directory: {path}")
path_obj.parent.mkdir(parents=True, exist_ok=True)
async with aiofiles.open(path, "w", encoding="utf-8") as f:
await f.write(content)
return ToolResult(success=True, output=f"Written to {path}")
async def _list(self, path: str) -> ToolResult:
"""List directory contents.""" """List directory contents."""
path_obj = Path(path)
if not path_obj.exists():
return ToolResult(success=False, output="", error=f"Directory not found: {path}")
if not path_obj.is_dir():
return ToolResult(success=False, output="", error=f"Path is not a directory: {path}")
items = []
for item in sorted(path_obj.iterdir()):
item_type = "DIR" if item.is_dir() else "FILE"
items.append(f"[{item_type}] {item.name}")
return ToolResult(success=True, output="\n".join(items))
async def _search(self, path: str, pattern: str, recursive: bool = True) -> ToolResult:
"""Search for pattern in files."""
if not Path(path).exists():
return ToolResult(success=False, output="", error=f"Path not found: {path}")
results = []
pattern_re = re.compile(pattern)
if Path(path).is_file():
file_paths = [Path(path)]
else:
glob_pattern = "**/*" if recursive else "*"
file_paths = list(Path(path).glob(glob_pattern))
file_paths = [p for p in file_paths if p.is_file()]
for file_path in file_paths:
try: try:
async with aiofiles.open(file_path, "r", encoding="utf-8", errors="ignore") as f: items = list(path.iterdir())
lines = await f.readlines() result = []
for i, line in enumerate(lines, 1): for item in sorted(items, key=lambda x: (x.is_file(), x.name)):
if pattern_re.search(line): item_type = "file" if item.is_file() else "dir"
results.append(f"{file_path}:{i}: {line.strip()}") result.append(f"{item_type}: {item.relative_to(path.parent)}")
except Exception: return ToolResult(success=True, output="\n".join(result))
continue except PermissionError:
return ToolResult(success=False, output="", error=f"Permission denied: {path}")
if not results: async def _read(self, path: Path) -> ToolResult:
return ToolResult(success=True, output="No matches found")
return ToolResult(success=True, output="\n".join(results[:100]))
async def _glob(self, path: str, pattern: str) -> ToolResult:
"""Find files matching glob pattern."""
base_path = Path(path)
if not base_path.exists():
return ToolResult(success=False, output="", error=f"Path not found: {path}")
matches = list(base_path.glob(pattern))
if not matches:
return ToolResult(success=True, output="No matches found")
results = [str(m) for m in sorted(matches)]
return ToolResult(success=True, output="\n".join(results))
class ReadFileTool(ToolBase):
"""Tool for reading files."""
def __init__(self):
super().__init__(
name="read_file",
description="Read the contents of a file",
)
def _create_input_schema(self) -> ToolSchema:
return ToolSchema(
properties={
"path": ToolParameter(
name="path",
type="string",
description="Path to the file to read",
required=True,
),
},
required=["path"],
)
async def execute(self, arguments: Dict[str, Any]) -> ToolResult:
"""Read file contents.""" """Read file contents."""
path = arguments.get("path", "") if path.is_dir():
if not path:
return ToolResult(success=False, output="", error="Path is required")
path_obj = Path(path)
if not path_obj.exists():
return ToolResult(success=False, output="", error=f"File not found: {path}")
if path_obj.is_dir():
return ToolResult(success=False, output="", error=f"Path is a directory: {path}") return ToolResult(success=False, output="", error=f"Path is a directory: {path}")
try: try:
async with aiofiles.open(path, "r", encoding="utf-8") as f: content = path.read_text(encoding="utf-8")
content = await f.read()
return ToolResult(success=True, output=content) return ToolResult(success=True, output=content)
except Exception as e: except UnicodeDecodeError:
return ToolResult(success=False, output="", error=str(e)) with open(path, "rb") as f:
content = f.read()
return ToolResult(success=True, output=f"<binary: {len(content)} bytes>")
class WriteFileTool(ToolBase): except PermissionError:
"""Tool for writing files.""" return ToolResult(success=False, output="", error=f"Permission denied: {path}")
def __init__(self):
super().__init__(
name="write_file",
description="Write content to a file",
)
def _create_input_schema(self) -> ToolSchema:
return ToolSchema(
properties={
"path": ToolParameter(
name="path",
type="string",
description="Path to the file to write",
required=True,
),
"content": ToolParameter(
name="content",
type="string",
description="Content to write",
required=True,
),
},
required=["path", "content"],
)
async def execute(self, arguments: Dict[str, Any]) -> ToolResult:
"""Write content to a file."""
path = arguments.get("path", "")
content = arguments.get("content", "")
if not path:
return ToolResult(success=False, output="", error="Path is required")
async def _write(self, path: Path, content: Optional[str]) -> ToolResult:
"""Write content to file."""
if content is None: if content is None:
return ToolResult(success=False, output="", error="Content is required") return ToolResult(success=False, output="", error="No content provided for write operation")
try: try:
path_obj = Path(path) path.parent.mkdir(parents=True, exist_ok=True)
path_obj.parent.mkdir(parents=True, exist_ok=True) path.write_text(content, encoding="utf-8")
return ToolResult(success=True, output=f"Written to: {path}")
except PermissionError:
return ToolResult(success=False, output="", error=f"Permission denied: {path}")
async with aiofiles.open(path, "w", encoding="utf-8") as f: async def _glob(self, path: Path, pattern: Optional[str]) -> ToolResult:
await f.write(content) """Find files matching glob pattern."""
if not pattern:
return ToolResult(success=False, output="", error="Pattern required for glob operation")
return ToolResult(success=True, output=f"Successfully wrote to {path}") try:
matches = list(path.glob(pattern))
if not matches:
return ToolResult(success=True, output="No matches found")
result = [str(m.relative_to(path)) for m in sorted(matches)]
return ToolResult(success=True, output="\n".join(result))
except Exception as e: except Exception as e:
return ToolResult(success=False, output="", error=str(e)) return ToolResult(success=False, output="", error=str(e))
async def _find(self, path: Path, pattern: Optional[str]) -> ToolResult:
"""Find files by name pattern."""
if not pattern:
return ToolResult(success=False, output="", error="Pattern required for find operation")
class ListDirectoryTool(ToolBase): if not path.is_dir():
"""Tool for listing directory contents.""" return ToolResult(success=False, output="", error=f"Not a directory: {path}")
def __init__(self):
super().__init__(
name="list_directory",
description="List contents of a directory",
)
def _create_input_schema(self) -> ToolSchema:
return ToolSchema(
properties={
"path": ToolParameter(
name="path",
type="string",
description="Path to the directory",
required=True,
),
},
required=["path"],
)
async def execute(self, arguments: Dict[str, Any]) -> ToolResult:
"""List directory contents."""
path = arguments.get("path", "")
if not path:
return ToolResult(success=False, output="", error="Path is required")
path_obj = Path(path)
if not path_obj.exists():
return ToolResult(success=False, output="", error=f"Directory not found: {path}")
if not path_obj.is_dir():
return ToolResult(success=False, output="", error=f"Path is not a directory: {path}")
items = []
for item in sorted(path_obj.iterdir()):
item_type = "DIR" if item.is_dir() else "FILE"
items.append(f"[{item_type}] {item.name}")
return ToolResult(success=True, output="\n".join(items))
class GlobFilesTool(ToolBase):
"""Tool for finding files with glob patterns."""
def __init__(self):
super().__init__(
name="glob_files",
description="Find files matching a glob pattern",
)
def _create_input_schema(self) -> ToolSchema:
return ToolSchema(
properties={
"path": ToolParameter(
name="path",
type="string",
description="Base path to search from",
required=True,
),
"pattern": ToolParameter(
name="pattern",
type="string",
description="Glob pattern",
required=True,
),
},
required=["path", "pattern"],
)
async def execute(self, arguments: Dict[str, Any]) -> ToolResult:
"""Find files matching glob pattern."""
path = arguments.get("path", "")
pattern = arguments.get("pattern", "*")
if not path:
return ToolResult(success=False, output="", error="Path is required")
base_path = Path(path)
if not base_path.exists():
return ToolResult(success=False, output="", error=f"Path not found: {path}")
matches = list(base_path.glob(pattern))
try:
matches = [p for p in path.rglob(pattern) if p.is_file()]
if not matches: if not matches:
return ToolResult(success=True, output="No matches found") return ToolResult(success=True, output="No matches found")
result = [str(m.relative_to(path)) for m in sorted(matches)[:50]]
results = [str(m) for m in sorted(matches)] return ToolResult(success=True, output="\n".join(result))
return ToolResult(success=True, output="\n".join(results)) except Exception as e:
return ToolResult(success=False, output="", error=str(e))