Based on the research and the issue you're describing, I can help explain the proper way to handle tags with async flows in Prefect 3.x. The issue you're experiencing is likely related to how async context managers work. For async flows, you need to use the async context manager version of tags. Here's the correct way to do it:

python
from prefect import flow, tags

@flow
async def my_flow():
    # Flow logic here
    pass

# Option 1: Using async with
async def main():
    async with tags("tag1", "tag2"):
        await my_flow()

# Option 2: Directly on the flow decorator
@flow(tags=["tag1", "tag2"])
async def my_flow_with_tags():
    # Flow logic here
    pass

The preferred and more straightforward approach is Option 2 - adding tags directly to the flow decorator. This ensures the tags are always associated with the flow regardless of how it's called. If you need to add tags dynamically at runtime, you can use the async context manager approach (Option 1), but make sure you're using it in an async context. If you're still seeing empty TagsContext inside

create_flow_run()

, it would be helpful to see: 1. The complete flow definition 2. How exactly you're calling the flow 3. Where/how you're checking the TagsContext This would help identify if there's a specific issue with how the context is being propagated in your case. Also, note that if you're using the flow in a deployment, you can add tags at the deployment level which will be merged with any tags specified on the flow. Would you mind sharing more details about your specific use case and where exactly you're trying to access the TagsContext? This would help me provide a more targeted solution.

You're absolutely right - I apologize for my previous incorrect response. The

ContextModel

(which

TagsContext

inherits from) is indeed a synchronous context manager, and this explains why you're seeing empty context when using it with

async with

. This appears to be a limitation in the current implementation. Here are the approaches that should work reliably: 1. Use tags directly on the flow decorator (most reliable):

python
@flow(tags=["tag1", "tag2"])
async def my_flow():
    # Flow logic here
    pass

2. Use a regular

with

statement before the await:

python
async def main():
    with tags("tag1", "tag2"):
        await my_flow()

Would you be willing to share the specific use case where you need to access the

TagsContext

inside

create_flow_run()

? There might be alternative approaches depending on what you're trying to achieve. Also, if you believe this is a limitation that should be addressed (i.e., better support for async context management of tags), it would be valuable to open an issue on the Prefect GitHub repository. The fact that the context manager is synchronous but needs to work with async flows could be something worth improving in future versions.