Spaces:
Sleeping
Sleeping
| """ | |
| 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() |