Coverage for src/qdrant_loader_mcp_server/fastmcp_app.py: 47%

1"""FastMCP application for the Qdrant Loader MCP server. 

2 

3This is the entry point for both transports: ``cli.py`` serves it over stdio 

4(``mcp.run``) and over HTTP (the module-level ``http_app`` under uvicorn). 

5 

6Heavy resources (SearchEngine, QueryProcessor, the Search/Intelligence handlers) 

7are created inside the lifespan — after the event loop is up — not at import time, 

8so importing this module stays cheap. 

9""" 

10 

11from __future__ import annotations 

12 

13import os 

14from collections.abc import AsyncIterator 

15from contextlib import asynccontextmanager 

16from pathlib import Path 

17from typing import Any 

18 

19from fastmcp import FastMCP 

20from starlette.middleware import Middleware 

21from starlette.middleware.cors import CORSMiddleware 

22from starlette.responses import JSONResponse 

23 

24from .utils import LoggingConfig 

25 

26logger = LoggingConfig.get_logger(__name__) 

27 

28 

29@asynccontextmanager 

30async def _lifespan(server: FastMCP) -> AsyncIterator[dict[str, Any]]: 

31 """Build heavy resources on startup; clean them up on shutdown 

32 Whatever this yields becomes ``ctx.lifespan_context`` inside every tool. 

33 """ 

34 # Lazy imports keep module import cheap 

35 from .config_loader import load_config, resolve_config_path 

36 from .mcp.intelligence_handler import IntelligenceHandler 

37 from .mcp.protocol import MCPProtocol 

38 from .mcp.search_handler import SearchHandler 

39 from .search.engine import SearchEngine 

40 from .search.processor import QueryProcessor 

41 

42 config_path_env = os.getenv("MCP_CONFIG") 

43 cli_config_path = Path(config_path_env) if config_path_env else None 

44 # Resolve once so the graph store (initialized lazily, per-request) uses the 

45 # exact same config file as the search engine below instead of re-deriving 

46 # its own path from Path.cwd(), which can silently diverge from this one. 

47 resolved_config_path = resolve_config_path(cli_config_path) 

48 config, _, _ = load_config(resolved_config_path or cli_config_path) 

49 

50 search_engine = SearchEngine() 

51 query_processor = QueryProcessor(config.openai) 

52 initialized = False 

53 

54 try: 

55 await search_engine.initialize(config.qdrant, config.openai, config.search) 

56 initialized = True 

57 logger.info("FastMCP search engine initialized", pid=os.getpid()) 

58 

59 # Reuse the existing handler for its reranker 

60 # Built after initialize() so the hybrid pipeline exists 

61 search_handler = SearchHandler( 

62 search_engine=search_engine, 

63 query_processor=query_processor, 

64 protocol=MCPProtocol(), 

65 reranking_config=config.reranking, 

66 ) 

67 

68 # Stateful: holds the cluster store shared with expand_cluster 

69 # build one instance for reuse 

70 intelligence_handler = IntelligenceHandler( 

71 search_engine=search_engine, 

72 protocol=MCPProtocol(), 

73 config_path=resolved_config_path, 

74 ) 

75 

76 yield { 

77 "search_engine": search_engine, 

78 "query_processor": query_processor, 

79 "config": config, 

80 "search_handler": search_handler, 

81 "intelligence_handler": intelligence_handler, 

82 } 

83 finally: 

84 if initialized: 

85 try: 

86 await search_engine.cleanup() 

87 except Exception: 

88 logger.error( 

89 "Error during FastMCP search engine cleanup", exc_info=True 

90 ) 

91 

92 

93# The FastMCP app. Imported by entry points later 

94mcp = FastMCP("Qdrant Loader MCP Server", lifespan=_lifespan) 

95 

96# Register tools onto the instance 

97# tool modules pull only fastmcp/pydantic, not the search engine 

98from .fastmcp_tools import register_all # noqa: E402 

99 

100register_all(mcp) 

101 

102 

103# ---- HTTP transport surface (used only when served as an ASGI app) ---- 

104 

105 

106@mcp.custom_route("/health", methods=["GET"]) 

107async def health_check(request): 

108 """Liveness probe. The lifespan completes before uvicorn serves traffic, 

109 so reaching this means the engine is initialized.""" 

110 return JSONResponse( 

111 { 

112 "status": "healthy", 

113 "transport": "http", 

114 "protocol": "mcp", 

115 "pid": os.getpid(), 

116 } 

117 ) 

118 

119 

120# CORS, secure-by-default 

121_cors_origins = os.getenv("CORS_ORIGINS", "") 

122if _cors_origins.strip(): 

123 _allow_origins = [o.strip() for o in _cors_origins.split(",") if o.strip()] 

124else: 

125 _allow_origins = ["http://localhost", "http://127.0.0.1"] 

126 

127_http_middleware = [ 

128 Middleware( 

129 CORSMiddleware, 

130 allow_origins=_allow_origins, 

131 allow_credentials=("*" not in _allow_origins), 

132 allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"], 

133 allow_headers=["*"], 

134 ) 

135] 

136 

137# Module-level ASGI app so uvicorn can import it with workers=N 

138# Tools and /health route must be registered first 

139http_app = mcp.http_app( 

140 path="/mcp", 

141 middleware=_http_middleware, 

142 # HTTP transport profile: plain JSON responses (no SSE) and stateless 

143 # request/response (no initialize/session handshake). This is the intended 

144 # production behavior for simple HTTP clients; stdio is unaffected. 

145 json_response=True, 

146 stateless_http=True, 

147)