ms-sentinel-mcp-server

Instructions

Implementation Reference

• tools/entra_id_tools.py:78-146 (handler)

  EntraIDGetUserTool class: defines the tool name 'entra_id_get_user' and implements the core handler logic in the 'run' method. Handles parameter extraction (user_id, upn, email), permission checks, optional user lookup by filter, and fetches user data from Microsoft Graph API.
  class EntraIDGetUserTool(EntraIDToolBase): """ Tool to get a user by object ID, UPN, or email address from Entra ID (Azure AD). Accepts any of: user_id, upn, or email. If user_id is not provided, resolves upn/email to user_id. """ name = "entra_id_get_user" description = ( "Get a user from Entra ID (Azure AD) by object ID, UPN, or email address." ) async def run(self, ctx: Context, **kwargs): self.check_graph_permissions() client = GraphApiClient() user_id = self._extract_param(kwargs, "user_id") upn = self._extract_param(kwargs, "upn") email = self._extract_param(kwargs, "email") if not user_id: filter_str = None if upn: filter_str = f"userPrincipalName eq '{upn}'" elif email: filter_str = f"mail eq '{email}'" if filter_str: url = f"{GRAPH_API_BASE}/users?$filter={filter_str}" try: # Use a unique name for this fetch to avoid duplicate function definition def fetch_user_by_filter(): for page in client.call_azure_rest_api("GET", url): users = page.get("value", []) if users: return users[0] return None user = await run_in_thread( fetch_user_by_filter, name="entra_id_get_user_lookup" ) if user and user.get("id"): user_id = user["id"] else: logger.error("No user found for filter: %s", filter_str) raise Exception(f"No user found for filter: {filter_str}") except requests.HTTPError as e: logger.error("Graph API error during user lookup: %s", e) if e.response.status_code == 403: raise Exception( "Permission denied: User.Read.All is required." ) from e raise else: logger.error("Missing required parameter: user_id, upn, or email") raise Exception("Missing required parameter: user_id, upn, or email") url = f"{GRAPH_API_BASE}/users/{user_id}" try: def fetch(): for page in client.call_azure_rest_api("GET", url): return page return await run_in_thread(fetch, name="entra_id_get_user") except requests.HTTPError as e: logger.error("Graph API error during user fetch: %s", e) if e.response.status_code == 403: raise Exception("Permission denied: User.Read.All is required.") from e raise
• tools/entra_id_tools.py:203-214 (registration)

  register_tools function registers the EntraIDGetUserTool (along with other Entra ID tools) by calling its register method on the MCP server instance.
  def register_tools(mcp): """ Register all Entra ID tools with the MCP server instance. Args: mcp: The MCP server instance. """ EntraIDListUsersTool.register(mcp) EntraIDGetUserTool.register(mcp) EntraIDListGroupsTool.register(mcp) EntraIDGetGroupTool.register(mcp)