Python SDK

Installation

uv (recommended)

uv add flow-like

pip

pip install flow-like

Optional extras

Extra	For	Install
lance	create_lance_connection()	uv add flow-like[lance]
langchain	LangChain wrappers	uv add flow-like[langchain]

Creating a Client

from flow_like import FlowLikeClient

# From environment variables (FLOW_LIKE_BASE_URL + FLOW_LIKE_PAT or FLOW_LIKE_API_KEY)
client = FlowLikeClient()

# Explicit PAT
client = FlowLikeClient(
    base_url="https://api.flow-like.com",
    pat="pat_myid.mysecret",
)

# Explicit API key
client = FlowLikeClient(
    base_url="https://api.flow-like.com",
    api_key="flk_appid.keyid.secret",
)

# Auto-detect token type
client = FlowLikeClient(
    base_url="https://api.flow-like.com",
    token="pat_myid.mysecret",  # or "flk_..."
)

Tip

Use the client as a context manager to ensure HTTP connections are properly closed:

async with FlowLikeClient(base_url="...", pat="...") as client:
    apps = await client.alist_apps()

Sync vs Async

Every method has both a synchronous and an asynchronous variant. Async methods are prefixed with a:

# Sync
llms = client.list_llms()

# Async
llms = await client.alist_llms()

Workflows

Trigger a workflow (streaming)

Returns a Server-Sent Events stream:

for event in client.trigger_workflow("app-id", "board-id", "start-node-id", {"key": "value"}):
    print(event.data)

Trigger a workflow (async)

Returns immediately with a run ID for polling:

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

Trigger an event

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

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

Execution Monitoring

# Check run status
status = client.get_run_status("run-id")
print(status.status)  # "running" | "completed" | "failed"

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

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

Credentials

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

# Get raw presign response (shared_credentials dict, db_path, etc.)
raw = client.get_db_credentials_raw("app-id", table_name="_default", access_mode="write")

Tables

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

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

LanceDB Connection

Note

Requires the lance extra: uv add flow-like[lance]

# Returns a ready-to-use lancedb.DBConnection
db = client.create_lance_connection("app-id", access_mode="write")

# Use with the LanceDB API
table = db.open_table("my-embeddings")
results = table.search([0.1, 0.2, 0.3]).limit(10).to_pandas()

Access modes:

• "read" (default): read-only credentials
• "write": full read-write credentials

Chat Completions

Select a model whose api_surface is ChatCompletions. Models without that field use the same surface for compatibility. The API returns HTTP 400 if the selected model requires a different surface.

chat-completions.py

models = client.list_llms()
model = next(
    (item for item in models if (item.api_surface or "ChatCompletions") == "ChatCompletions"),
    None,
)
if model is None:
    raise RuntimeError("This hub has no Chat Completions model.")

result = client.chat_completions(
    messages=[{"role": "user", "content": "Explain quantum computing in one sentence."}],
    bit_id=model.bit_id,
    temperature=0.7,
    max_tokens=200,
)
print(result.choices[0].message.content)

Streaming

for event in client.chat_completions(
    messages=[{"role": "user", "content": "Hello!"}],
    bit_id=model.bit_id,
    stream=True,
):
    print(event.data, end="")

Responses

Use responses() for a model whose api_surface is Responses. The method returns the raw Responses dictionary. aresponses() is its async counterpart.

responses.py

models = client.list_llms()
model = next((item for item in models if item.api_surface == "Responses"), None)
if model is None:
    raise RuntimeError("This hub has no Responses model.")

result = client.responses(
    input="Explain quantum computing in one sentence.",
    bit_id=model.bit_id,
    max_output_tokens=200,
)
print(result["output"])

For streaming, iterate client.responses(input, model.bit_id, stream=True). Each event contains an event name and its JSON data string. Keep the event type when handling progress and final output.

Usage tracking

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

Embeddings

Use list_embedding_models() to discover available embedding model bit_id values.

result = client.embed(bit_id="bit-id-for-embedding", input="Hello world")
print(result.embeddings)  # list[list[float]]

# Multiple inputs
result = client.embed(bit_id="bit-id-for-embedding", input=["Hello", "World"])

Model Discovery

# List all remotely-available LLMs
llms = client.list_llms()
for m in llms:
    print(m.bit_id, m.name, m.api_surface, m.context_length)

# List all remotely-available embedding models
embeddings = client.list_embedding_models()
for m in embeddings:
    print(m.bit_id, m.name, m.vector_length)

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

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

Each model returns a ModelInfo dataclass:

@dataclass
class ModelInfo:
    bit_id: str
    name: str
    description: str
    provider_name: str | None
    api_surface: str | None  # "ChatCompletions", "Responses", or None
    model_id: str | None
    context_length: int | None   # LLMs only
    vector_length: int | None    # Embeddings only
    languages: list[str]
    tags: list[str]

Board Management

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

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

# Create or 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 (discover runtime variables)
prerun = client.prerun_board("app-id", "board-id")
print(prerun.runtime_variables)

App Management

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

HTTP Sinks

response = client.trigger_http_sink(
    "app-id",
    "my/webhook/path",
    method="POST",
    body={"event": "user.created", "data": {"userId": "123"}},
)

Health Check

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

LangChain Integration

Tip

Requires the langchain extra: uv add flow-like[langchain]

Factory methods (recommended)

Create LangChain-compatible models from your existing client:

# Chat model
chat_model = client.as_langchain_chat(
    bit_id="your-model-bit-id",
    temperature=0.7,
    max_tokens=1024,
)

# Embeddings
embeddings = client.as_langchain_embeddings(bit_id="your-embedding-bit-id")

Standalone usage

from flow_like.langchain import FlowLikeChatModel, FlowLikeEmbeddings

chat_model = FlowLikeChatModel(
    base_url="https://api.flow-like.com",
    token="pat_myid.mysecret",
    bit_id="your-model-bit-id",
    temperature=0.7,
    max_tokens=1024,
)

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

With LangChain chains

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

chat_model = client.as_langchain_chat("your-model-bit-id")

chain = (
    ChatPromptTemplate.from_messages([
        ("system", "You are a helpful assistant."),
        ("human", "{input}"),
    ])
    | chat_model
    | StrOutputParser()
)

response = chain.invoke({"input": "What is Flow-Like?"})

RAG with LangChain

embeddings = client.as_langchain_embeddings("your-embedding-bit-id")

# Use with any LangChain vector store
vectors = embeddings.embed_documents([
    "Flow-Like is a visual workflow engine.",
    "It runs on your device.",
])
query_vector = embeddings.embed_query("What runs locally?")

Async support

LangChain async methods work out of the box:

response = await chat_model.ainvoke("Hello!")
vectors = await embeddings.aembed_documents(["text1", "text2"])

Requirements

• Python >= 3.10
• httpx >= 0.27.0 (installed automatically)

Error Handling

API responses outside the successful HTTP range raise the public SDK exceptions. Catch a specific error when recovery differs, then APIError for other HTTP failures:

api-errors.py

from flow_like import APIError, NotFoundError, RateLimitError

try:
    result = client.get_app("nonexistent-id")
except NotFoundError:
    print("App not found or unavailable to this caller.")
except RateLimitError:
    print("Request rate limited. Retry after the service permits it.")
except APIError as error:
    print(f"HTTP {error.status_code}: {error.response_body}")

ServerError covers HTTP 5xx and also inherits from APIError. Configuration and credential-format failures raise ConfigurationError and AuthenticationError. Transport failures, such as a connection error or request timeout, remain httpx.RequestError subclasses. A timeout does not establish whether a mutation committed, so check its result before retrying.