|
|
""" |
|
|
Search MCP Client - Find migration guides and documentation using Tavily MCP server. |
|
|
""" |
|
|
|
|
|
import logging |
|
|
from typing import List, Dict, Optional |
|
|
from mcp import ClientSession |
|
|
from mcp.client.stdio import stdio_client |
|
|
|
|
|
logger = logging.getLogger(__name__) |
|
|
|
|
|
|
|
|
class SearchMCPClient: |
|
|
""" |
|
|
Client for Tavily Search MCP server to find migration guides and best practices. |
|
|
""" |
|
|
|
|
|
def __init__(self, mcp_manager): |
|
|
""" |
|
|
Initialize Search MCP client. |
|
|
|
|
|
Args: |
|
|
mcp_manager: MCPManager instance |
|
|
""" |
|
|
self.mcp_manager = mcp_manager |
|
|
self.server_name = "tavily" |
|
|
|
|
|
logger.info("SearchMCPClient initialized") |
|
|
|
|
|
async def find_migration_guide(self, from_tech: str, to_tech: str, max_results: int = 5) -> List[Dict]: |
|
|
""" |
|
|
Find migration documentation for technology upgrade. |
|
|
|
|
|
Args: |
|
|
from_tech: Source technology (e.g., "Python 2.7") |
|
|
to_tech: Target technology (e.g., "Python 3.12") |
|
|
max_results: Maximum number of results to return |
|
|
|
|
|
Returns: |
|
|
List of search results with URLs and snippets |
|
|
""" |
|
|
try: |
|
|
server_params = self.mcp_manager.get_server_params(self.server_name) |
|
|
if not server_params: |
|
|
logger.warning(f"{self.server_name} MCP server not registered, returning empty results") |
|
|
return [] |
|
|
|
|
|
query = f"{from_tech} to {to_tech} migration guide best practices" |
|
|
|
|
|
async with stdio_client(server_params) as (read, write): |
|
|
async with ClientSession(read, write) as session: |
|
|
await session.initialize() |
|
|
|
|
|
result = await session.call_tool( |
|
|
"search", |
|
|
arguments={ |
|
|
"query": query, |
|
|
"max_results": max_results |
|
|
} |
|
|
) |
|
|
|
|
|
|
|
|
results = [] |
|
|
if result and hasattr(result, 'content'): |
|
|
for item in result.content: |
|
|
if hasattr(item, 'text'): |
|
|
results.append({ |
|
|
'title': item.text.get('title', ''), |
|
|
'url': item.text.get('url', ''), |
|
|
'snippet': item.text.get('snippet', ''), |
|
|
'score': item.text.get('score', 0) |
|
|
}) |
|
|
|
|
|
logger.info(f"Found {len(results)} migration guides for {from_tech} to {to_tech}") |
|
|
return results |
|
|
|
|
|
except Exception as e: |
|
|
logger.error(f"Error finding migration guide: {e}") |
|
|
return [] |
|
|
|
|
|
async def find_library_documentation(self, library_name: str, version: Optional[str] = None) -> List[Dict]: |
|
|
""" |
|
|
Find official documentation for a library. |
|
|
|
|
|
Args: |
|
|
library_name: Name of the library |
|
|
version: Optional specific version |
|
|
|
|
|
Returns: |
|
|
List of documentation links |
|
|
""" |
|
|
try: |
|
|
server_params = self.mcp_manager.get_server_params(self.server_name) |
|
|
if not server_params: |
|
|
logger.warning(f"{self.server_name} MCP server not registered, returning empty results") |
|
|
return [] |
|
|
|
|
|
query = f"{library_name} official documentation" |
|
|
if version: |
|
|
query += f" version {version}" |
|
|
|
|
|
async with stdio_client(server_params) as (read, write): |
|
|
async with ClientSession(read, write) as session: |
|
|
await session.initialize() |
|
|
|
|
|
result = await session.call_tool( |
|
|
"search", |
|
|
arguments={ |
|
|
"query": query, |
|
|
"max_results": 3 |
|
|
} |
|
|
) |
|
|
|
|
|
results = [] |
|
|
if result and hasattr(result, 'content'): |
|
|
for item in result.content: |
|
|
if hasattr(item, 'text'): |
|
|
results.append({ |
|
|
'title': item.text.get('title', ''), |
|
|
'url': item.text.get('url', ''), |
|
|
'snippet': item.text.get('snippet', '') |
|
|
}) |
|
|
|
|
|
logger.info(f"Found {len(results)} documentation links for {library_name}") |
|
|
return results |
|
|
|
|
|
except Exception as e: |
|
|
logger.error(f"Error finding library documentation: {e}") |
|
|
return [] |
|
|
|
|
|
async def find_best_practices(self, topic: str, language: str = "python") -> List[Dict]: |
|
|
""" |
|
|
Find best practices for a specific topic. |
|
|
|
|
|
Args: |
|
|
topic: Topic to search for (e.g., "database connection pooling") |
|
|
language: Programming language |
|
|
|
|
|
Returns: |
|
|
List of best practice resources |
|
|
""" |
|
|
try: |
|
|
server_params = self.mcp_manager.get_server_params(self.server_name) |
|
|
if not server_params: |
|
|
logger.warning(f"{self.server_name} MCP server not registered, returning empty results") |
|
|
return [] |
|
|
|
|
|
query = f"{language} {topic} best practices 2024" |
|
|
|
|
|
async with stdio_client(server_params) as (read, write): |
|
|
async with ClientSession(read, write) as session: |
|
|
await session.initialize() |
|
|
|
|
|
result = await session.call_tool( |
|
|
"search", |
|
|
arguments={ |
|
|
"query": query, |
|
|
"max_results": 5 |
|
|
} |
|
|
) |
|
|
|
|
|
results = [] |
|
|
if result and hasattr(result, 'content'): |
|
|
for item in result.content: |
|
|
if hasattr(item, 'text'): |
|
|
results.append({ |
|
|
'title': item.text.get('title', ''), |
|
|
'url': item.text.get('url', ''), |
|
|
'snippet': item.text.get('snippet', '') |
|
|
}) |
|
|
|
|
|
logger.info(f"Found {len(results)} best practice resources for {topic}") |
|
|
return results |
|
|
|
|
|
except Exception as e: |
|
|
logger.error(f"Error finding best practices: {e}") |
|
|
return [] |
|
|
|
|
|
async def find_security_vulnerabilities(self, pattern: str, language: str = "python") -> List[Dict]: |
|
|
""" |
|
|
Find information about security vulnerabilities in a code pattern. |
|
|
|
|
|
Args: |
|
|
pattern: Code pattern to check (e.g., "SQL string interpolation") |
|
|
language: Programming language |
|
|
|
|
|
Returns: |
|
|
List of security resources |
|
|
""" |
|
|
try: |
|
|
server_params = self.mcp_manager.get_server_params(self.server_name) |
|
|
if not server_params: |
|
|
logger.warning(f"{self.server_name} MCP server not registered, returning empty results") |
|
|
return [] |
|
|
|
|
|
query = f"{language} {pattern} security vulnerability CVE" |
|
|
|
|
|
async with stdio_client(server_params) as (read, write): |
|
|
async with ClientSession(read, write) as session: |
|
|
await session.initialize() |
|
|
|
|
|
result = await session.call_tool( |
|
|
"search", |
|
|
arguments={ |
|
|
"query": query, |
|
|
"max_results": 5 |
|
|
} |
|
|
) |
|
|
|
|
|
results = [] |
|
|
if result and hasattr(result, 'content'): |
|
|
for item in result.content: |
|
|
if hasattr(item, 'text'): |
|
|
results.append({ |
|
|
'title': item.text.get('title', ''), |
|
|
'url': item.text.get('url', ''), |
|
|
'snippet': item.text.get('snippet', ''), |
|
|
'severity': self._extract_severity(item.text.get('snippet', '')) |
|
|
}) |
|
|
|
|
|
logger.info(f"Found {len(results)} security resources for {pattern}") |
|
|
return results |
|
|
|
|
|
except Exception as e: |
|
|
logger.error(f"Error finding security vulnerabilities: {e}") |
|
|
return [] |
|
|
|
|
|
def _extract_severity(self, text: str) -> str: |
|
|
""" |
|
|
Extract severity level from text. |
|
|
|
|
|
Args: |
|
|
text: Text to analyze |
|
|
|
|
|
Returns: |
|
|
Severity level (critical, high, medium, low, unknown) |
|
|
""" |
|
|
text_lower = text.lower() |
|
|
if 'critical' in text_lower: |
|
|
return 'critical' |
|
|
elif 'high' in text_lower: |
|
|
return 'high' |
|
|
elif 'medium' in text_lower or 'moderate' in text_lower: |
|
|
return 'medium' |
|
|
elif 'low' in text_lower: |
|
|
return 'low' |
|
|
return 'unknown' |
|
|
|
