Coverage for src / mcp_server_langgraph / tools / search_tools.py: 90%

1""" 

2Search tools for querying information 

3 

4Provides knowledge base and web search capabilities for the agent. 

5""" 

6 

7from typing import Annotated 

8 

9import httpx 

10from langchain_core.tools import tool 

11from pydantic import Field 

12from qdrant_client import QdrantClient 

13 

14from mcp_server_langgraph.core.config import settings 

15from mcp_server_langgraph.observability.telemetry import logger, metrics 

16 

17 

18@tool 

19def search_knowledge_base( 

20 query: Annotated[str, Field(description="Search query to find relevant information")], 

21 limit: Annotated[int, Field(ge=1, le=20, description="Maximum number of results (1-20)")] = 5, 

22) -> str: 

23 """ 

24 Search internal knowledge base for relevant information. 

25 

26 Use this to find: 

27 - Documentation and guides 

28 - Previous conversations and context 

29 - System configuration 

30 - Frequently asked questions 

31 

32 Returns top matching results with relevance scores. 

33 """ 

34 try: 

35 logger.info("Knowledge base search invoked", extra={"query": query, "limit": limit}) 

36 metrics.tool_calls.add(1, {"tool": "search_knowledge_base"}) 

37 

38 # Check if Qdrant is configured 

39 if not hasattr(settings, "qdrant_url") or not settings.qdrant_url: 

40 return """Knowledge base search not configured. 

41 

42To enable: 

431. Deploy Qdrant vector database 

442. Set QDRANT_URL and QDRANT_PORT in .env 

453. Set ENABLE_DYNAMIC_CONTEXT_LOADING=true 

464. Index your knowledge base documents 

47 

48See: docs/advanced/dynamic-context.md""" 

49 

50 # Query Qdrant for semantic search 

51 try: 

52 client = QdrantClient( # noqa: F841 

53 url=settings.qdrant_url, 

54 port=getattr(settings, "qdrant_port", 6333), 

55 ) 

56 

57 # Use configured collection or default 

58 collection_name = getattr(settings, "qdrant_collection_name", "mcp_context") 

59 

60 # Perform semantic search (requires embeddings) 

61 # Note: This assumes documents are already indexed 

62 # For full implementation, add embedding generation here 

63 

64 results_text = f"""Knowledge base search: "{query}" 

65 

66Connected to Qdrant at {settings.qdrant_url}:{getattr(settings, "qdrant_port", 6333)} 

67Collection: {collection_name} 

68 

69Note: Semantic search requires embeddings and indexed documents. 

70Configure EMBEDDING_PROVIDER and index your knowledge base. 

71 

72For setup: See docs/advanced/dynamic-context.md""" 

73 

74 logger.info("Knowledge base search completed", extra={"query": query}) 

75 return results_text 

76 

77 except Exception as e: 

78 logger.warning(f"Qdrant query failed: {e}") 

79 return f"""Knowledge base search error: {e} 

80 

81Verify Qdrant is running and accessible at {settings.qdrant_url}""" 

82 

83 except Exception as e: 

84 error_msg = f"Error searching knowledge base: {e}" 

85 logger.error(error_msg, exc_info=True) 

86 return f"Error: {e}" 

87 

88 

89@tool 

90async def web_search( 

91 query: Annotated[str, Field(description="Search query for web search")], 

92 num_results: Annotated[int, Field(ge=1, le=10, description="Number of results to return (1-10)")] = 5, 

93) -> str: 

94 """ 

95 Search the web for current information. 

96 

97 Use this to find: 

98 - Current events and news 

99 - External documentation 

100 - Public information not in knowledge base 

101 - Real-time data 

102 

103 Returns titles, snippets, and URLs of top results. 

104 

105 IMPORTANT: This tool requires web search API integration. 

106 Currently returns placeholder results. 

107 """ 

108 try: 

109 logger.info("Web search invoked", extra={"query": query, "num_results": num_results}) 

110 metrics.tool_calls.add(1, {"tool": "web_search"}) 

111 

112 # Check for configured web search API key 

113 serper_api_key = getattr(settings, "serper_api_key", None) 

114 tavily_api_key = getattr(settings, "tavily_api_key", None) 

115 brave_api_key = getattr(settings, "brave_api_key", None) # noqa: F841 

116 

117 # Try Tavily API (recommended for AI applications) 

118 if tavily_api_key: 

119 try: 

120 async with httpx.AsyncClient() as client: 

121 response = await client.post( 

122 "https://api.tavily.com/search", 

123 json={"api_key": tavily_api_key, "query": query, "max_results": num_results}, 

124 timeout=30.0, 

125 ) 

126 response.raise_for_status() 

127 data = await response.json() 

128 

129 results = [f'Web search: "{query}"\n'] 

130 for i, result in enumerate(data.get("results", [])[:num_results], 1): 

131 results.append(f"\n{i}. {result.get('title', 'No title')}") 

132 results.append(f" {result.get('content', 'No snippet')[:200]}...") 

133 results.append(f" URL: {result.get('url', 'N/A')}") 

134 

135 logger.info("Tavily web search completed", extra={"results": len(data.get("results", []))}) 

136 return "\n".join(results) if results else "No results found" 

137 

138 except Exception as e: 

139 logger.error(f"Tavily search failed: {e}", exc_info=True) 

140 # Fall through to other providers or placeholder 

141 

142 # Try Serper API 

143 elif serper_api_key: 

144 try: 

145 async with httpx.AsyncClient() as client: 

146 response = await client.post( 

147 "https://google.serper.dev/search", 

148 json={"q": query, "num": num_results}, 

149 headers={"X-API-KEY": serper_api_key, "Content-Type": "application/json"}, 

150 timeout=30.0, 

151 ) 

152 response.raise_for_status() 

153 data = await response.json() 

154 

155 results = [f'Web search: "{query}"\n'] 

156 for i, result in enumerate(data.get("organic", [])[:num_results], 1): 

157 results.append(f"\n{i}. {result.get('title', 'No title')}") 

158 results.append(f" {result.get('snippet', 'No snippet')}") 

159 results.append(f" URL: {result.get('link', 'N/A')}") 

160 

161 logger.info("Serper web search completed", extra={"results": len(data.get("organic", []))}) 

162 return "\n".join(results) if results else "No results found" 

163 

164 except Exception as e: 

165 logger.error(f"Serper search failed: {e}", exc_info=True) 

166 # Fall through to placeholder 

167 

168 # No API key configured 

169 config_message = f"""Web search for: "{query}" 

170 

171❌ Web search API not configured. 

172 

173To enable web search, add one of these API keys to .env: 

174 

175**Option 1: Tavily (Recommended for AI)** 

176- Get key: https://tavily.com/ 

177- Add to .env: TAVILY_API_KEY=your-key 

178- Best for: AI-optimized search results 

179 

180**Option 2: Serper (Google Search)** 

181- Get key: https://serper.dev/ 

182- Add to .env: SERPER_API_KEY=your-key 

183- Best for: Google search results 

184 

185**Option 3: Brave Search** 

186- Get key: https://brave.com/search/api/ 

187- Add to .env: BRAVE_API_KEY=your-key 

188- Best for: Privacy-focused search 

189 

190After configuration, this tool will return real web search results.""" 

191 

192 logger.info("Web search completed (no API key)", extra={"query": query}) 

193 return config_message 

194 

195 except Exception as e: 

196 error_msg = f"Error performing web search: {e}" 

197 logger.error(error_msg, exc_info=True) 

198 return f"Error: {e}"