Coverage for gco_mcp / completions.py: 100.00%

1"""Argument completion for GCO resource templates (FastMCP 4). 

2 

3FastMCP 4 lets a server answer MCP ``completion/complete`` requests through a 

4single handler registered with ``mcp.add_completion_handler``. Registering the 

5handler is also what advertises the ``completions`` capability during 

6negotiation, so clients only send completion requests when this server can 

7answer them. 

8 

9Scope is deliberately limited to the static, registry-backed template 

10parameters — documentation names, example manifests, ADR ids, package README 

11slugs, and the project-config allowlist. Those complete from in-memory 

12metadata the resource modules already maintain, so a completion request never 

13costs an AWS round-trip. Live-state templates (``gco://``, ``tasks://``, 

14``mission://``, ``costs://``, ``images://``) would need per-keystroke network 

15calls and deliberately return no suggestions. 

16""" 

17 

18from __future__ import annotations 

19 

20from collections.abc import Callable 

21from typing import Any 

22 

23# The MCP protocol caps one completion response at 100 values; FastMCP 

24# truncates anyway, but capping here keeps the payload deterministic. 

25_MAX_COMPLETIONS = 100 

26 

27 

28def _doc_names() -> list[str]: 

29 from resources.docs import DOC_METADATA 

30 

31 return sorted(DOC_METADATA) 

32 

33 

34def _package_names() -> list[str]: 

35 from resources.docs import PACKAGE_DOC_METADATA 

36 

37 return sorted(PACKAGE_DOC_METADATA) 

38 

39 

40def _example_names() -> list[str]: 

41 from resources.docs import EXAMPLE_METADATA 

42 

43 return sorted(EXAMPLE_METADATA) 

44 

45 

46def _example_categories() -> list[str]: 

47 from resources.docs import EXAMPLE_METADATA 

48 

49 categories = { 

50 str(meta.get("category", "")) for meta in EXAMPLE_METADATA.values() if meta.get("category") 

51 } 

52 return sorted(categories) 

53 

54 

55def _doc_topics() -> list[str]: 

56 from resources.docs import DOC_METADATA 

57 

58 topics: set[str] = set() 

59 for meta in DOC_METADATA.values(): 

60 raw = meta.get("topics") 

61 if isinstance(raw, list): 

62 topics.update(str(topic) for topic in raw) 

63 return sorted(topics) 

64 

65 

66def _adr_ids() -> list[str]: 

67 from resources.docs import _adr_record_files 

68 

69 return [path.stem for path in _adr_record_files()] 

70 

71 

72def _config_filenames() -> list[str]: 

73 from resources.source import _CONFIG_FILES 

74 

75 return sorted(_CONFIG_FILES) 

76 

77 

78# One entry per completable template parameter: (template URI as registered, 

79# argument name) -> zero-argument provider returning the full candidate list. 

80_TEMPLATE_ARG_SOURCES: dict[tuple[str, str], Callable[[], list[str]]] = { 

81 ("docs://gco/docs/{doc_name}", "doc_name"): _doc_names, 

82 ("docs://gco/docs/by-related/{doc_name}", "doc_name"): _doc_names, 

83 ("docs://gco/docs/by-topic/{topic}", "topic"): _doc_topics, 

84 ("docs://gco/packages/{package_name}", "package_name"): _package_names, 

85 ("docs://gco/examples/{example_name}", "example_name"): _example_names, 

86 ("docs://gco/examples/by-category/{category}", "category"): _example_categories, 

87 ("docs://gco/adr/{adr_id}", "adr_id"): _adr_ids, 

88 ("source://gco/config/{filename}", "filename"): _config_filenames, 

89} 

90 

91 

92def _match(candidates: list[str], partial: str) -> list[str]: 

93 """Rank candidates for a partial value: prefix matches first, then substring.""" 

94 if not partial: 

95 return candidates[:_MAX_COMPLETIONS] 

96 lowered = partial.lower() 

97 prefix = [c for c in candidates if c.lower().startswith(lowered)] 

98 contains = [c for c in candidates if lowered in c.lower() and c not in prefix] 

99 return (prefix + contains)[:_MAX_COMPLETIONS] 

100 

101 

102async def _complete_argument(ref: Any, argument: Any, context: Any) -> list[str] | None: 

103 """Answer one ``completion/complete`` request. 

104 

105 ``ref`` is the SDK's ``PromptReference`` or ``ResourceTemplateReference``; 

106 only resource templates resolve here (this server registers no prompts). 

107 Unknown templates and arguments return ``None``, which FastMCP renders as 

108 an empty completion — never an error. 

109 """ 

110 template_uri = getattr(ref, "uri", None) 

111 arg_name = getattr(argument, "name", None) 

112 if not isinstance(template_uri, str) or not isinstance(arg_name, str): 

113 return None 

114 provider = _TEMPLATE_ARG_SOURCES.get((template_uri, arg_name)) 

115 if provider is None: 

116 return None 

117 try: 

118 candidates = provider() 

119 except Exception: # noqa: BLE001 — a completion must never break a session 

120 return None 

121 partial = getattr(argument, "value", "") or "" 

122 return _match(candidates, str(partial)) 

123 

124 

125def register_completions(mcp_instance: Any) -> None: 

126 """Register the argument-completion handler on the shared MCP server. 

127 

128 Called from ``run_mcp.py`` after every resource module has registered 

129 (the providers read registries owned by those modules). Calling it again 

130 replaces the handler, so reload-driven re-registration is idempotent. 

131 """ 

132 mcp_instance.add_completion_handler(_complete_argument)