code: def _run_plugin(run: _Run) -> Dict[str, Any]:
      from lightning.app.runners.cloud import _to_clean_dict
  
      """Create a run with the given name and entrypoint under the cloudspace with the given ID."""
      with tempfile.TemporaryDirectory() as tmpdir:
          download_path = os.path.join(tmpdir, "source.tar.gz")
          source_path = os.path.join(tmpdir, "source")
          os.makedirs(source_path)
  
          # Download the tarball
          try:
              logger.info(
                  f"Downloading plugin source: {run.source_code_url}"
              )
  
              # Sometimes the URL gets encoded, so we parse it here
              source_code_url = urlparse(run.source_code_url).geturl()
  
              response = requests.get(source_code_url)
  
              # TODO: Backoff retry a few times in case the URL is flaky
              response.raise_for_status()
  
              with open(download_path, "wb") as f:
                  f.write(response.content)
          except Exception as ex:
              raise HTTPException(
                  status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                  detail=f"Error downloading plugin source: {str(ex)}.",
              )
  
          # Extract
          try:
              logger.info("Extracting plugin source.")
  
              with tarfile.open(download_path, "r:gz") as tf:
                  tf.extractall(source_path)  # noqa: S202
          except Exception as ex:
              raise HTTPException(
                  status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                  detail=f"Error extracting plugin source: {str(ex)}.",
              )
  
          # Import the plugin
          try:
              logger.info(f"Importing plugin: {run.plugin_entrypoint}")
  
              plugin = _load_plugin_from_file(
                  os.path.join(source_path, run.plugin_entrypoint)
              )
          except Exception as ex:
              raise HTTPException(
                  status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                  detail=f"Error loading plugin: {str(ex)}.",
              )
  
          # Allow devs to add files to the app source
          if os.path.isdir(_PLUGIN_INTERNAL_DIR_PATH):
              shutil.copytree(
                  _PLUGIN_INTERNAL_DIR_PATH,
                  source_path,
                  dirs_exist_ok=True,
              )
  
          # Ensure that apps are dispatched from the temp directory
          cwd = os.getcwd()
          os.chdir(source_path)
  
          # Setup and run the plugin
          try:
              logger.info(
                  "Running plugin. "
                  f"project_id: {run.project_id}, cloudspace_id: {run.cloudspace_id}, cluster_id: {run.cluster_id}."
              )
  
              plugin._setup(
                  project_id=run.project_id,
                  cloudspace_id=run.cloudspace_id,
                  cluster_id=run.cluster_id,
                  source_app=run.source_app,
              )
              app_instance = plugin.run(**run.plugin_arguments)
              return _to_clean_dict(app_instance, True)
          except Exception as ex:
              raise HTTPException(
                  status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                  detail=f"Error running plugin: {str(ex)}.",
              )
          finally:
              os.chdir(cwd)
  
imports: import os
  import shutil
  import tarfile
  import tempfile
  from pathlib import Path
  from typing import Any, Dict
  from urllib.parse import urlparse
  import requests
  import uvicorn
  from fastapi import FastAPI, HTTPException, status
  from fastapi.middleware.cors import CORSMiddleware
  from lightning_cloud.openapi import Externalv1LightningappInstance
  from pydantic import BaseModel
  from lightning.app.utilities.app_helpers import Logger
  from lightning.app.utilities.component import _set_flow_context
  from lightning.app.utilities.enum import AppStage
  from lightning.app.utilities.load_app import _load_plugin_from_file
label_desc: Use image node_id 1 for calling a global function with example usage: _run_plugin(run) and returns: _to_clean_dict