fix

.github/workflows/ci.yml

@@ -22,6 +22,9 @@ jobs:
       - name: Lint
         run: ruff check app/ tests/
 
+      - name: Type check
+        run: mypy app/
+
       - name: Security scan
         run: bandit -r app/ -q
 

app/api/dependencies.py

@@ -5,7 +5,7 @@ from app.core.config import get_settings
 settings = get_settings()
 
 
-def verify_api_key(x_api_key: Optional[str] = Header(default=None)):
+def verify_api_key(x_api_key: Optional[str] = Header(default=None)) -> None:
     if not settings.api_key:
         # API_KEY not configured — open access (dev/HF Spaces mode)
         return

app/api/main.py

@@ -29,7 +29,7 @@ app.include_router(router, prefix="/api/v1")
 
 
 @app.get("/")
-def root():
+def root() -> dict[str, str]:
     return {
         "message": "AI Code Review Agent",
         "version": "0.1.0",

app/api/routes.py

@@ -4,6 +4,7 @@ from app.core.logger import get_logger
 from fastapi import APIRouter, HTTPException, Depends
 from app.api.dependencies import verify_api_key
 from app.api.limiter import limiter
+from typing import Any
 from app.core.metrics import metrics
 from app.models.repository import RepositoryRequest, RepositoryResponse
 from app.models.report import EngineeringReport
@@ -22,12 +23,12 @@ logger = get_logger(__name__)
 
 
 @router.get("/health")
-def health_check():
+def health_check() -> dict[str, str]:
     return {"status": "ok", "message": "AI Code Review Agent is running"}
 
 @limiter.limit("5/minute")
 @router.post("/analyze", response_model=EngineeringReport, dependencies=[Depends(verify_api_key)])
-async def analyze_repository(request: RepositoryRequest):
+async def analyze_repository(request: RepositoryRequest) -> EngineeringReport:
     """
     Full pipeline:
     1. Clone repository
@@ -75,7 +76,7 @@ async def analyze_repository(request: RepositoryRequest):
 
 @limiter.limit("10/minute")
 @router.post("/quick-analyze", response_model=RepositoryResponse, dependencies=[Depends(verify_api_key)])
-async def quick_analyze(request: RepositoryRequest):
+async def quick_analyze(request: RepositoryRequest) -> RepositoryResponse:
     """
     Only run repository analysis agent.
     Faster, for testing purposes.
@@ -100,6 +101,6 @@ async def quick_analyze(request: RepositoryRequest):
             delete_repository(local_path)
 
 @router.get("/metrics")
-def get_metrics():
+def get_metrics() -> dict[str, Any]:
     """Observability endpoint: agent run counts, error rates, latencies."""
     return metrics.snapshot()

app/core/config.py

@@ -1,12 +1,11 @@
 import tempfile
 import os
-from pydantic_settings import BaseSettings
-from pydantic import ConfigDict
+from pydantic_settings import BaseSettings, SettingsConfigDict
 from functools import lru_cache
 
 
 class Settings(BaseSettings):
-    model_config = ConfigDict(env_file=".env", case_sensitive=False, extra="ignore")
+    model_config = SettingsConfigDict(env_file=".env", case_sensitive=False, extra="ignore")
 
     # Set OPENROUTER_API_KEY as a HF Space Secret
     openrouter_api_key: str = ""

app/core/llm.py

@@ -2,6 +2,7 @@ import httpx
 from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
 from app.core.config import get_settings
 from app.core.logger import get_logger
+from typing import Optional
 
 settings = get_settings()
 logger = get_logger(__name__)
@@ -18,7 +19,7 @@ OPENROUTER_API_URL = "https://openrouter.ai/api/v1/chat/completions"
         extra={"attempt": rs.attempt_number, "error": str(rs.outcome.exception())},
     ),
 )
-async def call_llm(prompt: str, system_prompt: str = None, max_tokens: int = 2000) -> str:
+async def call_llm(prompt: str, system_prompt: Optional[str] = None, max_tokens: int = 2000) -> str:
     """Async call to OpenRouter API with retry logic."""
     messages = []
     if system_prompt:
@@ -45,4 +46,4 @@ async def call_llm(prompt: str, system_prompt: str = None, max_tokens: int = 200
         response.raise_for_status()
 
     data = response.json()
-    return data["choices"][0]["message"]["content"]
+    return str(data["choices"][0]["message"]["content"])

app/core/metrics.py

@@ -1,25 +1,27 @@
 import time
 from collections import defaultdict
 from threading import Lock
+from typing import Any
+
 
 class MetricsCollector:
-    def __init__(self):
+    def __init__(self) -> None:
         self._lock = Lock()
         self._counts: dict[str, int] = defaultdict(int)
         self._latencies: dict[str, list[float]] = defaultdict(list)
         self._errors: dict[str, int] = defaultdict(int)
         self._started_at = time.time()
 
-    def record_run(self, agent: str, duration_seconds: float, success: bool):
+    def record_run(self, agent: str, duration_seconds: float, success: bool) -> None:
         with self._lock:
             self._counts[agent] += 1
             self._latencies[agent].append(round(duration_seconds, 3))
             if not success:
                 self._errors[agent] += 1
 
-    def snapshot(self) -> dict:
+    def snapshot(self) -> dict[str, Any]:
         with self._lock:
-            agents = {}
+            agents: dict[str, Any] = {}
             for agent, count in self._counts.items():
                 lats = self._latencies[agent]
                 agents[agent] = {
@@ -33,4 +35,5 @@ class MetricsCollector:
                 "agents": agents,
             }
 
+
 metrics = MetricsCollector()

app/models/issue.py

@@ -1,5 +1,5 @@
 from pydantic import BaseModel, model_validator, field_validator
-from typing import List, Optional
+from typing import List, Optional, Any
 
 
 class Bug(BaseModel):
@@ -10,7 +10,7 @@ class Bug(BaseModel):
 
     @field_validator("file", mode="before")
     @classmethod
-    def coerce_file(cls, v):
+    def coerce_file(cls, v: Any) -> str:
         if isinstance(v, list):
             return ", ".join(str(f) for f in v) if v else "unknown"
         return v or "unknown"
@@ -23,7 +23,7 @@ class Warning(BaseModel):
 
     @field_validator("file", mode="before")
     @classmethod
-    def coerce_file(cls, v):
+    def coerce_file(cls, v: Any) -> str:
         if isinstance(v, list):
             return ", ".join(str(f) for f in v) if v else "unknown"
         return v or "unknown"
@@ -35,7 +35,7 @@ class Suggestion(BaseModel):
 
     @field_validator("file", mode="before")
     @classmethod
-    def coerce_file(cls, v):
+    def coerce_file(cls, v: Any) -> str:
         if isinstance(v, list):
             return ", ".join(str(f) for f in v) if v else "unknown"
         return v or "unknown"

app/tools/ast_parser.py

@@ -1,75 +1,53 @@
 import ast
 import os
-from typing import List, Dict
+from typing import Any
 
 
-def parse_python_file(file_path: str) -> Dict:
-    """
-    Parse a Python file using AST.
-    Returns functions, classes, and imports found.
-    """
-    result = {
+def parse_python_file(file_path: str) -> dict[str, Any]:
+    result: dict[str, Any] = {
         "file": file_path,
         "functions": [],
         "classes": [],
         "imports": [],
-        "errors": []
+        "errors": [],
     }
-
     try:
         with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
             source = f.read()
-
         tree = ast.parse(source)
-
         for node in ast.walk(tree):
-            # Extract functions
             if isinstance(node, ast.FunctionDef):
                 result["functions"].append({
                     "name": node.name,
                     "line": node.lineno,
                     "args": [arg.arg for arg in node.args.args],
-                    "docstring": ast.get_docstring(node) or ""
+                    "docstring": ast.get_docstring(node) or "",
                 })
-
-            # Extract classes
             elif isinstance(node, ast.ClassDef):
                 result["classes"].append({
                     "name": node.name,
                     "line": node.lineno,
-                    "docstring": ast.get_docstring(node) or ""
+                    "docstring": ast.get_docstring(node) or "",
                 })
-
-            # Extract imports
             elif isinstance(node, ast.Import):
                 for alias in node.names:
                     result["imports"].append(alias.name)
-
             elif isinstance(node, ast.ImportFrom):
-                module = node.module or ""
-                result["imports"].append(module)
-
+                result["imports"].append(node.module or "")
     except SyntaxError as e:
         result["errors"].append(f"SyntaxError: {str(e)}")
     except Exception as e:
         result["errors"].append(f"Error: {str(e)}")
-
     return result
 
 
-def parse_repository(local_path: str) -> List[Dict]:
-    """
-    Parse all Python files in a repository.
-    """
-    results = []
-
+def parse_repository(local_path: str) -> list[dict[str, Any]]:
+    results: list[dict[str, Any]] = []
     for root, dirs, files in os.walk(local_path):
         dirs[:] = [d for d in dirs if d not in {
             ".git", "__pycache__", ".venv", "venv", "node_modules"
         }]
         for file in files:
             if file.endswith(".py"):
-                file_path = os.path.join(root, file)
-                results.append(parse_python_file(file_path))
-
+                results.append(parse_python_file(os.path.join(root, file)))
     return results

app/tools/file_scanner.py

@@ -40,7 +40,7 @@ def scan_repository(local_path: str) -> Dict:
     Scan a repository and return metadata.
     """
     all_files = []
-    language_counts = {}
+    language_counts: dict[str, int] = {}
     total_lines = 0
     frameworks_found = set()
 
@@ -74,7 +74,7 @@ def scan_repository(local_path: str) -> Dict:
     # Detect primary language
     primary_language = "Unknown"
     if language_counts:
-        primary_language = max(language_counts, key=language_counts.get)
+        primary_language = max(language_counts, key=lambda k: language_counts[k])
 
     return {
         "all_files": all_files,

app/tools/github_tool.py

@@ -41,8 +41,8 @@ def clone_repository(github_url: str) -> str:
 
     logger.info("Clone complete", extra={"path": str(local_path), "size_mb": round(size_mb, 1)})
     return str(local_path)
-    
-def delete_repository(local_path: str):
+
+def delete_repository(local_path: str) -> None:
     path = Path(local_path)
     if path.exists():
         shutil.rmtree(path, ignore_errors=True)

app/tools/pytest_runner.py

@@ -1,32 +1,18 @@
 import subprocess   # nosec B404
-from typing import Dict
+from typing import Any
 
 
-def run_tests(local_path: str) -> Dict:
-    """
-    Run pytest on a repository and return results.
-    """
-    result = {
-        "passed": 0,
-        "failed": 0,
-        "errors": 0,
-        "output": "",
-        "success": False
+def run_tests(local_path: str) -> dict[str, Any]:
+    result: dict[str, Any] = {
+        "passed": 0, "failed": 0, "errors": 0, "output": "", "success": False
     }
-
     try:
         process = subprocess.run(  # nosec
             ["pytest", "--tb=short", "-q", "--no-header"],
-            cwd=local_path,
-            capture_output=True,
-            text=True,
-            timeout=120
+            cwd=local_path, capture_output=True, text=True, timeout=120
         )
-
         result["output"] = process.stdout + process.stderr
-
-        # Parse basic counts from output
-        output = result["output"]
+        output: str = str(result["output"])
         if "passed" in output:
             result["success"] = True
         for line in output.splitlines():
@@ -39,12 +25,10 @@ def run_tests(local_path: str) -> Dict:
                         result["failed"] = int(parts[i - 1])
                     elif part == "error" and i > 0:
                         result["errors"] = int(parts[i - 1])
-
     except subprocess.TimeoutExpired:
         result["output"] = "Tests timed out after 120 seconds"
     except FileNotFoundError:
         result["output"] = "pytest not found in repository"
     except Exception as e:
         result["output"] = f"Error running tests: {str(e)}"
-
     return result

app/tools/static_analyzer.py

@@ -1,66 +1,44 @@
 import subprocess   # nosec B404
-from typing import Dict
+from typing import Any
 
 
-def run_ruff(local_path: str) -> Dict:
-    """
-    Run Ruff linter on repository.
-    """
-    result = {"output": "", "issues": [], "success": False}
-
+def run_ruff(local_path: str) -> dict[str, Any]:
+    result: dict[str, Any] = {"output": "", "issues": [], "success": False}
     try:
         process = subprocess.run(  # nosec
             ["ruff", "check", ".", "--output-format=text"],
-            cwd=local_path,
-            capture_output=True,
-            text=True,
-            timeout=60
+            cwd=local_path, capture_output=True, text=True, timeout=60
         )
         result["output"] = process.stdout + process.stderr
         result["issues"] = [
-            line for line in result["output"].splitlines() if line.strip()
+            line for line in str(result["output"]).splitlines() if line.strip()
         ]
         result["success"] = True
-
     except Exception as e:
         result["output"] = f"Ruff error: {str(e)}"
-
     return result
 
 
-def run_bandit(local_path: str) -> Dict:
-    """
-    Run Bandit security checker on repository.
-    """
-    result = {"output": "", "issues": [], "success": False}
-
+def run_bandit(local_path: str) -> dict[str, Any]:
+    result: dict[str, Any] = {"output": "", "issues": [], "success": False}
     try:
         process = subprocess.run(  # nosec
             ["bandit", "-r", ".", "-f", "txt", "-q"],
-            cwd=local_path,
-            capture_output=True,
-            text=True,
-            timeout=60
+            cwd=local_path, capture_output=True, text=True, timeout=60
         )
         result["output"] = process.stdout + process.stderr
         result["issues"] = [
-            line for line in result["output"].splitlines() if line.strip()
+            line for line in str(result["output"]).splitlines() if line.strip()
         ]
         result["success"] = True
-
     except Exception as e:
         result["output"] = f"Bandit error: {str(e)}"
-
     return result
 
 
-def run_full_analysis(local_path: str) -> Dict:
-    """
-    Run all static analysis tools.
-    """
+def run_full_analysis(local_path: str) -> dict[str, Any]:
     ruff_results = run_ruff(local_path)
     bandit_results = run_bandit(local_path)
-
     return {
         "ruff": ruff_results,
         "bandit": bandit_results,

pyproject.toml

@@ -19,4 +19,14 @@ asyncio_mode = "auto"
 asyncio_default_fixture_loop_scope = "function"
 
 [tool.bandit]
-exclude_dirs = ["tests", ".venv"]
+exclude_dirs = ["tests", ".venv"]
+
+[tool.mypy]
+python_version = "3.11"
+strict = false
+ignore_missing_imports = true
+disallow_untyped_defs = true
+disallow_untyped_calls = false
+warn_return_any = true
+warn_unused_ignores = true
+exclude = ["tests/", "gradio_app/"]

requirements.txt

@@ -27,3 +27,4 @@ pytest-asyncio
 fastapi==0.115.12
 uvicorn==0.34.3
 tenacity==8.3.0
+mypy==1.10.0