Coverage for mcp/resources/self.py: 88%
100 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-15 15: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 ``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 ``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_MISSION,
58 FLAG_MODEL_UPLOAD,
59)
61# Import the live FastMCP instance so the resource handlers can hit
62# the same registry the rest of the server sees.
63sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
65# Project root used to build relative ``source_path`` strings.
66_PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
69# ---------------------------------------------------------------------------
70# Static gating table — kept in sync with the per-module ``if`` blocks.
71# ---------------------------------------------------------------------------
73_TOOL_GATING_TABLE: dict[str, str] = {
74 # mcp/tools/capacity.py — module-level gate for purchase tools
75 "reserve_capacity": FLAG_CAPACITY_PURCHASE,
76 # mcp/tools/models.py — model upload + destructive
77 "models_upload": FLAG_MODEL_UPLOAD,
78 "delete_model": FLAG_DESTRUCTIVE_OPERATIONS,
79 # mcp/tools/images.py — image-publish + destructive
80 "images_build": FLAG_IMAGE_PUBLISH,
81 "images_push": FLAG_IMAGE_PUBLISH,
82 "images_delete_tag": FLAG_DESTRUCTIVE_OPERATIONS,
83 "images_delete_repo": FLAG_DESTRUCTIVE_OPERATIONS,
84 "images_cleanup": FLAG_DESTRUCTIVE_OPERATIONS,
85 "images_prune": FLAG_DESTRUCTIVE_OPERATIONS,
86 # mcp/tools/stacks.py — deploy + destroy
87 "deploy_stack": FLAG_INFRASTRUCTURE_DEPLOY,
88 "deploy_all": FLAG_INFRASTRUCTURE_DEPLOY,
89 "bootstrap_cdk": FLAG_INFRASTRUCTURE_DEPLOY,
90 "destroy_stack": FLAG_INFRASTRUCTURE_DESTROY,
91 "destroy_all": FLAG_INFRASTRUCTURE_DESTROY,
92 # mcp/tools/mission.py — module-level gate
93 "mission_start": FLAG_MISSION,
94 "mission_status": FLAG_MISSION,
95 "mission_iterate": FLAG_MISSION,
96 "mission_checkpoint": FLAG_MISSION,
97 "mission_complete": FLAG_MISSION,
98 "mission_abort": FLAG_MISSION,
99 "mission_resume": FLAG_MISSION,
100 "mission_history": FLAG_MISSION,
101 "mission_list": FLAG_MISSION,
102}
105# ---------------------------------------------------------------------------
106# Helpers
107# ---------------------------------------------------------------------------
110def _make_not_found(message: str) -> Exception:
111 """Construct the best available "not found" exception.
113 Prefers :class:`fastmcp.exceptions.NotFoundError` because the
114 FastMCP error-handling middleware maps it to MCP error code
115 ``-32002`` for resource reads. Falls back through ``ResourceError``
116 and finally :class:`KeyError` so the resource layer still surfaces
117 a structured not-found regardless of the FastMCP build in use.
118 """
119 try:
120 from fastmcp.exceptions import NotFoundError
122 return NotFoundError(message)
123 except ImportError:
124 pass
125 try:
126 from fastmcp.exceptions import ResourceError
128 return ResourceError(message)
129 except ImportError:
130 pass
131 return KeyError(message)
134def _source_info_for_fn(fn: Any) -> tuple[str | None, int | None]:
135 """Return (project-root-relative path, 1-indexed first line) for ``fn``.
137 Walks :func:`inspect.unwrap` so the source location of the wrapped
138 function is reported rather than the audit decorator's wrapper.
139 Both halves can be ``None`` when the source is unavailable (built-
140 ins, dynamically generated functions); the index handler emits
141 ``null`` JSON for those cases.
142 """
143 try:
144 target = inspect.unwrap(fn)
145 except Exception:
146 target = fn
148 try:
149 src_path = inspect.getsourcefile(target)
150 except TypeError, OSError:
151 src_path = None
152 try:
153 _src_lines, src_lineno = inspect.getsourcelines(target)
154 except TypeError, OSError:
155 src_lineno = None
157 rel_path: str | None = None
158 if src_path:
159 try:
160 rel_path = str(Path(src_path).resolve().relative_to(_PROJECT_ROOT))
161 except ValueError:
162 # Tool defined outside the project tree (e.g. site-
163 # packages). Fall back to the absolute path.
164 rel_path = src_path
166 return rel_path, src_lineno
169async def _list_tools_async() -> list[Any]:
170 """Snapshot every registered tool, asynchronously.
172 The catch-all keeps a transient FastMCP error from blowing up the
173 introspection endpoint — an empty list is safer than a 500.
174 """
175 from server import mcp
177 try:
178 # ``_list_tools`` returns a ``Sequence[Tool]``; widen to
179 # ``list[Any]`` for the JSON-projection helpers below.
180 return list(await mcp._list_tools())
181 except Exception:
182 return []
185async def _list_resources_async() -> tuple[list[Any], list[Any]]:
186 """Snapshot static resources and resource templates, asynchronously."""
187 from server import mcp
189 try:
190 # ``_list_resources`` and ``_list_resource_templates`` return
191 # ``Sequence[Resource]`` and ``Sequence[ResourceTemplate]``
192 # respectively; widen to ``list[Any]`` so the JSON-projection
193 # helpers don't have to know FastMCP's concrete classes.
194 resources = list(await mcp._list_resources())
195 except Exception:
196 resources = []
197 try:
198 templates = list(await mcp._list_resource_templates())
199 except Exception:
200 templates = []
201 return resources, templates
204def _tool_to_dict(tool: Any) -> dict[str, Any]:
205 """Build the index entry shape from a FastMCP tool object."""
206 src_path, src_line = _source_info_for_fn(getattr(tool, "fn", None))
207 tags = getattr(tool, "tags", None) or set()
208 return {
209 "name": tool.name,
210 "description": getattr(tool, "description", "") or "",
211 "tags": sorted(str(t) for t in tags),
212 "source_path": src_path,
213 "source_line": src_line,
214 "gating_flag": _TOOL_GATING_TABLE.get(tool.name),
215 }
218def _resource_to_dict(resource: Any) -> dict[str, Any]:
219 """Build the index entry shape from a FastMCP static resource."""
220 src_path, src_line = _source_info_for_fn(getattr(resource, "fn", None))
221 tags = getattr(resource, "tags", None) or set()
222 return {
223 "uri": str(getattr(resource, "uri", "")),
224 "name": getattr(resource, "name", "") or "",
225 "description": getattr(resource, "description", "") or "",
226 "tags": sorted(str(t) for t in tags),
227 "source_path": src_path,
228 "source_line": src_line,
229 }
232def _template_to_dict(template: Any) -> dict[str, Any]:
233 """Build the index entry shape from a FastMCP resource template."""
234 src_path, src_line = _source_info_for_fn(getattr(template, "fn", None))
235 tags = getattr(template, "tags", None) or set()
236 return {
237 "uri_template": getattr(template, "uri_template", "") or "",
238 "name": getattr(template, "name", "") or "",
239 "description": getattr(template, "description", "") or "",
240 "tags": sorted(str(t) for t in tags),
241 "source_path": src_path,
242 "source_line": src_line,
243 }
246# ---------------------------------------------------------------------------
247# Resource handler bodies
248# ---------------------------------------------------------------------------
251async def _tools_index() -> str:
252 """Return the full tool index as a JSON string."""
253 tools = await _list_tools_async()
254 payload = {"tools": [_tool_to_dict(t) for t in tools]}
255 return json.dumps(payload, default=str)
258async def _tool_detail(tool_name: str) -> str:
259 """Return one tool's detail dict as JSON, or raise not-found."""
260 for tool in await _list_tools_async():
261 if tool.name == tool_name:
262 return json.dumps(_tool_to_dict(tool), default=str)
263 raise _make_not_found(f"tool {tool_name!r} is not registered")
266async def _resources_index() -> str:
267 """Return the full resource + template index as a JSON string."""
268 resources, templates = await _list_resources_async()
269 payload = {
270 "resources": [_resource_to_dict(r) for r in resources],
271 "templates": [_template_to_dict(t) for t in templates],
272 }
273 return json.dumps(payload, default=str)
276async def _feature_flags() -> str:
277 """Return the feature-flag table as a JSON string.
279 Each entry carries the flag's name, its always-False default
280 (gates default off until the operator opts in), and the list of
281 tool names the flag gates. The umbrella flag ``GCO_ENABLE_ALL_TOOLS``
282 appears with an empty ``gated_tools`` list because it overrides
283 every per-tool flag.
284 """
285 by_flag: dict[str, list[str]] = {flag: [] for flag in ALL_FLAGS}
286 for tool_name, flag in _TOOL_GATING_TABLE.items():
287 if flag in by_flag: 287 ↛ 286line 287 didn't jump to line 286 because the condition on line 287 was always true
288 by_flag[flag].append(tool_name)
290 flags_list: list[dict[str, Any]] = [
291 {
292 "name": FLAG_ALL_TOOLS,
293 "default": False,
294 "gated_tools": [],
295 }
296 ]
297 for flag in ALL_FLAGS:
298 flags_list.append(
299 {
300 "name": flag,
301 "default": False,
302 "gated_tools": sorted(by_flag.get(flag, [])),
303 }
304 )
306 return json.dumps({"flags": flags_list}, default=str)
309# ---------------------------------------------------------------------------
310# Registration
311# ---------------------------------------------------------------------------
314def register(mcp_instance: Any) -> None:
315 """Register the four self-indexing resource handlers.
317 Always-on. The handlers are pure functions of the live FastMCP
318 registry plus the static gating table above, so registering them
319 on import has no side effects beyond exposing the URIs.
320 """
321 mcp_instance.resource("mcp://gco/tools/index")(_tools_index)
322 mcp_instance.resource("mcp://gco/tools/{tool_name}")(_tool_detail)
323 mcp_instance.resource("mcp://gco/resources/index")(_resources_index)
324 mcp_instance.resource("mcp://gco/feature-flags")(_feature_flags)
327# Make the helpers reachable for tests without importing the
328# private leading-underscore symbols. The handler functions stay
329# private because they're driven through FastMCP's resource layer.
330__all__ = [
331 "register",
332]
335# Auto-cast helper: keep mypy quiet about ``Any`` returns in the
336# resource bodies (FastMCP's resource decorator types ``fn`` as
337# ``Callable[..., str | bytes | dict | list]``). Cast at the call
338# site rather than wrapping every helper in a string-only signature.
339cast # noqa: B018 - re-exported only to keep ``cast`` imported