Dynamic Per-User Tool Generation MCP Server

#!/usr/bin/env python3 """ Integration Test for Slot Resolution System This test verifies that the slot resolution system works correctly when integrated with the MCP server and can handle requests from AI agents and LLMs. """ import asyncio import json import logging from typing import Dict, Any # Setup logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) async def test_slot_resolution_with_names(): """ Test Case 1: Request with entity names (should be resolved to IDs) Simulates an AI agent sending a request with human-readable entity names like "high" for impact, "shivam" for assignee, etc. """ logger.info("=" * 80) logger.info("TEST 1: Slot Resolution with Entity Names") logger.info("=" * 80) from tool_execution_handler import get_execution_router router = get_execution_router() # Simulate request from AI agent with entity names request_data = { "subject": "Laptop not working - screen is black", "description": "My laptop screen went black suddenly. Need urgent help.", "impact": "high", # Entity name (should resolve to ID) "urgency": "high", # Entity name (should resolve to ID) "priority": "high", # Entity name (should resolve to ID) "assignee": "shivam", # User name (should resolve to technician ID) "category": "hardware", # Category name (should resolve to ID) "location": "bangalore", # Location name (should resolve to ID) } logger.info(f"Input Request (with entity names):") logger.info(json.dumps(request_data, indent=2)) # Execute the tool result = await router.execute( auth_token="test_user_token_12345", tool_name="create_request", **request_data ) logger.info(f"\nResult:") logger.info(json.dumps(result, indent=2)) # Verify the result if "error" in result: logger.error("❌ Test FAILED: Error in result") return False if result.get("status") == "disambiguation_required": logger.warning("⚠️ Disambiguation required (multiple matches found)") logger.info(f"Disambiguations: {json.dumps(result.get('disambiguations', {}), indent=2)}") return True # This is a valid outcome # Check if entity names were resolved to IDs expected_id_fields = ["impactId", "urgencyId", "priorityId", "technicianId", "categoryId", "locationId"] resolved_fields = [field for field in expected_id_fields if field in result] logger.info(f"\n✅ Resolved Fields: {resolved_fields}") if len(resolved_fields) > 0: logger.info("✅ TEST PASSED: Entity names were resolved to IDs") return True else: logger.warning("⚠️ TEST PARTIAL: No ID fields found (slot resolution may be disabled)") return True async def test_slot_resolution_with_ids(): """ Test Case 2: Request with entity IDs (should pass through unchanged) Simulates a request that already has IDs instead of names. Slot resolution should skip these fields. """ logger.info("\n" + "=" * 80) logger.info("TEST 2: Request with Entity IDs (Pass-through)") logger.info("=" * 80) from tool_execution_handler import get_execution_router router = get_execution_router() # Request with IDs already provided request_data = { "subject": "Network issue", "description": "Cannot connect to VPN", "impactId": 2, # Already an ID "urgencyId": 2, # Already an ID "priorityId": 3, # Already an ID "technicianId": 433, # Already an ID } logger.info(f"Input Request (with entity IDs):") logger.info(json.dumps(request_data, indent=2)) # Execute the tool result = await router.execute( auth_token="test_user_token_12345", tool_name="create_request", **request_data ) logger.info(f"\nResult:") logger.info(json.dumps(result, indent=2)) # Verify IDs are preserved if result.get("impactId") == 2 and result.get("urgencyId") == 2: logger.info("✅ TEST PASSED: Entity IDs passed through unchanged") return True else: logger.error("❌ TEST FAILED: Entity IDs were modified") return False async def test_slot_resolution_mixed(): """ Test Case 3: Request with mixed names and IDs Simulates a request with some fields as names and some as IDs. """ logger.info("\n" + "=" * 80) logger.info("TEST 3: Mixed Entity Names and IDs") logger.info("=" * 80) from tool_execution_handler import get_execution_router router = get_execution_router() # Mixed request request_data = { "subject": "Software installation request", "description": "Need Adobe Photoshop installed", "impact": "low", # Name (should resolve) "urgencyId": 3, # ID (should pass through) "assignee": "john", # Name (should resolve) "categoryId": 15, # ID (should pass through) } logger.info(f"Input Request (mixed):") logger.info(json.dumps(request_data, indent=2)) # Execute the tool result = await router.execute( auth_token="test_user_token_12345", tool_name="create_request", **request_data ) logger.info(f"\nResult:") logger.info(json.dumps(result, indent=2)) if "error" not in result: logger.info("✅ TEST PASSED: Mixed request handled successfully") return True else: logger.error("❌ TEST FAILED: Error in mixed request") return False async def test_slot_resolution_ambiguous(): """ Test Case 4: Request with ambiguous entity names Simulates a request where entity names match multiple records, requiring disambiguation. """ logger.info("\n" + "=" * 80) logger.info("TEST 4: Ambiguous Entity Names (Disambiguation)") logger.info("=" * 80) from tool_execution_handler import get_execution_router router = get_execution_router() # Request with potentially ambiguous names request_data = { "subject": "Test disambiguation", "description": "Testing ambiguous entity resolution", "assignee": "admin", # Common name, might have multiple matches "location": "office", # Generic name, might have multiple matches } logger.info(f"Input Request (potentially ambiguous):") logger.info(json.dumps(request_data, indent=2)) # Execute the tool result = await router.execute( auth_token="test_user_token_12345", tool_name="create_request", **request_data ) logger.info(f"\nResult:") logger.info(json.dumps(result, indent=2)) if result.get("status") == "disambiguation_required": logger.info("✅ TEST PASSED: Disambiguation correctly triggered") logger.info(f"Disambiguations provided: {list(result.get('disambiguations', {}).keys())}") return True elif "error" not in result: logger.info("✅ TEST PASSED: Request resolved without disambiguation") return True else: logger.error("❌ TEST FAILED: Error in ambiguous request") return False async def test_slot_resolution_invalid(): """ Test Case 5: Request with invalid/non-existent entity names Simulates a request with entity names that don't exist in the system. """ logger.info("\n" + "=" * 80) logger.info("TEST 5: Invalid Entity Names") logger.info("=" * 80) from tool_execution_handler import get_execution_router router = get_execution_router() # Request with invalid entity names request_data = { "subject": "Test invalid entities", "description": "Testing non-existent entity resolution", "impact": "super_ultra_critical", # Invalid impact value "assignee": "nonexistent_user_xyz123", # Non-existent user } logger.info(f"Input Request (with invalid names):") logger.info(json.dumps(request_data, indent=2)) # Execute the tool result = await router.execute( auth_token="test_user_token_12345", tool_name="create_request", **request_data ) logger.info(f"\nResult:") logger.info(json.dumps(result, indent=2)) # Should either return error or pass through original values logger.info("✅ TEST PASSED: Invalid entities handled gracefully") return True async def run_all_tests(): """Run all integration tests.""" logger.info("\n" + "=" * 80) logger.info("SLOT RESOLUTION INTEGRATION TEST SUITE") logger.info("=" * 80) logger.info("Testing slot resolution with AI agent/LLM scenarios\n") tests = [ ("Entity Names Resolution", test_slot_resolution_with_names), ("Entity IDs Pass-through", test_slot_resolution_with_ids), ("Mixed Names and IDs", test_slot_resolution_mixed), ("Ambiguous Names (Disambiguation)", test_slot_resolution_ambiguous), ("Invalid Entity Names", test_slot_resolution_invalid), ] results = [] for test_name, test_func in tests: try: result = await test_func() results.append((test_name, result)) except Exception as e: logger.error(f"❌ Test '{test_name}' raised exception: {e}", exc_info=True) results.append((test_name, False)) # Print summary logger.info("\n" + "=" * 80) logger.info("TEST SUMMARY") logger.info("=" * 80) passed = sum(1 for _, result in results if result) total = len(results) for test_name, result in results: status = "✅ PASSED" if result else "❌ FAILED" logger.info(f"{status}: {test_name}") logger.info(f"\nTotal: {passed}/{total} tests passed") logger.info("=" * 80) return passed == total if __name__ == "__main__": # Run the tests success = asyncio.run(run_all_tests()) if success: logger.info("\n🎉 All integration tests passed!") exit(0) else: logger.error("\n❌ Some tests failed") exit(1)