Spaces:
Sleeping
Sleeping
File size: 10,732 Bytes
9a6a5aa |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 |
"""
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() |