Skip to main content
Glama

MCP Agent Tracker

by Big0290
enhanced_mcp_tools.py•23.2 kB
""" Enhanced MCP Tools with Embedding Integration This module provides enhanced MCP tools that integrate the embedding system with your existing conversation intelligence tools. """ import time from typing import Dict, Any, Optional, List from datetime import datetime # Import the bridge from mcp_embedding_bridge import get_mcp_embedding_bridge, MCPEmbeddingBridge # Import existing MCP tools for compatibility from main import ( agent_interaction, get_conversation_summary, get_interaction_history, get_system_status, test_conversation_tracking ) # šŸš€ NEW: Import optimized prompt generator try: from optimized_prompt_generator import OptimizedPromptGenerator OPTIMIZED_PROMPTS_AVAILABLE = True print("šŸš€ Optimized prompt generator loaded in enhanced MCP tools") except ImportError: OPTIMIZED_PROMPTS_AVAILABLE = False print("āš ļø Optimized prompt generator not available in enhanced MCP tools") class EnhancedMCPTools: """ Enhanced MCP tools that integrate embeddings with existing functionality. """ def __init__(self): """Initialize enhanced MCP tools.""" self.bridge = get_mcp_embedding_bridge() self.tools_registry = { 'enhanced_agent_interaction': self.enhanced_agent_interaction, 'semantic_context_search': self.semantic_context_search, 'enhanced_conversation_summary': self.enhanced_conversation_summary, 'semantic_insights': self.semantic_insights, 'bridge_status': self.bridge_status, 'test_enhanced_integration': self.test_enhanced_integration, 'enhanced_prompt_generation': self.enhanced_prompt_generation, 'comprehensive_context_analysis': self.comprehensive_context_analysis, 'semantic_learning_enhancement': self.semantic_learning_enhancement, 'bridge_statistics': self.bridge_statistics, 'clear_enhanced_cache': self.clear_enhanced_cache } def get_available_tools(self) -> Dict[str, Any]: """Get list of available enhanced MCP tools.""" return { 'enhanced_tools': list(self.tools_registry.keys()), 'base_tools': [ 'agent_interaction', 'get_conversation_summary', 'get_interaction_history', 'get_system_status', 'test_conversation_tracking' ], 'total_tools': len(self.tools_registry) + 5, 'integration_status': self.bridge.get_bridge_status() } def enhanced_agent_interaction(self, prompt: str, use_semantic_search: bool = True, context_type: str = "smart", similarity_threshold: float = 0.7) -> Dict[str, Any]: """ Enhanced agent interaction with semantic context injection. Args: prompt: User prompt use_semantic_search: Whether to use semantic search context_type: Context type for enhancement similarity_threshold: Minimum similarity threshold Returns: Enhanced interaction result """ start_time = time.time() try: # Generate enhanced prompt with embeddings enhanced_prompt = self.bridge.generate_enhanced_prompt_with_embeddings( prompt, context_type, use_semantic_search, similarity_threshold ) # Use existing agent interaction base_response = agent_interaction(prompt) # Enhance context learning learning_result = self.bridge.enhance_context_learning_with_embeddings( prompt, enhanced_prompt, 0.8, context_type ) processing_time = time.time() - start_time return { 'status': 'success', 'original_prompt': prompt, 'enhanced_prompt': enhanced_prompt, 'base_response': base_response, 'semantic_enhancement': { 'enhancement_ratio': len(enhanced_prompt) / len(prompt) if prompt else 1.0, 'similarity_threshold': similarity_threshold, 'context_type': context_type }, 'learning_enhancement': learning_result, 'processing_time_ms': round(processing_time * 1000, 2), 'timestamp': datetime.now().isoformat() } except Exception as e: processing_time = time.time() - start_time return { 'status': 'error', 'error': str(e), 'original_prompt': prompt, 'processing_time_ms': round(processing_time * 1000, 2), 'timestamp': datetime.now().isoformat() } def semantic_context_search(self, query: str, context_type: str = "conversation", limit: int = 10, min_similarity: float = 0.7) -> Dict[str, Any]: """ Search for semantically similar contexts. Args: query: Search query context_type: Context type to search in limit: Maximum number of results min_similarity: Minimum similarity threshold Returns: Search results with semantic similarity """ try: # Use bridge for semantic search similar_contexts = self.bridge.enhance_caching_with_semantic_similarity( query, context_type, min_similarity ) # Limit results limited_results = similar_contexts[:limit] return { 'status': 'success', 'query': query, 'context_type': context_type, 'min_similarity': min_similarity, 'total_found': len(similar_contexts), 'results_returned': len(limited_results), 'results': limited_results, 'search_metadata': { 'search_type': 'semantic', 'vector_search_available': self.bridge.embedding_integration is not None, 'timestamp': datetime.now().isoformat() } } except Exception as e: return { 'status': 'error', 'error': str(e), 'query': query, 'timestamp': datetime.now().isoformat() } def enhanced_conversation_summary(self, session_id: Optional[str] = None, include_semantic_insights: bool = True) -> Dict[str, Any]: """ Enhanced conversation summary with semantic insights. Args: session_id: Optional session ID include_semantic_insights: Whether to include semantic analysis Returns: Enhanced conversation summary """ try: # Get base conversation summary base_summary = get_conversation_summary(session_id) enhanced_summary = { 'base_summary': base_summary, 'semantic_enhancements': {}, 'timestamp': datetime.now().isoformat() } if include_semantic_insights: try: # Get semantic insights for recent conversations recent_interactions = get_interaction_history(limit=5, session_id=session_id) if recent_interactions and 'interactions' in recent_interactions: # Analyze recent interactions for semantic patterns semantic_analysis = self._analyze_conversation_semantics(recent_interactions['interactions']) enhanced_summary['semantic_enhancements'] = semantic_analysis except Exception as e: enhanced_summary['semantic_enhancements']['error'] = str(e) return enhanced_summary except Exception as e: return { 'status': 'error', 'error': str(e), 'timestamp': datetime.now().isoformat() } def semantic_insights(self, user_message: str, context_type: str = "conversation", include_recommendations: bool = True) -> Dict[str, Any]: """ Get semantic insights for a user message. Args: user_message: User message to analyze context_type: Context type for analysis include_recommendations: Whether to include recommendations Returns: Semantic insights and analysis """ try: # Get comprehensive context context = self.bridge.get_comprehensive_context(user_message) # Get semantic insights from embedding system semantic_insights = {} if self.bridge.embedding_integration: try: semantic_insights = self.bridge.embedding_integration.get_semantic_insights( user_message, context_type ) except Exception as e: semantic_insights = {'error': str(e)} # Build insights response insights = { 'user_message': user_message, 'context_type': context_type, 'context_richness_score': context.get('bridge_enhancements', {}).get('context_richness_score', 0), 'comprehensive_context': context, 'semantic_insights': semantic_insights, 'timestamp': datetime.now().isoformat() } if include_recommendations: insights['recommendations'] = context.get('bridge_enhancements', {}).get('recommendations', []) return insights except Exception as e: return { 'status': 'error', 'error': str(e), 'user_message': user_message, 'timestamp': datetime.now().isoformat() } def bridge_status(self) -> Dict[str, Any]: """Get the current bridge integration status.""" return self.bridge.get_bridge_status() def test_enhanced_integration(self) -> Dict[str, Any]: """Test the enhanced integration between all systems.""" return self.bridge.test_bridge_integration() def enhanced_prompt_generation(self, user_message: str, context_type: str = "smart", use_semantic_search: bool = True, similarity_threshold: float = 0.7) -> Dict[str, Any]: """ Enhanced prompt generation with semantic context. Args: user_message: User message context_type: Context type use_semantic_search: Whether to use semantic search similarity_threshold: Minimum similarity threshold Returns: Enhanced prompt generation result """ try: start_time = time.time() # Generate enhanced prompt enhanced_prompt = self.bridge.generate_enhanced_prompt_with_embeddings( user_message, context_type, use_semantic_search, similarity_threshold ) processing_time = time.time() - start_time return { 'status': 'success', 'user_message': user_message, 'enhanced_prompt': enhanced_prompt, 'enhancement_metrics': { 'original_length': len(user_message), 'enhanced_length': len(enhanced_prompt), 'enhancement_ratio': len(enhanced_prompt) / len(user_message) if user_message else 1.0, 'processing_time_ms': round(processing_time * 1000, 2) }, 'context_type': context_type, 'semantic_search_used': use_semantic_search, 'similarity_threshold': similarity_threshold, 'timestamp': datetime.now().isoformat() } except Exception as e: return { 'status': 'error', 'error': str(e), 'user_message': user_message, 'timestamp': datetime.now().isoformat() } def comprehensive_context_analysis(self, user_message: str, session_id: Optional[str] = None, user_id: Optional[str] = None) -> Dict[str, Any]: """ Comprehensive context analysis using all available systems. Args: user_message: User message to analyze session_id: Optional session ID user_id: Optional user ID Returns: Comprehensive context analysis """ return self.bridge.get_comprehensive_context(user_message, session_id, user_id) def semantic_learning_enhancement(self, user_message: str, enhanced_prompt: str, response_quality: float, context_type: str = "conversation") -> Dict[str, Any]: """ Enhance learning with semantic capabilities. Args: user_message: User message enhanced_prompt: Enhanced prompt response_quality: Response quality rating context_type: Context type Returns: Learning enhancement result """ return self.bridge.enhance_context_learning_with_embeddings( user_message, enhanced_prompt, response_quality, context_type ) def bridge_statistics(self) -> Dict[str, Any]: """Get comprehensive statistics from the bridge.""" return self.bridge.get_bridge_statistics() def clear_enhanced_cache(self, context_type: Optional[str] = None) -> Dict[str, Any]: """Clear caches in both existing and embedding systems.""" return self.bridge.clear_bridge_cache(context_type) def _analyze_conversation_semantics(self, interactions: List[Dict[str, Any]]) -> Dict[str, Any]: """Analyze conversation interactions for semantic patterns.""" try: if not interactions: return {'error': 'No interactions to analyze'} # Extract text content from interactions texts = [] for interaction in interactions: if 'client_request' in interaction: texts.append(interaction['client_request']) if 'agent_response' in interaction: texts.append(interaction['agent_response']) if not texts: return {'error': 'No text content found in interactions'} # Analyze semantic patterns analysis = { 'total_interactions': len(interactions), 'total_text_segments': len(texts), 'semantic_patterns': {}, 'conversation_themes': [], 'technical_terms': [], 'timestamp': datetime.now().isoformat() } # Basic text analysis all_text = ' '.join(texts).lower() # Extract technical terms (simple heuristic) technical_terms = ['mcp', 'embedding', 'context', 'prompt', 'conversation', 'semantic'] found_terms = [term for term in technical_terms if term in all_text] analysis['technical_terms'] = found_terms # Extract conversation themes themes = [] if 'embedding' in all_text: themes.append('Embedding System') if 'context' in all_text: themes.append('Context Management') if 'prompt' in all_text: themes.append('Prompt Enhancement') if 'conversation' in all_text: themes.append('Conversation Intelligence') analysis['conversation_themes'] = themes return analysis except Exception as e: return {'error': str(e)} def call_tool(self, tool_name: str, **kwargs) -> Dict[str, Any]: """ Call a specific enhanced MCP tool. Args: tool_name: Name of the tool to call **kwargs: Tool-specific arguments Returns: Tool execution result """ if tool_name in self.tools_registry: try: return self.tools_registry[tool_name](**kwargs) except Exception as e: return { 'status': 'error', 'error': str(e), 'tool_name': tool_name, 'timestamp': datetime.now().isoformat() } else: return { 'status': 'error', 'error': f'Tool "{tool_name}" not found', 'available_tools': list(self.tools_registry.keys()), 'timestamp': datetime.now().isoformat() } # Global enhanced MCP tools instance _enhanced_mcp_tools = None def get_enhanced_mcp_tools() -> EnhancedMCPTools: """Get or create the global enhanced MCP tools instance.""" global _enhanced_mcp_tools if _enhanced_mcp_tools is None: _enhanced_mcp_tools = EnhancedMCPTools() return _enhanced_mcp_tools # Convenience functions for easy access def enhanced_agent_interaction(prompt: str, **kwargs) -> Dict[str, Any]: """Enhanced agent interaction with semantic context.""" tools = get_enhanced_mcp_tools() return tools.enhanced_agent_interaction(prompt, **kwargs) def semantic_context_search(query: str, **kwargs) -> Dict[str, Any]: """Search for semantically similar contexts.""" tools = get_enhanced_mcp_tools() return tools.semantic_context_search(query, **kwargs) def enhanced_conversation_summary(**kwargs) -> Dict[str, Any]: """Enhanced conversation summary with semantic insights.""" tools = get_enhanced_mcp_tools() return tools.enhanced_conversation_summary(**kwargs) def semantic_insights(user_message: str, **kwargs) -> Dict[str, Any]: """Get semantic insights for a user message.""" tools = get_enhanced_mcp_tools() return tools.semantic_insights(user_message, **kwargs) def get_enhanced_tools_status() -> Dict[str, Any]: """Get status of enhanced MCP tools.""" tools = get_enhanced_mcp_tools() return tools.get_available_tools() if __name__ == "__main__": # Test the enhanced MCP tools print("=== Testing Enhanced MCP Tools ===\n") tools = get_enhanced_mcp_tools() # Test tool availability available_tools = tools.get_available_tools() print("Available Enhanced Tools:") for tool in available_tools['enhanced_tools']: print(f" āœ… {tool}") print(f"\nTotal Enhanced Tools: {available_tools['total_tools']}") print(f"Integration Status: {available_tools['integration_status']['bridge_initialized']}") print("\n" + "="*50 + "\n") # Test enhanced agent interaction test_prompt = "How can I improve my MCP conversation system with embeddings?" print("Testing Enhanced Agent Interaction:") interaction_result = tools.enhanced_agent_interaction( test_prompt, use_semantic_search=True, context_type="technical" ) print(f"Status: {interaction_result['status']}") if interaction_result['status'] == 'success': print(f"Enhancement Ratio: {interaction_result['semantic_enhancement']['enhancement_ratio']:.2f}") print(f"Processing Time: {interaction_result['processing_time_ms']}ms") print("\n" + "="*50 + "\n") # Test semantic context search print("Testing Semantic Context Search:") search_result = tools.semantic_context_search( "MCP conversation system", context_type="conversation", limit=5 ) print(f"Status: {search_result['status']}") if search_result['status'] == 'success': print(f"Results Found: {search_result['total_found']}") print(f"Results Returned: {search_result['results_returned']}") print("\n" + "="*50 + "\n") # Test bridge status print("Testing Bridge Status:") bridge_status = tools.bridge_status() for key, value in bridge_status.items(): print(f" {key}: {value}") print("\n" + "="*50 + "\n") # Test comprehensive context analysis print("Testing Comprehensive Context Analysis:") context_result = tools.comprehensive_context_analysis(test_prompt) if 'error' not in context_result: richness_score = context_result.get('bridge_enhancements', {}).get('context_richness_score', 0) print(f"Context Richness Score: {richness_score:.2f}") recommendations = context_result.get('bridge_enhancements', {}).get('recommendations', []) if recommendations: print("Recommendations:") for rec in recommendations[:3]: # Show first 3 print(f" • {rec}") else: print(f"Error: {context_result['error']}") print("\n" + "="*50 + "\n") # Test enhanced prompt generation print("Testing Enhanced Prompt Generation:") prompt_result = tools.enhanced_prompt_generation( test_prompt, context_type="smart", use_semantic_search=True ) print(f"Status: {prompt_result['status']}") if prompt_result['status'] == 'success': metrics = prompt_result['enhancement_metrics'] print(f"Original Length: {metrics['original_length']}") print(f"Enhanced Length: {metrics['enhanced_length']}") print(f"Enhancement Ratio: {metrics['enhancement_ratio']:.2f}") print(f"Processing Time: {metrics['processing_time_ms']}ms") print("\n" + "="*50 + "\n") print("āœ… Enhanced MCP Tools Testing Complete!") print("\nYou can now use these enhanced tools in your MCP system:") print(" • enhanced_agent_interaction() - Enhanced AI interactions") print(" • semantic_context_search() - Semantic context search") print(" • enhanced_conversation_summary() - Rich conversation summaries") print(" • semantic_insights() - Semantic analysis and insights") print(" • comprehensive_context_analysis() - Full context analysis")

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/Big0290/MCP'

If you have feedback or need assistance with the MCP directory API, please join our Discord server