Coverage for src/qdrant_loader_mcp_server/fastmcp_tools/intelligence.py: 55%
33 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-20 10:15 +0000
1"""FastMCP cross-document intelligence tools.
3These delegate to the existing IntelligenceHandler (reusing its
4result-normalization logic) and unwrap the structured payload, rather than
5re-implementing it.
7The handler is stateful (its cluster store is shared with
8expand_cluster), so a single instance is built in the lifespan and reused.
9"""
11from __future__ import annotations
13from typing import Annotated, Any, Literal
15from fastmcp import Context, FastMCP
16from pydantic import Field
18from ._common import REQUEST_ID, unwrap
20SimilarityMetric = Literal[
21 "entity_overlap",
22 "topic_overlap",
23 "semantic_similarity",
24 "metadata_similarity",
25 "hierarchical_distance",
26 "content_features",
27]
28ClusterStrategy = Literal[
29 "mixed_features",
30 "entity_based",
31 "topic_based",
32 "project_based",
33 "hierarchical",
34 "adaptive",
35]
38def register_intelligence_tools(mcp: FastMCP) -> None:
39 @mcp.tool(annotations={"readOnlyHint": True})
40 async def analyze_relationships(
41 ctx: Context,
42 query: Annotated[
43 str,
44 Field(
45 description="Search query to get documents for analysis", min_length=1
46 ),
47 ],
48 limit: Annotated[
49 int,
50 Field(description="Maximum number of documents to analyze", ge=1, le=1000),
51 ] = 20,
52 source_types: Annotated[
53 list[str] | None,
54 Field(description="Optional list of source types to filter by"),
55 ] = None,
56 project_ids: Annotated[
57 list[str] | None,
58 Field(description="Optional list of project IDs to filter by"),
59 ] = None,
60 use_llm: Annotated[
61 bool, Field(description="Enable LLM validation for top pairs (budgeted)")
62 ] = False,
63 max_llm_pairs: Annotated[
64 int,
65 Field(
66 description="Maximum number of pairs to analyze with LLM", ge=0, le=100
67 ),
68 ] = 5,
69 overall_timeout_s: Annotated[
70 float,
71 Field(description="Overall analysis budget in seconds", ge=0, le=3600),
72 ] = 60,
73 max_pairs_total: Annotated[
74 int,
75 Field(
76 description="Maximum candidate pairs to analyze after tiering",
77 ge=0,
78 le=100000,
79 ),
80 ] = 1000,
81 text_window_chars: Annotated[
82 int,
83 Field(
84 description="Per-document text window size for lexical analysis",
85 ge=0,
86 le=10000,
87 ),
88 ] = 1000,
89 ) -> dict[str, Any]:
90 """Analyze relationships between documents."""
91 handler = ctx.lifespan_context["intelligence_handler"]
92 params = {
93 "query": query,
94 "limit": limit,
95 "source_types": source_types,
96 "project_ids": project_ids,
97 "use_llm": use_llm,
98 "max_llm_pairs": max_llm_pairs,
99 "overall_timeout_s": overall_timeout_s,
100 "max_pairs_total": max_pairs_total,
101 "text_window_chars": text_window_chars,
102 }
103 return unwrap(
104 await handler.handle_analyze_document_relationships(REQUEST_ID, params)
105 )
107 @mcp.tool(annotations={"readOnlyHint": True})
108 async def find_similar_documents(
109 ctx: Context,
110 target_query: Annotated[
111 str, Field(description="Query to find the target document", min_length=1)
112 ],
113 comparison_query: Annotated[
114 str,
115 Field(
116 description="Query to get documents to compare against", min_length=1
117 ),
118 ],
119 similarity_metrics: Annotated[
120 list[SimilarityMetric] | None,
121 Field(description="Similarity metrics to use"),
122 ] = None,
123 max_similar: Annotated[
124 int, Field(description="Maximum number of similar documents to return")
125 ] = 5,
126 source_types: Annotated[
127 list[str] | None,
128 Field(description="Optional list of source types to filter by"),
129 ] = None,
130 project_ids: Annotated[
131 list[str] | None,
132 Field(description="Optional list of project IDs to filter by"),
133 ] = None,
134 ) -> dict[str, Any]:
135 """Find documents similar to a target document using multiple similarity metrics."""
136 handler = ctx.lifespan_context["intelligence_handler"]
137 params = {
138 "target_query": target_query,
139 "comparison_query": comparison_query,
140 "similarity_metrics": similarity_metrics,
141 "max_similar": max_similar,
142 "source_types": source_types,
143 "project_ids": project_ids,
144 }
145 return unwrap(await handler.handle_find_similar_documents(REQUEST_ID, params))
147 @mcp.tool(annotations={"readOnlyHint": True})
148 async def detect_document_conflicts(
149 ctx: Context,
150 query: Annotated[
151 str,
152 Field(
153 description="Search query to get documents for conflict analysis",
154 min_length=1,
155 ),
156 ],
157 limit: Annotated[
158 int, Field(description="Maximum number of documents to analyze")
159 ] = 10,
160 source_types: Annotated[
161 list[str] | None,
162 Field(description="Optional list of source types to filter by"),
163 ] = None,
164 project_ids: Annotated[
165 list[str] | None,
166 Field(description="Optional list of project IDs to filter by"),
167 ] = None,
168 ) -> dict[str, Any]:
169 """Detect conflicts and contradictions between documents."""
170 handler = ctx.lifespan_context["intelligence_handler"]
171 params = {
172 "query": query,
173 "limit": limit,
174 "source_types": source_types,
175 "project_ids": project_ids,
176 }
177 return unwrap(
178 await handler.handle_detect_document_conflicts(REQUEST_ID, params)
179 )
181 @mcp.tool(annotations={"readOnlyHint": True})
182 async def find_complementary_content(
183 ctx: Context,
184 target_query: Annotated[
185 str, Field(description="Query to find the target document", min_length=1)
186 ],
187 context_query: Annotated[
188 str, Field(description="Query to get contextual documents", min_length=1)
189 ],
190 max_recommendations: Annotated[
191 int, Field(description="Maximum number of recommendations")
192 ] = 5,
193 source_types: Annotated[
194 list[str] | None,
195 Field(description="Optional list of source types to filter by"),
196 ] = None,
197 project_ids: Annotated[
198 list[str] | None,
199 Field(description="Optional list of project IDs to filter by"),
200 ] = None,
201 ) -> dict[str, Any]:
202 """Find content that complements a target document."""
203 handler = ctx.lifespan_context["intelligence_handler"]
204 params = {
205 "target_query": target_query,
206 "context_query": context_query,
207 "max_recommendations": max_recommendations,
208 "source_types": source_types,
209 "project_ids": project_ids,
210 }
211 return unwrap(
212 await handler.handle_find_complementary_content(REQUEST_ID, params)
213 )
215 @mcp.tool(annotations={"readOnlyHint": True})
216 async def cluster_documents(
217 ctx: Context,
218 query: Annotated[
219 str,
220 Field(
221 description="Search query to get documents for clustering", min_length=1
222 ),
223 ],
224 strategy: Annotated[
225 ClusterStrategy,
226 Field(
227 description="Clustering strategy (adaptive auto-selects the best one)"
228 ),
229 ] = "mixed_features",
230 max_clusters: Annotated[
231 int,
232 Field(description="Maximum number of clusters to create", ge=1, le=1000),
233 ] = 10,
234 min_cluster_size: Annotated[
235 int, Field(description="Minimum size for a cluster", ge=1, le=1000)
236 ] = 2,
237 limit: Annotated[
238 int,
239 Field(description="Maximum number of documents to cluster", ge=1, le=1000),
240 ] = 25,
241 source_types: Annotated[
242 list[str] | None,
243 Field(description="Optional list of source types to filter by"),
244 ] = None,
245 project_ids: Annotated[
246 list[str] | None,
247 Field(description="Optional list of project IDs to filter by"),
248 ] = None,
249 ) -> dict[str, Any]:
250 """Cluster documents based on similarity and relationships."""
251 handler = ctx.lifespan_context["intelligence_handler"]
252 params = {
253 "query": query,
254 "strategy": strategy,
255 "max_clusters": max_clusters,
256 "min_cluster_size": min_cluster_size,
257 "limit": limit,
258 "source_types": source_types,
259 "project_ids": project_ids,
260 }
261 return unwrap(await handler.handle_cluster_documents(REQUEST_ID, params))