Metadata-Version: 2.5
Name: oibot
Version: 2026.8.19
Summary: a lightweight bot framework
Requires-Python: >=3.13
Requires-Dist: aiohttp
Requires-Dist: cryptography
Description-Content-Type: text/markdown

# OiBot

OiBot is a lightweight, fully asynchronous QQ bot framework built on
[aiohttp](https://github.com/aio-libs/aiohttp). It wraps event delivery, plugin
dispatch, message construction, dependency injection, and the QQ Bot HTTP APIs
in a small, typed interface.

## Highlights

- Webhook and WebSocket event delivery.
- Automatic Webhook callback validation and WebSocket heartbeat,
  reconnection, and session resumption.
- Recursive plugin discovery with concurrent asynchronous dispatch.
- Event routing from handler type annotations, including union types.
- Composable synchronous or asynchronous matchers with `&`, `|`, and `~`.
- Matcher-result injection and lifecycle-aware dependency injection.
- Typed event models with convenient `reply(...)` and `defer(...)` methods.
- Plain-text, Markdown, keyboard, media, input-notification, and streaming
  messages.
- Built-in access-token caching, reply-sequence management, media upload, and
  local rate limiting.
- APIs for messages, recalls, interactions, group management, menus, panels,
  bot information, and rich media.

## Requirements

- Python 3.13 or later. The current source uses Python's type-parameter syntax.
- A QQ Bot application ID and application secret.

## Installation

Install the latest stable release from PyPI:

```sh
python -m pip install --upgrade oibot
```

For a project managed by uv:

```sh
uv add oibot --frozen
```

Install the latest development version from GitHub:

```sh
python -m pip install --upgrade 'oibot @ git+https://github.com/OrganRemoved/oibot.git'
```

Alternatively, install the branch archive:

```sh
python -m pip install --upgrade https://github.com/OrganRemoved/oibot/archive/refs/heads/main.zip
```

## Quick start

Create an importable plugin package next to the bot entry point:

```text
awesome_oibot/
├── bot.py
└── plugins/
    ├── __init__.py
    └── echo.py
```

### Start the bot

```python
import logging
from os import environ

from oibot.bot import OiBot


logging.basicConfig(
    level=logging.INFO,
    format="[%(asctime)s][%(levelname)s][%(module)s:%(funcName)s:%(lineno)d] %(message)s",
)


if __name__ == "__main__":
    OiBot(
        plugins="plugins",
        app_id=environ["OIBOT_APP_ID"],
        app_secret=environ["OIBOT_APP_SECRET"],
    ).webhook(
        path="/callback",
        host=environ.get("OIBOT_HOST", "0.0.0.0"),
        port=int(environ.get("OIBOT_PORT", "8080")),
    )
```

Configure the platform callback URL to point to the public HTTPS URL ending in
`/callback`, then run `python bot.py` from `awesome_oibot`. OiBot handles the
callback-validation request with the configured application secret.

### Write the smallest useful plugin

The following follows the minimal pattern used by the accompanying bot
plugins: one ordinary rule, one typed event parameter, and one async handler.
There is no plugin base class, manifest, registration call, or
framework-specific wrapper to maintain.

```python
from oibot.event.group_at_message_create import GroupAtMessageCreateEvent
from oibot.event.group_message_create import GroupMessageCreateEvent
from oibot.plugin import on


@on(lambda event: event.content.strip() in {"Oi", "oi"})
async def oi(
    event: GroupAtMessageCreateEvent | GroupMessageCreateEvent,
) -> None:
    await event.reply("Ciallo ~ (∠・ω< )⌒★")
```

This small plugin already demonstrates the central framework design:

- The handler annotation is the event subscription. The union receives both
  at-mention and full-group-message events without duplicate handlers.
- A matcher can be any synchronous or asynchronous callable. A lambda keeps a
  simple hot path simple; richer rules can use `Matcher` composition.
- `event.reply(...)` preserves the event context and lets the framework manage
  reply identifiers, sequence numbers, rate limits, and transport details.
- The decorated function is discovered automatically when OiBot imports the
  plugin directory.

## WebSocket transport

For WebSocket event delivery, import `Intents` from its concrete module and
combine the required flags with `|`:

```python
from os import environ

from oibot.bot import OiBot
from oibot.transport.websocket import Intents


if __name__ == "__main__":
    OiBot(
        plugins="plugins",
        app_id=environ["OIBOT_APP_ID"],
        app_secret=environ["OIBOT_APP_SECRET"],
    ).websocket(
        intents=Intents.GROUP_AND_C2C_EVENT | Intents.INTERACTION,
    )
```

`GROUP_AND_C2C_EVENT` is the default. Add `INTERACTION` for callback buttons.
Other flags include `GUILDS`, `GUILD_MEMBERS`, `GUILD_MESSAGES`,
`GUILD_MESSAGE_REACTIONS`, `DIRECT_MESSAGE`, `MESSAGE_AUDIT`, `FORUMS_EVENT`,
`AUDIO_ACTION`, and `PUBLIC_GUILD_MESSAGES`.

Both `webhook(...)` and `websocket(...)` are blocking entry points that own the
bot HTTP-client lifecycle. Inside a running handler, use `event.bot` for API
calls.

## Framework design

The event path is deliberately short and explicit:

```text
Webhook / WebSocket
        ↓
typed Event + reply/session context
        ↓
PluginManager
        ↓
type annotation → matcher → dependency graph
        ↓
async handler
```

This keeps transport concerns outside business plugins while preserving the
original event and bot instance all the way to the handler.

### Modules are plugins

`OiBot(plugins=...)` accepts one path or an iterable of paths. Directories are
scanned recursively and Python files beginning with `_` are ignored. An
imported module becomes a plugin when it exposes at least one `@on(...)`
handler or an `init(app)` lifecycle hook.

There is no plugin object graph to construct. A plugin remains an ordinary
Python module, so its functions, types, and tests stay easy to reuse. Handlers
from all loaded plugins are dispatched concurrently with `asyncio.TaskGroup`;
execution order is intentionally not part of the contract.

### Type annotations are event subscriptions

`@on(...)` inspects the handler signature. An event parameter annotated with a
concrete `Event` subclass selects that event, and a union selects several event
types. Annotating with the base `Event` creates a catch-all handler.

Import event classes from their concrete modules, for example
`oibot.event.friend_add`. The `oibot.event` package itself defines the base
`Event` and `EventType`.

This design provides three benefits at once:

- dispatch metadata lives next to the parameter that consumes the event;
- editors and type checkers know the exact event interface inside the handler;
- the framework can inject the matching event without string-based event names.

A handler must have at least one event-typed parameter. Without one, it has no
event subscription and will not run.

### Matchers can return context

A matcher may return a boolean-like value or a dictionary. A dictionary both
marks the rule as matched and supplies named values to the handler:

```python
from oibot.event.group_at_message_create import GroupAtMessageCreateEvent
from oibot.event.group_message_create import GroupMessageCreateEvent
from oibot.plugin import on


MessageEvent = GroupAtMessageCreateEvent | GroupMessageCreateEvent


def echo_rule(event: MessageEvent) -> dict[str, str] | None:
    content = event.content.strip()

    if content.startswith('echo '):
        return {'text': content.removeprefix('echo ').strip()}


@on(echo_rule)
async def echo(event: MessageEvent, *, text: str) -> None:
    await event.reply(text)
```

Only dictionary keys matching named handler parameters are injected. A handler
with `**kwargs` receives the complete matcher context.

For reusable rules, `Matcher` supports `&`, `|`, and `~`, with
`Matcher.all(...)` and `Matcher.any(...)` as named equivalents. Synchronous
rules are evaluated before asynchronous work; compatible asynchronous rules
are scheduled concurrently. Successful `&` branches merge their dictionaries,
which makes parsing, authorization, and feature-state rules independently
composable.

### Lifecycle hooks share the aiohttp application

A plugin may define `init(app)`. Use an asynchronous generator when a resource
needs deterministic setup and cleanup:

```python
from collections.abc import AsyncIterator

from aiohttp import ClientSession, web


async def init(app: web.Application) -> AsyncIterator[None]:
    async with ClientSession(base_url='https://example.com') as client:
        app['upstream_client'] = client
        yield
```

Synchronous and asynchronous generator functions and ordinary lifecycle
callables are supported. Generator cleanup runs when the transport shuts down.
The same `aiohttp.web.Application` is available as `bot.app`, allowing the
entry point and plugins to share routes, schedulers, connection pools, and
other process-wide resources without framework-specific containers.

## Dependency injection

Dependencies are parameter defaults declared with
`Dependency.from_provider(...)`. The provider may be a synchronous function,
coroutine function, generator, async generator, context-manager class, or
async-context-manager class.

The smallest resource dependency needs no wrapper function:

```python
from aiohttp import ClientSession

from oibot.event.group_at_message_create import GroupAtMessageCreateEvent
from oibot.event.group_message_create import GroupMessageCreateEvent
from oibot.plugin import Dependency, on


@on(lambda event: event.content.strip() == 'status')
async def status(
    event: GroupAtMessageCreateEvent | GroupMessageCreateEvent,
    *,
    session: ClientSession = Dependency.from_provider(ClientSession),
) -> None:
    async with session.get('https://example.com/health') as response:
        await event.reply(await response.text())
```

The framework enters `ClientSession` as an async context manager and closes it
after the handler finishes. The same lifecycle applies to custom generators
and context managers.

Providers may depend on other providers and may request the current event with
a type annotation. OiBot builds the dependency graph once when the decorator is
evaluated, then resolves it for each matching event:

- independent eager dependencies are resolved concurrently;
- a provider used several times in one graph is resolved once and cached;
- nested dependencies share the same cache and `AsyncExitStack`;
- generator and context-manager cleanup runs in reverse acquisition order; and
- `eager=False` injects an awaitable resolver, so an expensive browser, HTTP
  client, or other resource is created only when the handler actually needs it.

This keeps resource ownership local to a handler while retaining explicit
types and deterministic cleanup. Application-lifetime resources can instead
live in `bot.app`, so request-lifetime and process-lifetime dependencies remain
separate by design.

## Messages and conversations

Common message events expose `reply(...)`. A string is converted to a plain
text message, while the builders in `oibot.api.message` cover Markdown,
keyboards, rich media, references, and input notifications.

```python
from oibot.api.message import Button, Buttons, Keyboard, Markdown, Media, Message


await event.reply(
    Message.markdown(
        Markdown(
            Markdown.h2('OiBot'),
            'Choose an action:',
        ),
        keyboard=Keyboard(
            Buttons(
                Button.callback(
                    id='confirm',
                    label='Confirm',
                    visited_label='Confirmed',
                    data={'confirmed': True},
                ),
                Button.jump(
                    id='project',
                    label='Project',
                    data='https://github.com/OrganRemoved/oibot',
                ),
            )
        ),
    )
)


await event.reply(
    Message.media(
        Media.image(url='https://example.com/image.png'),
        content='An image',
    )
)
```

`Media.image(...)`, `Media.video(...)`, `Media.voice(...)`, and
`Media.file(...)` accept a URL or bytes through `data=...`. The high-level
`send_user_message(...)` and `send_group_message(...)` methods prepare and
upload byte content automatically, including multipart uploads when required.

### Turn event handling into a conversation

`defer(...)` sends a message and suspends only the current handler until the
same user replies. If the outgoing Markdown contains a keyboard, it waits for a
callback from that keyboard instead:

```python
from oibot.api.message import Button, Buttons, Keyboard, Markdown, Message


try:
    interaction = await event.defer(
        Message.markdown(
            Markdown('Continue?'),
            keyboard=Keyboard(
                Buttons(
                    Button.callback(
                        id='continue',
                        label='Continue',
                        data={'continue': True},
                    )
                )
            ),
        ),
        timeout=30,
    )
except TimeoutError:
    await event.reply('Timed out.')
else:
    if interaction.button_data['continue']:
        await interaction.reply('Continuing.')
```

OiBot namespaces callback IDs on a copy of the outgoing message, so concurrent
conversations do not collide. The session manager routes the matching event
directly to the suspended handler instead of broadcasting it to every plugin.
For a plain-text or media prompt, `defer(...)` waits for the same user's next
message.

The transport also records recent incoming message or event identifiers in a
bounded quote pool. When a reply does not specify `msg_id`, `event_id`, or
`msg_seq`, the sending layer acquires valid reply context automatically. This
keeps protocol bookkeeping out of plugin code while still allowing callers to
override every field explicitly.

## Calling APIs directly

During event handling, the complete API client is available as `event.bot`:

```python
from oibot.api.message import Message


await event.bot.send_group_message(
    group_openid=event.group_openid,
    message=Message.content('A proactive message'),
)

await event.bot.delete_group_message(
    group_openid=event.group_openid,
    message_id='MESSAGE_ID',
)
```

For a one-off API call outside a running transport, use `OiBot` as an async
context manager so its HTTP client is opened and closed correctly:

```python
import asyncio
from os import environ

from oibot.bot import OiBot


async def main() -> None:
    async with OiBot(
        app_id=environ['OIBOT_APP_ID'],
        app_secret=environ['OIBOT_APP_SECRET'],
    ) as bot:
        information = await bot.get_bot_information()
        print(information)


asyncio.run(main())
```

Access tokens are cached and refreshed before expiry. The API methods also use
local sliding-window limiters keyed by application and, where appropriate, by
user or group. This provides backpressure without forcing every plugin to
implement its own limiter.

## API coverage

`OiBot` composes the following mixins, so their methods are available directly
on every bot instance:

| Module | Main capabilities |
| --- | --- |
| [`oibot.api.access_token`](src/oibot/api/access_token.py) | Application access-token retrieval and caching |
| [`oibot.api.message`](src/oibot/api/message.py) | C2C and group messages, C2C streaming, Markdown, keyboards, and media builders |
| [`oibot.api.rich_media`](src/oibot/api/rich_media.py) | User and group rich-media creation and multipart upload |
| [`oibot.api.recall`](src/oibot/api/recall.py) | User, group, channel, and guild message recall |
| [`oibot.api.interaction`](src/oibot/api/interaction.py) | Interaction result acknowledgement |
| [`oibot.api.group_management`](src/oibot/api/group_management.py) | Group information, bot state, join requests, chat restrictions, and approval strategies |
| [`oibot.api.information`](src/oibot/api/information.py) | Bot information, guild listing, and URL-link generation |
| [`oibot.api.menu`](src/oibot/api/menu.py) | Global menu retrieval and update |
| [`oibot.api.panel`](src/oibot/api/panel.py) | Panel creation, retrieval, update, deletion, and target update |

Refer to the typed method signatures in each module for the exact request and
response fields.

## Structured events

The current package includes structured classes for these event families:

| Family | Event classes |
| --- | --- |
| Messages | `C2CMessageCreateEvent`, `GroupAtMessageCreateEvent`, `GroupMessageCreateEvent` |
| Message permissions | `C2CMsgReceiveEvent`, `C2CMsgRejectEvent`, `GroupMsgReceiveEvent`, `GroupMsgRejectEvent` |
| Friends | `FriendAddEvent`, `FriendDelEvent` |
| Group robot lifecycle | `GroupAddRobotEvent`, `GroupDelRobotEvent` |
| Group membership | `GroupMemberAddEvent`, `GroupMemberRemoveEvent`, `GroupJoinRequestEvent` |
| Interactions | `InteractionCreateEvent`, `C2CInteractionCreateEvent`, `GroupInteractionCreateEvent`, `GuildInteractionCreateEvent` |

Frequently used values such as authors, mentions, attachments, timestamps,
message scenes, and interaction payloads are exposed through typed properties.
Other payload fields fall back to the original event data. The full delivery
context remains available as `event.ctx`.

Unknown or not-yet-registered event types are represented by the base `Event`,
so a catch-all plugin can observe protocol additions without breaking event
dispatch.

## Troubleshooting

- **A plugin is not loaded:** start the process from the directory containing
  the plugin package, keep `plugins/__init__.py`, do not prefix the module name
  with `_`, and make sure the module contains `@on(...)` or `init(app)`.
- **A handler never runs:** annotate its event parameter, import the concrete
  event class, and enable the matching event or WebSocket intent in the QQ Bot
  configuration.
- **A callback button never arrives over WebSocket:** include
  `Intents.INTERACTION`.
- **An API method has no client session:** call it from a running transport or
  inside `async with OiBot(...)`.
- **A dependency is not resolved:** declare it as a parameter default with
  `Dependency.from_provider(...)` and keep the event annotation on the handler
  or provider.

Keep application credentials in environment variables or a secret manager; do
not commit them to source control. In production, expose Webhook endpoints over
HTTPS, normally through a reverse proxy.
