Coverage for mcp/resources/images.py: 90%

1"""Container image registry resources (images:// scheme) for the GCO MCP server. 

2 

3Each handler delegates to ``cli/images.py::ImageManager`` so the MCP 

4layer never re-implements ECR semantics. Handlers are sync — FastMCP's 

5@mcp.resource decorator handles sync resource handlers. 

6""" 

7 

8from __future__ import annotations 

9 

10import json 

11from typing import Any 

12 

13from server import mcp 

14 

15 

16def _get_manager() -> Any: 

17 """Lazy-import ``cli.images.get_image_manager`` so MCP server import 

18 doesn't pull boto3 prematurely. 

19 """ 

20 from cli.images import get_image_manager 

21 

22 return get_image_manager() 

23 

24 

25@mcp.resource("images://gco/index") 

26def images_index() -> str: 

27 """List every gco/* repository in ECR with summary metadata.""" 

28 try: 

29 repos = _get_manager().list_repos() 

30 except Exception as e: # noqa: BLE001 

31 return f"# Image Registry\n\nFailed to list repositories: {e}\n" 

32 

33 lines = ["# Image Registry — `gco/*` repositories\n"] 

34 if not repos: 

35 lines.append("No repositories found under the `gco/` prefix.") 

36 lines.append("") 

37 lines.append("Run `gco images init <name>` to create the first repo.") 

38 return "\n".join(lines) 

39 

40 lines.append("| Repository | Image Count | Tag Mutability | Created |") 

41 lines.append("| --- | --- | --- | --- |") 

42 for repo in repos: 

43 name = repo.get("name", "?") 

44 count = repo.get("image_count", "?") 

45 mutability = repo.get("tag_mutability", "?") 

46 created = repo.get("created_at", "?") 

47 lines.append(f"| `{name}` | {count} | {mutability} | {created} |") 

48 

49 lines.append("") 

50 lines.append("## Per-repo resources") 

51 for repo in repos: 

52 name = repo.get("name", "") 

53 if not name.startswith("gco/"): 53 ↛ 54line 53 didn't jump to line 54 because the condition on line 53 was never true

54 continue 

55 bare = name.removeprefix("gco/") 

56 lines.append(f"- `images://gco/{bare}/tags` — list every tag on `{name}`") 

57 lines.append("") 

58 lines.append("## Registry-wide resources") 

59 lines.append("- `images://gco/replication/status` — replication state across regions") 

60 return "\n".join(lines) 

61 

62 

63@mcp.resource("images://gco/{name}/tags") 

64def images_tags_resource(name: str) -> str: 

65 """List every tag on a single repository. 

66 

67 Args: 

68 name: Repository name (without the ``gco/`` prefix). 

69 """ 

70 try: 

71 rows = _get_manager().list_tags(name) 

72 except Exception as e: # noqa: BLE001 

73 return f"# Tags for `gco/{name}`\n\nFailed to list tags: {e}\n" 

74 

75 lines = [f"# Tags for `gco/{name}`\n"] 

76 if not rows: 

77 lines.append("No tags found on this repository.") 

78 return "\n".join(lines) 

79 

80 lines.append("| Tag | Digest | Pushed | Size (bytes) |") 

81 lines.append("| --- | --- | --- | --- |") 

82 for row in rows: 

83 tag = row.get("tag") or "(untagged)" 

84 digest = row.get("digest", "?") 

85 pushed = row.get("pushed_at", "?") 

86 size = row.get("size_bytes", "?") 

87 lines.append(f"| `{tag}` | `{digest}` | {pushed} | {size} |") 

88 

89 lines.append("") 

90 lines.append("## Per-tag resources") 

91 seen_tags: set[str] = set() 

92 for row in rows: 

93 tag = row.get("tag") 

94 if not tag or tag in seen_tags: 94 ↛ 95line 94 didn't jump to line 95 because the condition on line 94 was never true

95 continue 

96 seen_tags.add(tag) 

97 lines.append(f"- `images://gco/{name}/{tag}` — full describe for `gco/{name}:{tag}`") 

98 return "\n".join(lines) 

99 

100 

101@mcp.resource("images://gco/{name}/{tag}") 

102def images_describe_resource(name: str, tag: str) -> str: 

103 """Full ECR describe payload for a single tag, as JSON. 

104 

105 Args: 

106 name: Repository name (without the ``gco/`` prefix). 

107 tag: Image tag. 

108 """ 

109 try: 

110 result = _get_manager().describe(name, tag) 

111 except Exception as e: # noqa: BLE001 

112 return json.dumps({"error": str(e), "name": f"gco/{name}", "tag": tag}, indent=2) 

113 

114 if not result: 

115 return json.dumps({"error": "tag not found", "name": f"gco/{name}", "tag": tag}, indent=2) 

116 return json.dumps(result, indent=2, default=str) 

117 

118 

119@mcp.resource("images://gco/replication/status") 

120def images_replication_status_resource() -> str: 

121 """Registry-wide replication state across regions.""" 

122 try: 

123 rows = _get_manager().replication_status() 

124 except Exception as e: # noqa: BLE001 

125 return f"# Replication Status\n\nFailed to read replication state: {e}\n" 

126 

127 lines = ["# Replication Status — `gco/*` repositories\n"] 

128 if not rows: 

129 lines.append("No replication entries reported. The replication rule may be") 

130 lines.append("disabled, or no images have replicated yet.") 

131 lines.append("") 

132 lines.append("Run `gco images replication get` to inspect the configuration.") 

133 return "\n".join(lines) 

134 

135 lines.append("| Repository | Region | Status | Digest | Registry ID |") 

136 lines.append("| --- | --- | --- | --- | --- |") 

137 for row in rows: 

138 repo = row.get("repository", "?") 

139 region = row.get("region", "?") 

140 status = row.get("status", "?") 

141 digest = row.get("digest", "?") 

142 registry_id = row.get("registry_id", "?") 

143 lines.append(f"| `{repo}` | `{region}` | {status} | `{digest}` | {registry_id} |") 

144 return "\n".join(lines)