def METHOD_NAME(self, request, context):
    _suppress_unobserved_task_logging()
    # Configure the runtime so that the user program hooks up to Pulumi as appropriate.
    engine_address = request.args[0] if request.args else ""
    organization = request.organization if request.organization else "organization"
    reset_options(
        project=request.project,
        monitor_address=request.monitor_address,
        engine_address=engine_address,
        stack=request.stack,
        parallel=request.parallel,
        preview=request.dryRun,
        organization=organization,
    )
    if request.config:
        secret_keys = request.configSecretKeys if request.configSecretKeys else None
        set_all_config(request.config, secret_keys)
    # The strategy here is derived from sdk/python/cmd/pulumi-language-python-exec
    result = language_pb2.RunResponse()
    loop = asyncio.new_event_loop()
    loop.set_exception_handler(self._exception_handler)
    try:
        loop.run_until_complete(run_in_stack(self.program))
    except RunError as exn:
        msg = str(exn)
        log.error(msg)
        result.error = str(msg)
        return result
    except grpc.RpcError as exn:
        # If the monitor is unavailable, it is in the process of shutting down or has already
        # shut down. Don't emit an error if this is the case.
        # pylint: disable=no-member
        if exn.code() == grpc.StatusCode.UNAVAILABLE:
            log.debug("Resource monitor has terminated, shutting down.")
        else:
            msg = f"RPC error: {exn.details()}"
            log.error(msg)
            result.error = msg
            return result
    except Exception as exn:
        msg = str(
            f"python inline source runtime error: {exn}\n{traceback.format_exc()}"
        )
        log.error(msg)
        result.error = msg
        return result
    finally:
        # If there's an exception during `run_in_stack`, it may result in pending asyncio tasks remaining unresolved
        # at the time the loop is closed, which results in a `Task was destroyed but it is pending!` error being
        # logged to stdout. To avoid this, we collect all the unresolved tasks in the loop and cancel them before
        # closing the loop.
        pending = (
            # lint safety: we use the python version here to track deprecations
            # pylint: disable=no-member
            asyncio.Task.all_tasks(loop)
            if _py_version_less_than_3_7
            else asyncio.all_tasks(loop)
        )  # pylint: disable=no-member
        log.debug(f"Cancelling {len(pending)} tasks.")
        for task in pending:
            task.cancel()
            with suppress(asyncio.CancelledError):
                loop.run_until_complete(task)
        loop.close()
        sys.stdout.flush()
        sys.stderr.flush()
    return result