@ancient-wolf-19325 Hey, I want to implement a retry login upon x amount of fail in my code that will calls an API with a specific flag set to true instead of false(default). I am trying to store the retries in the context something like this:

# Get current execution context to check retry attempt
    ctx = fl.current_context()
    current_attempt = getattr(ctx.execution_state, 'retries', 0) + 1

but its not working. How can I store a value between task tries and set a custom logic based on its value ?

Flyte does not support directly storing or mutating custom values in the execution context between task retries. The context object (current_context) exposes read-only information about the current attempt, but you cannot persist custom state across retries this way. To persist state between retries, you should use Flyte's checkpointing mechanism, which allows you to save and restore state within a task execution, enabling custom retry logic based on the number of attempts or other criteria. See the checkpointing example for details on how to use this feature in your task code. For example, you can use current_context().checkpoint to read and write progress, and then implement your retry logic accordingly. See the code and explanation in the Flyte intratask checkpointing documentation and checkpoint.py example. Would you like a step-by-step example of how to implement this pattern? Sources: - https://github.com/flyteorg/flyte/blob/master/docs/user_guide/advanced_composition/intratask_checkpoints.md - https://github.com/flyteorg/flytesnacks/blob/master/examples/advanced_composition/advanced_composition/checkpoint.py 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

yes please @ancient-wolf-19325, I mean its enough for me if I can get the current retry number for that task and I can wire my custom logic with it. Can you show me that ?

It seems likely that you can access the current retry attempt for a task using Flyte's execution context. The recommended way is to use current_context().execution_state.attempt_number, which provides the current attempt (starting from 0). You can then use this value in your custom logic within the task. However, you cannot mutate this value or persist custom data in the context between retries—it's read-only and managed by Flyte. Here’s a minimal example:

from flytekit import task, current_context

@task(retries=3)
def my_task():
    attempt = current_context().execution_state.attempt_number
    print(f"Current attempt: {attempt}")
    # Your custom logic here

This will print the current retry attempt each time the task is retried. For more details, see the Flyte intratask checkpointing documentation. Would you like a more advanced example, such as combining this with checkpointing or custom retry logic? Sources: - Flyte intratask checkpointing documentation 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325 it says that it is not existing

current_attempt = ctx.execution_state.attempt_number\n                          ^^^^^^^^^^^^^^^^^^^\n      File \ │
│ "/opt/venv/lib/python3.12/site-packages/flytekit/core/context_manager.py\", line 330, in __getattr__\n        raise AssertionError(f\"{attr_name} not available as a parameter in Flyte context -  │
│ are you in right task-type?\")\n    AssertionError: EXECUTION_STATE not available as a parameter in Flyte context - are you in right task-type?\n\nMessage:\n\n    AssertionError: EXECUTION_STATE │
│  not available as a parameter in Flyte context - are you in right task-type?"}

It could be possible that the execution context (ctx.execution_state) is not available in all environments or modes—especially during local execution or outside of a Flyte task. The error message suggests that EXECUTION_STATE is not present in the current context, which is expected if you are not running inside a Flyte task container or are using an unsupported execution mode. This attribute is only reliably available during actual task execution on the Flyte backend, not in local runs or certain interactive/debugging scenarios (context_manager.py). If you need to test retry logic locally, you may need to simulate it manually, or ensure you are running the code in an environment where Flyte sets up the full execution context. Would you like more detail on how to structure your code to avoid this error, or on how to test retry logic in different environments? Sources: - context_manager.py 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325 this is the error

│ An unexpected error occurred: EXECUTION_STATE not available as a parameter in Flyte context - are you in right task-type?                                                                          │

The Flyte documentation I have ingested does not mention how to access the current retry attempt from the execution context in a way that is guaranteed to work across all environments, and specifically, it does not describe how to avoid the "EXECUTION_STATE not available as a parameter in Flyte context - are you in right task-type?" error. This error typically occurs if you try to access execution_state outside of a Flyte task execution environment, such as during local runs or in unsupported contexts. If you can provide more details about how and where you are running your code (e.g., locally, in a notebook, or on a Flyte backend), it may help in diagnosing the issue. If this doesn't help, please reach out to the Flyte support team for further assistance. 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.