The Python SDK

Start here

If you have never touched Conecto before, this section is the whole mental model. Five minutes here makes the rest of the page obvious.

What Conecto is

A chat widget that sits on your website. Visitors type into it; your team — or an AI agent, or your own code — answers. Everything below is a way of putting your code into that loop.

The five nouns

Almost every method in this SDK takes or returns one of these.

The three things you can build

Bot or plugin? If you want to write the words the visitor reads, build a bot. If you would rather let the AI talk and just give it access to your data — orders, subscriptions, stock levels — build a plugin. Most teams end up wanting a plugin.

Words this page uses

Jargon, defined once. Nothing here is Conecto-specific except the last two.

Which section do I need?

Install & authenticate

pip install conecto

Create a credential in Settings → Developers. You get a client ID (ck_…) and a secret (cs_…). The secret is shown once and stored hashed, so put it somewhere your code can read and your repository cannot.

export CONECTO_CLIENT_ID=ck_your_client_id
export CONECTO_SECRET=cs_your_secret

from conecto import Conecto

client = Conecto()                    # reads the environment
client = Conecto("ck_...", "cs_...")  # or pass them directly

print(client.ping()["workspace"]["name"])

Call ping() once at startup. It raises AuthenticationError immediately if the credential is wrong or revoked — much better than finding out mid-conversation. A credential can cover the workspace or be scoped to one widget; a scoped one simply cannot see other widgets' conversations.

Quickstart

Three things you can do in the first five minutes.

Read what is happening

for convo in client.conversations.list(status="open"):
    who = convo.visitor.email if convo.visitor else "anonymous"
    print(f"#{convo.id}  {who}  {convo.preview}")

Answer someone

convo = client.conversations.get(4821)
print(convo.messages[-1].body)          # what they said last
convo.reply("On it — give me one moment.")

Send something that is not text

from conecto import blocks

convo.reply("Here's how to reset it:", blocks=[
    blocks.image("https://cdn.you.com/reset.gif", alt="Reset flow"),
    blocks.buttons([
        blocks.reply_button("That worked"),
        blocks.link_button("Full guide", "https://docs.you.com/reset"),
    ]),
])

That is the shape of the whole library: a client with resources on it, objects that know how to act on themselves, and builders for anything structured.

The client

Build one and keep it. It pools connections and is safe to share across threads.

client = Conecto(
    client_id="ck_...",
    secret="cs_...",
    base_url="https://conecto.chat/api/v1",  # override for staging
    timeout=30.0,        # seconds per request
    max_retries=2,       # timeouts, 429s and 5xx
    app="acme-billing",  # added to the User-Agent; shows up in our logs
)

Environment

Resources

Version drift

Every response carries the server's API version. Log it at startup — if it moves and a field you use disappears, this is the first thing you will want to know.

print(client.api_version)   # what the server last reported
print(client.sdk_version)   # what this SDK was built against

limits = client.meta.schema()["limits"]
assert limits["blocks_per_message"] >= 4

client.meta.schema() returns the whole API as JSON — endpoints, events, block types, reserved actions, error codes and every limit — served by the same code that enforces them, so it cannot drift from reality.

Conversations

A thread between a visitor and your workspace. Messages have a sender of visitor, agent, bot or system.

Listing and reading

page = client.conversations.list(status="open", limit=25)

page.items                     # this page only
for convo in page:             # every page, fetched lazily
    ...

convo = client.conversations.get(4821)
convo.status, convo.visitor.email, convo.messages[-1].body

# Poll cheaply: only what is newer than what you already have.
fresh = client.conversations.messages(4821, since_id=last_seen_id)

Replying

convo.reply("I can refund order #1284 — confirm?",
            buttons=["Yes, refund it", "Talk to a human"])

A tap on a quick-reply button comes back as an ordinary visitor message containing that text, so your handler needs no special case for it.

Steering the thread

convo.typing()                  # "…is typing", expires after ~8s
convo.reply("Working on it…")

convo.reply("Escalating: refund over policy limit.", internal=True)
convo.handoff()                 # flag for a human; routing rules apply
convo.assign(user_id=12)        # ids come from client.members.list()

convo.close()
convo.reopen()

Writing to a closed thread raises ConversationClosed. Reopen it first, or catch it — see Errors.

Rich content

Any message can carry up to ten blocks: images and GIFs, video, audio, files, third-party embeds, card carousels, label/value lists, buttons and dividers. They render in the widget and in the agent's inbox.

Blocks are plain dicts on the wire, so you could hand-write them. The builders exist because a misspelled key becomes a customer telling you the image never appeared — they apply the same rules the server does, at the point where the traceback still points at your code.

from conecto import blocks

convo.reply("Here's your order:", blocks=[
    blocks.list_block([
        blocks.row("Placed", "12 July 2026"),
        blocks.row("Status", "Shipped"),
        blocks.row("Carrier", "DHL", subtitle="Tracking 4Z8871",
                   url="https://track.dhl.com/4Z8871"),
    ], title="Order #1042"),
    blocks.divider(),
    blocks.cards([
        blocks.card("Trail Runner 2",
                    subtitle="Road · Neutral",
                    price="89.00 USD",
                    badge="Back in stock",
                    image="https://cdn.you.com/tr2.jpg",
                    url="https://shop.you.com/p/tr2",
                    buttons=[blocks.link_button("Buy", "https://shop.you.com/cart")]),
    ]),
])

Every block type

Two kinds of button

blocks.reply_button("Yes, refund it")                    # sends that text back
blocks.reply_button("Yes, refund it", "confirm_refund")  # nice label, stable value
blocks.link_button("Read the policy", "https://you.com/refunds")  # opens a tab

Give a reply button an explicit value when the label should read well but the text your bot matches on should not change every time someone rewrites the copy.

Shipping a GIF

A .gif works. A muted, looping, autoplaying video is usually the better trade — a fraction of the bytes, and nobody can tell.

blocks.video("https://cdn.you.com/reset.mp4",
             autoplay=True, loop=True, muted=True)

Rules the builders enforce

• Every URL must be https. The widget runs on an https page, so anything else is blocked by the browser anyway.
• Ten blocks per message, ten cards, twelve list rows, six buttons.
• Long strings are truncated, not rejected. Structural mistakes raise BlockError.
• A YouTube link in a video block raises, and tells you to use embed.

Validation happens locally. reply(blocks=…) checks the list before sending, so a typo raises BlockError from your line rather than coming back as an HTTP 400 from inside the client.

Uploading files

No public URL for that GIF, receipt or spec sheet? Upload it (up to 10 MB) and use what you get back.

media = client.media.upload("celebrate.gif")     # a path
media = client.media.upload(open("a.pdf", "rb")) # a file object
media = client.media.upload(raw_bytes, filename="receipt.png")

convo.reply("All set!", blocks=[blocks.image(media)])

A Media can be passed straight to blocks.image, blocks.video or blocks.file.

Use media.url, not media.absolute_url. url is a path, resolved by the widget against its own origin, so the same message works in development, in a preview deploy and in production — and survives the day your domain changes.

Contacts & visitors

Contacts

The CRM. upsert is keyed on email, so it is safe to point a nightly sync at.

client.contacts.upsert(
    "maya@acme.io",
    name="Maya Silva",
    company="Acme",
    custom_fields={"plan": "scale", "mrr": "480"},
)

contact = client.contacts.find("maya@acme.io")     # or None
for c in client.contacts.list(query="acme"):
    print(c.email, c.custom_fields.get("plan"))

custom_fields is merged, not replaced — two jobs can each own their own keys without overwriting each other. Up to 30 keys per call.

Visitors

A browser session on a widget, keyed by (widget_id, session). The session id lives in the visitor's browser; read it client-side and send it to your backend.

visitor = client.visitors.get(widget_id=7, session=session_id)
visitor.email, visitor.verified

Proactive messages

Push to a session without waiting to be asked — shipping updates, trial nudges, cart recovery.

client.visitors.message(
    widget_id=7, session=session_id,
    body="Still thinking it over? Your cart is saved:",
    blocks=[blocks.cards([
        blocks.card("Trail Runner 2", price="89.00 USD",
                    image="https://cdn.you.com/tr2.jpg",
                    url="https://shop.you.com/cart"),
    ])],
)

It reuses the session's live conversation or opens one. An open widget shows it on the next poll; a closed one shows it on next open, with the unread treatment.

Tickets & articles

Tickets

ticket = client.tickets.create(
    email="maya@acme.io",
    message="Card was declined on renewal.",
    priority="high",          # low · normal · high · urgent
)

client.tickets.update(ticket.id, status="pending", assignee_user_id=12)
ticket.reply("We've fixed the card on file — try again?")   # emails the requester
ticket.reply("Billing confirmed the retry.", internal=True)  # private note

A public reply reopens a resolved ticket, because an answer to something marked done is almost always a new round.

Help-center articles

Useful in both directions: search it from a bot before opening a ticket, and push docs into it from whatever you author in.

hits = client.articles.list(query="refund", published=True)
if hits:
    convo.reply(f"This might help: “{hits[0].title}”")

client.articles.create("Shipping times", "<p>2-4 business days.</p>",
                       published=True)

HTML is cleaned server-side against an allowlist of safe tags — the same pipeline the dashboard editor uses — so scripts and styles are stripped before anything is stored.

Build a bot

A bot is a webhook and a reply. Bot handles the five things that are easy to get subtly wrong — signature verification, delivery deduplication, event routing, filtering out your own messages, and answering fast — so your code is the part that is yours.

from conecto import Conecto, Bot, blocks

client = Conecto()
bot = Bot(client, secret=WEBHOOK_SECRET)

@bot.on_message
def handle(ctx):
    text = ctx.text.lower()

    if "human" in text:
        ctx.reply("Of course — connecting you now.")
        ctx.handoff()
        return

    if "order" in text:
        if not ctx.verified:
            ctx.reply("What's the email on the order?", ask_email=True)
            return
        order = lookup_order(ctx.verified_email)     # your system
        ctx.reply("Here it is:", blocks=[
            blocks.list_block([
                blocks.row("Status", order["status"]),
                blocks.row("Carrier", order["carrier"]),
            ], title=order["number"]),
        ])
        return

    ctx.reply("I'm not sure about that one — let me get someone who is.")
    ctx.note(f"Bot could not classify: {ctx.text!r}")
    ctx.handoff()

Serving it

# Flask
app.add_url_rule("/conecto", view_func=bot.flask_view(), methods=["POST"])

# FastAPI (handlers run in a worker thread, so they never stall the loop)
app.post("/conecto")(bot.fastapi_route())

# Django (urls.py)
path("conecto/", csrf_exempt(bot.django_view()))

# No framework at all
from wsgiref.simple_server import make_server
make_server("", 8000, bot.wsgi_app()).serve_forever()

Then subscribe the endpoint:

hook = client.webhooks.create(
    "https://bots.you.com/conecto",
    ["message.created", "conversation.created", "conversation.rated"],
)
print(hook.secret)   # store this — it is how you verify deliveries

Handlers

What a handler gets

Running more than one worker? Deduplication is in-process by default, so two workers can each handle the same delivery once. Pass your own store — anything with add(id) -> bool, backed by Redis or your database.

class RedisSeen:
    def add(self, delivery_id):
        # True if new, False if we have handled it before
        return bool(redis.set(f"conecto:{delivery_id}", 1, nx=True, ex=86400))

bot = Bot(client, secret=WEBHOOK_SECRET, seen=RedisSeen())

Build a plugin

A bot answers webhooks. A plugin is the other direction: you declare what your systems can do, and the AI agent decides when to use them while it is talking to someone. You never write the conversation logic — only the lookup.

This is how a store, billing system or CRM we have never heard of becomes a first-class integration.

from conecto import Conecto, Plugin, parameter

plugin = Plugin("acme-store", base_url="https://api.acme.com/conecto")

@plugin.action(
    "catalog.search_products",
    description="Search the Acme catalog by keyword.",
    parameters=[parameter("query", required=True,
                          description="Search words from the visitor.")],
)
def search(req):
    hits = catalog.search(req.get("query"), limit=req.get("limit", 4))
    if not hits:
        return req.not_found()
    return req.ok(products=[
        {"title": p.name, "url": p.url, "image": p.image,
         "price_from": f"{p.price:.2f}", "currency": "USD",
         "available": p.stock > 0}
        for p in hits
    ])

@plugin.action(
    "orders.get_status",
    risk="verified_read",
    description="Status of one of the visitor's own orders.",
    parameters=[parameter("order_number", required=True)],
)
def order_status(req):
    order = orders.find(req.get("order_number"), email=req.verified_email)
    return req.ok(**order) if order else req.not_found()

Serve and register

# One route serves every action.
app.add_url_rule("/conecto/<path:action>", view_func=plugin.flask_view(),
                 methods=["POST"])

# Tell Conecto it exists. Idempotent — run it on every release.
integration = plugin.deploy(Conecto())
print(integration.signing_secret)   # set this in your environment

deploy() creates or updates the integration so it matches your code, deletes actions you removed, and installs it on your widgets. Nothing is exposed to the AI until it is installed.

The five answers

return req.ok(status="Shipped", carrier="DHL")   # success
return req.ok(result, blocks=[...])              # …plus rich content in the reply
return req.not_found()                           # nothing matched — a normal outcome
return req.verify_required()                     # "I need a proven identity"
return req.error("That subscription was already cancelled.")

Write error messages for the visitor, not for your logs: the AI may relay them. Returning a plain dict is also fine — it is treated as the result.

Risk levels

req.verified_email is the only identity to trust. An email in req.arguments is whatever the model parsed out of a chat. For verified_read and write the platform will not call you at all until we have proven who the visitor is, and the proven address is the one we hand you.

Reserved actions

A few names mean something to the platform itself. Declare one and your service wires into the feature that already exists for it.

Test it before a visitor does

client.integrations.run("acme-store", "catalog.search_products",
                        {"query": "trail shoes"})
# {'status': 'ok', 'result': {...}, 'blocks': []}

This runs the real call path — the signature, your credential, the checks that stop your URL pointing anywhere private, the timeout, the parsing — so what you see is exactly what the AI will get. Pass conversation_id= to run in the context of a live chat, which is the only way a verified action can succeed.

Writing a good description

The description is the highest-leverage string in your integration: it is what the AI reads to decide whether this action answers the question in front of it. Write it as an instruction to a capable colleague who cannot see your code.

Weak: “Order lookup.”
Strong: “Look up the status and delivery date of one of the visitor's own orders, by its order number. Use this whenever they ask where something is or when it will arrive.”

What the platform guarantees

• 8 second timeout, 128 KB response cap.
• Calls are signed with your signing_secret — same scheme as webhooks.
• Your URL is resolved once and pinned to a public IP; redirects are refused.
• Everything you return reaches the model as data, never as instructions, and secret-looking keys are stripped.
• Custom integrations require a paid (AI Agent) plan. 20 per workspace, 40 actions each.

Identity

Anything that touches one person's data waits until we know who they are. That happens one of two ways: an emailed code, or your site vouching for a user who is already logged in.

# your backend, right after your own auth check
client.visitors.identify(
    widget_id=7,
    session=conecto_session,          # read from the visitor's browser
    email=user.email,
    name=user.name,
    verified=True,                    # the vouch
    verify_hours=24,                  # max 72, default 12
    data={"plan": user.plan, "customer_since": "2024"},
)

# when they log out
client.visitors.unverify(widget_id=7, session=conecto_session)

While the vouch lasts, Conecto treats the email as proven: the widget knows their name, chats attach to the right contact, and flows that would otherwise demand an email code — order lookups, refunds, account data, your plugin's verified actions — proceed without one.

Only vouch for sessions your backend actually authenticated. Your attestation is treated as being as strong as our own email code, because you really did authenticate them. It is time-boxed and session-scoped, so a new browser needs a new vouch. The browser can never mark itself verified — that call is authenticated, and belongs on your server.

Webhooks & events

Bot does all of this for you. Reach for the raw helpers when you want to verify at the edge and process on a queue.

from conecto import webhooks

event = webhooks.parse(request.get_data(), request.headers, secret=SECRET)

event.id                  # the delivery id — dedupe on this
event.event               # "message.created"
event.conversation_id
event.text
event.is_visitor_message  # excludes your own bot's messages
event.visitor.verified

Verify against the raw body. Re-serializing JSON changes the bytes and the signature stops matching — this is the most common first-integration bug there is. Use request.get_data() in Flask, request.body in Django, await request.body() in FastAPI. Passing a parsed dict raises with this exact advice.

Events

Delivery semantics

Deliveries are at-least-once and unordered: if we cannot confirm you got one we send it again, and they will not always arrive in the order things happened. They also time out after 2.5 seconds — so answer 200 immediately and do the work afterwards.

• Every payload carries an id (also in X-Conecto-Delivery). Store it and skip ids you have already handled.
• Pass that same id as idempotency_key when you reply, and a redelivery can never make your bot speak twice.
• X-Conecto-Timestamp lets you reject stale deliveries; the SDK checks it with a five-minute tolerance by default.

signature = webhooks.sign(SECRET, raw_body)      # what we send
webhooks.verify(raw_body, header_value, SECRET)  # True / False, never raises

Integration calls we make to your plugin use the identical scheme with the integration's signing_secret — one verification routine covers both directions.

Errors & retries

Every failure raises a subclass of ConectoError, so you can catch exactly what you expect.

from conecto import ConversationClosed, RateLimited, NotFoundError

try:
    convo.reply("Refunded — you'll see it in 3-5 days.")
except ConversationClosed:
    convo.reopen()
    convo.reply("Refunded — you'll see it in 3-5 days.")
except NotFoundError:
    log.warning("conversation vanished")

Every exception carries code, message, status and the raw response.

Retries

Timeouts, rate limits (429) and server errors (5xx) are retried for you. Each retry waits a little longer than the last, with a random amount added so that a thousand clients recovering at once do not all come back in the same instant — and if we send a Retry-After, that wins.

Writes get retried too, which is normally dangerous. It is safe here because every write carries an idempotency key and a retry reuses the same one, so the server recognises the second attempt as the same call and does not repeat it.

client = Conecto(max_retries=0)   # opt out entirely

Limits worth knowing

client.meta.schema()["limits"] always has the live numbers.

Pagination

List endpoints return a Page you can use three ways.

page = client.contacts.list(query="acme")

page.items        # just this page — a list, so len() and [0] work
page.has_more     # is there another?

for contact in page:   # every page, fetched lazily as you consume
    ...

page.all()        # everything, collected into one list

Iterating is lazy: next(iter(page)) costs exactly one request no matter how many contacts exist. all() is fine for contacts and tickets; think twice on a busy workspace's conversations, where iterating and stopping early is usually what you meant.

Testing

The client takes a transport, which is the seam for tests. Anything with get, post, patch, delete and request works — no network, no credentials, no mocking library.

class FakeTransport:
    base_url = "https://conecto.test/api/v1"

    def __init__(self):
        self.calls = []
        self.responses = {}

    def request(self, method, path, **kw):
        self.calls.append((method, path, kw.get("json_body")))
        return self.responses.get(f"{method} {path}", {})

    def get(self, p, **kw): return self.request("GET", p, **kw)
    def post(self, p, json_body=None, **kw): return self.request("POST", p, json_body=json_body, **kw)
    def patch(self, p, json_body=None, **kw): return self.request("PATCH", p, json_body=json_body, **kw)
    def delete(self, p, **kw): return self.request("DELETE", p, **kw)


def test_bot_answers_refunds():
    transport = FakeTransport()
    client = Conecto(transport=transport)
    client.conversations.reply(1, "Refunded.")
    assert transport.calls[0][1] == "/conversations/1/messages/"

For bots and plugins, skip signature verification in tests rather than signing fixtures:

bot = Bot(client, verify=False)
plugin = Plugin("acme", base_url="https://x.test", verify_signatures=False)

Method reference

Conversations

Contacts

Visitors

Integrations

Tickets, articles, widgets, media, webhooks, meta

Models

Every returned object keeps the payload it was built from. Attribute access is the nice path; obj.raw is always there for a field this SDK version does not know about yet, so a release is never what stands between you and a new API field.

convo.status          # typed
convo.raw["status"]   # the same thing
convo["status"]       # shorthand for the raw payload
convo.to_dict()       # exactly what the API sent

Recipes

Deflect a ticket with your help center

@bot.on_message
def deflect(ctx):
    hits = ctx.client.articles.list(query=ctx.text, published=True)
    if hits:
        ctx.reply(f"This might help: “{hits[0].title}”. Did that solve it?",
                  buttons=["Solved it", "Open a ticket"])
    else:
        ctx.reply("Let's get this to the team.", ticket_form=True)

Qualify a lead and route it

@bot.on_message
def qualify(ctx):
    if not ctx.event.visitor or not ctx.event.visitor.email:
        ctx.reply("Happy to help — what's your work email?", ask_email=True)
        return

    ctx.client.contacts.upsert(
        ctx.event.visitor.email,
        custom_fields={"lead_source": "chat", "intent": classify(ctx.text)},
    )
    ctx.note("Qualified lead — routing to sales.")
    ctx.assign(SALES_USER_ID)
    ctx.handoff()

Rescue a bad rating

@bot.on_rating
def follow_up(ctx):
    rating = ctx.event.rating or {}
    if rating.get("score", 5) <= 2:
        ctx.client.tickets.create(
            email=ctx.event.visitor.email,
            message=f"Low CSAT ({rating['score']}/5): {rating.get('comment', '')}",
            priority="high",
        )

Nightly contact sync

for user in your_database.active_users():
    client.contacts.upsert(
        user.email, name=user.name, company=user.company,
        custom_fields={"plan": user.plan, "mrr": str(user.mrr)},
    )

A store we have never heard of

@plugin.action("catalog.search_products",
               description="Search the catalog by keyword.",
               parameters=[parameter("query", required=True)])
def search(req):
    hits = your_catalog.search(req.get("query"))
    return req.ok(products=[to_conecto(p) for p in hits]) if hits else req.not_found()

That is the whole integration. Product cards, the Home showcase, and the discipline that stops the model pasting raw URLs all come along with the reserved name.

Building something? Talk to us — we want people building on Conecto, and we prioritize what devs ask for. The SDK is MIT licensed and open source; issues and pull requests are welcome on GitHub.