flow-like 0.1.2

Python SDK for the Flow-Like API

flow-like

Python SDK for the Flow-Like API
Trigger workflows, manage files, query LanceDB, run chat completions & embeddings — all from Python.

⭐ Flow-Like on GitHub · 📖 Docs · 💬 Discord · 🌐 Website

---

Part of the Flow-Like ecosystem — a Rust-powered visual workflow engine that runs on your device. See the main repository for the full platform.

---

Installation

uv add flow-like

For LanceDB support:

uv add 'flow-like[lance]'

Quick Start

from flow_like import FlowLikeClient

# Auth via environment variables:
#   FLOW_LIKE_BASE_URL - API base URL (required)
#   FLOW_LIKE_PAT      - Personal Access Token (pat_xxx.yyy)
#   FLOW_LIKE_API_KEY   - API Key (flk_app.key.secret)
client = FlowLikeClient()

# Or provide credentials explicitly
client = FlowLikeClient(
    base_url="https://api.flow-like.com",
    pat="pat_xxx.yyy",
)

# Auto-detect token type
client = FlowLikeClient(
    base_url="https://api.flow-like.com",
    token="pat_xxx.yyy",  # or "flk_app.key.secret"
)

Usage

Trigger Workflow

# Synchronous (SSE streaming) — node_id specifies the start node
for event in client.trigger_workflow("app-id", "board-id", "start-node-id", {"key": "value"}):
    print(event.data)

# Asynchronous invocation
result = client.trigger_workflow_async("app-id", "board-id", "start-node-id", {"key": "value"})
print(result.run_id, result.poll_token)

Trigger Event

for event in client.trigger_event("app-id", "event-id", {"key": "value"}):
    print(event.data)

result = client.trigger_event_async("app-id", "event-id")

File Management

# List files
files = client.list_files("app-id")

# Upload
client.upload_file("app-id", open("data.csv", "rb"))

# Download
content = client.download_file("app-id", "path/to/file.csv")

# Delete
client.delete_file("app-id", "path/to/file.csv")

Database / LanceDB

# Get presigned credentials (resolved to uri + storage_options)
info = client.get_db_credentials("app-id", access_mode="read")
print(info.uri, info.storage_options)

# List tables
tables = client.list_tables("app-id")

# Query a table
result = client.query_table("app-id", "my-table", {"filter": "col > 5"})

# Get a LanceDB connection (requires flow-like[lance])
db = client.create_lance_connection("app-id", access_mode="write")

Execution Monitoring

status = client.get_run_status("run-id")
print(status.status)

poll = client.poll_execution("poll-token", after_sequence=0, timeout=30)
for event in poll.events:
    print(event)

Chat Completions

# bit_id identifies the model — use list_llms() to discover available ones
result = client.chat_completions(
    messages=[{"role": "user", "content": "Hello!"}],
    bit_id="bit-id-for-gpt4",
)
print(result.choices)

# Streaming
for event in client.chat_completions(
    messages=[{"role": "user", "content": "Hello!"}],
    bit_id="bit-id-for-gpt4",
    stream=True,
):
    print(event.data)

# Usage tracking
usage = client.get_usage()
print(usage)  # {"llm_price": ..., "embedding_price": ...}

Embeddings

# bit_id identifies the embedding model — use list_embedding_models() to discover
result = client.embed(bit_id="bit-id-for-embedding", input="Hello world")
print(result.embeddings)

Models / Bits

# List available LLMs (remote only)
llms = client.list_llms()
for m in llms:
    print(m.bit_id, m.name, m.provider_name)

# List embedding models (remote only)
embeddings = client.list_embedding_models()

# Search all bits
bits = client.search_bits(search="llama", bit_types=["Llm"])

# Get a specific bit by ID
bit = client.get_bit("some-bit-id")

Board Management

# List boards
boards = client.list_boards("app-id")

# Read a board
board = client.get_board("app-id", "board-id")

# Create / update a board
result = client.upsert_board("app-id", "board-id", name="My Board", description="Does things")
print(result.id)

# Delete a board
client.delete_board("app-id", "board-id")

# Pre-run analysis
prerun = client.prerun_board("app-id", "board-id")
print(prerun.runtime_variables)

HTTP Sink

response = client.trigger_http_sink("app-id", "my/webhook/path", method="POST", body={"data": 1})

App Management

apps = client.list_apps()
app = client.get_app("app-id")
new_app = client.create_app("My App", "Description")

Health Check

health = client.health()
print(health.healthy)

Async Support

All methods have async counterparts prefixed with a:

import asyncio
from flow_like import FlowLikeClient

async def main():
    async with FlowLikeClient(base_url="https://api.flow-like.com", pat="pat_xxx.yyy") as client:
        apps = await client.alist_apps()
        status = await client.ahealth()

asyncio.run(main())

Authentication

Type	Prefix	Header	Env Variable
PAT	pat_	Authorization: pat_{id}.{secret}	FLOW_LIKE_PAT
API Key	flk_	X-API-Key: flk_{app_id}.{key_id}.{secret}	FLOW_LIKE_API_KEY

LangChain Integration

Optional LangChain-compatible wrappers are available.

uv add flow-like[langchain]

from flow_like.langchain import FlowLikeChatModel, FlowLikeEmbeddings

# Option 1: Factory methods from an existing client (recommended)
chat_model = client.as_langchain_chat(
    bit_id="your-model-bit-id",
    temperature=0.7,
    max_tokens=1024,
)
embeddings = client.as_langchain_embeddings(bit_id="your-embedding-bit-id")

# Option 2: Standalone (requires base_url + token)
chat_model2 = FlowLikeChatModel(
    base_url="https://api.flow-like.com",
    token="pat_myid.mysecret",
    bit_id="your-model-bit-id",
    temperature=0.7,
    max_tokens=1024,
)

response = chat_model.invoke("Hello, how are you?")

embeddings = FlowLikeEmbeddings(
    base_url="https://api.flow-like.com",
    token="pat_myid.mysecret",
    bit_id="your-embedding-bit-id",
)

vectors = embeddings.embed_documents(["Hello world", "Goodbye world"])
query_vector = embeddings.embed_query("search query")

License

MIT