aiocqhttp SDK

aiocqhttp is a Python SDK for OneBot (formerly the CQHTTP plugin for 酷Q). It is designed to help Python developers build plugins using asynchronous I/O. It encapsulates web server logic and supports two primary communication methods:

1. HTTP: Standard HTTP communication.
2. Reverse WebSocket: Allows the SDK to act as a backend for multiple bot accounts simultaneously.

Requirements:

• Python 3.7 or higher.
• It is recommended to use a OneBot implementation that supports OneBot v11.

A 'Quick Operation' is a way to perform simple actions by returning a dictionary from your event handler. The SDK serializes this dictionary to JSON and returns it to CQHTTP as the response to the event report. This is useful for simple replies or group actions like banning/recalling messages.

Example of replying to a message via Quick Operation:

@bot.on_message('private')
async def _(event: Event):
    # This returns a 'reply' action to CQHTTP
    return {'reply': event.message}

The CQHttp class is the main entry point of the SDK. It encapsulates a Quart web server to handle incoming connections via HTTP or WebSocket.

• HTTP Communication: You must provide the api_root parameter so the bot knows where to send API requests (e.g., bot.send).
• WebSocket Communication: Supports multiple CQHTTP connections simultaneously. When using bot.send, the SDK automatically selects the correct account to send the message.

from aiocqhttp import CQHttp

bot = CQHttp(api_root='http://127.0.0.1:8080')

While asynchronous operations (using aiofiles, aiohttp, httpx.AsyncClient, etc.) are recommended, the SDK supports synchronous handler functions.

Synchronous handlers decorated with @bot.on_message will run in the asyncio loop's default executor (multi-threaded). To perform API calls within a synchronous handler, you must use the bot.sync attribute to access synchronous versions of the API methods.

@bot.on_message
def sync_handle_msg(event):
    import time
    time.sleep(5)  # Simulating blocking I/O
    # Use bot.sync for API calls in sync handlers
    bot.sync.send_private_msg(user_id=event.user_id, message='处理完毕')

Use bot.run() to start the backend. This is a wrapper around Quart.run(), so you can pass any standard Quart/hypercorn arguments (like host and port) directly to it.

bot.run(host='127.0.0.1', port=8080)

While bot.run() is available for development (running in debug mode by default), it is not recommended for production. Instead, use a professional ASGI server like Uvicorn, Daphne, or Hypercorn.

To deploy with Uvicorn, expose the bot.asgi attribute from your application module.

# main.py
from aiocqhttp import CQHttp
bot = CQHttp()

# Command to run with Uvicorn
# uvicorn --host 127.0.0.1 --port 8080 main:bot.asgi

To use aiocqhttp, ensure your environment meets the following requirements:

• Python Version: Python 3.7 or higher.
• OneBot Version: It is recommended to use a OneBot implementation that supports OneBot v11.

To connect CQHTTP to your bot via HTTP, follow these steps:

1. In your Python code, initialize CQHttp with the api_root pointing to the CQHTTP listener: bot = CQHttp(api_root='http://<cqhttp_host>:<cqhttp_port>').
2. In the CQHTTP configuration file, set post_url to http://<your_bot_host>:<your_bot_port>/.
3. Set use_http to true (this is the default).
4. Restart CQHTTP.

# bot.py configuration for HTTP mode
bot = CQHttp(api_root='http://127.0.0.1:5700')

Install the aiocqhttp package using pip. By default, it installs Quart and aiocqhttp. To install optional dependencies including ujson, use the [all] extra.

# Standard installation
pip install aiocqhttp

# Installation with optional dependencies (ujson)
pip install aiocqhttp[all]

The SDK registers several default routes (/, /ws, /ws/api, /ws/event) for CQHTTP service. You can add custom routes (e.g., for webhooks or admin panels) by interacting with the bot.server_app (which is a Quart instance).

If you encounter issues with relative paths (e.g., when loading templates via render_template), pass the import_name argument to the CQHttp constructor.

# Adding a custom route
@bot.server_app.route('/webhook')
async def webhook():
    pass

# Configuring import_name for template resolution
bot = CQHttp(__name__)

@bot.server_app.route('/admin')
async def admin():
    return await render_template('admin.html')

For simple features or testing, you can use the built-in default bot object provided by aiocqhttp.default instead of creating a CQHttp instance. This allows you to use decorators like @on_message and utility functions like send, api, and run directly.

If you need to modify the configuration of the default bot (e.g., changing the api_root), use the reconfigure_default_bot function.

from aiocqhttp.default import on_message, send, api, run
from aiocqhttp.default import reconfigure_default_bot

@on_message
async def handle_msg(event):
    await send(event, event.message)
    await api.send_private_msg(user_id=event.user_id, message='。。。')

run(host='127.0.0.1', port=8080)

# Reconfiguring the default bot
reconfigure_default_bot(api_root='http://127.0.0.1:8080')