""" Dynamic Tool Mapper for MCP Application This module provides dynamic mapping of tools and resources based on runtime MCP specs instead of hardcoded values. """ import re from typing import List, Dict, Any, Optional, Tuple from src.mcp.compact_utils import CompactSpecsUtils from config.conversation_config import conversation_config class DynamicToolMapper: """Maps tools and resources dynamically based on MCP specs.""" def __init__(self): self.compact_utils = CompactSpecsUtils() self.config = conversation_config.tool_mapping self._cached_mappings = {} def get_tool_type(self, tool_name: str) -> Optional[str]: """Determine tool type based on name and specs.""" if tool_name in self._cached_mappings: return self._cached_mappings[tool_name] # Check against patterns tool_lower = tool_name.lower() # Check challenge patterns for pattern in self.config.challenge_tool_patterns: if re.match(pattern, tool_lower, re.IGNORECASE): self._cached_mappings[tool_name] = "challenge" return "challenge" # Check skill patterns for pattern in self.config.skill_tool_patterns: if re.match(pattern, tool_lower, re.IGNORECASE): self._cached_mappings[tool_name] = "skill" return "skill" # Check member/user patterns for pattern in self.config.member_resource_patterns: if re.match(pattern, tool_lower, re.IGNORECASE): self._cached_mappings[tool_name] = "member" return "member" # Try to determine from tool spec description try: spec = self.compact_utils.get_tool_spec(tool_name) if spec and "description" in spec: description = spec["description"].lower() if any(word in description for word in ["challenge", "contest", "competition"]): self._cached_mappings[tool_name] = "challenge" return "challenge" elif any(word in description for word in ["skill", "technology", "competency"]): self._cached_mappings[tool_name] = "skill" return "skill" elif any(word in description for word in ["member", "user", "profile"]): self._cached_mappings[tool_name] = "member" return "member" except Exception: pass # Default fallback self._cached_mappings[tool_name] = "unknown" return "unknown" def get_resource_type(self, resource_name: str) -> Optional[str]: """Determine resource type based on name and specs.""" if resource_name in self._cached_mappings: return self._cached_mappings[resource_name] resource_lower = resource_name.lower() # Check member/user patterns for pattern in self.config.member_resource_patterns: if re.match(pattern, resource_lower, re.IGNORECASE): self._cached_mappings[resource_name] = "member" return "member" # Check challenge patterns for pattern in self.config.challenge_tool_patterns: if re.match(pattern, resource_lower, re.IGNORECASE): self._cached_mappings[resource_name] = "challenge" return "challenge" # Try to determine from resource spec try: spec = self.compact_utils.get_resource_spec(resource_name) if spec and "description" in spec: description = spec["description"].lower() if any(word in description for word in ["member", "user", "profile"]): self._cached_mappings[resource_name] = "member" return "member" elif any(word in description for word in ["challenge", "contest"]): self._cached_mappings[resource_name] = "challenge" return "challenge" elif any(word in description for word in ["skill", "technology"]): self._cached_mappings[resource_name] = "skill" return "skill" except Exception: pass self._cached_mappings[resource_name] = "unknown" return "unknown" def get_extractable_fields(self, tool_or_resource_name: str, tool_type: str = None) -> Dict[str, str]: """Get fields that can be extracted for entity creation.""" if tool_type is None: tool_type = self.get_tool_type(tool_or_resource_name) or self.get_resource_type(tool_or_resource_name) extractable_fields = {} try: # Try to get spec (tool or resource) spec = self.compact_utils.get_tool_spec(tool_or_resource_name) if not spec: spec = self.compact_utils.get_resource_spec(tool_or_resource_name) if spec and "parameters" in spec: parameters = spec["parameters"] # Look for ID fields for field_name, field_info in parameters.items(): field_lower = field_name.lower() # Check ID patterns if any(pattern.lower() in field_lower for pattern in self.config.id_field_patterns): if tool_type == "challenge": extractable_fields[field_name] = "challenge_id" elif tool_type == "skill": extractable_fields[field_name] = "skill_id" elif tool_type == "member": extractable_fields[field_name] = "user_id" # Check name patterns elif any(pattern.lower() in field_lower for pattern in self.config.name_field_patterns): if tool_type == "challenge": extractable_fields[field_name] = "challenge_name" elif tool_type == "skill": extractable_fields[field_name] = "skill_name" elif tool_type == "member": extractable_fields[field_name] = "user_name" # Check handle patterns (specific to members) elif any(pattern.lower() in field_lower for pattern in self.config.handle_field_patterns): extractable_fields[field_name] = "user_handle" except Exception: # Fallback to common patterns if tool_type == "challenge": extractable_fields = {"id": "challenge_id", "name": "challenge_name"} elif tool_type == "skill": extractable_fields = {"id": "skill_id", "name": "skill_name"} elif tool_type == "member": extractable_fields = {"handle": "user_handle", "id": "user_id"} return extractable_fields def get_available_tools_by_type(self) -> Dict[str, List[str]]: """Get all available tools categorized by type.""" tools_by_type = { "challenge": [], "skill": [], "member": [], "unknown": [] } try: available_tools = self.compact_utils.get_working_tools() for tool_name in available_tools: tool_type = self.get_tool_type(tool_name) if tool_type in tools_by_type: tools_by_type[tool_type].append(tool_name) else: tools_by_type["unknown"].append(tool_name) except Exception: pass return tools_by_type def get_available_resources_by_type(self) -> Dict[str, List[str]]: """Get all available resources categorized by type.""" resources_by_type = { "challenge": [], "skill": [], "member": [], "unknown": [] } try: available_resources = self.compact_utils.get_working_resources() for resource_name in available_resources: resource_type = self.get_resource_type(resource_name) if resource_type in resources_by_type: resources_by_type[resource_type].append(resource_name) else: resources_by_type["unknown"].append(resource_name) except Exception: pass return resources_by_type def extract_entities_from_data(self, data: Any, tool_or_resource_name: str) -> List[Tuple[str, str, str]]: """Extract entities from API response data.""" entities = [] tool_type = self.get_tool_type(tool_or_resource_name) or self.get_resource_type(tool_or_resource_name) extractable_fields = self.get_extractable_fields(tool_or_resource_name, tool_type) max_entities = conversation_config.entity_extraction.max_entities_per_tool try: if isinstance(data, list): # Handle list of items for item in data[:max_entities]: if isinstance(item, dict): for field_name, entity_type in extractable_fields.items(): if field_name in item and item[field_name]: entities.append(( str(item[field_name]), # name entity_type, # type str(item[field_name]) # value )) elif isinstance(data, dict): # Handle single item for field_name, entity_type in extractable_fields.items(): if field_name in data and data[field_name]: entities.append(( str(data[field_name]), # name entity_type, # type str(data[field_name]) # value )) except Exception as e: print(f"Error extracting entities from {tool_or_resource_name}: {e}") return entities def clear_cache(self): """Clear the mapping cache (useful when MCP specs change).""" self._cached_mappings.clear() # Global instance dynamic_tool_mapper = DynamicToolMapper()