diff --git a/codemcp/agno.py b/codemcp/agno.py index bc20606..2e1c811 100644 --- a/codemcp/agno.py +++ b/codemcp/agno.py @@ -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 @@ -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 @@ -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" ) @@ -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( diff --git a/codemcp/main.py b/codemcp/main.py index af42b9c..3cdbcdc 100644 --- a/codemcp/main.py +++ b/codemcp/main.py @@ -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]: @@ -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)): @@ -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) @@ -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. @@ -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" ) @@ -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" ) diff --git a/codemcp/tools/init_project.py b/codemcp/tools/init_project.py index 5a23884..74b49b7 100644 --- a/codemcp/tools/init_project.py +++ b/codemcp/tools/init_project.py @@ -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: diff --git a/codemcp/tools/write_file.py b/codemcp/tools/write_file.py index 25e6725..25665ea 100644 --- a/codemcp/tools/write_file.py +++ b/codemcp/tools/write_file.py @@ -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 @@ -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, @@ -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 )