Fully async framework for building VK Teams (VK Workspace) bots, inspired by aiogram 3
Project description
vkworkspace
Fully asynchronous framework for building bots on VK Teams (VK Workspace / VK SuperApp), inspired by aiogram 3.
A modern replacement for the official mail-ru-im/bot-python (mailru-im-bot) library.
🤖 LLM-Assisted Development: Feed
llm_full.mdto any LLM (ChatGPT, Claude, Gemini, etc.) and it will know the entire framework API — write handlers, keyboards, FSM dialogs, middleware, and more without reading the docs.
Table of Contents
- vkworkspace vs mailru-im-bot
- Why async?
- Features
- Installation
- Quick Start
- Bot Configuration
- Dispatcher
- Router & Event Types
- Filters
- Text Formatting
- Inline Keyboards
- FSM (Finite State Machine)
- Custom Middleware
- Message Methods
- Bot API Methods
- Handler Dependency Injection
- Mini-Apps + Bots
- Testing
- Examples
- Project Structure
- Requirements
- License
vkworkspace vs mailru-im-bot
The official mailru-im-bot library is synchronous, based on requests, and has not been updated in years. vkworkspace is a ground-up async rewrite with a modern developer experience.
Feature Comparison
| mailru-im-bot | vkworkspace | |
|---|---|---|
| HTTP client | requests (sync) |
httpx (async) |
| Models | raw dict |
Pydantic v2 |
| Router / Dispatcher | — | aiogram-style |
Magic filters (F) |
— | F.text, F.chat.type == "private" |
| FSM (state machine) | — | Memory + Redis backends |
| Middleware pipeline | — | inner/outer, per-event type |
| Inline keyboard builder | — | InlineKeyboardBuilder |
| Rate limiter | — | built-in token-bucket |
| Retry on 5xx | — | exponential backoff |
| SSL control | — | verify_ssl=False |
| Proxy support | — | proxy= parameter |
| Handler DI | — | auto-inject bot, state, command |
| Custom command prefixes | — | prefix=("/", "!", "") |
| Error handlers | — | @router.error() |
| Type hints | partial | full |
Speed Benchmarks
Tested on Python 3.11, VK Teams corporate instance. 5 rounds per discipline, median values:
mailru-im-bot vkworkspace winner
(requests sync) (httpx async)
Bot Info (self/get) 34 ms 33 ms ~tie
Send Text 82 ms 109 ms mailru-im-bot
Send Keyboard 97 ms 90 ms vkworkspace
Edit Message 57 ms 44 ms vkworkspace
Delete Message 33 ms 38 ms mailru-im-bot
Chat Info 43 ms 42 ms ~tie
Send Actions 30 ms 28 ms vkworkspace
BURST (10 msgs) 1110 ms 335 ms vkworkspace (3.3x!)
───────────────────────────────────────────
Score 2 4 vkworkspace wins
Key takeaway: For single sequential requests, both libraries are within network noise (~1 ms difference). But when you need to send multiple messages or handle concurrent users, async wins decisively — 3.3x faster on burst operations via asyncio.gather().
The framework overhead is zero in real conditions. Network latency dominates 99%+ of total time.
Why async?
VK Teams bots spend 99% of their time waiting for the network. Synchronous frameworks (requests) block the entire process on every API call. vkworkspace uses httpx.AsyncClient and asyncio — while one request waits for a response, the bot handles other messages.
Features
- 100% async —
httpx+asyncio, zero blocking calls, real concurrency - Aiogram-like API —
Router,Dispatcher,Fmagic filters, middleware, FSM - Type-safe — Pydantic v2 models for all API types, full type hints
- Flexible filtering —
Command,StateFilter,ChatTypeFilter,CallbackData,ReplyFilter,ForwardFilter, regex, magic filters - Custom command prefixes —
Command("start", prefix=("/", "!", ""))for any prefix style - FSM — Finite State Machine with Memory and Redis storage backends
- Middleware — inner/outer middleware pipeline for logging, auth, throttling
- Text formatting —
md,htmlhelpers for MarkdownV2/HTML +FormatBuilderfor offset/length +split_textfor long messages - Keyboard builder — fluent API for inline keyboards with button styles
- Rate limiter — built-in request throttling (
rate_limit=5= max 5 req/sec) - Retry on 5xx — automatic retry with exponential backoff on server errors
- SSL control — disable SSL verification for on-premise with self-signed certs
- FSM session timeout — auto-clear abandoned FSM forms after configurable inactivity
- Proxy support — route API requests through corporate proxy
- Edited message routing —
handle_edited_as_message=Trueto process edits as new messages - Lifecycle hooks —
on_startup/on_shutdownfor init/cleanup logic - Error handlers —
@router.error()to catch exceptions from any handler - Multi-bot polling —
dp.start_polling(bot1, bot2)to poll multiple bots - 9 event types — message, edited, deleted, pinned, unpinned, members join/leave, chat info, callbacks
- Python 3.11 — 3.14 support (free-threaded / no-GIL build not yet tested)
Installation
pip install vkworkspace
With Redis FSM storage:
pip install vkworkspace[redis]
Quick Start
import asyncio
from vkworkspace import Bot, Dispatcher, Router, F
from vkworkspace.filters import Command
from vkworkspace.types import Message
router = Router()
@router.message(Command("start"))
async def cmd_start(message: Message) -> None:
await message.answer("Hello! I'm your bot.")
@router.message(F.text)
async def echo(message: Message) -> None:
await message.answer(message.text)
async def main() -> None:
bot = Bot(token="YOUR_TOKEN", api_url="https://myteam.mail.ru/bot/v1")
dp = Dispatcher()
dp.include_router(router)
await dp.start_polling(bot)
asyncio.run(main())
API URL: Each company has its own VK Teams API endpoint. Check with your IT/integrator for the correct URL. Common examples:
https://myteam.mail.ru/bot/v1(Mail.ru default)https://api.teams.yourcompany.ru/bot/v1(on-premise)https://agent.mail.ru/bot/v1(SaaS variant)Even SaaS deployments may have custom URLs — always verify with your administrator.
Bot Configuration
bot = Bot(
token="YOUR_TOKEN",
api_url="https://myteam.mail.ru/bot/v1",
timeout=30.0, # Request timeout (seconds)
poll_time=60, # Long-poll timeout (seconds)
rate_limit=5.0, # Max 5 requests/sec (None = unlimited)
proxy="http://proxy:8080", # HTTP proxy for corporate networks
parse_mode="HTML", # Default parse mode for all messages (None = plain text)
retry_on_5xx=3, # Retry up to 3 times on 5xx errors (None = disabled)
verify_ssl=True, # Set False for self-signed certs (on-premise)
)
Rate Limiter
Built-in token-bucket rate limiter prevents API throttling:
# Max 10 requests per second — framework queues the rest automatically
bot = Bot(token="TOKEN", api_url="URL", rate_limit=10)
Proxy
Corporate networks often require a proxy. The proxy parameter applies only to API calls:
bot = Bot(
token="TOKEN",
api_url="https://api.internal.corp/bot/v1",
proxy="http://corp-proxy.internal:3128",
)
Retry on 5xx
Automatic retry with exponential backoff (1s, 2s, 4s, max 8s) on server errors:
bot = Bot(token="TOKEN", api_url="URL", retry_on_5xx=3) # 3 retries (default)
bot = Bot(token="TOKEN", api_url="URL", retry_on_5xx=None) # Disabled
SSL Verification
On-premise installations with self-signed certificates can disable SSL verification:
bot = Bot(token="TOKEN", api_url="https://internal.corp/bot/v1", verify_ssl=False)
Default Parse Mode
parse_mode can be set in two places — globally on Bot, or per-message on answer() / reply() / send_text():
from vkworkspace.enums import ParseMode
# ── 1. Global default on Bot ──────────────────────────────────────
# Applies to ALL send_text / answer / reply / edit_text / send_file
bot = Bot(
token="TOKEN",
api_url="https://myteam.mail.ru/bot/v1",
parse_mode=ParseMode.HTML, # or ParseMode.MARKDOWNV2
)
# parse_mode="HTML" is sent automatically — no need to pass it
await message.answer(f"{html.bold('Hello')}, world!")
# ── 2. Override per message ───────────────────────────────────────
await message.answer("*bold*", parse_mode=ParseMode.MARKDOWNV2)
# ── 3. Disable for a single message ──────────────────────────────
await message.answer("plain text", parse_mode=None)
Tip: IDE autocomplete works — type
parse_mode=ParseMode.and your editor will suggestHTMLandMARKDOWNV2.
Dispatcher
dp = Dispatcher(
storage=MemoryStorage(), # FSM storage (default: MemoryStorage)
fsm_strategy="user_in_chat", # FSM key strategy
handle_edited_as_message=False, # Route edits to @router.message handlers
)
# Lifecycle hooks
@dp.on_startup
async def on_start():
print("Bot started!")
@dp.on_shutdown
async def on_stop():
print("Bot stopped!")
# Include routers
dp.include_router(router)
dp.include_routers(admin_router, user_router)
# Start polling (supports multiple bots)
await dp.start_polling(bot)
await dp.start_polling(bot1, bot2, skip_updates=True)
Edited Message Routing
When handle_edited_as_message=True, edited messages are routed to @router.message handlers instead of @router.edited_message. Useful when you don't need to distinguish between new and edited messages:
dp = Dispatcher(handle_edited_as_message=True)
# This handler fires for BOTH new and edited messages
@router.message(F.text)
async def handle_text(message: Message) -> None:
await message.answer(f"Got: {message.text}")
Router & Event Types
router = Router(name="my_router")
# All 9 supported event types:
@router.message(...) # New message
@router.edited_message(...) # Edited message
@router.deleted_message(...) # Deleted message
@router.pinned_message(...) # Pinned message
@router.unpinned_message(...) # Unpinned message
@router.new_chat_members(...) # Users joined
@router.left_chat_members(...) # Users left
@router.changed_chat_info(...) # Chat info changed
@router.callback_query(...) # Button clicked
# Error handler — catches exceptions from any handler
@router.error()
async def on_error(event, error: Exception) -> None:
print(f"Error: {error}")
# Sub-routers for modular code
admin_router = Router(name="admin")
user_router = Router(name="user")
main_router = Router()
main_router.include_routers(admin_router, user_router)
Filters
Command Filter
from vkworkspace.filters import Command
# Basic
@router.message(Command("start"))
@router.message(Command("help", "info")) # Multiple commands
# Custom prefixes
@router.message(Command("start", prefix="/")) # Only /start
@router.message(Command("menu", prefix=("/", "!", ""))) # /menu, !menu, menu
# Regex commands
import re
@router.message(Command(re.compile(r"cmd_\d+"))) # /cmd_1, /cmd_42, ...
# Access command arguments in handler
@router.message(Command("ban"))
async def ban_user(message: Message, command: CommandObject) -> None:
user_to_ban = command.args # Text after "/ban "
await message.answer(f"Banned: {user_to_ban}")
CommandObject injected into handler:
prefix— matched prefix ("/","!", etc.)command— command name ("ban")args— arguments after command ("user123")raw_text— full message textmatch—re.Matchif regex pattern was used
Magic Filter (F)
from vkworkspace import F
@router.message(F.text) # Has text
@router.message(F.text == "hello") # Exact match
@router.message(F.text.startswith("hi")) # String methods
@router.message(F.from_user.user_id == "admin@co") # Nested attrs
@router.message(F.chat.type == "private") # Private chat
@router.message(F.chat.type.in_(["private", "group"]))
@router.callback_query(F.callback_data == "confirm")
State Filter
from vkworkspace.filters.state import StateFilter
@router.message(StateFilter(Form.name)) # Specific state
@router.message(StateFilter("*")) # Any state (non-None)
@router.message(StateFilter(None)) # No state (default)
Other Filters
from vkworkspace.filters import (
CallbackData, ChatTypeFilter, RegexpFilter,
ReplyFilter, ForwardFilter, RegexpPartsFilter,
)
# Callback data
@router.callback_query(CallbackData("confirm"))
@router.callback_query(CallbackData(re.compile(r"^action_\d+$")))
# Chat type
@router.message(ChatTypeFilter("private"))
@router.message(ChatTypeFilter(["private", "group"]))
# Regex on message text
@router.message(RegexpFilter(r"\d{4}"))
# Reply / Forward detection
@router.message(ReplyFilter()) # Message is a reply
@router.message(ForwardFilter()) # Message contains forwards
# Regex on reply/forward text
@router.message(RegexpPartsFilter(r"urgent|asap"))
async def on_urgent(message: Message, regexp_parts_match) -> None:
await message.answer("Forwarded message contains urgent text!")
# Combine filters with &, |, ~
@router.message(ChatTypeFilter("private") & Command("secret"))
Custom Filters
from vkworkspace.filters.base import BaseFilter
class IsAdmin(BaseFilter):
async def __call__(self, event, **kwargs) -> bool:
admins = await event.bot.get_chat_admins(event.chat.chat_id)
return any(a.user_id == event.from_user.user_id for a in admins)
@router.message(IsAdmin())
async def admin_only(message: Message) -> None:
await message.answer("Admin panel")
Text Formatting
VK Teams supports MarkdownV2 and HTML formatting. Use the md and html helpers to build formatted messages safely:
from vkworkspace.utils.text import md, html, split_text
# ── MarkdownV2 ──
text = f"{md.bold('Status')}: {md.escape(user_input)}"
await message.answer(text, parse_mode="MarkdownV2")
md.bold("text") # *text*
md.italic("text") # _text_
md.underline("text") # __text__
md.strikethrough("text") # ~text~
md.code("x = 1") # `x = 1`
md.pre("code", "python") # ```python\ncode\n```
md.link("Click", "https://example.com") # [Click](https://example.com)
md.quote("quoted text") # >quoted text
md.mention("user@company.ru") # @\[user@company\.ru\]
md.escape("price: $100") # Escapes special chars
# ── HTML ──
text = f"{html.bold('Status')}: {html.escape(user_input)}"
await message.answer(text, parse_mode="HTML")
html.bold("text") # <b>text</b>
html.italic("text") # <i>text</i>
html.underline("text") # <u>text</u>
html.strikethrough("text") # <s>text</s>
html.code("x = 1") # <code>x = 1</code>
html.pre("code", "python") # <pre><code class="python">code</code></pre>
html.link("Click", "https://example.com") # <a href="...">Click</a>
html.quote("quoted text") # <blockquote>quoted text</blockquote>
html.mention("user@company.ru") # @[user@company.ru]
html.ordered_list(["a", "b"]) # <ol><li>a</li><li>b</li></ol>
html.unordered_list(["a", "b"]) # <ul><li>a</li><li>b</li></ul>
# ── Split long text ──
# VK Teams may lag on messages > 4096 chars; split_text breaks them up
for chunk in split_text(long_text):
await message.answer(chunk)
# Custom limit
for chunk in split_text(long_text, max_length=2000):
await message.answer(chunk)
Text Builder (aiogram-style)
Composable formatting nodes with auto-escaping and automatic parse_mode. No need to think about which parse mode to use — as_kwargs() handles it:
from vkworkspace.utils.text import Text, Bold, Italic, Code, Link, Pre, Quote, Mention
# Compose text from nodes — strings are auto-escaped
content = Text(
Bold("Order #42"), "\n",
"Status: ", Italic("processing"), "\n",
"Total: ", Code("$99.99"),
)
await message.answer(**content.as_kwargs()) # text + parse_mode="HTML" auto
# Nesting works
content = Bold(Italic("bold italic")) # <b><i>bold italic</i></b>
# Operator chaining
content = "Hello, " + Bold("World") + "!" # Text node
await message.answer(**content.as_kwargs())
# Render as MarkdownV2 instead
await message.answer(**content.as_kwargs("MarkdownV2"))
Available nodes: Text, Bold, Italic, Underline, Strikethrough, Code, Pre, Link, Mention, Quote, Raw
Warning: Do not mix string helpers (
md.*/html.*) with node builder — raw strings inside nodes get auto-escaped, soText(md.bold("x"))produces literal*x*, not bold. UseBold("x")instead.
Format Builder (offset/length)
VK Teams also supports formatting via the format parameter — a JSON dict with offset/length ranges instead of markup. FormatBuilder makes it easy:
from vkworkspace.utils import FormatBuilder
# By offset/length
fb = FormatBuilder("Hello, World! Click here.")
fb.bold(0, 5) # "Hello" bold
fb.italic(7, 6) # "World!" italic
fb.link(14, 10, url="https://example.com") # "Click here" as link
await bot.send_text(chat_id, fb.text, format_=fb.build())
# By substring (auto-find offset)
fb = FormatBuilder("Order #42 is ready! Visit https://shop.com")
fb.bold_text("Order #42")
fb.italic_text("ready")
fb.link_text("https://shop.com", url="https://shop.com")
await bot.send_text(chat_id, fb.text, format_=fb.build())
Supported styles: bold, italic, underline, strikethrough, link, mention, inline_code, pre, ordered_list, unordered_list, quote. All methods support chaining.
Inline Keyboards
from vkworkspace.utils.keyboard import InlineKeyboardBuilder
from vkworkspace.enums import ButtonStyle
builder = InlineKeyboardBuilder()
builder.button(text="Yes", callback_data="confirm", style=ButtonStyle.PRIMARY)
builder.button(text="No", callback_data="cancel", style=ButtonStyle.ATTENTION)
builder.button(text="Maybe", callback_data="maybe")
builder.adjust(2, 1) # Row 1: 2 buttons, Row 2: 1 button
await message.answer("Are you sure?", inline_keyboard_markup=builder.as_markup())
# Copy & modify
builder2 = builder.copy()
builder2.button(text="Extra", callback_data="extra")
FSM (Finite State Machine)
from vkworkspace.fsm import StatesGroup, State, FSMContext
from vkworkspace.filters.state import StateFilter
from vkworkspace.fsm.storage.memory import MemoryStorage
class Form(StatesGroup):
name = State()
age = State()
@router.message(Command("start"))
async def start(message: Message, state: FSMContext) -> None:
await state.set_state(Form.name)
await message.answer("What is your name?")
@router.message(StateFilter(Form.name), F.text)
async def process_name(message: Message, state: FSMContext) -> None:
await state.update_data(name=message.text)
await state.set_state(Form.age)
await message.answer("How old are you?")
@router.message(StateFilter(Form.age), F.text)
async def process_age(message: Message, state: FSMContext) -> None:
data = await state.get_data()
await message.answer(f"Name: {data['name']}, Age: {message.text}")
await state.clear()
# Storage backends
dp = Dispatcher(storage=MemoryStorage()) # In-memory (dev)
from vkworkspace.fsm.storage.redis import RedisStorage
dp = Dispatcher(storage=RedisStorage()) # Redis (prod)
FSMContext methods:
await state.get_state()— current state (orNone)await state.set_state(Form.name)— transition to stateawait state.get_data()— get stored data dictawait state.update_data(key=value)— merge into dataawait state.set_data({...})— replace data entirelyawait state.clear()— clear state and data
FSM Session Timeout
Prevent users from getting stuck in abandoned FSM forms. When enabled, the middleware automatically clears expired FSM sessions:
dp = Dispatcher(
storage=MemoryStorage(),
session_timeout=300, # 5 minutes — clear FSM if user is inactive
)
If a user starts a form and doesn't finish within 5 minutes, the next message clears the state and goes through as a normal message — no more "Enter your age" after 3 days of silence.
Disabled by default (session_timeout=None).
Custom Middleware
from vkworkspace import BaseMiddleware
class LoggingMiddleware(BaseMiddleware):
async def __call__(self, handler, event, data):
print(f"Event from: {event.from_user.user_id}")
result = await handler(event, data)
print(f"Handled OK")
return result
class ThrottleMiddleware(BaseMiddleware):
def __init__(self, delay: float = 1.0):
self.delay = delay
self.last: dict[str, float] = {}
async def __call__(self, handler, event, data):
import time
uid = getattr(event, "from_user", None)
uid = uid.user_id if uid else "unknown"
now = time.monotonic()
if now - self.last.get(uid, 0) < self.delay:
return None # Skip handler — too fast
self.last[uid] = now
return await handler(event, data)
# Register on specific event observer
router.message.middleware.register(LoggingMiddleware())
router.message.middleware.register(ThrottleMiddleware(delay=0.5))
router.callback_query.middleware.register(LoggingMiddleware())
Message Methods
# In handler:
await message.answer("Hello!") # Send to same chat
await message.reply("Reply to you!") # Reply to this message
await message.edit_text("Updated text") # Edit this message
await message.delete() # Delete this message
await message.pin() # Pin this message
await message.unpin() # Unpin this message
await message.answer_file(file=InputFile("photo.jpg")) # Send file
await message.answer_voice(file=InputFile("audio.ogg")) # Send voice
Bot API Methods
bot = Bot(token="TOKEN", api_url="URL")
# Self
await bot.get_me()
# Messages
await bot.send_text(chat_id, "Hello!", parse_mode="HTML")
await bot.send_text(chat_id, "Hi", reply_msg_id=msg_id) # Reply
await bot.send_text(chat_id, "Hi", reply_msg_id=[id1, id2]) # Multi-reply
await bot.send_text(chat_id, "Hi", forward_chat_id=cid, forward_msg_id=mid) # Forward
await bot.send_text(chat_id, "Hi", request_id="unique-123") # Idempotent send
await bot.send_text_with_deeplink(chat_id, "Open", deeplink="payload") # Deeplink
await bot.edit_text(chat_id, msg_id, "Updated")
await bot.delete_messages(chat_id, msg_id)
await bot.send_file(chat_id, file=InputFile("photo.jpg"), caption="Look!")
await bot.send_voice(chat_id, file=InputFile("voice.ogg"))
await bot.answer_callback_query(query_id, "Done!", show_alert=True)
# Chat management
await bot.get_chat_info(chat_id)
await bot.get_chat_admins(chat_id)
await bot.get_chat_members(chat_id)
await bot.get_blocked_users(chat_id)
await bot.get_pending_users(chat_id)
await bot.set_chat_title(chat_id, "New Title")
await bot.set_chat_about(chat_id, "Description")
await bot.set_chat_rules(chat_id, "Rules")
await bot.set_chat_avatar(chat_id, file=InputFile("avatar.png"))
await bot.block_user(chat_id, user_id, del_last_messages=True)
await bot.unblock_user(chat_id, user_id)
await bot.resolve_pending(chat_id, approve=True, user_id=uid)
await bot.add_chat_members(chat_id, members=[uid1, uid2]) # Add members
await bot.delete_chat_members(chat_id, members=[uid1, uid2]) # Remove members
await bot.pin_message(chat_id, msg_id)
await bot.unpin_message(chat_id, msg_id)
await bot.send_actions(chat_id, "typing")
# Files
await bot.get_file_info(file_id)
# Threads
thread = await bot.threads_add(chat_id, msg_id) # thread.thread_id — the thread's own chat ID
await bot.threads_autosubscribe(chat_id, enable=True, with_existing=True)
Threads
VK Teams threads work differently from Telegram topics — each thread gets its
own chat ID. Thread messages arrive as ordinary newMessage events but
with parent_topic set.
# Create a thread under a message
thread = await bot.threads_add(chat_id, msg_id)
# thread.thread_id = "XXXXXXX@chat.agent" — the new thread's own chat ID
# Enable autosubscribe so the bot receives thread replies
await bot.threads_autosubscribe(chat_id, enable=True)
# Reply into a thread (convenience method)
@router.message(Command("discuss"))
async def start_thread(message: Message):
await message.answer_thread("Let's discuss here!")
# Handle messages sent inside a thread
@router.message(F.is_thread_message)
async def on_thread_msg(message: Message):
# message.chat.chat_id = thread's own chat ID
# message.thread_root_chat_id = original group/channel chat ID
# message.thread_root_message_id = root message ID (int)
await message.answer("Saw your thread reply!")
Channel comments are thread messages. When a bot is added to a channel it automatically receives all post comments as
newMessageevents withis_thread_message == Trueandthread_root_chat_idequal to the channel's chat ID — no extra subscription needed.
Handler Dependency Injection
Handlers receive only the parameters they declare. The framework inspects the signature and injects available values:
@router.message(Command("info"))
async def full_handler(
message: Message, # The message object
bot: Bot, # Bot instance
state: FSMContext, # FSM context (if storage configured)
command: CommandObject, # Parsed command (from Command filter)
raw_event: dict, # Raw VK Teams event dict
event_type: str, # "message", "callback_query", etc.
) -> None:
...
# Minimal — only take what you need
@router.message(F.text)
async def simple(message: Message) -> None:
await message.answer(message.text)
Mini-Apps + Bots
VK Teams mini-apps (WebView apps inside the client) cannot send notifications or messages to users. The official recommendation is to use a bot as the notification channel.
Typical pattern:
- Bot sends a message with a mini-app link (with parameters for deep-linking)
- Mini-app opens, identifies the user via
GetSelfId/GetAuth(JS Bridge) - Mini-app communicates with its own backend (can be the same server as the bot)
- For notifications back to the user — the bot sends messages via the Bot API
# Bot sends a deep-link to a mini-app
miniapp_url = "https://u.myteam.mail.ru/miniapp/my-app-id?order=42"
await message.answer(
f"Open your order: {miniapp_url}",
)
Mini-apps are registered via Metabot (
/newapp) — the same bot used for bot management (/newbot).
Testing
The project includes two types of tests:
Unit Tests (mocked)
pip install -e ".[dev]"
pytest tests/test_bot_api.py -v
98 tests covering all Bot API methods and FSM with mocked HTTP transport — no real API calls.
Live API Tests
Run against a real VK Teams instance to verify all methods work in your environment:
python tests/test_bot_live.py \
--token "YOUR_BOT_TOKEN" \
--api-url "https://myteam.mail.ru/bot/v1" \
--chat "GROUP_CHAT_ID"
Or via environment variables:
export VKWS_TOKEN="..."
export VKWS_API_URL="https://myteam.mail.ru/bot/v1"
export VKWS_CHAT_ID="12345@chat.agent"
python tests/test_bot_live.py
The live test sends real messages, tests all endpoints, and cleans up after itself. Methods unsupported on your installation (e.g. threads) are automatically skipped.
Examples
| Example | Description |
|---|---|
| echo_bot.py | Basic commands + text echo |
| Features | |
| keyboards.py | Inline keyboards, callbacks, CallbackDataFactory, pagination |
| formatting.py | MarkdownV2/HTML + Text builder + FormatBuilder |
| fsm.py | Multi-step dialog with FSM + session timeout |
| files.py | 8 ways to send files + voice conversion |
| typing_actions.py | Typing indicator: decorator, context manager, one-shot |
| middleware.py | Custom middleware (logging, access control) |
| multi_router.py | Modular sub-routers, chat events, ReplyFilter, ForwardFilter |
| error_handling.py | Error handlers, lifecycle hooks, edited message routing |
| custom_prefix.py | Custom command prefixes, regex commands, argument parsing |
| proxy.py | Corporate proxy + rate limiter + retry on 5xx + SSL control |
| Integrations | |
| server_with_bot.py | BotServer — HTTP API for non-Python systems |
| fastapi_bot.py | Bot inside FastAPI with shared Bot instance and DI |
| redis_listener.py | RedisListener — consume tasks from Redis Streams |
| zabbix_alert.py | Zabbix alerts with feedback chain (FSM + BotServer) |
| chatops.py | ChatOps with RBAC, Scheduler, audit log |
| report.py | BI reports: run_sync() for Oracle/pandas, Scheduler |
Project Structure
vkworkspace/
├── client/ # Async HTTP client (httpx), rate limiter, proxy
├── types/ # Pydantic v2 models (Message, Chat, User, CallbackQuery, ...)
├── enums/ # EventType, ChatType, ParseMode, ButtonStyle, ...
├── dispatcher/ # Dispatcher, Router, EventObserver, Middleware pipeline
├── filters/ # Command, StateFilter, CallbackData, ChatType, Regexp
├── fsm/ # StatesGroup, State, FSMContext, Memory/Redis storage
└── utils/ # InlineKeyboardBuilder, magic filter (F), text formatting (md, html, split_text)
Requirements
- Python 3.11+
- httpx >= 0.28.1
- pydantic >= 2.6
- magic-filter >= 1.0.12
License
MIT LICENSE.
Keywords: VK Teams bot, VK Workspace bot, VK WorkSpace, VK SuperApp, vkteams, workspace.vk.ru, workspacevk, vkworkspace, myteam.mail.ru, agent.mail.ru, mail-ru-im-bot, mailru-im-bot, bot-python, VK Teams API, VK Teams framework, async bot framework, Python bot VK Teams, aiogram VK Teams, httpx bot, Pydantic bot framework
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file vkworkspace-1.8.5.tar.gz.
File metadata
- Download URL: vkworkspace-1.8.5.tar.gz
- Upload date:
- Size: 174.2 kB
- Tags: Source
- Uploaded using Trusted Publishing? Yes
- Uploaded via: twine/6.1.0 CPython/3.13.7
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
c9a7193330868ff10d6eef3c4938f4fb7f251b5871a7dc281da0be8eb256c817
|
|
| MD5 |
c278b060144c6a8c4520753183cf2481
|
|
| BLAKE2b-256 |
8aa4886bd15480ca3b49416e31590f6fdd147d21f0143b0d36e41570582b393f
|
Provenance
The following attestation bundles were made for vkworkspace-1.8.5.tar.gz:
Publisher:
release.yml on TimmekHW/vkworkspace
-
Statement:
-
Statement type:
https://in-toto.io/Statement/v1 -
Predicate type:
https://docs.pypi.org/attestations/publish/v1 -
Subject name:
vkworkspace-1.8.5.tar.gz -
Subject digest:
c9a7193330868ff10d6eef3c4938f4fb7f251b5871a7dc281da0be8eb256c817 - Sigstore transparency entry: 1002374962
- Sigstore integration time:
-
Permalink:
TimmekHW/vkworkspace@b937bc273b3c39ae33f792f928d4c4de90317776 -
Branch / Tag:
refs/tags/v1.8.5 - Owner: https://github.com/TimmekHW
-
Access:
public
-
Token Issuer:
https://token.actions.githubusercontent.com -
Runner Environment:
github-hosted -
Publication workflow:
release.yml@b937bc273b3c39ae33f792f928d4c4de90317776 -
Trigger Event:
push
-
Statement type:
File details
Details for the file vkworkspace-1.8.5-py3-none-any.whl.
File metadata
- Download URL: vkworkspace-1.8.5-py3-none-any.whl
- Upload date:
- Size: 93.2 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? Yes
- Uploaded via: twine/6.1.0 CPython/3.13.7
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
5d69f659f150193e93ca730d5df04ebab0847d5c5f4de7919e0e7bdd2f60b36f
|
|
| MD5 |
d657ee4c60e445a479545a9f80f65582
|
|
| BLAKE2b-256 |
bd80f0cc4aae0f1af37cf95a7b4893a3fc42f2c5a402f4e5aaa4c789a626ab20
|
Provenance
The following attestation bundles were made for vkworkspace-1.8.5-py3-none-any.whl:
Publisher:
release.yml on TimmekHW/vkworkspace
-
Statement:
-
Statement type:
https://in-toto.io/Statement/v1 -
Predicate type:
https://docs.pypi.org/attestations/publish/v1 -
Subject name:
vkworkspace-1.8.5-py3-none-any.whl -
Subject digest:
5d69f659f150193e93ca730d5df04ebab0847d5c5f4de7919e0e7bdd2f60b36f - Sigstore transparency entry: 1002374968
- Sigstore integration time:
-
Permalink:
TimmekHW/vkworkspace@b937bc273b3c39ae33f792f928d4c4de90317776 -
Branch / Tag:
refs/tags/v1.8.5 - Owner: https://github.com/TimmekHW
-
Access:
public
-
Token Issuer:
https://token.actions.githubusercontent.com -
Runner Environment:
github-hosted -
Publication workflow:
release.yml@b937bc273b3c39ae33f792f928d4c4de90317776 -
Trigger Event:
push
-
Statement type: