Coverage for gco_mcp/resources/self.py: 89.08%
105 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +0000
1"""Self-indexing resources (``mcp://gco/...``) for the GCO MCP server.
3Four templates that surface the live MCP catalog through resource URIs
4so introspection clients (and AI assistants) can list every registered
5tool and resource template at a glance, plus the feature-flag map that
6gates each gated tool. Always-on — no feature flag gates these.
8* ``mcp://gco/tools/index`` — full tool index. Returns JSON shaped as
9 ``{"tools": [{"name", "description", "tags", "source_path",
10 "source_line", "gating_flag"}, ...]}``. ``source_path`` is project-
11 root-relative; ``source_line`` is the 1-indexed first line of the
12 wrapped function. ``gating_flag`` is the ``GCO_ENABLE_*`` constant
13 that gates the tool, or ``null`` when the tool is always-on.
14* ``mcp://gco/tools/{tool_name}`` — single-tool detail. Same shape as
15 one element of the index. Raises :class:`fastmcp.exceptions.NotFoundError`
16 for unknown names so the FastMCP error-handling middleware maps it to
17 MCP error code ``-32002``.
18* ``mcp://gco/resources/index`` — index of every static resource and
19 resource template. Returns ``{"resources": [{"uri", "name",
20 "description", "tags", "source_path", "source_line"}, ...],
21 "templates": [{"uri_template", "name", "description", ...}, ...]}``.
22* ``mcp://gco/feature-flags`` — the umbrella + per-tool flag table.
23 Returns ``{"flags": [{"name", "default", "gated_tools": [...]},
24 ...]}``. The ``gated_tools`` list is the static map below, kept in
25 sync by hand with the ``if is_enabled(...)`` blocks at the top of
26 each ``gco_mcp/tools/*.py`` module. The ``mission`` family lives in a
27 module-level ``if`` so its nine tools all gate together; image and
28 destructive tools use multiple-flag combinations.
30Tool-name → flag inference uses a static ``_TOOL_GATING_TABLE`` rather
31than re-parsing the source modules at request time. That table is
32short, easy to keep in sync, and cheap to read; the alternative — AST-
33walking each ``gco_mcp/tools/*.py`` module on every list call — would
34either thrash the disk on every introspection or grow a layer of
35caches we'd then have to invalidate. The map is exercised in
36``tests/test_mcp_self_resources.py`` so any drift between it and the
37real gating bodies trips a test failure rather than a silent
38documentation lie.
39"""
41from __future__ import annotations
43import inspect
44import json
45import sys
46from pathlib import Path
47from typing import Any, cast
49from feature_flags import (
50 ALL_FLAGS,
51 FLAG_ALL_TOOLS,
52 FLAG_CAPACITY_PURCHASE,
53 FLAG_DESTRUCTIVE_OPERATIONS,
54 FLAG_IMAGE_PUBLISH,
55 FLAG_INFRASTRUCTURE_DEPLOY,
56 FLAG_INFRASTRUCTURE_DESTROY,
57 FLAG_LOCAL_METRICS,
58 FLAG_LOCAL_STORAGE_SYNC,
59 FLAG_MISSION,
60 FLAG_MODEL_UPLOAD,
61 FLAG_SEMANTIC_PROGRESS,
62)
64# Import the live FastMCP instance so the resource handlers can hit
65# the same registry the rest of the server sees.
66sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
68# Project root used to build relative ``source_path`` strings.
69_PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
72# ---------------------------------------------------------------------------
73# Static gating table — kept in sync with the per-module ``if`` blocks.
74# ---------------------------------------------------------------------------
76_TOOL_GATING_TABLE: dict[str, str] = {
77 # gco_mcp/tools/capacity.py — purchase + destructive tools
78 "reserve_capacity": FLAG_CAPACITY_PURCHASE,
79 "create_reservation": FLAG_CAPACITY_PURCHASE,
80 "cancel_reservation": FLAG_DESTRUCTIVE_OPERATIONS,
81 # gco_mcp/tools/models.py / storage.py — local model-data upload + deletion
82 "models_upload": FLAG_MODEL_UPLOAD,
83 "upload_to_regional_bucket": FLAG_MODEL_UPLOAD,
84 "delete_model": FLAG_DESTRUCTIVE_OPERATIONS,
85 # gco_mcp/tools/images.py — image-publish + destructive
86 "images_build": FLAG_IMAGE_PUBLISH,
87 "images_push": FLAG_IMAGE_PUBLISH,
88 "images_mirror": FLAG_IMAGE_PUBLISH,
89 "images_delete_tag": FLAG_DESTRUCTIVE_OPERATIONS,
90 "images_delete_repo": FLAG_DESTRUCTIVE_OPERATIONS,
91 "images_cleanup": FLAG_DESTRUCTIVE_OPERATIONS,
92 "images_prune": FLAG_DESTRUCTIVE_OPERATIONS,
93 # gco_mcp/tools/stacks.py — deploy + destroy
94 "deploy_stack": FLAG_INFRASTRUCTURE_DEPLOY,
95 "deploy_all": FLAG_INFRASTRUCTURE_DEPLOY,
96 "bootstrap_cdk": FLAG_INFRASTRUCTURE_DEPLOY,
97 "addons_install": FLAG_INFRASTRUCTURE_DEPLOY,
98 "destroy_stack": FLAG_INFRASTRUCTURE_DESTROY,
99 "destroy_all": FLAG_INFRASTRUCTURE_DESTROY,
100 # Other destructive module-level gates
101 "delete_job": FLAG_DESTRUCTIVE_OPERATIONS,
102 "delete_inference": FLAG_DESTRUCTIVE_OPERATIONS,
103 "delete_template": FLAG_DESTRUCTIVE_OPERATIONS,
104 "delete_webhook": FLAG_DESTRUCTIVE_OPERATIONS,
105 "delete_nodepool": FLAG_DESTRUCTIVE_OPERATIONS,
106 "analytics_user_remove": FLAG_DESTRUCTIVE_OPERATIONS,
107 "monitoring_user_remove": FLAG_DESTRUCTIVE_OPERATIONS,
108 "cancel_queue_job": FLAG_DESTRUCTIVE_OPERATIONS,
109 "task_prune": FLAG_DESTRUCTIVE_OPERATIONS,
110 # Local filesystem and model-scoring readers
111 "metrics_from_local_file": FLAG_LOCAL_METRICS,
112 "metrics_semantic_progress": FLAG_SEMANTIC_PROGRESS,
113 # gco_mcp/tools/storage.py — local filesystem transfer
114 "sync_storage_bucket": FLAG_LOCAL_STORAGE_SYNC,
115 # gco_mcp/tools/mission.py — module-level gate
116 "mission_start": FLAG_MISSION,
117 "mission_status": FLAG_MISSION,
118 "mission_iterate": FLAG_MISSION,
119 "mission_checkpoint": FLAG_MISSION,
120 "mission_complete": FLAG_MISSION,
121 "mission_abort": FLAG_MISSION,
122 "mission_resume": FLAG_MISSION,
123 "mission_history": FLAG_MISSION,
124 "mission_list": FLAG_MISSION,
125}
128# ---------------------------------------------------------------------------
129# Helpers
130# ---------------------------------------------------------------------------
133def _make_not_found(message: str) -> Exception:
134 """Construct the best available "not found" exception.
136 Prefers :class:`fastmcp.exceptions.NotFoundError` because the
137 FastMCP error-handling middleware maps it to MCP error code
138 ``-32002`` for resource reads. Falls back through ``ResourceError``
139 and finally :class:`KeyError` so the resource layer still surfaces
140 a structured not-found regardless of the FastMCP build in use.
141 """
142 try:
143 from fastmcp.exceptions import NotFoundError
145 return NotFoundError(message)
146 except ImportError:
147 pass
148 try:
149 from fastmcp.exceptions import ResourceError
151 return ResourceError(message)
152 except ImportError:
153 pass
154 return KeyError(message)
157def _source_info_for_fn(fn: Any) -> tuple[str | None, int | None]:
158 """Return (project-root-relative path, 1-indexed first line) for ``fn``.
160 Walks :func:`inspect.unwrap` so the source location of the wrapped
161 function is reported rather than the audit decorator's wrapper.
162 Both halves can be ``None`` when the source is unavailable (built-
163 ins, dynamically generated functions); the index handler emits
164 ``null`` JSON for those cases.
165 """
166 try:
167 target = inspect.unwrap(fn)
168 except Exception:
169 target = fn
171 try:
172 src_path = inspect.getsourcefile(target)
173 except TypeError, OSError:
174 src_path = None
176 # Resolve the first line number. Prefer the code object's
177 # ``co_firstlineno``: it is always present for real Python functions
178 # and lambdas and equals what ``inspect.getsourcelines`` reports for
179 # them, but (unlike ``getsourcelines``) it needs no re-read of the
180 # on-disk source. That robustness matters under pytest's assertion-
181 # rewriting import hook, where re-reading the source of a function
182 # defined in a rewritten module raises ``OSError`` on newer CPython
183 # 3.14 patch releases and would otherwise drop the line to ``None``.
184 # Fall back to ``getsourcelines`` for the rare callable that exposes
185 # a source location but no ``__code__``, and to ``None`` for built-ins.
186 src_lineno: int | None = None
187 code = getattr(target, "__code__", None)
188 co_firstlineno = getattr(code, "co_firstlineno", None)
189 if isinstance(co_firstlineno, int):
190 src_lineno = co_firstlineno
191 else:
192 try:
193 _src_lines, src_lineno = inspect.getsourcelines(target)
194 except TypeError, OSError:
195 src_lineno = None
197 rel_path: str | None = None
198 if src_path:
199 try:
200 rel_path = str(Path(src_path).resolve().relative_to(_PROJECT_ROOT))
201 except ValueError:
202 # Tool defined outside the project tree (e.g. site-
203 # packages). Fall back to the absolute path.
204 rel_path = src_path
206 return rel_path, src_lineno
209async def _list_tools_async() -> list[Any]:
210 """Snapshot every registered tool, asynchronously.
212 The catch-all keeps a transient FastMCP error from blowing up the
213 introspection endpoint — an empty list is safer than a 500.
214 """
215 from server import mcp
217 try:
218 # ``_list_tools`` returns a ``Sequence[Tool]``; widen to
219 # ``list[Any]`` for the JSON-projection helpers below.
220 return list(await mcp._list_tools())
221 except Exception:
222 return []
225async def _list_resources_async() -> tuple[list[Any], list[Any]]:
226 """Snapshot static resources and resource templates, asynchronously."""
227 from server import mcp
229 try:
230 # ``_list_resources`` and ``_list_resource_templates`` return
231 # ``Sequence[Resource]`` and ``Sequence[ResourceTemplate]``
232 # respectively; widen to ``list[Any]`` so the JSON-projection
233 # helpers don't have to know FastMCP's concrete classes.
234 resources = list(await mcp._list_resources())
235 except Exception:
236 resources = []
237 try:
238 templates = list(await mcp._list_resource_templates())
239 except Exception:
240 templates = []
241 return resources, templates
244def _tool_to_dict(tool: Any) -> dict[str, Any]:
245 """Build the index entry shape from a FastMCP tool object."""
246 src_path, src_line = _source_info_for_fn(getattr(tool, "fn", None))
247 tags = getattr(tool, "tags", None) or set()
248 return {
249 "name": tool.name,
250 "description": getattr(tool, "description", "") or "",
251 "tags": sorted(str(t) for t in tags),
252 "source_path": src_path,
253 "source_line": src_line,
254 "gating_flag": _TOOL_GATING_TABLE.get(tool.name),
255 }
258def _resource_to_dict(resource: Any) -> dict[str, Any]:
259 """Build the index entry shape from a FastMCP static resource."""
260 src_path, src_line = _source_info_for_fn(getattr(resource, "fn", None))
261 tags = getattr(resource, "tags", None) or set()
262 return {
263 "uri": str(getattr(resource, "uri", "")),
264 "name": getattr(resource, "name", "") or "",
265 "description": getattr(resource, "description", "") or "",
266 "tags": sorted(str(t) for t in tags),
267 "source_path": src_path,
268 "source_line": src_line,
269 }
272def _template_to_dict(template: Any) -> dict[str, Any]:
273 """Build the index entry shape from a FastMCP resource template."""
274 src_path, src_line = _source_info_for_fn(getattr(template, "fn", None))
275 tags = getattr(template, "tags", None) or set()
276 return {
277 "uri_template": getattr(template, "uri_template", "") or "",
278 "name": getattr(template, "name", "") or "",
279 "description": getattr(template, "description", "") or "",
280 "tags": sorted(str(t) for t in tags),
281 "source_path": src_path,
282 "source_line": src_line,
283 }
286# ---------------------------------------------------------------------------
287# Resource handler bodies
288# ---------------------------------------------------------------------------
291async def _tools_index() -> str:
292 """Return the full tool index as a JSON string."""
293 tools = await _list_tools_async()
294 payload = {"tools": [_tool_to_dict(t) for t in tools]}
295 return json.dumps(payload, default=str)
298async def _tool_detail(tool_name: str) -> str:
299 """Return one tool's detail dict as JSON, or raise not-found."""
300 for tool in await _list_tools_async():
301 if tool.name == tool_name:
302 return json.dumps(_tool_to_dict(tool), default=str)
303 raise _make_not_found(f"tool {tool_name!r} is not registered")
306async def _resources_index() -> str:
307 """Return the full resource + template index as a JSON string."""
308 resources, templates = await _list_resources_async()
309 payload = {
310 "resources": [_resource_to_dict(r) for r in resources],
311 "templates": [_template_to_dict(t) for t in templates],
312 }
313 return json.dumps(payload, default=str)
316async def _feature_flags() -> str:
317 """Return the feature-flag table as a JSON string.
319 Each entry carries the flag's name, its always-False default
320 (gates default off until the operator opts in), and the list of
321 tool names the flag gates. The umbrella flag ``GCO_ENABLE_ALL_TOOLS``
322 appears with an empty ``gated_tools`` list because it overrides
323 every per-tool flag.
324 """
325 by_flag: dict[str, list[str]] = {flag: [] for flag in ALL_FLAGS}
326 for tool_name, flag in _TOOL_GATING_TABLE.items():
327 if flag in by_flag: 327 ↛ 326line 327 didn't jump to line 326 because the condition on line 327 was always true
328 by_flag[flag].append(tool_name)
330 flags_list: list[dict[str, Any]] = [
331 {
332 "name": FLAG_ALL_TOOLS,
333 "default": False,
334 "gated_tools": [],
335 }
336 ]
337 for flag in ALL_FLAGS:
338 flags_list.append(
339 {
340 "name": flag,
341 "default": False,
342 "gated_tools": sorted(by_flag.get(flag, [])),
343 }
344 )
346 return json.dumps({"flags": flags_list}, default=str)
349# ---------------------------------------------------------------------------
350# Registration
351# ---------------------------------------------------------------------------
354def register(mcp_instance: Any) -> None:
355 """Register the four self-indexing resource handlers.
357 Always-on. The handlers are pure functions of the live FastMCP
358 registry plus the static gating table above, so registering them
359 on import has no side effects beyond exposing the URIs.
360 """
361 mcp_instance.resource("mcp://gco/tools/index")(_tools_index)
362 mcp_instance.resource("mcp://gco/tools/{tool_name}")(_tool_detail)
363 mcp_instance.resource("mcp://gco/resources/index")(_resources_index)
364 mcp_instance.resource("mcp://gco/feature-flags")(_feature_flags)
367# Make the helpers reachable for tests without importing the
368# private leading-underscore symbols. The handler functions stay
369# private because they're driven through FastMCP's resource layer.
370__all__ = [
371 "register",
372]
375# Auto-cast helper: keep mypy quiet about ``Any`` returns in the
376# resource bodies (FastMCP's resource decorator types ``fn`` as
377# ``Callable[..., str | bytes | dict | list]``). Cast at the call
378# site rather than wrapping every helper in a string-only signature.
379cast # noqa: B018 - re-exported only to keep ``cast`` imported