Async Telegram bot framework for Python.
Routing, middleware, scenes, keyboards, webhooks, and a built-in test harness,
with a public record of which Bot API methods are verified.
Installation · Quickstart · Features · Testing · Examples · Docs · Contributing
- Async handlers. Every handler is a plain
async defthat receives aContext. - Middleware at the core. Routing, scenes, logging, and rate limiting are all middleware, so they compose the same way.
- Typed. The package ships type hints (
py.typed) and is checked withmypy --strict. - Testable.
TestBotandUpdateslet you simulate updates and assert on replies without touching the network. - Honest about coverage. A bundled table records the verification status of each tracked Bot API method.
- Light. The only runtime dependencies are
httpxandtyping-extensions.
Status: alpha (
0.1.0). The API may change between minor versions. See the changelog.
pip install wizardgramRequires Python 3.10 or newer. See the installation guide for virtual environments and development extras.
-
Create a bot with @BotFather (send
/newbot) and copy the token. -
Export the token. Keep it out of source control.
# macOS / Linux export WIZARDGRAM_TOKEN="123456:ABC-your-token"
# Windows PowerShell $env:WIZARDGRAM_TOKEN = "123456:ABC-your-token"
-
Save this as
bot.pyand runpython bot.py:import os import wizardgram bot = wizardgram.Bot(token=os.environ["WIZARDGRAM_TOKEN"]) @bot.command("start") async def start(ctx: wizardgram.Context) -> None: await ctx.reply("Hello!") if __name__ == "__main__": bot.run()
-
Open your bot in Telegram and send
/start.
bot.run() uses long polling. It keeps polling after network or Telegram errors and backs off between retries, up to 30 seconds.
Four decorators cover the common cases. Each one is also available as a direct call: bot.command("help", handler).
@bot.command("help") # /help
async def help_(ctx): ...
@bot.hears(r"(?i)^ping$") # text matching a regex; ctx.match holds the match
async def ping(ctx): await ctx.reply("pong")
@bot.action(r"^vote:(\d+)$") # callback data matching a regex
async def vote(ctx): ...
@bot.on("edited_message") # any Telegram update type
async def edited(ctx): ...Handlers run in registration order. The first one that matches handles the update.
A Context wraps each update and exposes what you usually need: ctx.text, ctx.chat, ctx.from_user, ctx.match, ctx.callback_query, ctx.update_type, and ctx.scene.
| Helper | What it does |
|---|---|
reply(text, **kwargs) |
Send a message to the current chat |
reply_with_keyboard(text, keyboard) |
Send a message with a keyboard attached |
reply_with_photo(...), reply_with_document(...) |
Upload media (file path, bytes, or file object) |
edit_text(...), edit_caption(...), edit_reply_markup(...) |
Edit the message that triggered the update |
delete_message() |
Delete a message |
answer_callback_query(text) |
Acknowledge an inline button press |
send_chat_action("typing") |
Show a typing or upload indicator |
Middleware wraps every update. Do work before and after await next_(), or skip it to stop the update.
import logging
from wizardgram import Context, Next
@bot.middleware
async def log_updates(ctx: Context, next_: Next) -> None:
logging.info("update=%s type=%s", ctx.update_id, ctx.update_type)
await next_()You can also register with bot.use(fn), which returns the bot so calls can be chained. An exception raised inside middleware is logged and the chain continues, so one faulty middleware does not take the bot down.
from wizardgram import Keyboard
# Inline keyboard
keyboard = (
Keyboard.inline()
.button("Confirm", callback_data="confirm")
.button("Cancel", callback_data="cancel")
.row()
.button("Docs", url="https://github.com/daddymaou/wizardgram")
.build()
)
await ctx.reply("Choose:", reply_markup=keyboard)
# Reply keyboard
keyboard = Keyboard.reply().button("Share contact", request_contact=True).resize().one_time().build()A Scene is an ordered list of step handlers. A Stage keeps track of which step each chat is on. Steps share data through ctx.scene.state.
from wizardgram import Context, Scene, Stage
async def ask_name(ctx: Context) -> None:
ctx.scene.state["name"] = ctx.text or ""
await ctx.reply("What city are you from?")
await ctx.scene.next()
async def finish(ctx: Context) -> None:
await ctx.reply(f"Thanks, {ctx.scene.state['name']}. You're signed up.")
await ctx.scene.leave()
bot.use(Stage([Scene("signup", [ask_name, finish])]).middleware())
@bot.command("signup")
async def signup(ctx: Context) -> None:
await ctx.scene.enter("signup")
await ctx.reply("What is your name?")Note: scene state is held by
MemoryStateStore, so it is not persistent. Progress is lost when the process restarts.
handle_webhook accepts any request object with an async json() method. Here it is with FastAPI:
import os
from fastapi import FastAPI, Request
from wizardgram import Bot
bot = Bot(token=os.environ["WIZARDGRAM_TOKEN"])
app = FastAPI()
@app.post("/telegram/webhook")
async def telegram_webhook(request: Request) -> dict[str, bool]:
return await bot.handle_webhook(request)Register the URL once with await bot.set_webhook("https://example.com/telegram/webhook"), and remove it with await bot.delete_webhook(). Call await bot.close() on shutdown. Run the full example with uvicorn examples.webhook_bot:app (install fastapi and uvicorn separately).
@bot.command("photo")
async def photo(ctx: Context) -> None:
await ctx.reply_with_photo("photo.jpg", caption="Attached")Files can be a path, raw bytes, a (filename, bytes) tuple, or an open binary file.
- Flood control. On HTTP 429 the transport waits for the
retry_afterTelegram sends and retries, up tomax_retries(default 3). - Per-chat throttling.
Bot(token, min_interval_ms=1000)enforces a minimum gap between API calls to the same chat. - Timeouts.
Bot(token, timeout=30.0)sets the request timeout. - Clear errors.
TelegramError(witherror_code,description,retry_after) andNetworkErrorboth inherit fromWizardgramError.
from wizardgram import TelegramError
try:
await ctx.reply("hi")
except TelegramError as exc:
print(exc.error_code, exc.description)TestBot records outgoing API calls and never opens a network connection, so handler tests run fast and offline.
from wizardgram import Context, TestBot, Updates
async def test_hello() -> None:
bot = TestBot()
@bot.command("hello")
async def hello(ctx: Context) -> None:
await ctx.reply("world")
reply = await bot.simulate(Updates.command("hello"))
assert reply is not None
assert reply.text == "world"Updates.text(...), Updates.command(...), Updates.callback(...), and Updates.photo(...) build realistic updates. bot.calls lists every API method your handler invoked, along with its parameters.
wizardgram keeps a table of the Bot API methods it tracks and how each was checked. At 0.1.0 it holds 34 methods, all marked verified, including sendMessage, sendPhoto, editMessageText, setWebhook, banChatMember, and answerCallbackQuery. Unknown method names report unverified.
from wizardgram import status
print(status("sendMessage").value) # "verified"Print the whole table from the command line:
wizardgram| Area | Confidence | How it was verified |
|---|---|---|
| Core send methods | Verified | Listed in the bundled method status table |
| Update polling | Verified | Listed in the bundled method status table |
| Webhook methods | Verified | Listed in the bundled method status table |
| Callback queries | Verified | Listed in the bundled method status table |
| Chat administration | Verified | Listed in the bundled method status table |
| Multipart uploads | Inferred | Exercised with mocked HTTP requests |
See the coverage guide for details.
Set WIZARDGRAM_TOKEN, then run any example.
| File | What it shows | Run |
|---|---|---|
echo_bot.py |
Command and text handlers | python examples/echo_bot.py |
menu_bot.py |
Inline keyboard and callback queries | python examples/menu_bot.py |
scene_bot.py |
Two-step signup scene | python examples/scene_bot.py |
middleware_bot.py |
Logging and rate limiting | python examples/middleware_bot.py |
webhook_bot.py |
FastAPI webhook endpoint | uvicorn examples.webhook_bot:app |
src/wizardgram/
bot.py Bot: polling, webhooks, handler registration
context.py Context: per-update reply and edit helpers
router.py command / hears / action / on routing
middleware.py Middleware chain
fsm.py Scene, Stage, MemoryStateStore
keyboard.py Inline and reply keyboard builders
transport.py httpx transport: retries, flood control, uploads
coverage.py Bot API verification table and CLI
testing.py TestBot, Updates, MockMessage
errors.py WizardgramError, TelegramError, NetworkError
types.py Telegram TypedDicts
examples/ Runnable example bots
tests/ pytest suite
docs/ MkDocs documentation
Guides live in docs/: installation, quickstart, and guides for routing, context, middleware, keyboards, scenes, and coverage.
Contributions are welcome. Open a branch, make your change, and send a pull request.
git clone https://github.com/daddymaou/wizardgram.git
cd wizardgram
pip install -e ".[dev]"
ruff check . && ruff format --check . && mypy src/wizardgram && pytestRead CONTRIBUTING.md and the Code of Conduct first. To report a security issue, see SECURITY.md.
If wizardgram saves you time, a star helps other developers find it.
If you use wizardgram in research, cite it using CITATION.cff.