What is FlowKitX?

FlowKitX is a Python library that makes writing async workflows more readable and structured.

It provides lazy pipelines: operations (like API calls) are queued and executed only when awaited. This lets you chain calls and process data fluently without blocking the event loop.

It’s not magic — it doesn’t convert blocking code to async or bypass asyncio fundamentals. It’s built for I/O-bound workflows, especially APIs and data pipelines.

Why use it?

Traditional async code can get messy:

async def fetch_user_data(user_id): async with aiohttp.ClientSession() as session: async with session.get(f"https://api.example.com/users/{user_id}") as response: user = await response.json() async with session.get("https://api.example.com/posts", params={"userId": user_id}) as posts_response: posts = await posts_response.json() return {"user": user, "posts": posts}

With FlowKitX:

@flowkit.simple def fetch_user_data(user_id): user = flowkit.get(f"https://api.example.com/users/{user_id}").json() posts = flowkit.get("https://api.example.com/posts", params={"userId": user_id}).json() return {"user": user, "posts": posts}

result = await fetch_user_data(123)

Cleaner, more readable, and fully composable.

Features

Fluent API – chain operations with .pipe()

Lazy execution – operations run only when awaited

Type-safe – full IDE support and type hints

Framework-ready – examples for FastAPI, Django, Flask

Minimal boilerplate – built on httpx

Examples Data Processing @flowkit.simple def fetch_user_data(user_id: int): return (flowkit.get(f"https://jsonplaceholder.typicode.com/users/{user_id}") .json() .pipe(lambda user: {*user, "processed": True}))

Concurrent API Calls @flowkit.simple def fetch_dashboard(): users = flowkit.get("https://jsonplaceholder.typicode.com/users").json() posts = flowkit.get("https://jsonplaceholder.typicode.com/posts").json() comments = flowkit.get("https://jsonplaceholder.typicode.com/comments").json()

    return {"users": users, "posts": posts, "comments": comments}  # executes concurrently

When to Use

API-heavy applications

Data pipelines & processing workflows

Teams wanting consistent async patterns

Rapid prototyping

Not for: CPU-bound tasks or making synchronous code non-blocking.

Installation pip install flowkitx==1.0.1

import flowkit

result = await flowkit.get("https://jsonplaceholder.typicode.com/users/1").json() print(result)

Github - https://github.com/Dawaman43/flowkit

Show HN: FlowKitX – Fluent async workflows with lazy execution for Python