Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

Any tips on how to further investigate root cause for:
```Unexpected error: ValueError('Could not infer an active Flow context.')```

Hey <@U01PY53305D> -- you'll have to provide more information than this. Generally this means you did

```@task
def foo():
   pass

with Flow() as flow:
    pass

foo()```

Hi Jay - any chance you could post a minimum example of your flow code?

Typically this error happens when tasks are called outside flow context block. For example
```from prefect import task

@task
def my_task():
        print ('hi')

my_task_ = my_task()```

(note that it's _called_ not _initialized_)

Hey <@U01PY53305D> -- let's continue this conversation in the main thread so other people can benefit and help as well.

I've tried to improve that error message to help narrow down your problem, can you install my patch with `pip install git+<https://github.com/PrefectHQ/prefect.git@improve-flow-context-err>` then try again?

sure. thanks.  pip install failed.  too much STDERR to post here.

`pip install <https://github.com/PrefectHQ/prefect/archive/improve-flow-context-err.zip>` should work too

something to do with my pip trying to pull from our CodeArtifact pypi

(that second install failed for me actually :disappointed: looks like that's not an easy option)

there's nothing wrong with calling a task from within a task, correct?

If you do that you need to call `task.run()` not `task()`

Otherwise there wouldn't be a flow context (as this error says :smile: )

Hopefully adding the task name will help clarify things.. perhaps I should add even more information?

```@task()
def mytask2():
    print("task 2")

@task()
def mytask1():
    print("task 1")
    mytask2()

with Flow('myflow') as flow:
    mytask1()```
reasonable?

That will break -- you'll need to call `mytask2.run()`

When you call a task with `task()` you are creating an edge while defining a flow, not running it immediately.

```@task()
def mytask2():
    print("task 2")
    return {'status': 'good'}

@task()
def mytask1():
    print("task 1")
    t2_results = mytask2.run()
    print(f"task 2 returned status '{t2_results['status']}'")

with Flow('myflow') as flow:
    mytask1()```
ok, so like this.

When you do it this way though, the Prefect API will not know about `mytask2` so you won't get to see it in the UI and such.

Because of the returning of some result or because of running a task from within a task?

It's no longer a part of the 'flow' that we track -- it's just code you're executing in a task

so then why decorate it with @task?
should i make just make it a regular function?

You definitely could. Decorating it with `@task` might be misleading in this case. The only reason to do so is if you intend to add it to the flow later.