splunk-mcp

Instructions

Implementation Reference

• splunk_mcp.py:455-542 (handler)

  The main handler function for the 'current_user' tool. It connects to Splunk, determines the current username from environment or context, retrieves the user object, extracts roles and other properties handling different formats, and returns a dictionary with user details including username, real_name, email, roles, capabilities, default_app, and type.
  @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()
• splunk_mcp.py:298-332 (helper)

  Helper function get_splunk_connection() used by current_user to establish connection to the Splunk service supporting both token and username/password authentication.
  def get_splunk_connection() -> splunklib.client.Service: """ Get a connection to the Splunk service. Supports both username/password and token-based authentication. If SPLUNK_TOKEN is set, it will be used for authentication and username/password will be ignored. Returns: splunklib.client.Service: Connected Splunk service """ try: if SPLUNK_TOKEN: logger.debug(f"🔌 Connecting to Splunk at {SPLUNK_SCHEME}://{SPLUNK_HOST}:{SPLUNK_PORT} using token authentication") service = splunklib.client.connect( host=SPLUNK_HOST, port=SPLUNK_PORT, scheme=SPLUNK_SCHEME, verify=VERIFY_SSL, token=f"Bearer {SPLUNK_TOKEN}" ) else: username = os.environ.get("SPLUNK_USERNAME", "admin") logger.debug(f"🔌 Connecting to Splunk at {SPLUNK_SCHEME}://{SPLUNK_HOST}:{SPLUNK_PORT} as {username}") service = splunklib.client.connect( host=SPLUNK_HOST, port=SPLUNK_PORT, username=username, password=SPLUNK_PASSWORD, scheme=SPLUNK_SCHEME, verify=VERIFY_SSL ) logger.debug(f"✅ Connected to Splunk successfully") return service except Exception as e: logger.error(f"❌ Failed to connect to Splunk: {str(e)}") raise