Python SDK

The fluidtalk package is the official Python client for the FluidTalk Characters API. It wraps every endpoint your connector token can call, returns the inner data payload as light objects, and maps HTTP errors to typed exceptions.

This page is the method-by-method reference. For the concepts behind a character — sessions, people, bubbles, triggers, follow-ups, comment threads — see Concepts. For the underlying HTTP shapes, see the API reference. For how usage is metered, see Billing.

Install

bash

pip install "fluidtalk>=2"

The client requires Python 3.8+ and pulls in httpx as its only runtime dependency.

Construct a client

python

from fluidtalk import FluidTalk

ft = FluidTalk(token="ftc_live_8f3c...")

Your token authenticates as one character. There is no account-level key and no character_id field — the token is the character. You name the platform in each call, and the same token works across every platform the character runs on. Get the token from the character's platform settings in the FluidTalk dashboard; it is shown once, so treat it like a password. See Authentication for details.

The connector owns platform I/O

The API generates the character's messages and tracks the relationship; it never posts anything itself. Reading and sending on Instagram, X, Reddit, Fanvue, TikTok, or anywhere else is your connector's job — how you connect to the platform is your concern.

Constructor parameters

Parameter	Type	Default	Description
`token`	`str`	— (required)	Your per-character connector token (`ftc_live_...`). Sent automatically as the `X-Connector-Token` header.
`base_url`	`str`	`https://api-talk.fluidvip.com`	API origin. The default is production; you should not need to change it.
`timeout`	`float`	`60.0`	Per-request timeout in seconds. Each generated turn runs an LLM, so the default is generous.

python

ft = FluidTalk(
    token="ftc_live_8f3c...",
    base_url="https://api-talk.fluidvip.com",
    timeout=60.0,
)

All requests target the versioned Characters API under /api/v1/characters, so the effective endpoint base is https://api-talk.fluidvip.com/api/v1/characters. You pass the bare origin; the client appends the /api/v1/characters prefix for you.

Lifecycle

The client opens a pooled HTTP connection. Close it when you are done, or use it as a context manager so it closes automatically:

python

with FluidTalk(token="ftc_live_8f3c...") as ft:
    reply = ft.chat(platform="instagram", handle="mark", message="hey ava")
# connection closed here

# or, explicitly:
ft = FluidTalk(token="ftc_live_8f3c...")
try:
    ...
finally:
    ft.close()

Methods

A constructed client exposes the character's actions directly, plus a ft.followups namespace for the follow-up pull-queue. Each method calls one endpoint and returns the inner data as a light object (see Return values).

Method	Endpoint	Returns
`ft.chat(platform, handle, message="", image_url=None, session_id=None, own_username=None)`	`POST /chat`	`ChatReply`
`ft.event(platform, handle, event_type="purchase", external_event_id=, amount=None, currency=None)`	`POST /events`	`EventResult`
`ft.trigger(platform, handle, event_id=, external_event_id=, context=None, own_username=None)`	`POST /triggers`	`TriggerResult`
`ft.followups.list(platform, own_username=None, limit=100)`	`GET /followups`	`list[Followup]`
`ft.followups.ack(followup_id)`	`POST /followups/{id}/ack`	`AckResult`
`ft.comment(platform, post_ref, caption=None, image_urls=None, author_handle=None)`	`POST /comments`	`CommentResult`
`ft.comment_reply(platform, post_ref, replier_handle=, reply_text="", parent_comment_ref=None)`	`POST /comments/reply`	`CommentReplyResult`
`ft.inbound_media(platform, data, filename=None, content_type=None)`	`POST /inbound-media`	`InboundMedia`

`ft.chat`

The lead sent the character a DM — get the character's reply. The session for (platform, handle) is auto-created or resumed, so you rarely pass session_id.

python

reply = ft.chat(
    platform="instagram",
    handle="mark",
    message="hey ava, loved your latest post",
)

for bubble in reply.bubbles:
    # send bubble.text on the platform, pausing bubble.delay_ms before each
    send_dm("mark", bubble.text, image_url=bubble.image_url)

print("session:", reply.session_id)

The reply is a list of bubbles — the character may send several short chat messages with human-like delays. Send them in order. If the conversation is already closed, bubbles is [].

Parameters:

platform — the platform the DM arrived on (required).
handle — the lead's @username (required).
message — the text the lead sent.
image_url — an image the lead sent; it is described via vision and folded into the turn.
session_id — resume a specific conversation. Omit to auto-resume or create one for (platform, handle).
own_username — the bot account that received this DM, for multi-account dedup.

If dedup is on and another account already claimed this lead, the call returns ignored=True and you send nothing:

python

reply = ft.chat(platform="instagram", handle="mark", message="hi", own_username="ava.daily")
if reply.ignored:
    print("ignored:", reply.ignore_reason)   # "lead_claimed_by_other_account"

A generated turn may carry a billing object — see Billing. See the Chat reference.

`ft.event`

Report that money moved (purchase, refund, or chargeback). The API authoritatively flips the lead's lifecycle; your connector only reports the fact. external_event_id is the idempotency key — re-delivering the same event applies exactly once.

python

res = ft.event(
    platform="instagram",
    handle="mark",
    event_type="purchase",          # "purchase" | "refund" | "chargeback"
    external_event_id="stripe_evt_9f12",
    amount=24.0,
    currency="USD",
)
print(res.stage, res.sale_count, res.total_spend)
if res.deduped:
    print("already applied — not re-counted")

See the Events reference.

`ft.trigger`

Fire a conversation trigger / entry event — a story reaction, a new follower, a custom event. If the lead has no live chat, the character opens the conversation reacting to the event; if a chat already exists, the event becomes the next turn's stimulus. event_id is the entry-point key configured in the dashboard (an unknown one is a safe no-op); external_event_id is the idempotency key.

python

res = ft.trigger(
    platform="instagram",
    handle="mark",
    event_id="story_reaction",
    external_event_id="ig_story_react_55",
    context="🔥",                    # a short detail for the opener
)

if res.opened:
    for bubble in res.bubbles:
        send_dm("mark", bubble.text)

Parameters: platform, handle, event_id, external_event_id (all required), plus optional context (a short label/emoji) and own_username (dedup). See the Triggers reference.

`ft.followups.list`

Pull this connector's pending proactive follow-ups, oldest first. The character periodically queues "hey stranger" rekindles for dormant leads; you pull them, send each on the platform, then ack. This is a pull-queue.

python

for f in ft.followups.list(platform="instagram", limit=100):
    send_dm(f.handle, f.message)
    ft.followups.ack(f.id)

Parameters: platform (required for character-token callers), optional own_username (a claimed row is delivered only to that account), and limit (default 100, max 500). See the Follow-ups reference.

`ft.followups.ack`

Acknowledge that a pulled follow-up was delivered (pending → delivered). Terminal and idempotent.

python

res = ft.followups.ack("fu_7c2a1e")
print(res.delivered_at)

`ft.comment`

Generate a top-level public comment for a post. Your connector posts the returned text — the API does not.

python

res = ft.comment(
    platform="instagram",
    post_ref="https://instagram.com/p/Cxyz123",
    caption="sunset hike 🌄",
    image_urls=["https://cdn.example.com/post.jpg"],
    author_handle="mark",
)

if res.comment:
    post_comment("https://instagram.com/p/Cxyz123", res.comment)
    print("recognized lead:", res.recognized_lead)   # cross-surface awareness

post_ref (the platform's unique id/URL for the post) is required. When comments aren't enabled for this character/platform, comment is unset and the result is ignored. See the Comments reference.

`ft.comment_reply`

Generate a reply to a reply under the character's comment. Threaded — it can fire multiple times in a thread, to the original poster or to other commenters.

python

res = ft.comment_reply(
    platform="instagram",
    post_ref="https://instagram.com/p/Cxyz123",
    replier_handle="mark",
    reply_text="haha you're funny, where do you post more?",
)

if res.decision != "skip" and res.reply:
    post_reply("https://instagram.com/p/Cxyz123", res.reply)

The character decides whether to engage: decision is "engage", "drive_to_dm", or "skip", and reply is None when it chooses not to reply. Optional parent_comment_ref and post context can be supplied. See the Comments reference.

`ft.inbound_media`

Added in 2.1.0. Upload the bytes of a photo the lead sent, and get a permanent URL to pass to chat as image_url. For platforms that hand you a file rather than a link we can fetch — a Telegram file_id you download with your own bot token, or an Instagram CDN URL that is signed and expires.

python

up = ft.inbound_media(
    platform="instagram",
    data=raw_bytes,            # raw bytes; the base64 is done for you
    filename="photo.jpg",
)

ft.chat(platform="instagram", handle="mark", message="what do you think? 😏", image_url=up.url)

Max 10 MB decoded — larger raises ApiError (413). Skip this entirely when you already have a publicly-fetchable URL: pass it straight to chat as image_url. See the Media reference.

Return values

Methods return small objects that mirror the inner data of each API response (the { "data": ..., "request_id": ... } envelope is unwrapped for you). Timestamps are strings (ISO-8601 UTC). Each object ignores unknown fields, so a server that adds a field will not break an older SDK.

`ChatReply`

Field	Type	Notes
`session_id`	`str \| None`	the conversation thread; `None` when an inbound was ignored by dedup
`bubbles`	`list[Bubble]`	the reply, in send order; `[]` when the conversation is closed
`ignored`	`bool`	`True` when multi-account dedup dropped this inbound
`ignore_reason`	`str \| None`	e.g. `"lead_claimed_by_other_account"`
`billing`	`Billing \| None`	present when billing is enabled

`Bubble`

Field	Type	Notes
`text`	`str`	the message to send
`delay_ms`	`int`	human-like pause before sending this bubble
`image_url`	`str \| None`	a photo to attach, if the character sent one

`EventResult`

Field	Type	Notes
`ok`	`bool`
`deduped`	`bool`	`True` when this `external_event_id` was already applied
`event_type`	`str`	`"purchase"` / `"refund"` / `"chargeback"`
`stage`	`str`	the lead's lifecycle stage after the event
`sale_count`	`int`
`total_spend`	`float`

`TriggerResult`

Field	Type	Notes
`ok`	`bool`
`deduped`	`bool`	`True` on a re-delivered `external_event_id`
`opened`	`bool`	`True` when the character opened a cold conversation
`session_id`	`str \| None`
`bubbles`	`list[Bubble]`	the opener, when `opened` is `True`
`ignored`	`str \| None`	`"event not configured"` when `event_id` is unknown
`billing`	`Billing \| None`	present when an opener was generated and billing is enabled

`Followup`

Field	Type	Notes
`id`	`str`	pass to `followups.ack`
`handle`	`str`	the dormant lead to re-engage
`platform`	`str`
`kind`	`str`	the follow-up category
`message`	`str`	the rekindle text to send
`created_at`	`str`	ISO-8601 UTC

`AckResult`

Field	Type	Notes
`ok`	`bool`
`deduped`	`bool`	`True` when the follow-up was already acked
`delivered_at`	`str`	ISO-8601 UTC

`CommentResult`

Field	Type	Notes
`ok`	`bool`
`comment`	`str \| None`	the comment text to post; unset when comments aren't configured
`aware`	`dict`	cross-surface awareness summary
`recognized_lead`	`bool`	`True` when the commenter is a known Person
`thread_id`	`str \| None`
`billing`	`Billing \| None`	present when a comment was generated and billing is enabled

`CommentReplyResult`

Field	Type	Notes
`ok`	`bool`
`decision`	`str`	`"engage"` / `"drive_to_dm"` / `"skip"`
`reply`	`str \| None`	the reply text; `None` when the character chooses not to reply
`reason`	`str \| None`	why the character made that decision
`aware`	`dict \| None`	cross-surface awareness summary
`billing`	`Billing \| None`	present when a reply was generated and billing is enabled

`Billing`

Attached to any generating response when billing is enabled. See Billing.

Field	Type	Notes
`tokens`	`float`	what this message cost you = real cost × plan rate (1 token = $1, e.g. `0.00127932`)
`tokens_used`	`float`	the raw model cost of this turn, in tokens (real $ before plan rate)
`balance_after`	`float`	wallet token balance after the charge
`collected`	`bool`	whether the charge was applied

Errors

The client raises a typed exception for every non-success status. Import them from the package:

python

from fluidtalk import (
    ApiError,              # base class for all of them
    AuthError,             # 401 — missing/invalid/revoked token
    PaymentRequiredError,  # 402 — wallet can't cover the turn
    PermissionError,       # 403 — forbidden
    NotFoundError,         # 404 — unknown character/lead/follow-up (or not yours)
    ConflictError,         # 409 — conflicting state
    ValidationError,       # 422 — a field failed validation
    RateLimitError,        # 429 — too many requests
)

Exception	HTTP status	`error.code`	When
`AuthError`	401	`invalid_token`	Missing, invalid, or revoked connector token
`PaymentRequiredError`	402	`payment_required`	Billing is on and the wallet can't cover this turn — raised before any model call
`PermissionError`	403	`forbidden`	The action isn't allowed for this character/token
`NotFoundError`	404	`not_found`	The resource doesn't exist, or belongs to another owner
`ConflictError`	409	`conflict`	The request conflicts with current state
`ValidationError`	422	`validation_error`	The body is well-formed but a field failed validation
`RateLimitError`	429	`rate_limited`	Too many requests — honor `Retry-After`, then retry
`ApiError`	any other (400, 500, …)	`invalid_request` / `internal_error`	Catch-all base; also the parent of every class above

Every exception carries the HTTP status, the error.code, the human-readable error.message, and the request_id from the response body. Handle the specific ones you care about and let ApiError cover the rest:

python

from fluidtalk import PaymentRequiredError, RateLimitError, ApiError

try:
    reply = ft.chat(platform="instagram", handle="mark", message="hey ava")
except PaymentRequiredError:
    print("Top up the character's wallet — no turn was generated, no charge made.")
except RateLimitError as e:
    print("Rate limited — back off and retry.")
except ApiError as e:
    print(f"API error {e.status} ({e.code}): {e.message}  request_id={e.request_id}")

Because PaymentRequiredError is raised before the model runs, you are never charged for a refused turn — handle it by topping up the wallet and surfacing insufficient balance to the operator. See Errors, Billing, and Rate limits.

Full worked example

Drive an inbound DM, report a purchase, generate a public comment, and clear the follow-up queue — all for the character "Ava" on Instagram.

python

from fluidtalk import FluidTalk, PaymentRequiredError

with FluidTalk(token="ftc_live_8f3c...") as ft:
    # 1. A lead DM'd Ava — generate her reply (bubbles, in order).
    reply = ft.chat(platform="instagram", handle="mark", message="hey ava, love your posts")
    for bubble in reply.bubbles:
        send_dm("mark", bubble.text, image_url=bubble.image_url)
    print("session:", reply.session_id)

    # 2. Mark bought something — report it (idempotent on external_event_id).
    ev = ft.event(
        platform="instagram",
        handle="mark",
        event_type="purchase",
        external_event_id="stripe_evt_9f12",
        amount=24.0,
    )
    print("stage now:", ev.stage, "| sales:", ev.sale_count)

    # 3. Generate a public comment for one of Ava's posts.
    cm = ft.comment(
        platform="instagram",
        post_ref="https://instagram.com/p/Cxyz123",
        caption="sunset hike 🌄",
        author_handle="mark",
    )
    if cm.comment:
        post_comment("https://instagram.com/p/Cxyz123", cm.comment)

    # 4. Deliver any queued proactive follow-ups, then ack each.
    for f in ft.followups.list(platform="instagram"):
        send_dm(f.handle, f.message)
        ft.followups.ack(f.id)

Generating a client from the OpenAPI spec (optional)

If you prefer to generate your own typed client instead of using this SDK, the API publishes an OpenAPI specification. Any standard OpenAPI code generator can produce a client from it; remember to send your token in the X-Connector-Token header and to address the character by token (there is no character_id). For most integrations the fluidtalk package is the simplest path.

Python SDK ​

Install ​

Construct a client ​

Constructor parameters ​

Lifecycle ​

Methods ​

ft.chat ​

ft.event ​

ft.trigger ​

ft.followups.list ​

ft.followups.ack ​

ft.comment ​

ft.comment_reply ​

ft.inbound_media ​

Return values ​

ChatReply ​

Bubble ​

EventResult ​

TriggerResult ​

Followup ​

AckResult ​

CommentResult ​

CommentReplyResult ​

Billing ​

Errors ​

Full worked example ​

Generating a client from the OpenAPI spec (optional) ​

See also ​

Python SDK

Install

Construct a client

Constructor parameters

Lifecycle

Methods

`ft.chat`

`ft.event`

`ft.trigger`

`ft.followups.list`

`ft.followups.ack`

`ft.comment`

`ft.comment_reply`

`ft.inbound_media`

Return values

`ChatReply`

`Bubble`

`EventResult`

`TriggerResult`

`Followup`

`AckResult`

`CommentResult`

`CommentReplyResult`

`Billing`

Errors

Full worked example

Generating a client from the OpenAPI spec (optional)

See also