Coverage for gco_mcp / resources / self.py: 100.00%
95 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +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_CONFIG_MANAGEMENT,
54 FLAG_DESTRUCTIVE_OPERATIONS,
55 FLAG_IMAGE_PUBLISH,
56 FLAG_INFRASTRUCTURE_DEPLOY,
57 FLAG_INFRASTRUCTURE_DESTROY,
58 FLAG_LOCAL_METRICS,
59 FLAG_LOCAL_STORAGE_SYNC,
60 FLAG_MISSION,
61 FLAG_MODEL_UPLOAD,
62 FLAG_SEMANTIC_PROGRESS,
63 FLAG_SWARM,
64)
66# Import the live FastMCP instance so the resource handlers can hit
67# the same registry the rest of the server sees.
68sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
70# Project root used to build relative ``source_path`` strings.
71_PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
74# ---------------------------------------------------------------------------
75# Static gating table — kept in sync with the per-module ``if`` blocks.
76# ---------------------------------------------------------------------------
78_TOOL_GATING_TABLE: dict[str, str] = {
79 # gco_mcp/tools/capacity.py — purchase + destructive tools
80 "reserve_capacity": FLAG_CAPACITY_PURCHASE,
81 "create_reservation": FLAG_CAPACITY_PURCHASE,
82 "cancel_reservation": FLAG_DESTRUCTIVE_OPERATIONS,
83 # gco_mcp/tools/models.py / storage.py — local model-data upload + deletion
84 "models_upload": FLAG_MODEL_UPLOAD,
85 "upload_to_regional_bucket": FLAG_MODEL_UPLOAD,
86 "delete_model": FLAG_DESTRUCTIVE_OPERATIONS,
87 # gco_mcp/tools/images.py — image-publish + destructive
88 "images_build": FLAG_IMAGE_PUBLISH,
89 "images_push": FLAG_IMAGE_PUBLISH,
90 "images_mirror": FLAG_IMAGE_PUBLISH,
91 "images_delete_tag": FLAG_DESTRUCTIVE_OPERATIONS,
92 "images_delete_repo": FLAG_DESTRUCTIVE_OPERATIONS,
93 "images_cleanup": FLAG_DESTRUCTIVE_OPERATIONS,
94 "images_prune": FLAG_DESTRUCTIVE_OPERATIONS,
95 # gco_mcp/tools/stacks.py — deploy + destroy
96 "deploy_stack": FLAG_INFRASTRUCTURE_DEPLOY,
97 "deploy_all": FLAG_INFRASTRUCTURE_DEPLOY,
98 "bootstrap_cdk": FLAG_INFRASTRUCTURE_DEPLOY,
99 "addons_install": FLAG_INFRASTRUCTURE_DEPLOY,
100 "destroy_stack": FLAG_INFRASTRUCTURE_DESTROY,
101 "destroy_all": FLAG_INFRASTRUCTURE_DESTROY,
102 # gco_mcp/tools/stacks.py — managed deployment config
103 "list_deployment_regions": FLAG_CONFIG_MANAGEMENT,
104 "add_deployment_region": FLAG_CONFIG_MANAGEMENT,
105 "remove_deployment_region": FLAG_CONFIG_MANAGEMENT,
106 "set_deployment_region": FLAG_CONFIG_MANAGEMENT,
107 "set_eks_endpoint_access": FLAG_CONFIG_MANAGEMENT,
108 "set_mission_default_model": FLAG_CONFIG_MANAGEMENT,
109 "set_capacity_advisor_default_model": FLAG_CONFIG_MANAGEMENT,
110 "set_claude_code_default_model": FLAG_CONFIG_MANAGEMENT,
111 "set_codex_default_model": FLAG_CONFIG_MANAGEMENT,
112 "set_codex_reasoning_effort": FLAG_CONFIG_MANAGEMENT,
113 "set_opencode_default_model": FLAG_CONFIG_MANAGEMENT,
114 # Other destructive module-level gates
115 "delete_job": FLAG_DESTRUCTIVE_OPERATIONS,
116 # gco_mcp/tools/slurm.py — scancel
117 "slurm_cancel": FLAG_DESTRUCTIVE_OPERATIONS,
118 "delete_inference": FLAG_DESTRUCTIVE_OPERATIONS,
119 "delete_template": FLAG_DESTRUCTIVE_OPERATIONS,
120 "delete_webhook": FLAG_DESTRUCTIVE_OPERATIONS,
121 "delete_nodepool": FLAG_DESTRUCTIVE_OPERATIONS,
122 "analytics_user_remove": FLAG_DESTRUCTIVE_OPERATIONS,
123 "monitoring_user_remove": FLAG_DESTRUCTIVE_OPERATIONS,
124 "cancel_queue_job": FLAG_DESTRUCTIVE_OPERATIONS,
125 "task_prune": FLAG_DESTRUCTIVE_OPERATIONS,
126 # Local filesystem and model-scoring readers
127 "metrics_from_local_file": FLAG_LOCAL_METRICS,
128 "metrics_semantic_progress": FLAG_SEMANTIC_PROGRESS,
129 # gco_mcp/tools/storage.py — local filesystem transfer
130 "sync_storage_bucket": FLAG_LOCAL_STORAGE_SYNC,
131 # gco_mcp/tools/mission.py — module-level gate
132 "mission_start": FLAG_MISSION,
133 "mission_status": FLAG_MISSION,
134 "mission_iterate": FLAG_MISSION,
135 "mission_checkpoint": FLAG_MISSION,
136 "mission_complete": FLAG_MISSION,
137 "mission_abort": FLAG_MISSION,
138 "mission_resume": FLAG_MISSION,
139 "mission_history": FLAG_MISSION,
140 "mission_list": FLAG_MISSION,
141 "mission_memory_search": FLAG_MISSION,
142 # gco_mcp/tools/swarm.py — swarm supervision (orchestrator-of-missions)
143 "swarm_start": FLAG_SWARM,
144 "swarm_iterate": FLAG_SWARM,
145 "swarm_status": FLAG_SWARM,
146 "swarm_abort": FLAG_SWARM,
147 "swarm_list": FLAG_SWARM,
148 "swarm_plan": FLAG_SWARM,
149}
152# ---------------------------------------------------------------------------
153# Helpers
154# ---------------------------------------------------------------------------
157def _make_not_found(message: str) -> Exception:
158 """Construct the pinned FastMCP resource-not-found exception."""
159 from fastmcp.exceptions import NotFoundError
161 return NotFoundError(message)
164def _source_info_for_fn(fn: Any) -> tuple[str | None, int | None]:
165 """Return (project-root-relative path, 1-indexed first line) for ``fn``.
167 Walks :func:`inspect.unwrap` so the source location of the wrapped
168 function is reported rather than the audit decorator's wrapper.
169 Both halves can be ``None`` when the source is unavailable (built-
170 ins, dynamically generated functions); the index handler emits
171 ``null`` JSON for those cases.
172 """
173 try:
174 target = inspect.unwrap(fn)
175 except Exception:
176 target = fn
178 try:
179 src_path = inspect.getsourcefile(target)
180 except TypeError, OSError:
181 src_path = None
183 # Resolve the first line number. Prefer the code object's
184 # ``co_firstlineno``: it is always present for real Python functions
185 # and lambdas and equals what ``inspect.getsourcelines`` reports for
186 # them, but (unlike ``getsourcelines``) it needs no re-read of the
187 # on-disk source. That robustness matters under pytest's assertion-
188 # rewriting import hook, where re-reading the source of a function
189 # defined in a rewritten module raises ``OSError`` on newer CPython
190 # 3.14 patch releases and would otherwise drop the line to ``None``.
191 # Fall back to ``getsourcelines`` for the rare callable that exposes
192 # a source location but no ``__code__``, and to ``None`` for built-ins.
193 src_lineno: int | None = None
194 code = getattr(target, "__code__", None)
195 co_firstlineno = getattr(code, "co_firstlineno", None)
196 if isinstance(co_firstlineno, int):
197 src_lineno = co_firstlineno
198 else:
199 try:
200 _src_lines, src_lineno = inspect.getsourcelines(target)
201 except TypeError, OSError:
202 src_lineno = None
204 rel_path: str | None = None
205 if src_path:
206 try:
207 rel_path = str(Path(src_path).resolve().relative_to(_PROJECT_ROOT))
208 except ValueError:
209 # Tool defined outside the project tree (e.g. site-
210 # packages). Fall back to the absolute path.
211 rel_path = src_path
213 return rel_path, src_lineno
216async def _list_tools_async() -> list[Any]:
217 """Snapshot every registered tool, asynchronously.
219 The catch-all keeps a transient FastMCP error from blowing up the
220 introspection endpoint — an empty list is safer than a 500.
221 """
222 from server import mcp
224 try:
225 # ``_list_tools`` returns a ``Sequence[Tool]``; widen to
226 # ``list[Any]`` for the JSON-projection helpers below.
227 return list(await mcp._list_tools())
228 except Exception:
229 return []
232async def _list_resources_async() -> tuple[list[Any], list[Any]]:
233 """Snapshot static resources and resource templates, asynchronously."""
234 from server import mcp
236 try:
237 # ``_list_resources`` and ``_list_resource_templates`` return
238 # ``Sequence[Resource]`` and ``Sequence[ResourceTemplate]``
239 # respectively; widen to ``list[Any]`` so the JSON-projection
240 # helpers don't have to know FastMCP's concrete classes.
241 resources = list(await mcp._list_resources())
242 except Exception:
243 resources = []
244 try:
245 templates = list(await mcp._list_resource_templates())
246 except Exception:
247 templates = []
248 return resources, templates
251def _tool_to_dict(tool: Any) -> dict[str, Any]:
252 """Build the index entry shape from a FastMCP tool object."""
253 src_path, src_line = _source_info_for_fn(getattr(tool, "fn", None))
254 tags = getattr(tool, "tags", None) or set()
255 return {
256 "name": tool.name,
257 "description": getattr(tool, "description", "") or "",
258 "tags": sorted(str(t) for t in tags),
259 "source_path": src_path,
260 "source_line": src_line,
261 "gating_flag": _TOOL_GATING_TABLE.get(tool.name),
262 }
265def _resource_to_dict(resource: Any) -> dict[str, Any]:
266 """Build the index entry shape from a FastMCP static resource."""
267 src_path, src_line = _source_info_for_fn(getattr(resource, "fn", None))
268 tags = getattr(resource, "tags", None) or set()
269 return {
270 "uri": str(getattr(resource, "uri", "")),
271 "name": getattr(resource, "name", "") or "",
272 "description": getattr(resource, "description", "") or "",
273 "tags": sorted(str(t) for t in tags),
274 "source_path": src_path,
275 "source_line": src_line,
276 }
279def _template_to_dict(template: Any) -> dict[str, Any]:
280 """Build the index entry shape from a FastMCP resource template."""
281 src_path, src_line = _source_info_for_fn(getattr(template, "fn", None))
282 tags = getattr(template, "tags", None) or set()
283 return {
284 "uri_template": getattr(template, "uri_template", "") or "",
285 "name": getattr(template, "name", "") or "",
286 "description": getattr(template, "description", "") or "",
287 "tags": sorted(str(t) for t in tags),
288 "source_path": src_path,
289 "source_line": src_line,
290 }
293# ---------------------------------------------------------------------------
294# Resource handler bodies
295# ---------------------------------------------------------------------------
298async def _tools_index() -> str:
299 """Return the full tool index as a JSON string."""
300 tools = await _list_tools_async()
301 payload = {"tools": [_tool_to_dict(t) for t in tools]}
302 return json.dumps(payload, default=str)
305async def _tool_detail(tool_name: str) -> str:
306 """Return one tool's detail dict as JSON, or raise not-found."""
307 for tool in await _list_tools_async():
308 if tool.name == tool_name:
309 return json.dumps(_tool_to_dict(tool), default=str)
310 raise _make_not_found(f"tool {tool_name!r} is not registered")
313async def _resources_index() -> str:
314 """Return the full resource + template index as a JSON string."""
315 resources, templates = await _list_resources_async()
316 payload = {
317 "resources": [_resource_to_dict(r) for r in resources],
318 "templates": [_template_to_dict(t) for t in templates],
319 }
320 return json.dumps(payload, default=str)
323async def _feature_flags() -> str:
324 """Return the feature-flag table as a JSON string.
326 Each entry carries the flag's name, its always-False default
327 (gates default off until the operator opts in), and the list of
328 tool names the flag gates. The umbrella flag ``GCO_ENABLE_ALL_TOOLS``
329 appears with an empty ``gated_tools`` list because it overrides
330 every per-tool flag.
331 """
332 by_flag: dict[str, list[str]] = {flag: [] for flag in ALL_FLAGS}
333 for tool_name, flag in _TOOL_GATING_TABLE.items():
334 # The table is executable registry metadata: drift must fail loudly
335 # rather than silently omitting a gated tool from introspection.
336 by_flag[flag].append(tool_name)
338 flags_list: list[dict[str, Any]] = [
339 {
340 "name": FLAG_ALL_TOOLS,
341 "default": False,
342 "gated_tools": [],
343 }
344 ]
345 for flag in ALL_FLAGS:
346 flags_list.append(
347 {
348 "name": flag,
349 "default": False,
350 "gated_tools": sorted(by_flag.get(flag, [])),
351 }
352 )
354 return json.dumps({"flags": flags_list}, default=str)
357# ---------------------------------------------------------------------------
358# Registration
359# ---------------------------------------------------------------------------
362def register(mcp_instance: Any) -> None:
363 """Register the four self-indexing resource handlers.
365 Always-on. The handlers are pure functions of the live FastMCP
366 registry plus the static gating table above, so registering them
367 on import has no side effects beyond exposing the URIs.
368 """
369 mcp_instance.resource("mcp://gco/tools/index")(_tools_index)
370 mcp_instance.resource("mcp://gco/tools/{tool_name}")(_tool_detail)
371 mcp_instance.resource("mcp://gco/resources/index")(_resources_index)
372 mcp_instance.resource("mcp://gco/feature-flags")(_feature_flags)
375# Make the helpers reachable for tests without importing the
376# private leading-underscore symbols. The handler functions stay
377# private because they're driven through FastMCP's resource layer.
378__all__ = [
379 "register",
380]
383# Auto-cast helper: keep mypy quiet about ``Any`` returns in the
384# resource bodies (FastMCP's resource decorator types ``fn`` as
385# ``Callable[..., str | bytes | dict | list]``). Cast at the call
386# site rather than wrapping every helper in a string-only signature.
387cast # noqa: B018 - re-exported only to keep ``cast`` imported