Skip to content
Open
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
15 changes: 8 additions & 7 deletions codemcp/agno.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,12 @@
import asyncio
import sys
from typing import Union
from typing import Any, NoReturn, Union
from urllib.parse import quote

import click
from agno.agent import Agent
from agno.api.playground import PlaygroundEndpointCreate, create_playground_endpoint
from agno.agent.agent import Agent
from agno.api.playground import create_playground_endpoint
from agno.api.schemas.playground import PlaygroundEndpointCreate
from agno.cli.console import console
from agno.cli.settings import agno_cli_settings
from agno.tools.mcp import MCPTools
Expand All @@ -22,8 +23,8 @@ async def serve_playground_app_async(
host: str = "localhost",
port: int = 7777,
reload: bool = False,
prefix="/v1",
**kwargs,
prefix: str = "/v1",
**kwargs: Any,
):
import os
import signal
Expand Down Expand Up @@ -60,7 +61,7 @@ async def serve_playground_app_async(
console.print(panel)

# Define our custom signal handler that exits immediately
def handle_exit(sig, frame):
def handle_exit(sig: int, frame: Any) -> NoReturn:
logger.info(
"Received shutdown signal - exiting immediately without waiting for connections"
)
Expand Down Expand Up @@ -93,7 +94,7 @@ def handle_exit(sig, frame):
async def main(hello_world: bool = False):
async with MCPTools(f"{sys.executable} -m codemcp") as codemcp:
# TODO: cli-ify the model
from agno.models.anthropic import Claude
from agno.models.anthropic.claude import Claude

# from agno.models.google import Gemini
agent = Agent(
Expand Down
34 changes: 16 additions & 18 deletions codemcp/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,18 +14,10 @@
from starlette.routing import Mount

from .mcp import mcp
from .tools.chmod import chmod # noqa: F401
from .tools.edit_file import edit_file # noqa: F401
from .tools.glob import glob # noqa: F401
from .tools.grep import grep # noqa: F401
from .tools.init_project import init_project # noqa: F401
from .tools.ls import ls # noqa: F401
from .tools.mv import mv # noqa: F401
from .tools.read_file import read_file # noqa: F401
from .tools.rm import rm # noqa: F401
from .tools.run_command import run_command # noqa: F401
from .tools.think import think # noqa: F401
from .tools.write_file import write_file # noqa: F401

# pyright: reportUnusedImport=false
# pylint: disable=unused-import
# Import tools to register them with MCP - these are used indirectly


def get_files_respecting_gitignore(dir_path: Path, pattern: str = "**/*") -> List[Path]:
Expand All @@ -44,7 +36,7 @@ def get_files_respecting_gitignore(dir_path: Path, pattern: str = "**/*") -> Lis
all_dirs = [dir_path] + [p for p in all_paths if p.is_dir()]

# Find all .gitignore files in the directory and subdirectories
gitignore_specs = {}
gitignore_specs: dict[Path, pathspec.GitIgnoreSpec] = {}

# Process .gitignore files from root to leaf directories
for directory in sorted(all_dirs, key=lambda d: str(d)):
Expand Down Expand Up @@ -268,13 +260,15 @@ def init_codemcp_project(path: str, python: bool = False) -> str:
files_to_add = []

# Function to replace placeholders in a string
def replace_placeholders(text):
def replace_placeholders(text: str) -> str:
for placeholder, value in replacements.items():
text = text.replace(placeholder, value)
return text

# Function to process a file from template directory to output directory
def process_file(template_file, template_root, output_root):
def process_file(
template_file: Path, template_root: Path, output_root: Path
) -> Optional[Path]:
# Get the relative path from template root
rel_path = template_file.relative_to(template_root)

Expand Down Expand Up @@ -382,7 +376,9 @@ def init(path: str, python: bool) -> None:
@click.argument("command", type=str)
@click.argument("args", nargs=-1)
@click.option("--path", type=click.Path(), default=".", help="Project directory path")
def run(command: str, args: tuple, path: str) -> None:
def run_command_cli(
command: str, args: tuple[str, ...], path: str
) -> None: # Renamed from run to avoid redeclaration
"""Run a command defined in codemcp.toml without doing git commits.

The command should be defined in the [commands] section of codemcp.toml.
Expand Down Expand Up @@ -529,8 +525,9 @@ def run() -> None:
# Set up a signal handler to exit immediately on Ctrl+C
import os
import signal
from typing import Any, NoReturn

def handle_exit(sig, frame):
def handle_exit(sig: int, frame: Any) -> NoReturn:
logging.info(
"Received shutdown signal - exiting immediately without waiting for connections"
)
Expand Down Expand Up @@ -577,9 +574,10 @@ def serve(host: str, port: int, cors_origin: List[str]) -> None:

import os
import signal
from typing import Any, NoReturn

# Register a custom signal handler that will take precedence and exit immediately
def handle_exit(sig, frame):
def handle_exit(sig: int, frame: Any) -> NoReturn:
logging.info(
"Received shutdown signal - exiting immediately without waiting for connections"
)
Expand Down
3 changes: 2 additions & 1 deletion codemcp/tools/init_project.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,8 @@ def _slugify(text: str) -> str:
return text[:50]


def _generate_command_docs(command_docs: Dict[str, str]) -> str:
# pyright: reportUnusedFunction=false
def _generate_command_docs(command_docs: Dict[str, str]) -> str: # noqa: F811
"""Generate documentation for commands from the command_docs dictionary.

Args:
Expand Down
9 changes: 6 additions & 3 deletions codemcp/tools/write_file.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import json
import logging
import os
from typing import Any

from ..code_command import run_formatter_without_commit
from ..common import normalize_file_path
Expand All @@ -24,7 +25,7 @@
@mcp.tool()
async def write_file(
path: str,
content: str | dict | list | None = None,
content: str | dict[str, Any] | list[Any] | None = None,
description: str | None = None,
chat_id: str | None = None,
commit_hash: str | None = None,
Expand Down Expand Up @@ -68,10 +69,12 @@ async def write_file(
else:
content_str = content or ""

# Normalize newlines
# Normalize newlines if content is a string
# content_str could be dict/list that was passed directly from the content parameter

content_str = (
content_str.replace("\r\n", "\n")
if isinstance(content_str, str)
if isinstance(content_str, str) # pyright: ignore[reportUnnecessaryIsInstance]
else content_str
)

Expand Down
Loading