flowith

An unofficial async Python client for the Flowith.io web application.

Disclaimer: This library reverse-engineers the Flowith.io browser API. It is not affiliated with or endorsed by Flowith. Use responsibly.

Features

• Text generation — single-turn and multi-turn, streaming and non-streaming
• Image generation — Gemini, GPT Image, and more
• Video generation — Seedance, Kling, etc.
• File upload — POST /file/store with multipart support
• Conversation management — create, list, rename, delete
• Flow canvas — read node graph for any conversation
• Credits & models — fetch balance, browse model catalog
• Templates — list and filter prompt templates
• Profile & subscriptions — read/update profile, list grants
• Supabase content — notices, projects, discussions, oracle stream
• Daily rewards — claim and check status
• Account auth — OAuth URL, auth user, refresh token
• Real-time SSE — listen on /user_stream/stream for live generation events
• Async-first — built on aiohttp, zero blocking calls

Installation

pip install flowith_webapi

Requires Python 3.10+.

Authentication

1. Sign in at flowith.io with Google.
2. After the OAuth redirect, copy the access_token from the URL hash:

   https://flowith.io/#access_token=eyJhbGci...&...
   

3. Your user ID (sub claim in the JWT) is also needed for SSE streaming — it is parsed automatically from the JWT if you do not supply it explicitly.

To enable token rotation, also capture the refresh token from Supabase. Generate the OAuth URL with access_type=offline and prompt=consent so Google issues one on the first consent screen:

url = client.get_oauth_url(
    provider="google",
    redirect_to="https://flowith.io",
    access_type="offline",
    prompt="consent",
)
print(url)

import os
TOKEN   = os.environ["FLOWITH_TOKEN"]
USER_ID = os.environ["FLOWITH_USER_ID"]          # optional — parsed from JWT
REFRESH = os.environ.get("FLOWITH_REFRESH_TOKEN", "")

Keep these secret — they grant full access to your Flowith account.

How it works (SSE architecture)

The Flowith web app maintains two persistent SSE connections:

Both connections send the JWT as a bare token in the authorization header (no Bearer prefix) — this matches what the edge server expects.

When generate() is called the client:

1. Creates the Supabase conversation and user/AI node rows.
2. POSTs the completion request to /completion/async (returns immediately).
3. Listens on /user_stream/stream for events whose nodeId matches the AI node.
4. Falls back to polling flow_node on Supabase if the stream closes early.

Quick start

import asyncio
from flowith_webapi import FlowithClient

async def main():
    async with FlowithClient(TOKEN, user_id=USER_ID, refresh_token=REFRESH) as client:
        r = await client.generate("What is the capital of Poland?")
        print(r.text)   # "Warsaw"

asyncio.run(main())

Usage

Single-turn generation

r = await client.generate("Explain quantum entanglement in one sentence.")
print(r.text)
print(r.model)     # model ID used
print(r.conv_id)   # conversation UUID

Streaming

async for chunk in client.generate_stream("Write me a haiku about async Python."):
    print(chunk.text_delta, end="", flush=True)
print()

Multi-turn session

session = client.start_conversation(model="gemini-3.1-pro-preview")

r1 = await session.send("My name is Alice.")
r2 = await session.send("What is my name?")
print(r2.text)   # "Your name is Alice."

async for chunk in session.send_stream("Tell me a fun fact about Alice."):
    print(chunk.text_delta, end="", flush=True)
print()

Image generation

from flowith_webapi import ImageModel, AspectRatio

result = await client.generate_image(
    "A ripe banana on a marble surface, studio lighting",
    model=ImageModel.GEMINI_3_1_FLASH_IMAGE,
    aspect_ratio=AspectRatio.SQUARE,
)
await result.image.save("./outputs")
print(result.image.url)

Video generation

from flowith_webapi import VideoModel

video = await client.generate_video(
    "A cat walking on a sunny beach",
    model=VideoModel.SEEDANCE_2_FAST,
    timeout=300.0,
)
await video.save("./outputs")
print(video.video_url)

File upload

# Upload a local file and get a public CDN URL
record = await client.upload_file("my_game.html")
print(record.url)   # https://r2-bucket.flowith.net/f/…/my_game.html

# Reference the file in a generation
r = await client.generate(
    f"Review this HTML file: [[file:html|my_game.html|{record.url}]]"
)

# Rotate the access token (uses stored refresh token if not supplied)
if REFRESH:
    await client.refresh_access_token()

Conversations

# List recent conversations
convs = await client.list_conversations(limit=10)
for c in convs:
    print(c.conv_id, c.title)

# Read the flow-canvas node graph
nodes = await client.get_flow_nodes(conv_id)
for n in nodes:
    print(n.node_type, n.text[:80])

# Rename / delete
await client.rename_conversation(conv_id, "New Title")
await client.delete_conversation(conv_id)

Raw SSE streams

# Generation-data stream (/user_stream/stream) — heartbeats filtered out
async for evt in client.stream_user_stream(timeout=60):
    print(evt)

# Keepalive stream (/user_stream/sse) — heartbeats included
async for evt in client.stream_user_sse(timeout=60):
    print(evt)   # {"heartbeat": 1779127465201}

Credits & models

credits = await client.get_credits()
total = sum(c.remain_quota for c in credits)
print(f"Credits: {total:.2f}")

models = await client.list_models()
for m in models:
    print(m.model_id, m.title)

# Active subscriptions
subs = await client.list_subscription_user_own()
print(len(subs))

# Profile
profile = await client.get_user_profile()
print(profile.language)

API reference

FlowithClient

ConversationSession

Return types

Error handling

from flowith_webapi import (
    AuthenticationError,
    RateLimitError,
    ValidationError,
    APIError,
    TimeoutError,
    FileUploadError,
)

try:
    r = await client.generate("Hello!")
except AuthenticationError:
    print("Invalid or expired token.")
except RateLimitError as e:
    print(f"Rate limited — retry in {e.retry_after_s}s")
except TimeoutError:
    print("Generation timed out.")
except APIError as e:
    print(f"API error {e.status_code}: {e}")

Logging

import flowith_webapi
flowith_webapi.set_log_level("DEBUG")   # DEBUG | INFO | WARNING | ERROR

Related projects

• 1MinAI-API — sister project for 1min.AI
• Claude-API — unofficial Claude.ai client