blackbirdtalk

The official Python library for Blackbird bots.

pip install blackbirdtalk

Python 3.11 or newer. One dependency (httpx).

from blackbirdtalk import Bot, Button

bot = Bot()  # reads BLACKBIRD_TOKEN; or Bot("bb_...")

@bot.command("start")
async def start(msg):
    await msg.answer("Welcome!", buttons=[[Button("Pay ₹49", callback="pay_49", style="main", icon="card")]])

@bot.button("pay_49")
async def pay(tap):
    await tap.answer("Payment started")

bot.run()

Get the token in the Blackbird app: Settings › Bot Studio › your bot › Overview. Set it in your terminal so it stays out of your code:

Or start from a ready project: blackbirdtalk new mybot. Check a token, its webhook and commands any time with blackbirdtalk check.

Contents

• Handlers
• Buttons
• Questions with ask()
• Files
• Groups
• Private bots
• Webhooks
• No async: SyncBot
• Testing your bot
• Errors and retries
• Every method
• Updates and versions

Handlers

The first handler that matches an update runs, in the order you wrote them. Messages from one chat are handled one at a time, in order; different chats run in parallel.

@bot.command("quiz", "q")                 # /quiz physics -> msg.args == "physics"
async def quiz(msg): ...

@bot.message(regex=r"(\d+) apples")       # msg.match[1] == "3"
async def apples(msg): ...

@bot.message(text=["hi", "hello"])
async def hi(msg): ...

@bot.message(kind="image")                # image, video, voice, doc, text
async def photo(msg): ...

@bot.message(chat="group", when=lambda m: len(m.text or "") > 500)
async def long_text(msg): ...

@bot.message()                            # everything else
async def fallback(msg): ...

@bot.on_error
async def oops(error, update):
    print("handler failed:", error)

A message has msg.answer() (same chat), msg.reply() (quoting it), msg.answer_file(), msg.edit(), msg.delete(), msg.pin(), msg.typing(), msg.ask() and msg.download().

Buttons

Blackbird buttons have styles, icons, colours, sizes, a confirm step and copy-to-clipboard. The library checks every rule the server checks before it sends, so a mistake fails on your machine with the button's row and position.

from blackbirdtalk import Button, Buttons

menu = Buttons(
    [Button("Buy ₹49", callback="buy", style="success", icon="card", confirm="Pay ₹49?")],
    [Button("Coupon", copy="SAVE20", icon="copy"), Button("Site", url="https://example.com", icon="link")],
    [Button("Brand colour", callback="x", color="#5B3FD9", size="pill")],
)
print(menu.preview())          # see it in the terminal
await msg.answer("Pick one", buttons=menu)

@bot.button(prefix="buy")      # also: exact data, a list, regex=, when=
async def buy(tap):
    await tap.answer("Done!")              # a short note (alert=True for a dialog)
    await tap.edit("Paid ✓", buttons=[])   # change the message in place

style: main, normal, success, warning, danger. size: pill, normal, full. icon: one of 40 (blackbirdtalk.BUTTON_ICONS). Up to 8 rows, 8 per row. Taps you don't answer are answered for you, so the spinner never hangs.

Keyboard buttons (under the text box) send their text; they can also ask for the person's location or contact:

from blackbirdtalk import Keyboard, RemoveKeyboard

await msg.answer("When?", keyboard=Keyboard([["Today", "Tomorrow"],
                                             [Button("Share location", request="location")]], one_time=True))
await msg.answer("Thanks", reply_markup=RemoveKeyboard())

Questions with ask()

Write a conversation top to bottom. Each person's answers come back to their own question, even with many people at once.

from blackbirdtalk import AskTimeout

@bot.command("register")
async def register(msg):
    try:
        name = await msg.ask("Your name?", placeholder="Full name")
        tap = await msg.ask("Class?", buttons=[[Button("10", callback="10"), Button("12", callback="12")]])
    except AskTimeout:                    # default 5 minutes; timeout=...
        return await msg.answer("Stopped. Send /register again.")
    await msg.answer(f"Saved {name.text}, class {tap.data}")

A text answer comes back as a Message, a tap as a Callback.

Files

await msg.answer_file("report.pdf", caption="Your report")        # a path
await msg.answer_file(InputFile(data, filename="chart.png"))      # bytes made in memory
await bot.send_document(chat_id, file_id=old.file.file_id)        # send again, no upload

@bot.message(kind="doc")
async def got(msg):
    data = await msg.download()            # bytes; or msg.download("saved.pdf")

Groups

Add the bot to a group and give it the powers it needs (approve, remove, delete, pin ...); msg.chat.my_powers lists what it has.

@bot.on_join_request()
async def check(req):
    await (req.approve() if "blackbird" in (req.answer or "").lower() else req.decline())

@bot.on_member(lambda m: m.event == "joined")
async def welcome(change):
    await change.chat.send("Welcome " + ", ".join(u.name for u in change.users))

@bot.on_my_membership(lambda m: m.status == "added")
async def added(change): ...

Private bots

Choose who can use the bot (everyone, owner, or chosen people) from the code or from Bot Studio. People can ask for access; you decide.

await bot.set_access(audience="chosen", message="For our class only", allow=[user_id])

@bot.on_access_request()
async def asked(req):
    await req.approve()        # or req.decline()

Webhooks

Polling (bot.run()) needs nothing but the token. On a server with HTTPS you can use a webhook instead:

app = bot.webhook_app(secret="long-random-text", path="/hook")   # ASGI: uvicorn mybot:app
await bot.set_webhook("https://your.domain/hook", secret_token="long-random-text")   # once

FastAPI, aiohttp or anything else:

@api.post("/hook")
async def hook(request: Request):
    status = await bot.handle_webhook(await request.body(), request.headers, secret=SECRET)
    return Response(status_code=status)

SyncBot.wsgi_app() does the same for gunicorn, Flask and Django. The secret header is checked in constant time; requests without it get 401.

No async: SyncBot

Same handlers and helpers with plain functions:

from blackbirdtalk import SyncBot

bot = SyncBot()

@bot.command("start")
def start(msg):
    msg.answer("Hi!")

bot.run()

Testing your bot

FakeBlackbird stands in for the server: no token, no network.

from blackbirdtalk.testing import FakeBlackbird

async def test_pay():
    fake = FakeBlackbird(bot)
    await fake.send("/start")
    assert fake.last("sendMessage")["text"] == "Welcome!"
    await fake.tap(fake.last_message(), "pay_49")
    assert fake.last("answerCallback")["text"] == "Payment started"

Errors and retries

Every error is a BlackbirdError with .code, .description and .method: BadRequest, Unauthorized (token wrong or replaced), Forbidden (the person hasn't started the bot, or blocked it), NotFound, Conflict (another copy is polling with this token), RateLimited (.retry_after), ServerError, NetworkError, and MarkupError for a button mistake caught before sending.

"Too many requests" is waited out by itself (up to max_retry_wait), and server errors and lost connections are retried with back-off. Retries never send a message twice: each send carries an id the server uses to drop repeats.

Every method

All 25 Bot API methods are on the bot with snake_case names (send_message, edit_message, answer_callback, set_commands, get_chat, remove_member, set_access ...), typed, with docs in your editor. Telegram-style names are accepted by the server too, so code ported from a Telegram bot mostly works.

New fields from the server show up on objects at once (obj["new_field"], or obj.to_dict()), and a method or parameter newer than your library can still be used: await bot.send_message(chat_id, "hi", new_param=1). The full reference: https://blackbirdtalk.com/bots.

Updates and versions

The library prints one line a day when a newer version is out (turn it off with Bot(check_updates=False) or BLACKBIRD_NO_UPDATE_CHECK=1). Update with pip install -U blackbirdtalk. Versions: a new minor (1.2.0) adds things, a patch (1.2.1) fixes things; your code keeps working across both. The JavaScript library @blackbirdtalk/bot has the same version number and features.

MIT licence.