@task(max_retries=3, retry_delay=timedelta(seconds=10))
def send_task(id):
...
...
if config.get("status") == "unconnect":
        client = prefect.client.client.Client()
        flow_run_id = prefect.context.flow_run_id
        skip_state = prefect.engine.state.Skipped()
        logger.info(flow_run_id)
        result = client.set_flow_run_state(flow_run_id=flow_run_id, state=skip_state)
        logger.info("--------------")
        logger.info(result)
        raise prefect.engine.signals.ENDRUN(skip_state)