taskflow

library6.3.0pypypi✓ verified 85d ago

Taskflow is a Python 3.10+ library for structured state management, task orchestration, and error handling in complex asynchronous workflows. It enables defining tasks and flows using decorators or classes, facilitating robust and scalable application development. The current version is 6.2.0, with an active development cadence that includes significant API changes in major releases.

workflow database devops testing data

pip install taskflow

package ↗homepage ↗

INSTALL

IMPORT

SIG · TASKFLOW

taskflow

workflowpythonv6.3.0

Install

6.4s avg

Import

584ms

Disk

72MB

Pass rate

10/ 10

Env Coverage10 / 10

glibc

3.9–3.13

musl

3.9–3.13

Install & Compatibility

Where this runs

tested against v6.3.0 · pip install

no network on importno background threads

Install × environment matrix

Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux

musl

py 3.10–3.910 runs

installs and imports cleanly · install 0.0s · import 0.598s · 68.3MB

glibc

py 3.10–3.910 runs

installs and imports cleanly · install 6.4s · import 0.570s · 70MB

72MB installed

● package 72MB

Code

Verified usage

Verified import paths — ran on the pinned version, not inferred.

task

✓ from taskflow import task

flow

✓ from taskflow import flow

Flow

✓ from taskflow import Flow

Task

✓ from taskflow import Task

FlowConfig

✓ from taskflow import FlowConfig

This quickstart demonstrates defining and executing a simple asynchronous task and flow using Taskflow's decorator API. It showcases the basic structure for creating a workflow and running it with asyncio.

import asyncio from taskflow import task, flow @task async def greet_task(name: str) -> str: """A simple task that greets a given name.""" print(f"Task received: {name}") return f"Hello, {name}!" @flow async def greeting_flow(input_name: str) -> str: """A flow that orchestrates the greeting task.""" # Inputs/outputs are now dictionaries in v6+ greeting_message = await greet_task(input_name) print(f"Flow processed: {greeting_message}") return f"Flow finished with: {greeting_message}" async def main(): name_to_greet = "World" final_output = await greeting_flow(name_to_greet) print(f"Final output from flow: {final_output}") if __name__ == "__main__": asyncio.run(main())

taskflow --version

Debug

Known issues

breakingTask and Flow input/output declarations changed significantly in v6.0.0. Previously, inputs were declared as direct keyword arguments (e.g., `@task(name=str)`). Now, they must be dictionaries (e.g., `@task(inputs={"name": str})`).

fix

Update task and flow decorators/constructors to use `inputs` and `outputs` parameters as dictionaries, mapping argument names to types.

affects: >=6.0.0

breakingThe `LocalThreadPoolExecutor` and `ProcessThreadPoolExecutor` options for `FlowConfig` were removed in v6.0.0. The default executor is now `AsyncIOExecutor`.

fix

Remove references to the deprecated executors. If non-asyncio concurrency is needed, consider integrating external solutions or implement a custom executor that wraps `AsyncIOExecutor`.

affects: >=6.0.0

breakingConfiguration parameters for `Flow` and `Task` are no longer passed directly to their constructors or decorators. All configuration is now done via `FlowConfig` or `TaskConfig` objects.

fix

When instantiating `Flow` or `Task` classes, pass a `config` object (e.g., `Flow(config=FlowConfig(max_concurrency=5))`). For decorators, use `flow(config=FlowConfig(...))`.

affects: >=6.0.0

deprecatedThe base classes `AbstractFlow` and `AbstractTask` were removed in v6.0.0. Users subclassing these should now subclass `Flow` and `Task` directly.

fix

Migrate custom flow and task implementations to directly inherit from `taskflow.Flow` and `taskflow.Task` respectively.

affects: >=6.0.0

gotchaTaskflow v6.x introduced stricter type enforcement. Mismatched type annotations between task signatures, input/output declarations, and actual data types can lead to runtime `TypeError` exceptions.

fix

Ensure all type annotations in task and flow definitions are accurate and consistent with the data types being passed and returned. Validate data types at runtime if external inputs are untrusted.

affects: >=6.0.0

Errors

Common errors & fixes

TypeError: task() got an unexpected keyword argument 'my_input'

Attempting to declare task inputs as direct keyword arguments in Taskflow v6.x or newer, which is the pre-v6 syntax.

fix

Declare inputs using the `inputs` dictionary parameter: `@task(inputs={'my_input': str})`.

AttributeError: 'FlowConfig' object has no attribute 'executor' or similar configuration errors.

Accessing or configuring executors (e.g., `LocalThreadPoolExecutor`) or other flow/task settings using pre-v6 methods or referencing removed components.

fix

The `LocalThreadPoolExecutor` and `ProcessThreadPoolExecutor` were removed. Use the `AsyncIOExecutor` (default) or implement a custom one. For general config, pass a `FlowConfig` instance.

ModuleNotFoundError: No module named 'taskflow'

The `taskflow` library is not installed in the current Python environment or the environment is not activated.

fix

Install the library using pip: `pip install taskflow`. Ensure your virtual environment is activated if you are using one.

TypeError: Flow.__init__() got an unexpected keyword argument 'max_concurrency'

In Taskflow v6.x+, direct configuration arguments for `Flow` or `Task` constructors (like `max_concurrency`) were removed.

fix

Pass configuration parameters via a `FlowConfig` object: `my_flow = Flow(config=FlowConfig(max_concurrency=5))`.

Upgrade

Version history

6.3.0latest on PyPI · released May 18, 2026

Audit

Dependencies

No dependency data recorded yet.

Agent activity

33 hits · last 30 days

node

OpenAI (training)

Resources

packagehttps://pypi.org/project/taskflow/ ↗

homepagehttps://docs.openstack.org/taskflow ↗