Workspace ONE UEM MCP Server

""" Workspace ONE UEM MCP Server This MCP server provides access to commonly used Workspace ONE UEM APIs for device management, user management, and system operations. Authentication supports both OAuth2 and Basic Auth methods. """ from fastmcp import FastMCP import httpx from typing import Optional, Literal import os from datetime import datetime, timedelta # Initialize FastMCP server mcp = FastMCP("Workspace ONE UEM") # Global configuration BASE_URL = os.getenv("WS1_UEM_BASE_URL", "") # e.g., https://cn1506.awmdm.com API_KEY = os.getenv("WS1_UEM_API_KEY", "") # Tenant code CLIENT_ID = os.getenv("WS1_UEM_CLIENT_ID", "") CLIENT_SECRET = os.getenv("WS1_UEM_CLIENT_SECRET", "") TOKEN_URL = os.getenv("WS1_UEM_TOKEN_URL", "") # e.g., https://na.uemauth.vmwservices.com/connect/token USERNAME = os.getenv("WS1_UEM_USERNAME", "") PASSWORD = os.getenv("WS1_UEM_PASSWORD", "") # Token cache _access_token = None _token_expiry = None async def get_oauth_token() -> str: """Get or refresh OAuth access token.""" global _access_token, _token_expiry # Return cached token if still valid if _access_token and _token_expiry and datetime.now() < _token_expiry: return _access_token # Get new token 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"}, ) response.raise_for_status() data = response.json() _access_token = data["access_token"] # Set expiry with 5 minute buffer expires_in = data.get("expires_in", 3600) _token_expiry = datetime.now() + timedelta(seconds=expires_in - 300) return _access_token async def make_request( method: str, endpoint: str, use_oauth: bool = True, api_version: str = "2", **kwargs ) -> dict: """ Make an authenticated request to Workspace ONE UEM API. Args: method: HTTP method (GET, POST, PUT, DELETE) endpoint: API endpoint path (e.g., '/api/mdm/devices/search') use_oauth: Use OAuth if True, Basic Auth if False api_version: API version (1 or 2) **kwargs: Additional arguments for httpx request """ url = f"{BASE_URL}{endpoint}" headers = { "Accept": f"application/json;version={api_version}", "Content-Type": "application/json", "aw-tenant-code": API_KEY, } if use_oauth and TOKEN_URL: token = await get_oauth_token() headers["Authorization"] = f"Bearer {token}" elif USERNAME and PASSWORD: # Use Basic Auth pass # httpx will handle this with auth parameter async with httpx.AsyncClient() as client: if use_oauth or not (USERNAME and PASSWORD): response = await client.request(method, url, headers=headers, **kwargs) else: response = await client.request( method, url, headers=headers, auth=(USERNAME, PASSWORD), **kwargs ) response.raise_for_status() # Some endpoints return empty responses if response.status_code == 204 or not response.content: return {"status": "success"} return response.json() # =========================== # Device Management (MDM) APIs # =========================== @mcp.tool() async def search_devices( search_by: Optional[Literal["Serialnumber", "Udid", "Macaddress", "ImeiNumber", "EasId"]] = None, search_value: Optional[str] = None, user: Optional[str] = None, model: Optional[str] = None, platform: Optional[str] = None, last_seen_days: Optional[int] = None, ownership: Optional[Literal["C", "E", "S"]] = None, lgid: Optional[int] = None, page: int = 0, page_size: int = 500, ) -> dict: """ Search for devices in Workspace ONE UEM. This is one of the most commonly used APIs for finding and listing devices. You can search by specific identifiers or use filters to find groups of devices. Args: search_by: Type of search (Serialnumber, Udid, Macaddress, ImeiNumber, EasId) search_value: Value to search for (required if search_by is specified) user: Filter by username model: Filter by device model platform: Filter by platform (e.g., Apple, Android, WindowsPC) last_seen_days: Filter devices seen within this many days ownership: Filter by ownership type (C=Corporate, E=Employee, S=Shared) lgid: Filter by Location Group ID (Organization Group) page: Page number for pagination page_size: Number of results per page (max 500) Returns: Device search results with device details Examples: # Search by serial number search_devices(search_by="Serialnumber", search_value="C02ABC123") # Find all Apple devices search_devices(platform="Apple") # Find devices not seen in 30 days search_devices(last_seen_days=30) """ params = {"page": page, "pagesize": page_size} if search_by and search_value: params["searchby"] = search_by params["id"] = search_value if user: params["user"] = user if model: params["model"] = model if platform: params["platform"] = platform if last_seen_days: params["lastseen"] = last_seen_days if ownership: params["ownership"] = ownership if lgid: params["lgid"] = lgid return await make_request("GET", "/api/mdm/devices/search", params=params) @mcp.tool() async def get_device_details( search_by: Literal["Serialnumber", "Udid", "Macaddress", "ImeiNumber", "EasId", "DeviceId"], search_value: str, ) -> dict: """ Get detailed information about a specific device. This returns comprehensive device information including hardware details, OS version, compliance status, enrollment info, network details, and more. Args: search_by: Type of identifier to search by search_value: The identifier value Returns: Detailed device information Examples: # Get device by serial number get_device_details(search_by="Serialnumber", search_value="C02ABC123") # Get device by UDID get_device_details(search_by="Udid", search_value="12345678-ABCD...") """ return await make_request( "GET", f"/api/mdm/devices", params={"searchby": search_by, "id": search_value} ) @mcp.tool() async def send_device_command( command: Literal[ "DeviceLock", "DeviceWipe", "EnterpriseWipe", "ClearPasscode", "DeviceQuery", "SyncDevice", "RestartDevice", "ShutDownDevice" ], search_by: Literal["Serialnumber", "Udid", "Macaddress", "ImeiNumber", "DeviceId"], search_value: str, ) -> dict: """ Send a command to a specific device. Common commands include lock, wipe, sync, restart, and device query. Use with caution for destructive commands like DeviceWipe. Args: command: The command to send search_by: Type of identifier search_value: The identifier value Returns: Command execution result Examples: # Lock a device send_device_command("DeviceLock", "Serialnumber", "C02ABC123") # Request device info sync send_device_command("DeviceQuery", "Serialnumber", "C02ABC123") """ return await make_request( "POST", f"/api/mdm/devices/commands", params={ "command": command, "searchby": search_by, "id": search_value } ) @mcp.tool() async def bulk_device_command( command: Literal[ "DeviceLock", "DeviceWipe", "EnterpriseWipe", "DeviceQuery", "SyncDevice" ], serial_numbers: list[str], ) -> dict: """ Send a command to multiple devices at once. This is much more efficient than sending individual commands when you need to perform the same action on multiple devices. Args: command: The command to send to all devices serial_numbers: List of device serial numbers Returns: Bulk command result with accepted and failed items Example: bulk_device_command("DeviceQuery", ["SN001", "SN002", "SN003"]) """ return await make_request( "POST", f"/api/mdm/devices/commands/bulk", params={"command": command, "searchby": "Serialnumber"}, json={"BulkValues": {"Value": serial_numbers}} ) @mcp.tool() async def get_device_compliance( search_by: Literal["Serialnumber", "Udid", "Macaddress", "ImeiNumber", "DeviceId"], search_value: str, ) -> dict: """ Get compliance status and details for a specific device. Returns information about whether the device meets compliance policies, which policies are violated, and compliance history. Args: search_by: Type of identifier search_value: The identifier value Returns: Device compliance information """ return await make_request( "GET", f"/api/mdm/devices/compliance", params={"searchby": search_by, "id": search_value} ) @mcp.tool() async def get_device_profiles(device_id: int) -> dict: """ Get all profiles assigned to a specific device. Shows configuration profiles, restrictions, and settings applied to the device. Args: device_id: The UEM device ID (obtained from device search/details) Returns: List of profiles assigned to the device """ return await make_request("GET", f"/api/mdm/devices/{device_id}/profiles") # =========================== # Organization Group (System) APIs # =========================== @mcp.tool() async def search_organization_groups( name: Optional[str] = None, group_id: Optional[str] = None, type: Optional[str] = None, ) -> dict: """ Search for organization groups (OGs) in Workspace ONE UEM. Organization groups are used to organize devices, users, and content in a hierarchical structure. Args: name: Search by organization group name group_id: Search by group ID type: Filter by group type Returns: Organization group search results Example: search_organization_groups(name="Sales Department") """ params = {} if name: params["name"] = name if group_id: params["groupid"] = group_id if type: params["type"] = type return await make_request("GET", "/api/system/groups/search", params=params) @mcp.tool() async def get_organization_group_details(og_id: int) -> dict: """ Get detailed information about a specific organization group. Args: og_id: The organization group ID Returns: Organization group details """ return await make_request("GET", f"/api/system/groups/{og_id}") # =========================== # User Management (System) APIs # =========================== @mcp.tool() async def search_users( firstname: Optional[str] = None, lastname: Optional[str] = None, email: Optional[str] = None, username: Optional[str] = None, organizationgroup_id: Optional[int] = None, page: int = 0, page_size: int = 500, ) -> dict: """ Search for users in Workspace ONE UEM. Args: firstname: Filter by first name lastname: Filter by last name email: Filter by email address username: Filter by username organizationgroup_id: Filter by organization group page: Page number for pagination page_size: Number of results per page Returns: User search results Example: search_users(email="john.doe@company.com") """ params = {"page": page, "pagesize": page_size} if firstname: params["firstname"] = firstname if lastname: params["lastname"] = lastname if email: params["email"] = email if username: params["username"] = username if organizationgroup_id: params["organizationgroupid"] = organizationgroup_id return await make_request("GET", "/api/system/users/search", params=params) @mcp.tool() async def get_user_details(user_id: int) -> dict: """ Get detailed information about a specific user. Args: user_id: The UEM user ID Returns: User details including enrollment info and assigned devices """ return await make_request("GET", f"/api/system/users/{user_id}") @mcp.tool() async def get_user_devices(user_id: int) -> dict: """ Get all devices enrolled by a specific user. Args: user_id: The UEM user ID Returns: List of devices enrolled by the user """ return await make_request("GET", f"/api/system/users/{user_id}/devices") # =========================== # Tag Management (System) APIs # =========================== @mcp.tool() async def get_tags(organization_group_id: int) -> dict: """ Get all tags available in a specific organization group. Tags are commonly used to categorize and organize devices for automation, assignment, and reporting purposes. Args: organization_group_id: The organization group ID Returns: List of available tags Example: get_tags(organization_group_id=123) """ return await make_request( "GET", f"/api/system/groups/{organization_group_id}/tags" ) @mcp.tool() async def add_device_tag(device_id: int, tag_id: int) -> dict: """ Add a tag to a specific device. Args: device_id: The UEM device ID tag_id: The tag ID to add Returns: Operation result """ return await make_request( "POST", f"/api/mdm/devices/{device_id}/tags/{tag_id}" ) @mcp.tool() async def remove_device_tag(device_id: int, tag_id: int) -> dict: """ Remove a tag from a specific device. Args: device_id: The UEM device ID tag_id: The tag ID to remove Returns: Operation result """ return await make_request( "DELETE", f"/api/mdm/devices/{device_id}/tags/{tag_id}" ) # =========================== # Application Management (MAM) APIs # =========================== @mcp.tool() async def search_apps( application_name: Optional[str] = None, bundle_id: Optional[str] = None, category: Optional[str] = None, platform: Optional[str] = None, status: Optional[str] = None, organization_group_id: Optional[int] = None, ) -> dict: """ Search for applications in Workspace ONE UEM. Args: application_name: Filter by application name bundle_id: Filter by bundle/package ID category: Filter by category platform: Filter by platform (Apple, Android, etc.) status: Filter by status (Active, Inactive, etc.) organization_group_id: Filter by organization group Returns: Application search results Example: search_apps(application_name="Microsoft Teams", platform="Apple") """ params = {} if application_name: params["applicationname"] = application_name if bundle_id: params["bundleid"] = bundle_id if category: params["category"] = category if platform: params["platform"] = platform if status: params["status"] = status if organization_group_id: params["organizationgroupid"] = organization_group_id return await make_request("GET", "/api/mam/apps/search", params=params) @mcp.tool() async def get_device_apps(device_id: int) -> dict: """ Get all applications installed on a specific device. Args: device_id: The UEM device ID Returns: List of installed applications """ return await make_request("GET", f"/api/mdm/devices/{device_id}/apps") # =========================== # Event and Audit (System) APIs # =========================== @mcp.tool() async def get_device_events( device_id: int, page: int = 0, page_size: int = 100, ) -> dict: """ Get event history for a specific device. Shows enrollment events, command history, compliance changes, and other important device lifecycle events. Args: device_id: The UEM device ID page: Page number for pagination page_size: Number of results per page Returns: Device event history """ return await make_request( "GET", f"/api/mdm/devices/{device_id}/events", params={"page": page, "pagesize": page_size} ) @mcp.tool() async def search_events( user: Optional[str] = None, module: Optional[str] = None, event_type: Optional[str] = None, start_date: Optional[str] = None, end_date: Optional[str] = None, page: int = 0, page_size: int = 500, ) -> dict: """ Search system-wide events and audit logs. Useful for tracking admin activities, system changes, and troubleshooting. Args: user: Filter by username module: Filter by module (e.g., Device, Application, Admin) event_type: Filter by event type start_date: Start date (YYYY-MM-DD format) end_date: End date (YYYY-MM-DD format) page: Page number page_size: Number of results per page Returns: Event search results Example: search_events(module="Device", start_date="2024-01-01", end_date="2024-01-31") """ params = {"page": page, "pagesize": page_size} if user: params["user"] = user if module: params["module"] = module if event_type: params["eventtype"] = event_type if start_date: params["startdate"] = start_date if end_date: params["enddate"] = end_date return await make_request("GET", "/api/system/events/search", params=params) # =========================== # Smart Groups APIs # =========================== @mcp.tool() async def get_smart_groups(organization_group_id: Optional[int] = None) -> dict: """ Get smart groups (assignment groups) from Workspace ONE UEM. Smart groups are dynamic groups of devices based on criteria like platform, ownership, tags, etc. They're used for targeted deployments. Args: organization_group_id: Filter by organization group (optional) Returns: List of smart groups """ params = {} if organization_group_id: params["organizationgroupid"] = organization_group_id return await make_request("GET", "/api/mdm/smartgroups/search", params=params) @mcp.tool() async def get_smart_group_devices(smart_group_id: int) -> dict: """ Get all devices in a specific smart group. Args: smart_group_id: The smart group ID Returns: List of devices in the smart group """ return await make_request( "GET", f"/api/mdm/smartgroups/{smart_group_id}/devices" ) # =========================== # Helper/Utility Tools # =========================== @mcp.tool() async def get_api_version() -> dict: """ Get Workspace ONE UEM API version information. Useful for verifying connectivity and API compatibility. Returns: API version details """ return await make_request("GET", "/api/system/info", api_version="1") if __name__ == "__main__": # Run the MCP server mcp.run()