Application API

Document version: 0.2 β€” Last updated: 2026-10-08 β€” Status: πŸ”΄ UNDER REVIEW

Import concrete applications from kajenn_bot_application. Call management and sending APIs from trusted application code after startup. Grammar classes are imported from their corresponding provider modules.

Configuration and input examples are in Getting started, Telegram bots and WhatsApp bots.

Shared conversation and task APIs

class kajenn_bot_application.BotBaseApplication(*, persistence_route=None, webhook_url=None, client=None, **kwargs)

Base for concrete messaging applications with class-owned bot grammars.

Parameters:

  • persistence_route (str | None)

  • webhook_url (str | None)

  • client (httpx.AsyncClient | None)

  • kwargs (Any)

get_bot_registration(code)

Return a copy of the trusted registration, including its credentials.

Parameters:

code (str)

Return type:

dict[str, Any]

async activate_bot(code)

Activate a saved bot; only receivers configure a webhook.

Parameters:

code (str)

Return type:

RoutingClass

async create_conversation(bot_code, *, participants, route, context=None, expires_at=None)

Persist an independent conversation; participants include user_id and chat_id.

Parameters:

  • bot_code (str)

  • participants (list[dict[str, Any]])

  • route (str)

  • context (dict[str, Any] | None)

  • expires_at (float | None)

Return type:

dict[str, Any]

async get_conversation(bot_code, conversation_id)

Read one conversation through the application persistence route.

Parameters:

  • bot_code (str)

  • conversation_id (str)

Return type:

dict[str, Any]

async send_conversation_message(bot_code, conversation_id, user_id, text, *, buttons=None, chat_id=None)

Send to one participant; buttons map labels to routed action names.

Parameters:

  • bot_code (str)

  • conversation_id (str)

  • user_id (int | str)

  • text (str)

  • buttons (dict[str, str] | None)

  • chat_id (int | str | None)

Return type:

Any

async update_conversation_context(bot_code, conversation_id, context, *, revision)

Replace context only when the caller’s snapshot is still current.

Parameters:

  • bot_code (str)

  • conversation_id (str)

  • context (dict[str, Any])

  • revision (int)

Return type:

dict[str, Any]

async close_conversation(bot_code, conversation_id, *, state='closed')

Conclude or cancel a general conversation; admission decisions use admin callbacks.

Parameters:

  • bot_code (str)

  • conversation_id (str)

  • state (str)

Return type:

None

async send_text(bot_code, chat_id, text)

Send long plain text in ordered chunks, preserving every character.

Parameters:

  • bot_code (str)

  • chat_id (int | str)

  • text (str)

Return type:

list[Any]

async send_document(bot_code, chat_id, document, *, filename=None, caption='')

Send a file_id, provider-fetchable URL or bytes with a filename.

Parameters:

  • bot_code (str)

  • chat_id (int | str)

  • document (str | bytes)

  • filename (str | None)

  • caption (str)

Return type:

Any

async send_announcement(bot_code, chat_ids, text)

Send once per distinct destination and report complete, partial or failed delivery.

Parameters:

  • bot_code (str)

  • chat_ids (list[int | str])

  • text (str)

Return type:

list[dict[str, Any]]

async queue_announcement(bot_code, chat_ids, text)

Stage an announcement in kajenn.tasks; results are kept in the task spool.

Parameters:

  • bot_code (str)

  • chat_ids (list[int | str])

  • text (str)

Return type:

str

async schedule_reminder(bot_code, chat_id, text, *, when, conversation_id=None, user_id=None)

Persist a one-shot reminder, optionally bound to an open conversation participant.

Parameters:

  • bot_code (str)

  • chat_id (int | str)

  • text (str)

  • when (datetime)

  • conversation_id (str | None)

  • user_id (int | str | None)

Return type:

str

async get_reminder(code)

Read this application’s reminder schedule and delivery state.

Parameters:

code (str)

Return type:

dict[str, Any]

async cancel_reminder(code)

Cancel a pending reminder; False means it has already left pending state.

Parameters:

code (str)

Return type:

bool

get_bot_registration returns trusted registration data, including credentials. Do not serialize it into logs or public responses. For persistence contracts, see Persistence.

Telegram

class kajenn_bot_application.TelegramBotApplication(*, persistence_route=None, webhook_url=None, client=None, **kwargs)

Own bot instances, sending directly and optionally receiving webhooks.

register_bot is a trusted in-process API, not a public HTTP route. client optionally supplies an httpx client (owned by the caller). Constructor registry/URL options override the application’s grammar values.

Parameters:

  • persistence_route (str | None)

  • webhook_url (str | None)

  • client (httpx.AsyncClient | None)

  • kwargs (Any)

async register_bot(*, code, bot_class, token, name='', icon=None, config=None)

Validate, durably register, then activate a bot created with BotFather.

Failed persistence leaves the bot inactive. Failed webhook activation leaves its registration available for activate_bot or startup. Duplicate codes and tokens within this application are rejected. A separate send-only application can use the same token without replacing its webhook.

Parameters:

  • code (str)

  • bot_class (type[RoutingClass])

  • token (str)

  • name (str)

  • icon (str | None)

  • config (dict[str, Any] | None)

Return type:

RoutingClass

async send_message(bot_code, chat_id, text, *, reply_markup=None)

Send a plain-text message to a known Telegram chat.

Parameters:

  • bot_code (str)

  • chat_id (int)

  • text (str)

  • reply_markup (dict[str, Any] | None)

Return type:

Any

async send_typing(bot_code, chat_id)

Emit one typing indication; no background refresh loop is started.

Parameters:

  • bot_code (str)

  • chat_id (int)

Return type:

Any

async send_media(bot_code, chat_id, kind, media, *, filename=None, caption='')

Send a document, photo, video, audio, voice or animation by reference or upload.

Parameters:

  • bot_code (str)

  • chat_id (int)

  • kind (str)

  • media (str | bytes)

  • filename (str | None)

  • caption (str)

Return type:

Any

async send_poll(bot_code, chat_id, question, options, *, is_anonymous=True, allows_multiple_answers=False, route=None)

Send and track a native regular poll; optionally route result events to the bot.

Parameters:

  • bot_code (str)

  • chat_id (int)

  • question (str)

  • options (list[str])

  • is_anonymous (bool)

  • allows_multiple_answers (bool)

  • route (str | None)

Return type:

Any

async get_poll(bot_code, poll_id)

Read persisted poll totals and the latest received answer per voter.

Parameters:

  • bot_code (str)

  • poll_id (str)

Return type:

dict[str, Any]

async stop_poll(bot_code, poll_id)

Close a tracked native poll and save its final totals.

Parameters:

  • bot_code (str)

  • poll_id (str)

Return type:

Any

class kajenn_bot_application.telegram.TelegramBotGrammar

The Telegram application’s registry and optional public webhook location.

telegram = <genro_builders.builder._decorators._DeclarativeMarker object>
class kajenn_bot_application.telegram.TelegramBotInstanceGrammar

Shared access options for Telegram bot classes.

WhatsApp Business

class kajenn_bot_application.WhatsAppBotApplication(*, api_version=None, app_secret=None, verify_token=None, **kwargs)

Receive authenticated WhatsApp events and send through Cloud API.

Parameters:

  • api_version (str | None)

  • app_secret (str | None)

  • verify_token (str | None)

  • kwargs (Any)

async register_bot(*, code, bot_class, token, phone_number_id, business_account_id, name='', icon=None, config=None)

Register an already provisioned business number; never alter its subscriptions.

Parameters:

  • code (str)

  • bot_class (type[RoutingClass])

  • token (str)

  • phone_number_id (str)

  • business_account_id (str)

  • name (str)

  • icon (str | None)

  • config (dict[str, Any] | None)

Return type:

RoutingClass

async send_message(bot_code, chat_id, text)

Submit one free-form text inside a known open service window.

Parameters:

  • bot_code (str)

  • chat_id (str)

  • text (str)

Return type:

Any

async send_template(bot_code, chat_id, *, name, language, components=None)

Submit an explicitly selected approved template; approval is checked by Meta.

Parameters:

  • bot_code (str)

  • chat_id (str)

  • name (str)

  • language (str)

  • components (list[dict[str, Any]] | None)

Return type:

Any

async send_buttons(bot_code, chat_id, text, buttons)

Submit at most three label-to-payload reply buttons inside the service window.

Parameters:

  • bot_code (str)

  • chat_id (str)

  • text (str)

  • buttons (dict[str, str])

Return type:

Any

async send_media(bot_code, chat_id, kind, media, *, filename=None, caption='')

Submit media by provider ID, HTTPS URL or uploaded bytes.

Parameters:

  • bot_code (str)

  • chat_id (str)

  • kind (str)

  • media (str | bytes)

  • filename (str | None)

  • caption (str)

Return type:

Any

async get_message(bot_code, message_id)

Read API acceptance and the most advanced delivery status received.

Parameters:

  • bot_code (str)

  • message_id (str)

Return type:

dict[str, Any]

async send_announcement(bot_code, chat_ids, text='', *, template=None)

Submit explicit text or a template once per destination and collect outcomes.

Parameters:

  • bot_code (str)

  • chat_ids (list[Any])

  • text (str)

  • template (dict[str, Any] | None)

Return type:

list[dict[str, Any]]

async queue_announcement(bot_code, chat_ids, text='', *, template=None)

Queue an announcement whose sending eligibility is checked at execution time.

Parameters:

  • bot_code (str)

  • chat_ids (list[Any])

  • text (str)

  • template (dict[str, Any] | None)

Return type:

str

async schedule_reminder(bot_code, chat_id, text='', *, when, conversation_id=None, user_id=None, template=None)

Schedule explicit text or a template; record API acceptance separately from delivery.

Parameters:

  • bot_code (str)

  • chat_id (Any)

  • text (str)

  • when (datetime)

  • conversation_id (str | None)

  • user_id (Any)

  • template (dict[str, Any] | None)

Return type:

str

class kajenn_bot_application.whatsapp.WhatsAppBotGrammar

Application-wide connection and persistence settings.

whatsapp = <genro_builders.builder._decorators._DeclarativeMarker object>
class kajenn_bot_application.whatsapp.WhatsAppBotInstanceGrammar

Shared admission options plus explicit templates for administrator notices.

notifications = <genro_builders.builder._decorators._DeclarativeMarker object>

Shared instance grammar

class kajenn_bot_application.bot.BotInstanceGrammar

Common instance options; bot grammars inherit and add their own elements.

access = <genro_builders.builder._decorators._DeclarativeMarker object>