Coverage for src/secchi/mcp_server.py: 82%
82 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-04 22:15 +0000
1"""Model Context Protocol server for Secchi package intelligence."""
3from __future__ import annotations
5import json
6import logging
7from typing import Any
9import httpx
10from mcp.server.mcpserver import MCPServer
12from secchi import __version__
13from secchi.config import find_config
14from secchi.errors import SecchiError
15from secchi.export import export_package_json
16from secchi.models import PackageRef
17from secchi.services.intelligence import IntelligenceResult, PackageIntelligenceService
18from secchi.services.resolver import parse_package_spec, resolve_package
19from secchi.workflows import check as check_workflow
20from secchi.workflows import compare as compare_workflow
21from secchi.workflows import inspect as inspect_workflow
22from secchi.workflows import report as report_workflow
23from secchi.workflows import search as search_workflow
25server = MCPServer(
26 name="secchi",
27 title="Secchi Package Intelligence",
28 description=(
29 "Explore package health, adoption, dependencies, releases, and "
30 "ecosystem signals across supported registries."
31 ),
32 version=__version__,
33)
34_intelligence_service = PackageIntelligenceService()
35logger = logging.getLogger(__name__)
38async def _resolve_refs(package: str, registry: str | None) -> list[PackageRef]:
39 if registry is not None:
40 return [parse_package_spec(package, registry)]
41 return await resolve_package(package)
44def _package_result(result: IntelligenceResult) -> dict[str, Any]:
45 if result.error or result.info is None or result.derived is None:
46 return {
47 "package": result.ref.name,
48 "registry": result.ref.registry.value,
49 "error": result.error.message
50 if result.error
51 else "No package data returned.",
52 }
53 return json.loads(
54 export_package_json(
55 result.info,
56 result.derived,
57 result.ref,
58 result.ref.project_name,
59 result.warnings,
60 )
61 )
64@server.tool(
65 name="inspect_package",
66 title="Inspect package intelligence",
67 description=(
68 "Fetch normalized package intelligence. Without a registry, resolve "
69 "exact matches across supported ecosystems; with a registry, inspect "
70 "that specific package source."
71 ),
72 structured_output=True,
73)
74async def inspect_package(
75 package: str,
76 registry: str | None = None,
77 refresh: bool = False,
78) -> dict[str, Any]:
79 """Return health, adoption, release, dependency, and repository signals."""
80 refs = await _resolve_refs(package, registry)
81 if not refs:
82 return {
83 "query": package,
84 "matches": [],
85 "message": "No exact package matches found.",
86 }
87 intelligence = await inspect_workflow.run(
88 refs, refresh=refresh, service=_intelligence_service
89 )
90 return {
91 "query": package,
92 "matches": [
93 _package_result(result) for result in intelligence.results.values()
94 ],
95 }
98@server.tool(
99 name="search_packages",
100 title="Search package ecosystems",
101 description="Search supported package registries and return ranked normalized matches.",
102 structured_output=True,
103)
104async def search_packages(
105 query: str,
106 registry: str | None = None,
107 limit: int = 10,
108) -> dict[str, Any]:
109 """Search one registry or all supported registries."""
110 if limit < 1 or limit > 50:
111 raise ValueError("limit must be between 1 and 50")
112 results = await search_workflow.run(query, registry=registry, limit=limit)
113 return {
114 "query": query,
115 "results": [
116 {
117 "name": result.name,
118 "registry": result.registry.value,
119 "version": result.version,
120 "description": result.description,
121 "score": result.score,
122 "exact": result.exact,
123 }
124 for result in results
125 ],
126 }
129@server.tool(
130 name="inspect_project",
131 title="Inspect configured project",
132 description="Load one project from secchi.toml and return its combined registry report.",
133 structured_output=True,
134)
135async def inspect_project(
136 project: str,
137 config: str | None = None,
138 refresh: bool = False,
139) -> dict[str, Any]:
140 """Return project-wide intelligence using the same report pipeline as the CLI."""
141 config_path = find_config(config)
142 if config_path is None:
143 raise ValueError(
144 "No Secchi config found. Provide config or create secchi.toml."
145 )
146 output = await report_workflow.run(
147 project_name=project,
148 config=str(config_path),
149 format_name="json",
150 output="-",
151 refresh=refresh,
152 service=_intelligence_service,
153 )
154 return json.loads(output.content)
157@server.tool(
158 name="check_package",
159 title="Evaluate package policy",
160 description="Evaluate a package against minimum health and repository CI policies.",
161 structured_output=True,
162)
163async def check_package(
164 package: str,
165 registry: str | None = None,
166 min_health: int = 70,
167 require_ci: bool = False,
168 refresh: bool = False,
169) -> dict[str, Any]:
170 """Return policy results without terminating the MCP server on failure."""
171 if min_health < 0 or min_health > 100:
172 raise ValueError("min_health must be between 0 and 100")
173 refs = await _resolve_refs(package, registry)
174 if not refs:
175 return {
176 "query": package,
177 "matches": [],
178 "message": "No exact package matches found.",
179 }
181 matches: list[dict[str, Any]] = []
182 for ref in refs:
183 item: dict[str, Any] = {
184 "package": ref.name,
185 "registry": ref.registry.value,
186 }
187 try:
188 check_result = await check_workflow.run(
189 ref,
190 min_health=min_health,
191 require_ci=require_ci,
192 refresh=refresh,
193 service=_intelligence_service,
194 )
195 item["passed"] = check_result.passed
196 item["checks"] = [
197 {"name": check.name, "passed": check.passed, "detail": check.detail}
198 for check in check_result.checks
199 ]
200 item["warnings"] = [
201 {"source": warning.source, "message": warning.message}
202 for warning in check_result.warnings
203 ]
204 except (
205 SecchiError,
206 httpx.HTTPError,
207 OSError,
208 ValueError,
209 KeyError,
210 TypeError,
211 ) as exc:
212 item["passed"] = False
213 item["error"] = str(exc)
214 logger.debug(
215 "MCP policy check failed for %s (%s): %s",
216 ref.name,
217 ref.registry.value,
218 exc,
219 exc_info=True,
220 )
221 matches.append(item)
222 return {"query": package, "matches": matches}
225@server.tool(
226 name="compare_packages",
227 title="Compare package choices",
228 description=(
229 "Compare two or more package choices using health, adoption momentum, "
230 "community, release recency, and data completeness. Returns advisory "
231 "recommendations with evidence and confidence; it never installs packages."
232 ),
233 structured_output=True,
234)
235async def compare_packages(
236 packages: list[str],
237 registry: str | None = None,
238 refresh: bool = False,
239) -> dict[str, Any]:
240 """Return a ranked, evidence-backed package selection recommendation."""
241 if len(packages) < 2:
242 raise ValueError("packages must contain at least two package references")
243 if len(packages) > 20:
244 raise ValueError("packages must contain no more than 20 package references")
245 comparison = await compare_workflow.run(
246 packages,
247 registry=registry,
248 refresh=refresh,
249 service=_intelligence_service,
250 )
251 return {"query": packages, **comparison.as_dict()}
254def main() -> None:
255 """Run the MCP server over stdio for local agent integrations."""
256 server.run(transport="stdio")
259if __name__ == "__main__":
260 main()