splunk-mcp

Instructions

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

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