Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions codemcp/hot_reload_entry.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,13 @@
#!/usr/bin/env python3
# pyright: reportUnknownMemberType=false

import asyncio
import functools
import logging
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
Expand Down Expand Up @@ -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
Expand Down
10 changes: 5 additions & 5 deletions codemcp/tools/edit_file.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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:
Expand All @@ -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,
Expand All @@ -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")
Expand Down
12 changes: 7 additions & 5 deletions codemcp/tools/glob.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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":
Expand Down
25 changes: 16 additions & 9 deletions codemcp/tools/grep.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -120,15 +120,19 @@ 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:
logging.exception(f"Error executing git grep: {e!s}")
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:
Expand Down Expand Up @@ -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":
Expand All @@ -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
8 changes: 4 additions & 4 deletions codemcp/tools/init_project.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
import logging
import os
import re
from typing import Dict, Optional
from typing import Any, Dict, List, Optional

import tomli

Expand Down Expand Up @@ -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}")

Expand Down Expand Up @@ -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:
Expand Down
19 changes: 10 additions & 9 deletions codemcp/tools/ls.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand All @@ -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)

Expand Down Expand Up @@ -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:
Expand All @@ -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)
Expand All @@ -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
Expand All @@ -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 = "",
Expand Down
9 changes: 5 additions & 4 deletions codemcp/tools/read_file.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
#!/usr/bin/env python3

import os
from typing import List

from ..common import (
MAX_LINE_LENGTH,
Expand Down Expand Up @@ -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(
Expand All @@ -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)

Expand Down
6 changes: 3 additions & 3 deletions e2e/test_chmod.py
Original file line number Diff line number Diff line change
Expand Up @@ -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}",
)


Expand Down
Loading