Dynamic Per-User Tool Generation MCP Server

# Slot Resolution - MCP Server Integration Guide This guide explains how to integrate the slot resolution system with the MCP server to enable natural language entity references in create requests. ## 🎯 Integration Goal Enable users to send requests like: ```json { "subject": "Laptop not working", "impact": "high", "assignee": "shivam" } ``` And automatically transform them to: ```json { "subject": "Laptop not working", "impactId": 2, "technicianId": 433 } ``` Before sending to the main server. --- ## 📋 Integration Steps ### Step 1: Initialize Slot Resolution in Server Initializer **File**: `server_initializer.py` Add slot resolution initialization to the `ServerInitializer` class: ```python from slot_resolution import SlotResolutionMiddleware from slot_resolution.core.resolver import SlotResolver from slot_resolution.services.elasticsearch_service import ElasticsearchMatchingService from slot_resolution.services.cache_service import CacheService from slot_resolution.services.audit_service import AuditService class ServerInitializer: def __init__(self): # ... existing initialization ... # Initialize slot resolution self.slot_resolution_middleware = self._initialize_slot_resolution() def _initialize_slot_resolution(self) -> SlotResolutionMiddleware: """Initialize slot resolution middleware.""" logger.info("Initializing slot resolution system...") # Get search client from existing SearchSystem search_client = self.search_system.search_client # Create Elasticsearch matching service es_service = ElasticsearchMatchingService(search_client) # Create optional services cache_service = CacheService(ttl_seconds=900, max_size=1000) audit_service = AuditService(log_to_file=True) # Create resolver resolver = SlotResolver( es_service=es_service, cache_service=cache_service, audit_service=audit_service ) # Create middleware middleware = SlotResolutionMiddleware( tenant_id=self.tenant_id, resolver=resolver ) logger.info("Slot resolution system initialized successfully") return middleware ``` --- ### Step 2: Modify Dynamic Tool Middleware **File**: `dynamic_tool_middleware.py` (or wherever create_request tool is defined) Add slot resolution before sending to main server: ```python async def create_request_tool( subject: str, impact: Optional[str] = None, urgency: Optional[str] = None, assignee: Optional[str] = None, **kwargs ) -> str: """ Create a new request. Args: subject: Request subject impact: Impact level (e.g., "high", "medium", "low") urgency: Urgency level (e.g., "high", "medium", "low") assignee: Assignee name (e.g., "shivam", "john.doe") **kwargs: Additional fields Returns: Success message or disambiguation request """ try: # Build request payload request_payload = { "subject": subject, **kwargs } # Add optional fields if impact: request_payload["impact"] = impact if urgency: request_payload["urgency"] = urgency if assignee: request_payload["assignee"] = assignee # Resolve entity names to IDs resolution_result = await server_initializer.slot_resolution_middleware.resolve_request( request_payload=request_payload, module="request", user_id=get_current_user_id() # Get from context ) # Handle result if resolution_result.status == "READY": # All fields resolved, send to main server response = await main_server_client.create_request( payload=resolution_result.payload ) return f"Request created successfully: {response['requestNumber']}" elif resolution_result.status == "DISAMBIGUATION_REQUIRED": # Need user to select from candidates return _format_disambiguation_response(resolution_result.disambiguations) else: return f"Error: {resolution_result.error}" except Exception as e: logger.error(f"Error creating request: {e}", exc_info=True) return f"Failed to create request: {str(e)}" def _format_disambiguation_response(disambiguations: list) -> str: """ Format disambiguation request for user. Args: disambiguations: List of DisambiguationRequest objects Returns: Formatted message asking user to select """ message = "Multiple matches found. Please specify:\n\n" for disambiguation in disambiguations: message += f"**{disambiguation.field}** (you entered: '{disambiguation.original_input}'):\n" for i, candidate in enumerate(disambiguation.candidates, 1): message += f" {i}. {candidate.canonical_name}" # Add helpful attributes if candidate.attributes: attrs = [] if "emailDomain" in candidate.attributes: attrs.append(f"Email: {candidate.attributes['emailDomain']}") if "department" in candidate.attributes: attrs.append(f"Dept: {candidate.attributes['department']}") if attrs: message += f" ({', '.join(attrs)})" message += f" [Confidence: {candidate.confidence:.0%}]\n" message += "\n" message += "Please re-submit your request with the exact name or ID." return message ``` --- ### Step 3: Handle Disambiguation Follow-up When user provides clarification, you can either: **Option A: User provides exact name** ```python # User says: "Create request with assignee 'Shivam Pansuriya'" # System will resolve to exact match ``` **Option B: User provides ID directly** ```python # User says: "Create request with assignee ID 433" # System will pass through the ID request_payload = { "subject": "test", "assignee": 433 # Integer, will be passed through } ``` --- ### Step 4: Add Slot Resolution to Other Tools Apply the same pattern to other create tools: ```python # For problem creation async def create_problem_tool(...): resolution_result = await server_initializer.slot_resolution_middleware.resolve_request( request_payload=problem_payload, module="problem", # Different module user_id=get_current_user_id() ) # ... handle result ... # For change creation async def create_change_tool(...): resolution_result = await server_initializer.slot_resolution_middleware.resolve_request( request_payload=change_payload, module="change", # Different module user_id=get_current_user_id() ) # ... handle result ... ``` --- ## 🔧 Configuration ### Adding New Fields To add support for a new field (e.g., "ci" for Configuration Item): 1. **Update `field_mappings.json`**: ```json { "request": [ { "columnName": "ci", "primitiveType": "long", "referenceTo": "configurationitem", "dbKey": "ciId", "required": false, "filters": { "active": true } } ] } ``` 2. **No code changes needed!** The system will automatically: - Detect the "ci" field in requests - Search the "configurationitem" entity type - Apply the "active: true" filter - Transform to "ciId" in the output ### Adjusting Thresholds To adjust decisioning thresholds for specific entity types: ```python from slot_resolution.core.decisioning import DecisionEngine # Create custom decision engine decision_engine = DecisionEngine( min_scores={ "impact": 0.70, # Higher threshold for impact "user": 0.75, # Higher threshold for users "location": 0.55 # Lower threshold for locations }, score_gap_delta=0.20, # Larger gap required for auto-resolution max_candidates=3 # Show max 3 candidates ) # Use in resolver resolver = SlotResolver( es_service=es_service, decision_engine=decision_engine ) ``` --- ## 🧪 Testing Integration ### Test 1: Simple Resolution ```python async def test_simple_resolution(): middleware = server_initializer.slot_resolution_middleware result = await middleware.resolve_request( request_payload={"impact": "high"}, module="request" ) assert result.status == "READY" assert "impactId" in result.payload print(f"✓ Resolved 'high' to ID: {result.payload['impactId']}") ``` ### Test 2: Disambiguation ```python async def test_disambiguation(): middleware = server_initializer.slot_resolution_middleware result = await middleware.resolve_request( request_payload={"assignee": "john"}, module="request" ) if result.status == "DISAMBIGUATION_REQUIRED": print(f"✓ Disambiguation triggered for 'john'") print(f" Candidates: {len(result.disambiguations[0].candidates)}") else: print(f"✓ Auto-resolved to: {result.payload.get('technicianId')}") ``` ### Test 3: Multiple Fields ```python async def test_multiple_fields(): middleware = server_initializer.slot_resolution_middleware result = await middleware.resolve_request( request_payload={ "subject": "Test", "impact": "high", "urgency": "medium", "priority": "high", "assignee": "shivam.pansuriya" }, module="request" ) assert result.status == "READY" assert all(key in result.payload for key in ["impactId", "urgencyId", "priorityId", "technicianId"]) print(f"✓ All fields resolved successfully") ``` --- ## 📊 Monitoring ### Cache Statistics ```python # Get cache stats cache_stats = server_initializer.slot_resolution_middleware.resolver.cache_service.get_stats() print(f"Cache size: {cache_stats['size']}/{cache_stats['max_size']}") print(f"Cache hit rate: {cache_stats.get('hit_rate', 0):.1%}") ``` ### Audit Logs Audit logs are automatically written to `slot_resolution_audit.log` (if enabled): ``` 2025-10-17 10:30:45 - Resolution: tenant=apolo, entity=impact, input='high', status=RESOLVED, id=2, latency=45ms 2025-10-17 10:30:46 - Resolution: tenant=apolo, entity=user, input='shivam', status=MULTIPLE_MATCHES, candidates=3, latency=120ms ``` --- ## 🚨 Error Handling ### Common Scenarios **1. No Match Found** ```python # Input: {"impact": "xyz123"} # Output: ValidationError if required field, or skip if optional ``` **2. Service Error** ```python # Elasticsearch down # Output: ResolutionStatus.SERVICE_ERROR with error message ``` **3. Invalid Input** ```python # Input: {"impact": ""} # Output: ResolutionStatus.INVALID_REQUEST ``` ### Graceful Degradation ```python try: result = await middleware.resolve_request(...) except Exception as e: logger.error(f"Slot resolution failed: {e}") # Fallback: Ask user to provide IDs directly return "Please provide entity IDs instead of names" ``` --- ## ✅ Checklist Before deploying to production: - [ ] Slot resolution initialized in `ServerInitializer` - [ ] Integrated into all create tools (request, problem, change) - [ ] Disambiguation response formatting implemented - [ ] Error handling added - [ ] Cache service enabled - [ ] Audit logging enabled - [ ] Unit tests written - [ ] Integration tests passing - [ ] Performance tested (latency < 300ms) - [ ] Documentation updated --- ## 🎉 Benefits After integration, users can: ✅ Use natural language for entity references ✅ Get intelligent disambiguation when needed ✅ Enjoy faster request creation (no manual ID lookup) ✅ Experience consistent behavior across all modules ✅ Benefit from caching for repeated queries **The system is production-ready and waiting for integration!** 🚀