Kroger MCP Server

Instructions

    Test if the current authentication token is valid.
    
    Returns:
        Dictionary indicating authentication status
    

Implementation Reference

• src/kroger_mcp/tools/profile_tools.py:55-99 (handler)

  The main execution logic for the 'test_authentication' tool. It checks the validity of the current authentication token using the shared authenticated client, provides detailed status including refresh token availability, and handles errors gracefully.

  @mcp.tool()
  async def test_authentication(ctx: Context = None) -> Dict[str, Any]:
      """
      Test if the current authentication token is valid.
      
      Returns:
          Dictionary indicating authentication status
      """
      if ctx:
          await ctx.info("Testing authentication token validity")
      
      try:
          client = get_authenticated_client()
          is_valid = client.test_current_token()
          
          if ctx:
              await ctx.info(f"Authentication test result: {'valid' if is_valid else 'invalid'}")
          
          result = {
              "success": True,
              "token_valid": is_valid,
              "message": f"Authentication token is {'valid' if is_valid else 'invalid'}"
          }
          
          # Check for refresh token availability
          if hasattr(client.client, 'token_info') and client.client.token_info:
              has_refresh_token = "refresh_token" in client.client.token_info
              result["has_refresh_token"] = has_refresh_token
              result["can_auto_refresh"] = has_refresh_token
              
              if has_refresh_token:
                  result["message"] += ". Token can be automatically refreshed when it expires."
              else:
                  result["message"] += ". No refresh token available - will need to re-authenticate when token expires."
          
          return result
          
      except Exception as e:
          if ctx:
              await ctx.error(f"Error testing authentication: {str(e)}")
          return {
              "success": False,
              "error": str(e),
              "token_valid": False
          }

• src/kroger_mcp/server.py:76-76 (registration)

  The registration point where profile_tools.register_tools(mcp) is called, which defines and registers the test_authentication tool using FastMCP's @mcp.tool() decorator.

  profile_tools.register_tools(mcp)

• src/kroger_mcp/tools/shared.py:56-118 (helper)

  Supporting helper function get_authenticated_client() that provides the authenticated KrogerAPI client instance, handling token loading, validation, and refresh. Called directly in test_authentication.

  def get_authenticated_client() -> KrogerAPI:
      """Get or create a user-authenticated client for cart operations
      
      This function attempts to load an existing token or prompts for authentication.
      In an MCP context, the user needs to explicitly call start_authentication and
      complete_authentication tools to authenticate.
      
      Returns:
          KrogerAPI: Authenticated client
          
      Raises:
          Exception: If no valid token is available and authentication is required
      """
      global _authenticated_client
      
      if _authenticated_client is not None and _authenticated_client.test_current_token():
          # Client exists and token is still valid
          return _authenticated_client
      
      # Clear the reference if token is invalid
      _authenticated_client = None
      
      try:
          load_and_validate_env(["KROGER_CLIENT_ID", "KROGER_CLIENT_SECRET", "KROGER_REDIRECT_URI"])
          
          # Try to load existing user token first
          token_file = ".kroger_token_user.json"
          token_info = load_token(token_file)
          
          if token_info:
              # Create a new client with the loaded token
              _authenticated_client = KrogerAPI()
              _authenticated_client.client.token_info = token_info
              _authenticated_client.client.token_file = token_file
              
              if _authenticated_client.test_current_token():
                  # Token is valid, use it
                  return _authenticated_client
              
              # Token is invalid, try to refresh it
              if "refresh_token" in token_info:
                  try:
                      _authenticated_client.authorization.refresh_token(token_info["refresh_token"])
                      # If refresh was successful, return the client
                      if _authenticated_client.test_current_token():
                          return _authenticated_client
                  except Exception:
                      # Refresh failed, need to re-authenticate
                      _authenticated_client = None
          
          # No valid token available, need user-initiated authentication
          raise Exception(
              "Authentication required. Please use the start_authentication tool to begin the OAuth flow, "
              "then complete it with the complete_authentication tool."
          )
      except Exception as e:
          if "Authentication required" in str(e):
              # This is an expected error when authentication is needed
              raise
          else:
              # Other unexpected errors
              raise Exception(f"Authentication failed: {str(e)}")