initialize ruff and mypy

.github/workflows/check-code.yaml

@@ -7,7 +7,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ['3.10']
+        python-version: ['3.12']
     runs-on: ubuntu-latest
     steps:
       - name: Checkout Code
@@ -28,7 +28,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ['3.10']
+        python-version: ['3.12']
     steps:
     - uses: actions/checkout@v5
     - name: Set up Python ${{ matrix.python-version }}
@@ -47,7 +47,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ['3.10']
+        python-version: ['3.12']
     steps:
     - uses: actions/checkout@v5
     - name: Set up Python ${{ matrix.python-version }}
@@ -66,7 +66,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ['3.10']
+        python-version: ['3.12']
     steps:
     - uses: actions/checkout@v5
     - name: Set up Python ${{ matrix.python-version }}

.github/workflows/publish-pypi.yaml

@@ -15,7 +15,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ['3.10']
+        python-version: ['3.12']
     steps:
       - name: Checkout code
         uses: actions/checkout@v5

.github/workflows/publish-testpypi.yaml

@@ -15,7 +15,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ['3.10']
+        python-version: ['3.12']
     steps:
       - name: Checkout code
         uses: actions/checkout@v5

CONFIGURATION.md

@@ -130,7 +130,7 @@ Then, add one import at the top of your agent to trigger the patching:
 
 ```python
 try:
-    import kaizen.auto
+    import kaizen.auto # noqa: F401
 except ImportError:
     pass
 

demo/filesystem/server.py

@@ -10,7 +10,7 @@
 import json
 import mimetypes
 from pathlib import Path
-from typing import List, Optional, Dict, Any
+from typing import List, Optional, Dict, Any, TypedDict
 from datetime import datetime
 import glob
 import fnmatch
@@ -26,7 +26,7 @@
 
 def normalize_path(p: str) -> str:
     """Normalize a path to use forward slashes and resolve it."""
-    return str(Path(p).resolve()).replace('\\', '/')
+    return str(Path(p).resolve()).replace("\\", "/")
 
 
 def expand_home(p: str) -> str:
@@ -52,15 +52,15 @@ def validate_path(file_path: str) -> str:
 
     # Check if path is within any allowed directory
     for allowed_dir in allowed_directories:
-        if resolved.startswith(allowed_dir + '/') or resolved == allowed_dir:
+        if resolved.startswith(allowed_dir + "/") or resolved == allowed_dir:
             return resolved
 
     raise ValueError(f"Access denied: {file_path} is outside allowed directories")
 
 
-def format_size(size_bytes: int) -> str:
+def format_size(size_bytes: float) -> str:
     """Format file size in human-readable format."""
-    for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
+    for unit in ["B", "KB", "MB", "GB", "TB"]:
         if size_bytes < 1024.0:
             return f"{size_bytes:.1f}{unit}"
         size_bytes /= 1024.0
@@ -83,46 +83,46 @@ async def get_file_stats(file_path: str) -> Dict[str, Any]:
 
 async def read_file_content(file_path: str) -> str:
     """Read file content as text."""
-    with open(file_path, 'r', encoding='utf-8') as f:
+    with open(file_path, "r", encoding="utf-8") as f:
         return f.read()
 
 
 async def write_file_content(file_path: str, content: str) -> None:
     """Write content to file."""
     # Ensure parent directory exists
-    os.makedirs(os.path.dirname(file_path) or '.', exist_ok=True)
-    with open(file_path, 'w', encoding='utf-8') as f:
+    os.makedirs(os.path.dirname(file_path) or ".", exist_ok=True)
+    with open(file_path, "w", encoding="utf-8") as f:
         f.write(content)
 
 
 async def tail_file(file_path: str, n: int) -> str:
     """Read last n lines of a file."""
-    with open(file_path, 'r', encoding='utf-8') as f:
+    with open(file_path, "r", encoding="utf-8") as f:
         lines = f.readlines()
-    return ''.join(lines[-n:])
+    return "".join(lines[-n:])
 
 
 async def head_file(file_path: str, n: int) -> str:
     """Read first n lines of a file."""
-    with open(file_path, 'r', encoding='utf-8') as f:
+    with open(file_path, "r", encoding="utf-8") as f:
         lines = []
         for i, line in enumerate(f):
             if i >= n:
                 break
             lines.append(line)
-    return ''.join(lines)
+    return "".join(lines)
 
 
 async def apply_file_edits(file_path: str, edits: List[Dict[str, str]], dry_run: bool = False) -> str:
     """Apply edits to a file and return a diff-style result."""
-    with open(file_path, 'r', encoding='utf-8') as f:
+    with open(file_path, "r", encoding="utf-8") as f:
         content = f.read()
 
     original_content = content
 
     for edit in edits:
-        old_text = edit['oldText']
-        new_text = edit['newText']
+        old_text = edit["oldText"]
+        new_text = edit["newText"]
 
         if old_text not in content:
             raise ValueError(f"Text to replace not found: {old_text[:50]}...")
@@ -135,7 +135,7 @@ async def apply_file_edits(file_path: str, edits: List[Dict[str, str]], dry_run:
         content = content.replace(old_text, new_text)
 
     if not dry_run:
-        with open(file_path, 'w', encoding='utf-8') as f:
+        with open(file_path, "w", encoding="utf-8") as f:
             f.write(content)
 
     # Generate diff
@@ -156,20 +156,18 @@ async def apply_file_edits(file_path: str, edits: List[Dict[str, str]], dry_run:
             diff_lines.append(f"+ {new_line.rstrip()}")
 
     status = "Dry run - no changes made" if dry_run else "Changes applied successfully"
-    return f"{status}\n\n" + '\n'.join(diff_lines)
+    return f"{status}\n\n" + "\n".join(diff_lines)
 
 
-async def search_files_recursive(
-    root_path: str, pattern: str, exclude_patterns: List[str] = None
-) -> List[str]:
+async def search_files_recursive(root_path: str, pattern: str, exclude_patterns: List[str] | None = None) -> List[str]:
     """Recursively search for files matching a pattern."""
     if exclude_patterns is None:
         exclude_patterns = []
 
     results = []
 
     # Handle glob patterns
-    if '**' in pattern:
+    if "**" in pattern:
         # Use glob for recursive patterns
         glob_pattern = os.path.join(root_path, pattern)
         matches = glob.glob(glob_pattern, recursive=True)
@@ -247,17 +245,13 @@ async def read_media_file(path: str) -> Dict[str, str]:
         mime_type = "application/octet-stream"
 
     # Read file as binary and encode to base64
-    with open(valid_path, 'rb') as f:
-        data = base64.b64encode(f.read()).decode('utf-8')
+    with open(valid_path, "rb") as f:
+        data = base64.b64encode(f.read()).decode("utf-8")
 
     return {
         "mimeType": mime_type,
         "data": data,
-        "type": "image"
-        if mime_type.startswith("image/")
-        else "audio"
-        if mime_type.startswith("audio/")
-        else "blob",
+        "type": "image" if mime_type.startswith("image/") else "audio" if mime_type.startswith("audio/") else "blob",
     }
 
 
@@ -369,13 +363,19 @@ async def list_directory_with_sizes(path: str, sortBy: str = "name") -> str:
         path: Path to the directory to list
         sortBy: Sort entries by 'name' or 'size' (default: 'name')
     """
-    valid_path = validate_path(path)
-    entries = os.listdir(valid_path)
+
+    class Entry(TypedDict):
+        name: str
+        is_dir: bool
+        size: int
+
+    valid_path: str = validate_path(path)
+    entries: list[str] = os.listdir(valid_path)
 
     # Collect entry details
-    detailed_entries = []
-    for entry in entries:
-        entry_path = os.path.join(valid_path, entry)
+    detailed_entries: list[Entry] = []
+    for path in entries:
+        entry_path = os.path.join(valid_path, path)
         is_dir = os.path.isdir(entry_path)
 
         try:
@@ -384,26 +384,27 @@ async def list_directory_with_sizes(path: str, sortBy: str = "name") -> str:
         except Exception:
             size = 0
 
-        detailed_entries.append({'name': entry, 'is_dir': is_dir, 'size': size})
+        detailed_entries.append({"name": path, "is_dir": is_dir, "size": size})
 
     # Sort entries
-    if sortBy == 'size':
-        detailed_entries.sort(key=lambda x: x['size'], reverse=True)
+    if sortBy == "size":
+        detailed_entries.sort(key=lambda x: x["size"], reverse=True)
     else:
-        detailed_entries.sort(key=lambda x: x['name'])
+        detailed_entries.sort(key=lambda x: x["name"])
 
     # Format output
     formatted = []
     for entry in detailed_entries:
-        prefix = "[DIR]" if entry['is_dir'] else "[FILE]"
-        name = entry['name'].ljust(30)
-        size_str = "" if entry['is_dir'] else format_size(entry['size']).rjust(10)
+        is_dir = entry["is_dir"]
+        prefix = "[DIR]" if is_dir else "[FILE]"
+        name = entry["name"].ljust(30)
+        size_str = "" if is_dir else format_size(entry["size"]).rjust(10)
         formatted.append(f"{prefix} {name} {size_str}")
 
     # Add summary
-    total_files = sum(1 for e in detailed_entries if not e['is_dir'])
-    total_dirs = sum(1 for e in detailed_entries if e['is_dir'])
-    total_size = sum(e['size'] for e in detailed_entries if not e['is_dir'])
+    total_files = sum(1 for e in detailed_entries if not e["is_dir"])
+    total_dirs = sum(1 for e in detailed_entries if e["is_dir"])
+    total_size = sum(e["size"] for e in detailed_entries if not e["is_dir"])
 
     formatted.append("")
     formatted.append(f"Total: {total_files} files, {total_dirs} directories")
@@ -413,7 +414,7 @@ async def list_directory_with_sizes(path: str, sortBy: str = "name") -> str:
 
 
 @mcp.tool()
-async def directory_tree(path: str, excludePatterns: List[str] = None) -> str:
+async def directory_tree(path: str, excludePatterns: List[str] | None = None) -> str:
     """
     Get a recursive tree view of files and directories as a JSON structure.
     Each entry includes 'name', 'type' (file/directory), and 'children' for directories.
@@ -452,10 +453,10 @@ def build_tree(current_path: str, root_path: str) -> List[Dict[str, Any]]:
                 continue
 
             is_dir = os.path.isdir(entry_path)
-            entry_data = {'name': entry, 'type': 'directory' if is_dir else 'file'}
+            entry_data: dict[str, str | list] = {"name": entry, "type": "directory" if is_dir else "file"}
 
             if is_dir:
-                entry_data['children'] = build_tree(entry_path, root_path)
+                entry_data["children"] = build_tree(entry_path, root_path)
 
             result.append(entry_data)
 
@@ -489,7 +490,7 @@ async def move_file(source: str, destination: str) -> str:
 
 
 @mcp.tool()
-async def search_files(path: str, pattern: str, excludePatterns: List[str] = None) -> str:
+async def search_files(path: str, pattern: str, excludePatterns: List[str] | None = None) -> str:
     """
     Recursively search for files and directories matching a pattern.
     The patterns should be glob-style patterns that match paths relative to the working directory.
@@ -549,23 +550,9 @@ async def list_allowed_directories() -> str:
 def main():
     """Main entry point for the server."""
     parser = argparse.ArgumentParser(description="Secure MCP Filesystem Server")
-    parser.add_argument(
-        "allowed_directories",
-        nargs="+",
-        help="Directories to allow access to"
-    )
-    parser.add_argument(
-        "--port",
-        type=int,
-        default=8112,
-        help="Port to run the SSE server on (default: 8112)"
-    )
-    parser.add_argument(
-        "--transport",
-        choices=["sse", "stdio"],
-        default="sse",
-        help="Transport capability to expose (default: sse)"
-    )
+    parser.add_argument("allowed_directories", nargs="+", help="Directories to allow access to")
+    parser.add_argument("--port", type=int, default=8112, help="Port to run the SSE server on (default: 8112)")
+    parser.add_argument("--transport", choices=["sse", "stdio"], default="sse", help="Transport capability to expose (default: sse)")
 
     args = parser.parse_args()
 

docs/LOW_CODE_TRACING.md

@@ -49,7 +49,7 @@ Add one import at the top of your agent:
 
 ```python
 try:
-    import kaizen.auto
+    import kaizen.auto # noqa: F401
 except ImportError:
     pass
 
@@ -77,7 +77,7 @@ Use this when you are tracing **raw API calls** directly using the `openai` libr
 
 ```python
 try:
-    import kaizen.auto
+    import kaizen.auto # noqa: F401
 except ImportError:
     pass
 
@@ -99,7 +99,7 @@ Use this when using **LiteLLM** to abstract across multiple providers. Kaizen tr
 
 ```python
 try:
-    import kaizen.auto
+    import kaizen.auto # noqa: F401
 except ImportError:
     pass
 
@@ -128,7 +128,7 @@ Use this for **agentic workflows** built with HuggingFace's `smolagents`. Kaizen
 
 ```python
 try:
-    import kaizen.auto
+    import kaizen.auto # noqa: F401
 except ImportError:
     pass
 
@@ -150,7 +150,7 @@ Use this for the **OpenAI Agents framework** (`agents`). Kaizen traces the high-
 
 ```python
 try:
-    import kaizen.auto
+    import kaizen.auto # noqa: F401
 except ImportError:
     pass