Coverage for mcp/resources/source.py: 87%

1"""Source code resources (source:// scheme) for the GCO MCP server.""" 

2 

3from pathlib import Path 

4 

5from server import mcp 

6 

7PROJECT_ROOT = Path(__file__).parent.parent.parent 

8 

9_SOURCE_DIRS = { 

10 "gco": PROJECT_ROOT / "gco", 

11 "cli": PROJECT_ROOT / "cli", 

12 "lambda": PROJECT_ROOT / "lambda", 

13 "mcp": PROJECT_ROOT / "mcp", 

14 "scripts": PROJECT_ROOT / "scripts", 

15 "demo": PROJECT_ROOT / "demo", 

16 "dockerfiles": PROJECT_ROOT / "dockerfiles", 

17} 

18_SKIP_DIRS = { 

19 "__pycache__", 

20 ".git", 

21 "cdk.out", 

22 "node_modules", 

23 "kubectl-applier-simple-build", 

24 "helm-installer-build", 

25} 

26_SOURCE_EXTENSIONS = {".py", ".yaml", ".yml", ".json", ".txt", ".toml", ".cfg", ".sh", ".md"} 

27 

28_CONFIG_FILES = { 

29 "pyproject.toml", 

30 "cdk.json", 

31 "app.py", 

32 "Dockerfile.dev", 

33 ".gitlab-ci.yml", 

34 ".pre-commit-config.yaml", 

35 ".yamllint.yml", 

36 ".checkov.yaml", 

37 ".kics.yaml", 

38 ".gitleaks.toml", 

39 ".semgrepignore", 

40 ".dockerignore", 

41 ".gitignore", 

42} 

43 

44 

45def _list_source_files(base: Path) -> list[Path]: 

46 """Walk a directory and return all source files, skipping noise.""" 

47 files = [] 

48 for p in sorted(base.rglob("*")): 

49 if any(skip in p.parts for skip in _SKIP_DIRS): 

50 continue 

51 if p.is_file() and p.suffix in _SOURCE_EXTENSIONS: 

52 files.append(p) 

53 return files 

54 

55 

56@mcp.resource("source://gco/index") 

57def source_index() -> str: 

58 """List all source code files available for reading, grouped by package.""" 

59 sections = ["# GCO Source Code Index\n"] 

60 sections.append("## Project Config") 

61 for name in sorted(_CONFIG_FILES): 

62 if (PROJECT_ROOT / name).is_file(): 62 ↛ 61line 62 didn't jump to line 61 because the condition on line 62 was always true

63 sections.append(f"- `source://gco/config/{name}`") 

64 for pkg, base in _SOURCE_DIRS.items(): 

65 if not base.is_dir(): 65 ↛ 66line 65 didn't jump to line 66 because the condition on line 65 was never true

66 continue 

67 files = _list_source_files(base) 

68 if not files: 68 ↛ 69line 68 didn't jump to line 69 because the condition on line 68 was never true

69 continue 

70 sections.append(f"\n## {pkg}/ ({len(files)} files)") 

71 for f in files: 

72 rel = f.relative_to(PROJECT_ROOT) 

73 sections.append(f"- `source://gco/file/{rel}`") 

74 return "\n".join(sections) 

75 

76 

77@mcp.resource("source://gco/config/{filename}") 

78def config_file_resource(filename: str) -> str: 

79 """Read a top-level project config file (pyproject.toml, cdk.json, etc.).""" 

80 if filename not in _CONFIG_FILES: 

81 return f"Not available. Allowed: {', '.join(sorted(_CONFIG_FILES))}" 

82 path = PROJECT_ROOT / filename 

83 if not path.is_file(): 83 ↛ 84line 83 didn't jump to line 84 because the condition on line 83 was never true

84 return f"File '{filename}' not found." 

85 return path.read_text() 

86 

87 

88@mcp.resource("source://gco/file/{filepath*}") 

89def source_file_resource(filepath: str) -> str: 

90 """Read any source file by its path relative to the project root.""" 

91 path = (PROJECT_ROOT / filepath).resolve() 

92 if not str(path).startswith(str(PROJECT_ROOT.resolve())): 92 ↛ 93line 92 didn't jump to line 93 because the condition on line 92 was never true

93 return "Access denied: path is outside the project." 

94 if any(skip in path.parts for skip in _SKIP_DIRS): 

95 return "Access denied: path is in a skipped directory." 

96 if not path.is_file(): 

97 return f"File '{filepath}' not found." 

98 if path.suffix not in _SOURCE_EXTENSIONS: 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true

99 return f"File type '{path.suffix}' not served. Allowed: {', '.join(_SOURCE_EXTENSIONS)}" 

100 return path.read_text()