From a7e0bb625a50b93b83c694ef59fba0724d44046a Mon Sep 17 00:00:00 2001 From: "Edward Z. Yang" Date: Wed, 26 Mar 2025 05:49:53 +0800 Subject: [PATCH] Update [ghstack-poisoned] --- codemcp/hot_reload_entry.py | 8 ++++++-- codemcp/tools/edit_file.py | 10 +++++----- codemcp/tools/glob.py | 12 +++++++----- codemcp/tools/grep.py | 25 ++++++++++++++++--------- codemcp/tools/init_project.py | 8 ++++---- codemcp/tools/ls.py | 19 ++++++++++--------- codemcp/tools/read_file.py | 9 +++++---- e2e/test_chmod.py | 6 +++--- 8 files changed, 56 insertions(+), 41 deletions(-) diff --git a/codemcp/hot_reload_entry.py b/codemcp/hot_reload_entry.py index 145f4976..9056f5b3 100644 --- a/codemcp/hot_reload_entry.py +++ b/codemcp/hot_reload_entry.py @@ -1,4 +1,5 @@ #!/usr/bin/env python3 +# pyright: reportUnknownMemberType=false import asyncio import functools @@ -6,7 +7,7 @@ import os import sys from asyncio import Future, Queue, Task -from typing import Any, Optional, Tuple +from typing import Any, Dict, Optional, Tuple, cast from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client @@ -162,8 +163,11 @@ async def _run_manager_task( break if command == "call": + # Use explicit type cast for arguments to satisfy the type checker + tool_args = cast(Dict[str, Any], args) + # pyright: ignore[reportUnknownMemberType] result = await session.call_tool( - name="codemcp", arguments=args + name="codemcp", arguments=tool_args ) # This is the only error case FastMCP can # faithfully re-propagate, see diff --git a/codemcp/tools/edit_file.py b/codemcp/tools/edit_file.py index f6c92fd6..e648598b 100644 --- a/codemcp/tools/edit_file.py +++ b/codemcp/tools/edit_file.py @@ -7,7 +7,7 @@ import os import re from difflib import SequenceMatcher -from typing import Any +from typing import Any, Dict, List, Tuple from ..common import get_edit_snippet from ..file_utils import ( @@ -54,7 +54,7 @@ async def apply_edit( file_path: str, old_string: str, new_string: str, -) -> tuple[list[dict[str, Any]], str]: +) -> Tuple[List[Dict[str, Any]], str]: """Apply an edit to a file using robust matching strategies. Args: @@ -74,11 +74,11 @@ async def apply_edit( # For creating a new file, just return the new content if not old_string.strip(): updated_file = new_string - old_lines = [] + old_lines: List[str] = [] new_lines = new_string.split("\n") # Create a simple patch structure - patch = [ + patch: List[Dict[str, Any]] = [ { "oldStart": 1, "oldLines": 0, @@ -98,7 +98,7 @@ async def apply_edit( updated_file = content # Create a useful diff/patch structure - patch = [] + patch: List[Dict[str, Any]] = [] if content != updated_file: # Only create a patch if there were actual changes old_lines = old_string.split("\n") new_lines = new_string.split("\n") diff --git a/codemcp/tools/glob.py b/codemcp/tools/glob.py index a70a900b..c98114e8 100644 --- a/codemcp/tools/glob.py +++ b/codemcp/tools/glob.py @@ -4,7 +4,7 @@ import logging import os from pathlib import Path -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional, Tuple from ..common import normalize_file_path @@ -71,14 +71,16 @@ async def glob( loop = asyncio.get_event_loop() # Get file stats asynchronously - stats = [] + stats: List[Optional[os.stat_result]] = [] for match in matches: - stat = await loop.run_in_executor( + file_stat = await loop.run_in_executor( None, lambda m=match: os.stat(m) if os.path.exists(m) else None ) - stats.append(stat) + stats.append(file_stat) - matches_with_stats = list(zip(matches, stats, strict=False)) + matches_with_stats: List[Tuple[Path, Optional[os.stat_result]]] = list( + zip(matches, stats, strict=False) + ) # In tests, sort by filename for deterministic results if os.environ.get("NODE_ENV") == "test": diff --git a/codemcp/tools/grep.py b/codemcp/tools/grep.py index b4110b96..fde90dfb 100644 --- a/codemcp/tools/grep.py +++ b/codemcp/tools/grep.py @@ -3,7 +3,7 @@ import logging import os import subprocess -from typing import Any +from typing import Any, Dict, List, Optional, Tuple from ..common import normalize_file_path from ..git import is_git_repository @@ -120,7 +120,11 @@ async def git_grep( matches = [line.strip() for line in result.stdout.split() if line.strip()] # Convert to absolute paths - matches = [os.path.join(absolute_path, match) for match in matches] + matches = [ + os.path.join(absolute_path, match) + for match in matches + if isinstance(match, str) + ] return matches except subprocess.SubprocessError as e: @@ -128,7 +132,7 @@ async def git_grep( raise -def render_result_for_assistant(output: dict[str, Any]) -> str: +def render_result_for_assistant(output: Dict[str, Any]) -> str: """Render the results in a format suitable for the assistant. Args: @@ -182,14 +186,16 @@ async def grep_files( loop = asyncio.get_event_loop() # Get file stats asynchronously - stats = [] + stats: List[Optional[os.stat_result]] = [] for match in matches: - stat = await loop.run_in_executor( + file_stat = await loop.run_in_executor( None, lambda m=match: os.stat(m) if os.path.exists(m) else None ) - stats.append(stat) + stats.append(file_stat) - matches_with_stats = list(zip(matches, stats, strict=False)) + matches_with_stats: List[Tuple[str, Optional[os.stat_result]]] = list( + zip(matches, stats, strict=False) + ) # In tests, sort by filename for deterministic results if os.environ.get("NODE_ENV") == "test": @@ -201,12 +207,13 @@ async def grep_files( matches = [match for match, _ in matches_with_stats] # Prepare output - output = { + output: Dict[str, Any] = { "filenames": matches[:MAX_RESULTS], "numFiles": len(matches), } # Add formatted result for assistant - output["resultForAssistant"] = render_result_for_assistant(output) + formatted_result = render_result_for_assistant(output) + output["resultForAssistant"] = formatted_result return output diff --git a/codemcp/tools/init_project.py b/codemcp/tools/init_project.py index 20a97549..8bf96634 100644 --- a/codemcp/tools/init_project.py +++ b/codemcp/tools/init_project.py @@ -4,7 +4,7 @@ import logging import os import re -from typing import Dict, Optional +from typing import Any, Dict, List, Optional import tomli @@ -50,7 +50,7 @@ def _generate_command_docs(command_docs: Dict[str, str]) -> str: if not command_docs: return "" - docs = [] + docs: List[str] = [] for cmd_name, doc in command_docs.items(): docs.append(f"\n- {cmd_name}: {doc}") @@ -207,8 +207,8 @@ async def init_project( project_prompt = "" command_help = "" - command_docs = {} - rules_config = {} + command_docs: Dict[str, str] = {} + rules_config: Dict[str, Any] = {} # We've already confirmed that codemcp.toml exists try: diff --git a/codemcp/tools/ls.py b/codemcp/tools/ls.py index be0168a5..42603661 100644 --- a/codemcp/tools/ls.py +++ b/codemcp/tools/ls.py @@ -2,6 +2,7 @@ import asyncio import os +from typing import List, Optional from ..access import check_edit_permission from ..common import normalize_file_path @@ -67,7 +68,7 @@ async def ls_directory(directory_path: str, chat_id: str | None = None) -> str: return f"{TRUNCATED_MESSAGE}{tree_output}" -async def list_directory(initial_path: str) -> list[str]: +async def list_directory(initial_path: str) -> List[str]: """List all files and directories recursively. Args: @@ -77,12 +78,12 @@ async def list_directory(initial_path: str) -> list[str]: A list of relative paths to files and directories """ - results = [] + results: List[str] = [] loop = asyncio.get_event_loop() # Use a function to perform the directory listing asynchronously - async def list_dir_async(): - queue = [initial_path] + async def list_dir_async() -> List[str]: + queue: List[str] = [initial_path] while queue and len(results) <= MAX_FILES: path = queue.pop(0) @@ -145,10 +146,10 @@ def __init__(self, name: str, path: str, node_type: str): self.name = name self.path = path self.type = node_type - self.children = [] + self.children: List[TreeNode] = [] -def create_file_tree(sorted_paths: list[str]) -> list[TreeNode]: +def create_file_tree(sorted_paths: List[str]) -> List[TreeNode]: """Create a file tree from a list of paths. Args: @@ -158,7 +159,7 @@ def create_file_tree(sorted_paths: list[str]) -> list[TreeNode]: A list of TreeNode objects representing the root of the tree """ - root = [] + root: List[TreeNode] = [] for path in sorted_paths: parts = path.split(os.sep) @@ -173,7 +174,7 @@ def create_file_tree(sorted_paths: list[str]) -> list[TreeNode]: is_last_part = i == len(parts) - 1 # Check if this node already exists at this level - existing_node = None + existing_node: Optional[TreeNode] = None for node in current_level: if node.name == part: existing_node = node @@ -196,7 +197,7 @@ def create_file_tree(sorted_paths: list[str]) -> list[TreeNode]: def print_tree( - tree: list[TreeNode], + tree: List[TreeNode], level: int = 0, prefix: str = "", cwd: str = "", diff --git a/codemcp/tools/read_file.py b/codemcp/tools/read_file.py index 0f2a162b..2f392b90 100644 --- a/codemcp/tools/read_file.py +++ b/codemcp/tools/read_file.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 import os +from typing import List from ..common import ( MAX_LINE_LENGTH, @@ -75,7 +76,7 @@ async def read_file_content( selected_lines = all_lines[line_offset : line_offset + max_lines] # Process lines (truncate long lines) - processed_lines = [] + processed_lines: List[str] = [] for line in selected_lines: if len(line) > MAX_LINE_LENGTH: processed_lines.append( @@ -85,10 +86,10 @@ async def read_file_content( processed_lines.append(line) # Add line numbers (1-indexed) - numbered_lines = [] + numbered_lines: List[str] = [] for i, line in enumerate(processed_lines): - line_num = line_offset + i + 1 # 1-indexed line number - numbered_lines.append(f"{line_num:6}\t{line.rstrip()}") + line_number = line_offset + i + 1 # 1-indexed line number + numbered_lines.append(f"{line_number:6}\t{line.rstrip()}") content = "\n".join(numbered_lines) diff --git a/e2e/test_chmod.py b/e2e/test_chmod.py index 08bf3482..3422e604 100644 --- a/e2e/test_chmod.py +++ b/e2e/test_chmod.py @@ -137,9 +137,9 @@ async def test_chmod_error_handling(self): ) # Check for either error message (from main.py or chmod.py) self.assertTrue( - "unsupported chmod mode" in error_text.lower() or - "mode must be either 'a+x' or 'a-x'" in error_text.lower(), - f"Expected an error about invalid mode, but got: {error_text}" + "unsupported chmod mode" in error_text.lower() + or "mode must be either 'a+x' or 'a-x'" in error_text.lower(), + f"Expected an error about invalid mode, but got: {error_text}", )