test_connection.py•9.34 kB
#!/usr/bin/env python3
"""
Workspace ONE UEM MCP Server Test Script
This script tests your configuration and verifies connectivity to Workspace ONE UEM.
Run this before adding the server to Claude Desktop to ensure everything works.
Usage:
python test_connection.py
"""
import asyncio
import os
from dotenv import load_dotenv
import httpx
from datetime import datetime, timedelta
# Load environment variables
load_dotenv()
# Get configuration
BASE_URL = os.getenv("WS1_UEM_BASE_URL", "")
API_KEY = os.getenv("WS1_UEM_API_KEY", "")
CLIENT_ID = os.getenv("WS1_UEM_CLIENT_ID", "")
CLIENT_SECRET = os.getenv("WS1_UEM_CLIENT_SECRET", "")
TOKEN_URL = os.getenv("WS1_UEM_TOKEN_URL", "")
USERNAME = os.getenv("WS1_UEM_USERNAME", "")
PASSWORD = os.getenv("WS1_UEM_PASSWORD", "")
# ANSI color codes
GREEN = "\033[92m"
RED = "\033[91m"
YELLOW = "\033[93m"
BLUE = "\033[94m"
RESET = "\033[0m"
def print_header(text):
"""Print a formatted header."""
print(f"\n{BLUE}{'=' * 60}{RESET}")
print(f"{BLUE}{text.center(60)}{RESET}")
print(f"{BLUE}{'=' * 60}{RESET}\n")
def print_success(text):
"""Print success message."""
print(f"{GREEN}✓ {text}{RESET}")
def print_error(text):
"""Print error message."""
print(f"{RED}✗ {text}{RESET}")
def print_warning(text):
"""Print warning message."""
print(f"{YELLOW}⚠ {text}{RESET}")
def print_info(text):
"""Print info message."""
print(f" {text}")
async def test_oauth_token():
"""Test OAuth token retrieval."""
try:
print_info("Testing OAuth authentication...")
async with httpx.AsyncClient() as client:
response = await client.post(
TOKEN_URL,
data={
"grant_type": "client_credentials",
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=10.0,
)
response.raise_for_status()
data = response.json()
if "access_token" in data:
print_success("OAuth token retrieved successfully")
print_info(f"Token expires in: {data.get('expires_in', 'unknown')} seconds")
return data["access_token"]
else:
print_error("OAuth response missing access_token")
return None
except httpx.HTTPStatusError as e:
print_error(f"OAuth authentication failed: {e.response.status_code}")
print_info(f"Response: {e.response.text}")
return None
except Exception as e:
print_error(f"OAuth error: {str(e)}")
return None
async def test_api_call(use_oauth=True, token=None):
"""Test a basic API call."""
try:
endpoint = "/api/system/info"
url = f"{BASE_URL}{endpoint}"
headers = {
"Accept": "application/json;version=1",
"aw-tenant-code": API_KEY,
}
if use_oauth and token:
print_info("Testing API call with OAuth...")
headers["Authorization"] = f"Bearer {token}"
auth = None
else:
print_info("Testing API call with Basic Auth...")
auth = (USERNAME, PASSWORD)
async with httpx.AsyncClient() as client:
response = await client.get(
url,
headers=headers,
auth=auth,
timeout=10.0,
)
response.raise_for_status()
data = response.json()
print_success("API call successful")
if "ProductName" in data:
print_info(f"Product: {data.get('ProductName')}")
if "Version" in data:
print_info(f"Version: {data.get('Version')}")
return True
except httpx.HTTPStatusError as e:
print_error(f"API call failed: {e.response.status_code}")
print_info(f"Response: {e.response.text}")
return False
except Exception as e:
print_error(f"API error: {str(e)}")
return False
async def test_device_search():
"""Test device search API."""
try:
print_info("Testing device search...")
# Get OAuth token if configured
token = None
if CLIENT_ID and CLIENT_SECRET and TOKEN_URL:
token = await test_oauth_token()
if not token:
return False
endpoint = "/api/mdm/devices/search"
url = f"{BASE_URL}{endpoint}"
headers = {
"Accept": "application/json;version=2",
"aw-tenant-code": API_KEY,
}
if token:
headers["Authorization"] = f"Bearer {token}"
auth = None
else:
auth = (USERNAME, PASSWORD)
async with httpx.AsyncClient() as client:
response = await client.get(
url,
headers=headers,
auth=auth,
params={"page": 0, "pagesize": 5},
timeout=10.0,
)
response.raise_for_status()
data = response.json()
print_success("Device search successful")
device_count = data.get("Total", 0)
print_info(f"Total devices: {device_count}")
if device_count > 0 and "Devices" in data:
print_info(f"Sample device: {data['Devices'][0].get('DeviceFriendlyName', 'Unknown')}")
return True
except httpx.HTTPStatusError as e:
print_error(f"Device search failed: {e.response.status_code}")
print_info(f"Response: {e.response.text}")
return False
except Exception as e:
print_error(f"Device search error: {str(e)}")
return False
async def main():
"""Main test function."""
print_header("Workspace ONE UEM MCP Server Test")
# Step 1: Check configuration
print_header("Step 1: Checking Configuration")
config_ok = True
if BASE_URL:
print_success(f"Base URL: {BASE_URL}")
else:
print_error("Base URL not configured (WS1_UEM_BASE_URL)")
config_ok = False
if API_KEY:
print_success(f"API Key: {API_KEY[:10]}...")
else:
print_error("API Key not configured (WS1_UEM_API_KEY)")
config_ok = False
# Check authentication method
oauth_configured = bool(CLIENT_ID and CLIENT_SECRET and TOKEN_URL)
basic_configured = bool(USERNAME and PASSWORD)
if oauth_configured:
print_success("OAuth credentials configured")
print_info(f"Client ID: {CLIENT_ID[:10]}...")
print_info(f"Token URL: {TOKEN_URL}")
elif basic_configured:
print_warning("Using Basic Auth (OAuth recommended for production)")
print_info(f"Username: {USERNAME}")
else:
print_error("No authentication configured")
print_error("Configure either OAuth (CLIENT_ID/CLIENT_SECRET) or Basic Auth (USERNAME/PASSWORD)")
config_ok = False
if not config_ok:
print_error("\nConfiguration incomplete. Please check your .env file.")
return
# Step 2: Test OAuth (if configured)
if oauth_configured:
print_header("Step 2: Testing OAuth Authentication")
token = await test_oauth_token()
if not token:
print_error("OAuth authentication failed. Check credentials and token URL.")
return
else:
print_header("Step 2: OAuth Not Configured (Skipping)")
token = None
# Step 3: Test API connectivity
print_header("Step 3: Testing API Connectivity")
if oauth_configured and token:
api_ok = await test_api_call(use_oauth=True, token=token)
elif basic_configured:
api_ok = await test_api_call(use_oauth=False)
else:
api_ok = False
if not api_ok:
print_error("API connectivity test failed. Check credentials and network.")
return
# Step 4: Test device search
print_header("Step 4: Testing Device Search API")
search_ok = await test_device_search()
# Final summary
print_header("Test Summary")
if config_ok and api_ok and search_ok:
print_success("All tests passed! ✓")
print_info("\nYour configuration is working correctly.")
print_info("You can now add this server to Claude Desktop.")
print_info("\nNext steps:")
print_info("1. Add the server to your Claude Desktop config")
print_info("2. Restart Claude Desktop")
print_info("3. Try asking Claude to search for devices")
else:
print_error("Some tests failed. Please review the errors above.")
print_info("\nTroubleshooting tips:")
print_info("- Verify all credentials in .env file")
print_info("- Check BASE_URL has no trailing slash")
print_info("- Ensure API key is from the correct OG")
print_info("- Verify OAuth client is enabled in UEM")
print_info("- Check network connectivity to UEM instance")
if __name__ == "__main__":
asyncio.run(main())