Coverage for gco_mcp / resources / self.py: 100.00%
95 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +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 # Other destructive module-level gates
114 "delete_job": FLAG_DESTRUCTIVE_OPERATIONS,
115 "delete_inference": FLAG_DESTRUCTIVE_OPERATIONS,
116 "delete_template": FLAG_DESTRUCTIVE_OPERATIONS,
117 "delete_webhook": FLAG_DESTRUCTIVE_OPERATIONS,
118 "delete_nodepool": FLAG_DESTRUCTIVE_OPERATIONS,
119 "analytics_user_remove": FLAG_DESTRUCTIVE_OPERATIONS,
120 "monitoring_user_remove": FLAG_DESTRUCTIVE_OPERATIONS,
121 "cancel_queue_job": FLAG_DESTRUCTIVE_OPERATIONS,
122 "task_prune": FLAG_DESTRUCTIVE_OPERATIONS,
123 # Local filesystem and model-scoring readers
124 "metrics_from_local_file": FLAG_LOCAL_METRICS,
125 "metrics_semantic_progress": FLAG_SEMANTIC_PROGRESS,
126 # gco_mcp/tools/storage.py — local filesystem transfer
127 "sync_storage_bucket": FLAG_LOCAL_STORAGE_SYNC,
128 # gco_mcp/tools/mission.py — module-level gate
129 "mission_start": FLAG_MISSION,
130 "mission_status": FLAG_MISSION,
131 "mission_iterate": FLAG_MISSION,
132 "mission_checkpoint": FLAG_MISSION,
133 "mission_complete": FLAG_MISSION,
134 "mission_abort": FLAG_MISSION,
135 "mission_resume": FLAG_MISSION,
136 "mission_history": FLAG_MISSION,
137 "mission_list": FLAG_MISSION,
138 "mission_memory_search": FLAG_MISSION,
139 # gco_mcp/tools/swarm.py — swarm supervision (orchestrator-of-missions)
140 "swarm_start": FLAG_SWARM,
141 "swarm_iterate": FLAG_SWARM,
142 "swarm_status": FLAG_SWARM,
143 "swarm_abort": FLAG_SWARM,
144 "swarm_list": FLAG_SWARM,
145 "swarm_plan": FLAG_SWARM,
146}
149# ---------------------------------------------------------------------------
150# Helpers
151# ---------------------------------------------------------------------------
154def _make_not_found(message: str) -> Exception:
155 """Construct the pinned FastMCP resource-not-found exception."""
156 from fastmcp.exceptions import NotFoundError
158 return NotFoundError(message)
161def _source_info_for_fn(fn: Any) -> tuple[str | None, int | None]:
162 """Return (project-root-relative path, 1-indexed first line) for ``fn``.
164 Walks :func:`inspect.unwrap` so the source location of the wrapped
165 function is reported rather than the audit decorator's wrapper.
166 Both halves can be ``None`` when the source is unavailable (built-
167 ins, dynamically generated functions); the index handler emits
168 ``null`` JSON for those cases.
169 """
170 try:
171 target = inspect.unwrap(fn)
172 except Exception:
173 target = fn
175 try:
176 src_path = inspect.getsourcefile(target)
177 except TypeError, OSError:
178 src_path = None
180 # Resolve the first line number. Prefer the code object's
181 # ``co_firstlineno``: it is always present for real Python functions
182 # and lambdas and equals what ``inspect.getsourcelines`` reports for
183 # them, but (unlike ``getsourcelines``) it needs no re-read of the
184 # on-disk source. That robustness matters under pytest's assertion-
185 # rewriting import hook, where re-reading the source of a function
186 # defined in a rewritten module raises ``OSError`` on newer CPython
187 # 3.14 patch releases and would otherwise drop the line to ``None``.
188 # Fall back to ``getsourcelines`` for the rare callable that exposes
189 # a source location but no ``__code__``, and to ``None`` for built-ins.
190 src_lineno: int | None = None
191 code = getattr(target, "__code__", None)
192 co_firstlineno = getattr(code, "co_firstlineno", None)
193 if isinstance(co_firstlineno, int):
194 src_lineno = co_firstlineno
195 else:
196 try:
197 _src_lines, src_lineno = inspect.getsourcelines(target)
198 except TypeError, OSError:
199 src_lineno = None
201 rel_path: str | None = None
202 if src_path:
203 try:
204 rel_path = str(Path(src_path).resolve().relative_to(_PROJECT_ROOT))
205 except ValueError:
206 # Tool defined outside the project tree (e.g. site-
207 # packages). Fall back to the absolute path.
208 rel_path = src_path
210 return rel_path, src_lineno
213async def _list_tools_async() -> list[Any]:
214 """Snapshot every registered tool, asynchronously.
216 The catch-all keeps a transient FastMCP error from blowing up the
217 introspection endpoint — an empty list is safer than a 500.
218 """
219 from server import mcp
221 try:
222 # ``_list_tools`` returns a ``Sequence[Tool]``; widen to
223 # ``list[Any]`` for the JSON-projection helpers below.
224 return list(await mcp._list_tools())
225 except Exception:
226 return []
229async def _list_resources_async() -> tuple[list[Any], list[Any]]:
230 """Snapshot static resources and resource templates, asynchronously."""
231 from server import mcp
233 try:
234 # ``_list_resources`` and ``_list_resource_templates`` return
235 # ``Sequence[Resource]`` and ``Sequence[ResourceTemplate]``
236 # respectively; widen to ``list[Any]`` so the JSON-projection
237 # helpers don't have to know FastMCP's concrete classes.
238 resources = list(await mcp._list_resources())
239 except Exception:
240 resources = []
241 try:
242 templates = list(await mcp._list_resource_templates())
243 except Exception:
244 templates = []
245 return resources, templates
248def _tool_to_dict(tool: Any) -> dict[str, Any]:
249 """Build the index entry shape from a FastMCP tool object."""
250 src_path, src_line = _source_info_for_fn(getattr(tool, "fn", None))
251 tags = getattr(tool, "tags", None) or set()
252 return {
253 "name": tool.name,
254 "description": getattr(tool, "description", "") or "",
255 "tags": sorted(str(t) for t in tags),
256 "source_path": src_path,
257 "source_line": src_line,
258 "gating_flag": _TOOL_GATING_TABLE.get(tool.name),
259 }
262def _resource_to_dict(resource: Any) -> dict[str, Any]:
263 """Build the index entry shape from a FastMCP static resource."""
264 src_path, src_line = _source_info_for_fn(getattr(resource, "fn", None))
265 tags = getattr(resource, "tags", None) or set()
266 return {
267 "uri": str(getattr(resource, "uri", "")),
268 "name": getattr(resource, "name", "") or "",
269 "description": getattr(resource, "description", "") or "",
270 "tags": sorted(str(t) for t in tags),
271 "source_path": src_path,
272 "source_line": src_line,
273 }
276def _template_to_dict(template: Any) -> dict[str, Any]:
277 """Build the index entry shape from a FastMCP resource template."""
278 src_path, src_line = _source_info_for_fn(getattr(template, "fn", None))
279 tags = getattr(template, "tags", None) or set()
280 return {
281 "uri_template": getattr(template, "uri_template", "") or "",
282 "name": getattr(template, "name", "") or "",
283 "description": getattr(template, "description", "") or "",
284 "tags": sorted(str(t) for t in tags),
285 "source_path": src_path,
286 "source_line": src_line,
287 }
290# ---------------------------------------------------------------------------
291# Resource handler bodies
292# ---------------------------------------------------------------------------
295async def _tools_index() -> str:
296 """Return the full tool index as a JSON string."""
297 tools = await _list_tools_async()
298 payload = {"tools": [_tool_to_dict(t) for t in tools]}
299 return json.dumps(payload, default=str)
302async def _tool_detail(tool_name: str) -> str:
303 """Return one tool's detail dict as JSON, or raise not-found."""
304 for tool in await _list_tools_async():
305 if tool.name == tool_name:
306 return json.dumps(_tool_to_dict(tool), default=str)
307 raise _make_not_found(f"tool {tool_name!r} is not registered")
310async def _resources_index() -> str:
311 """Return the full resource + template index as a JSON string."""
312 resources, templates = await _list_resources_async()
313 payload = {
314 "resources": [_resource_to_dict(r) for r in resources],
315 "templates": [_template_to_dict(t) for t in templates],
316 }
317 return json.dumps(payload, default=str)
320async def _feature_flags() -> str:
321 """Return the feature-flag table as a JSON string.
323 Each entry carries the flag's name, its always-False default
324 (gates default off until the operator opts in), and the list of
325 tool names the flag gates. The umbrella flag ``GCO_ENABLE_ALL_TOOLS``
326 appears with an empty ``gated_tools`` list because it overrides
327 every per-tool flag.
328 """
329 by_flag: dict[str, list[str]] = {flag: [] for flag in ALL_FLAGS}
330 for tool_name, flag in _TOOL_GATING_TABLE.items():
331 # The table is executable registry metadata: drift must fail loudly
332 # rather than silently omitting a gated tool from introspection.
333 by_flag[flag].append(tool_name)
335 flags_list: list[dict[str, Any]] = [
336 {
337 "name": FLAG_ALL_TOOLS,
338 "default": False,
339 "gated_tools": [],
340 }
341 ]
342 for flag in ALL_FLAGS:
343 flags_list.append(
344 {
345 "name": flag,
346 "default": False,
347 "gated_tools": sorted(by_flag.get(flag, [])),
348 }
349 )
351 return json.dumps({"flags": flags_list}, default=str)
354# ---------------------------------------------------------------------------
355# Registration
356# ---------------------------------------------------------------------------
359def register(mcp_instance: Any) -> None:
360 """Register the four self-indexing resource handlers.
362 Always-on. The handlers are pure functions of the live FastMCP
363 registry plus the static gating table above, so registering them
364 on import has no side effects beyond exposing the URIs.
365 """
366 mcp_instance.resource("mcp://gco/tools/index")(_tools_index)
367 mcp_instance.resource("mcp://gco/tools/{tool_name}")(_tool_detail)
368 mcp_instance.resource("mcp://gco/resources/index")(_resources_index)
369 mcp_instance.resource("mcp://gco/feature-flags")(_feature_flags)
372# Make the helpers reachable for tests without importing the
373# private leading-underscore symbols. The handler functions stay
374# private because they're driven through FastMCP's resource layer.
375__all__ = [
376 "register",
377]
380# Auto-cast helper: keep mypy quiet about ``Any`` returns in the
381# resource bodies (FastMCP's resource decorator types ``fn`` as
382# ``Callable[..., str | bytes | dict | list]``). Cast at the call
383# site rather than wrapping every helper in a string-only signature.
384cast # noqa: B018 - re-exported only to keep ``cast`` imported