Coverage for src/qdrant_loader_mcp_server/fastmcp_tools/expand.py: 57%

1"""FastMCP expansion tools (lazy-loading). 

2 

3Delegate to existing handlers — SearchHandler for document/chunk expansion, 

4IntelligenceHandler for cluster expansion — and unwrap the structured payload. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import Annotated, Any 

10 

11from fastmcp import Context, FastMCP 

12from pydantic import Field 

13 

14from ._common import REQUEST_ID, unwrap 

15 

16 

17def register_expand_tools(mcp: FastMCP) -> None: 

18 @mcp.tool(annotations={"readOnlyHint": True}) 

19 async def expand_document( 

20 ctx: Context, 

21 document_id: Annotated[ 

22 str, 

23 Field( 

24 description="The ID of the document to expand and retrieve full content", 

25 min_length=1, 

26 ), 

27 ], 

28 include_metadata: Annotated[ 

29 bool, Field(description="Include detailed metadata (optional)") 

30 ] = True, 

31 include_hierarchy: Annotated[ 

32 bool, 

33 Field(description="Include hierarchy information for Confluence documents"), 

34 ] = True, 

35 include_attachments: Annotated[ 

36 bool, Field(description="Include attachment information if available") 

37 ] = True, 

38 ) -> dict[str, Any]: 

39 """Retrieve full document content by document ID for lazy loading.""" 

40 handler = ctx.lifespan_context["search_handler"] 

41 params = { 

42 "document_id": document_id, 

43 "include_metadata": include_metadata, 

44 "include_hierarchy": include_hierarchy, 

45 "include_attachments": include_attachments, 

46 } 

47 return unwrap(await handler.handle_expand_document(REQUEST_ID, params)) 

48 

49 @mcp.tool(annotations={"readOnlyHint": True}) 

50 async def expand_chunk_context( 

51 ctx: Context, 

52 document_id: Annotated[ 

53 str, Field(description="Unique identifier of the document.", min_length=1) 

54 ], 

55 chunk_index: Annotated[ 

56 int, 

57 Field(description="Index of the target chunk within the document.", ge=0), 

58 ], 

59 window_size: Annotated[ 

60 int, 

61 Field( 

62 description="Number of chunks to include before and after the target chunk.", 

63 ge=0, 

64 ), 

65 ] = 2, 

66 ) -> dict[str, Any]: 

67 """Retrieve neighboring chunks within the same document based on chunk_index.""" 

68 handler = ctx.lifespan_context["search_handler"] 

69 params = { 

70 "document_id": document_id, 

71 "chunk_index": chunk_index, 

72 "window_size": window_size, 

73 } 

74 return unwrap(await handler.handle_expand_chunk_context(REQUEST_ID, params)) 

75 

76 @mcp.tool(annotations={"readOnlyHint": True}) 

77 async def expand_cluster( 

78 ctx: Context, 

79 cluster_id: Annotated[ 

80 str, 

81 Field( 

82 description="The ID of the cluster to expand and retrieve all documents", 

83 min_length=1, 

84 ), 

85 ], 

86 cluster_session_id: Annotated[ 

87 str, 

88 Field(description="UUID representing a clustering session", min_length=1), 

89 ], 

90 limit: Annotated[ 

91 int, 

92 Field( 

93 description="Maximum number of documents to return from cluster", ge=1 

94 ), 

95 ] = 20, 

96 offset: Annotated[ 

97 int, Field(description="Number of documents to skip for pagination", ge=0) 

98 ] = 0, 

99 include_metadata: Annotated[ 

100 bool, Field(description="Include detailed metadata for each document") 

101 ] = True, 

102 ) -> dict[str, Any]: 

103 """Retrieve all documents from a specific cluster for lazy loading.""" 

104 handler = ctx.lifespan_context["intelligence_handler"] 

105 params = { 

106 "cluster_id": cluster_id, 

107 "cluster_session_id": cluster_session_id, 

108 "limit": limit, 

109 "offset": offset, 

110 "include_metadata": include_metadata, 

111 } 

112 return unwrap(await handler.handle_expand_cluster(REQUEST_ID, params))