splunk-mcp

Instructions

Implementation Reference

• splunk_mcp.py:455-542 (handler)

  The handler function for the 'current_user' tool, decorated with @mcp.tool() for automatic registration. It retrieves the current Splunk user's details including username, real name, email, roles, capabilities, default app, and type by connecting to the Splunk service and querying user information.
  @mcp.tool() async def current_user() -> Dict[str, Any]: """ Get information about the currently authenticated user. This endpoint retrieves: - Basic user information (username, real name, email) - Assigned roles - Default app settings - User type Returns: Dict[str, Any]: Dictionary containing user information """ try: service = get_splunk_connection() logger.info("👤 Fetching current user information...") # First try to get username from environment variable current_username = os.environ.get("SPLUNK_USERNAME", "admin") logger.debug(f"Using username from environment: {current_username}") # Try to get additional context information try: # Get the current username from the /services/authentication/current-context endpoint current_context_resp = service.get("/services/authentication/current-context", **{"output_mode":"json"}).body.read() current_context_obj = json.loads(current_context_resp) if "entry" in current_context_obj and len(current_context_obj["entry"]) > 0: context_username = current_context_obj["entry"][0]["content"].get("username") if context_username: current_username = context_username logger.debug(f"Using username from current-context: {current_username}") except Exception as context_error: logger.warning(f"⚠️ Could not get username from current-context: {str(context_error)}") try: # Get the current user by username current_user = service.users[current_username] # Ensure roles is a list roles = [] if hasattr(current_user, 'roles') and current_user.roles: roles = list(current_user.roles) else: # Try to get from content if hasattr(current_user, 'content'): roles = current_user.content.get("roles", []) else: roles = current_user.get("roles", []) if roles is None: roles = [] elif isinstance(roles, str): roles = [roles] # Determine how to access user properties if hasattr(current_user, 'content') and isinstance(current_user.content, dict): user_info = { "username": current_user.name, "real_name": current_user.content.get('realname', "N/A") or "N/A", "email": current_user.content.get('email', "N/A") or "N/A", "roles": roles, "capabilities": current_user.content.get('capabilities', []) or [], "default_app": current_user.content.get('defaultApp', "search") or "search", "type": current_user.content.get('type', "user") or "user" } else: user_info = { "username": current_user.name, "real_name": current_user.get("realname", "N/A") or "N/A", "email": current_user.get("email", "N/A") or "N/A", "roles": roles, "capabilities": current_user.get("capabilities", []) or [], "default_app": current_user.get("defaultApp", "search") or "search", "type": current_user.get("type", "user") or "user" } logger.info(f"✅ Successfully retrieved current user information: {current_user.name}") return user_info except KeyError: logger.error(f"❌ User not found: {current_username}") raise ValueError(f"User not found: {current_username}") except Exception as e: logger.error(f"❌ Error getting current user: {str(e)}") raise
• splunk_mcp.py:455-455 (registration)

  The @mcp.tool() decorator registers the current_user function as an MCP tool.
  @mcp.tool()