Dynamic Per-User Tool Generation MCP Server

""" Slot Resolution System - Complete Example Demonstrates how to use the slot resolution system to transform user requests from entity names to IDs. """ import asyncio import logging from elasticsearch_search_lib import SearchClient from slot_resolution import SlotResolutionMiddleware from slot_resolution.core.resolver import SlotResolver from slot_resolution.core.decisioning import DecisionEngine from slot_resolution.services.elasticsearch_service import ElasticsearchMatchingService from slot_resolution.services.cache_service import CacheService from slot_resolution.services.audit_service import AuditService from slot_resolution.services.alias_service import AliasService from slot_resolution.services.exact_match_service import ExactMatchService # Setup logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) async def main(): """Run complete slot resolution example.""" print("=" * 80) print("SLOT RESOLUTION SYSTEM - COMPLETE EXAMPLE") print("=" * 80) # ======================================================================== # STEP 1: Initialize Components # ======================================================================== print("\n[1] Initializing components...") # Create Elasticsearch search client search_client = SearchClient(tenant_id="apolo") print(" ✓ Elasticsearch search client created") # Create services es_service = ElasticsearchMatchingService(search_client) exact_match_service = ExactMatchService() alias_service = AliasService() cache_service = CacheService(ttl_seconds=900, max_size=1000) audit_service = AuditService(log_to_file=False) print(" ✓ Services initialized") # Create decision engine with custom thresholds decision_engine = DecisionEngine( min_scores={ "impact": 0.65, "urgency": 0.65, "priority": 0.65, "user": 0.70 }, score_gap_delta=0.15, max_candidates=5 ) print(" ✓ Decision engine configured") # Create slot resolver resolver = SlotResolver( es_service=es_service, exact_match_service=exact_match_service, alias_service=alias_service, cache_service=cache_service, audit_service=audit_service, decision_engine=decision_engine ) print(" ✓ Slot resolver created") # Create middleware middleware = SlotResolutionMiddleware( tenant_id="apolo", resolver=resolver ) print(" ✓ Middleware initialized") # ======================================================================== # STEP 2: Example 1 - Successful Resolution # ======================================================================== print("\n[2] Example 1: Successful Resolution") print("-" * 80) request_1 = { "subject": "Laptop not working", "impact": "high", "urgency": "medium", "priority": "high" } print(f"Input Request: {request_1}") result_1 = await middleware.resolve_request( request_payload=request_1, module="request", user_id="user123" ) print(f"\nResult Status: {result_1.status}") if result_1.status == "READY": print(f"Transformed Payload: {result_1.payload}") print(f"\nAudit Trail:") for field, audit in result_1.audit_trail.items(): print(f" {field}: {audit}") # ======================================================================== # STEP 3: Example 2 - With Assignee (Potential Disambiguation) # ======================================================================== print("\n[3] Example 2: Request with Assignee") print("-" * 80) request_2 = { "subject": "Password reset needed", "impact": "low", "assignee": "shivam", "requester": "john.doe" } print(f"Input Request: {request_2}") result_2 = await middleware.resolve_request( request_payload=request_2, module="request", user_id="user123" ) print(f"\nResult Status: {result_2.status}") if result_2.status == "READY": print(f"Transformed Payload: {result_2.payload}") elif result_2.status == "DISAMBIGUATION_REQUIRED": print(f"Disambiguation Required for {len(result_2.disambiguations)} field(s):") for disambiguation in result_2.disambiguations: print(f"\n Field: {disambiguation.field}") print(f" Target: {disambiguation.target_field}") print(f" Original Input: '{disambiguation.original_input}'") print(f" Candidates:") for i, candidate in enumerate(disambiguation.candidates, 1): print(f" {i}. {candidate.canonical_name} (ID: {candidate.id}, Confidence: {candidate.confidence:.2f})") if candidate.attributes: print(f" Attributes: {candidate.attributes}") # ======================================================================== # STEP 4: Example 3 - Multiple Modules # ======================================================================== print("\n[4] Example 3: Different Module (Problem)") print("-" * 80) problem_request = { "subject": "Network outage", "impact": "high", "urgency": "high", "priority": "critical", "assignee": "network.admin" } print(f"Input Request: {problem_request}") result_3 = await middleware.resolve_request( request_payload=problem_request, module="problem", user_id="user123" ) print(f"\nResult Status: {result_3.status}") if result_3.status == "READY": print(f"Transformed Payload: {result_3.payload}") # ======================================================================== # STEP 5: Example 4 - Alias Resolution # ======================================================================== print("\n[5] Example 4: Alias Resolution") print("-" * 80) request_4 = { "subject": "Email not working", "impact": "hi", # Alias for "high" "urgency": "med", # Alias for "medium" "priority": "p2" # Alias for "high" } print(f"Input Request: {request_4}") print("Note: Using aliases 'hi', 'med', 'p2'") result_4 = await middleware.resolve_request( request_payload=request_4, module="request", user_id="user123" ) print(f"\nResult Status: {result_4.status}") if result_4.status == "READY": print(f"Transformed Payload: {result_4.payload}") # ======================================================================== # STEP 6: Cache Statistics # ======================================================================== print("\n[6] Cache Statistics") print("-" * 80) cache_stats = cache_service.get_stats() print(f"Cache Size: {cache_stats['size']}/{cache_stats['max_size']}") print(f"TTL: {cache_stats['ttl_seconds']} seconds") # ======================================================================== # STEP 7: Configuration Info # ======================================================================== print("\n[7] Configuration Information") print("-" * 80) available_modules = middleware.config_loader.get_available_modules() print(f"Available Modules: {', '.join(available_modules)}") for module in available_modules: mappings = middleware.config_loader.get_module_mappings(module) print(f"\n{module.upper()} Module:") print(f" Fields: {len(mappings)}") for mapping in mappings[:3]: # Show first 3 print(f" - {mapping.column_name} → {mapping.reference_to} ({mapping.db_key})") if len(mappings) > 3: print(f" ... and {len(mappings) - 3} more") print("\n" + "=" * 80) print("EXAMPLE COMPLETED SUCCESSFULLY") print("=" * 80) if __name__ == "__main__": """Run the example.""" asyncio.run(main())