nonebot2 Documentation

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

📦 nonebot-adapter-{adapter-name}
├── 📂 nonebot
│   ├── 📂 adapters
│   │   ├── 📂 {adapter-name}
│   │   │   ├── 📜 __init__.py
│   │   │   ├── 📜 adapter.py
│   │   │   ├── 📜 bot.py
│   │   │   ├── 📜 config.py
│   │   │   ├── 📜 event.py
│   │   │   └── 📜 message.py
├── 📜 pyproject.toml
└── 📜 README.md

nb adapter create

from nonebot.utils import logger_wrapper

log = logger_wrapper("your_adapter_name")

from .log import log

log("DEBUG", "A DEBUG log.")
log("INFO", "A INFO log.")

try:
    ...
except Exception as e:
    log("ERROR", "something error.", e)

from pydantic import BaseModel

class Config(BaseModel):
    xxx_id: str
    xxx_token: str

from typing import Any
from typing_extensions import override

from nonebot.drivers import Driver
from nonebot import get_plugin_config
from nonebot.adapters import Adapter as BaseAdapter

from .config import Config

class Adapter(BaseAdapter):
    @override
    def __init__(self, driver: Driver, **kwargs: Any):
        super().__init__(driver, **kwargs)
        # 读取适配器所需的配置项
        self.adapter_config: Config = get_plugin_config(Config)

    @classmethod
    @override
    def get_name(cls) -> str:
        """适配器名称"""
        return "your_adapter_name"

/* Detailed source-code truncated for AI context efficiency. */

from nonebot import get_plugin_config
from nonebot.drivers import (
    Request,
    ASGIMixin,
    WebSocket,
    HTTPServerSetup,
    WebSocketServerSetup
)

class Adapter(BaseAdapter):
    @override
    def __init__(self, driver: Driver, **kwargs: Any):
        super().__init__(driver, **kwargs)
        self.adapter_config: Config = get_plugin_config(Config)
        self.setup()

    def setup(self) -> None:
        if not isinstance(self.driver, ASGIMixin):
            raise RuntimeError(
                f"Current driver {self.config.driver} doesn't support asgi server!"
                f"{self.get_name()} Adapter need a asgi server driver to work."
            )
        # 建立服务端路由
        # HTTP Webhook 路由
        http_setup = HTTPServerSetup(
            URL("your_webhook_url"),  # 路由地址
            "POST",  # 接收的方法
            "WEBHOOK name",  # 路由名称
            self._handle_http,  # 处理函数
        )
        self.setup_http_server(http_setup)

        # 反向 Websocket 路由
        ws_setup = WebSocketServerSetup(
            URL("your_websocket_url"),  # 路由地址
            "WebSocket name",  # 路由名称
            self._handle_ws,  # 处理函数
        )
        self.setup_websocket_server(ws_setup)


    async def _handle_http(self, request: Request) -> Response:
        """HTTP 路由处理函数，只有一个类型为 Request 的参数，且返回值类型为 Response"""
        ...
        return Response(
            status_code=200,  # 状态码
            headers={"something": "something"},  # 响应头
            content="xxx",  # 响应内容
        )

    async def _handle_ws(self, websocket: WebSocket) -> Any:
        """WebSocket 路由处理函数，只有一个类型为 WebSocket 的参数"""
        ...

from .bot import Bot

class Adapter(BaseAdapter):

    def _handle_connect(self):
        bot_id = ...  # 通过配置或者平台 API 等方式，获取到 Bot 的 ID
        bot = Bot(self, self_id=bot_id)  # 实例化 Bot
        self.bot_connect(bot)  # 建立 Bot 连接

    def _handle_disconnect(self):
        self.bot_disconnect(bot)  # 断开 Bot 连接

import asyncio
from typing import Any, Dict

from nonebot.compat import type_validate_python

from .bot import Bot
from .event import Event
from .log import log

class Adapter(BaseAdapter):

    @classmethod
    def payload_to_event(cls, payload: Dict[str, Any]) -> Event:
        """根据平台事件的特性，转换平台 payload 为具体 Event

        Event 模型继承自 pydantic.BaseModel，具体请参考 pydantic 文档
        """

        # 做一层异常处理，以应对平台事件数据的变更
        try:
            return type_validate_python(your_event_class, payload)
        except Exception as e:
            # 无法正常解析为具体 Event 时，给出日志提示
            log(
                "WARNING",
                f"Parse event error: {str(payload)}",
            )
            # 也可以尝试转为基础 Event 进行处理
            return type_validate_python(Event, payload)


    async def _forward(self, bot: Bot):

        payload: Dict[str, Any]  # 接收到的事件数据
        event = self.payload_to_event(payload)
        # 让 bot 对事件进行处理
        asyncio.create_task(bot.handle_event(event))

from typing import Any
from typing_extensions import override

from nonebot.drivers import Request, WebSocket

from .bot import Bot

class Adapter(BaseAdapter):

    @override
    async def _call_api(self, bot: Bot, api: str, **data: Any) -> Any:
        log("DEBUG", f"Calling API <y>{api}</y>")  # 给予日志提示
        platform_data = your_handle_data_method(data)  # 自行将数据转为平台所需要的格式

        # 采用 HTTP 请求的方式，需要构造一个 Request 对象
        request = Request(
            method="GET",  # 请求方法
            url=api,  # 接口地址
            headers=...,  # 请求头，通常需要包含鉴权信息
            params=platform_data,  # 自行处理数据的传输形式
            # json=platform_data,
            # data=platform_data,
        )
        # 发送请求，返回结果
        return await self.driver.request(request)


        # 采用 WebSocket 通信的方式，可以直接调用 send 方法发送数据
        # 通过某种方式获取到 bot 对应的 websocket 对象
        ws: WebSocket = your_get_websocket_method(bot.self_id)

        await ws.send_text(platform_data)  # 发送 str 类型的数据
        await ws.send_bytes(platform_data)  # 发送 bytes 类型的数据
        await ws.send(platform_data)  # 是以上两种方式的合体

        # 接收并返回结果，同样的，也有 str 和 bytes 的区别
        return await ws.receive_text()
        return await ws.receive_bytes()
        return await ws.receive()

from typing import TYPE_CHECKING, Any, Union
from typing_extensions import override

from nonebot.message import handle_event
from nonebot.adapters import Bot as BaseBot

from .event import Event
from .message import Message, MessageSegment

if TYPE_CHECKING:
    from .adapter import Adapter


class Bot(BaseBot):
    """
    your_adapter_name 协议 Bot 适配。
    """

    @override
    def __init__(self, adapter: Adapter, self_id: str, **kwargs: Any):
        super().__init__(adapter, self_id)
        self.adapter: Adapter = adapter
        # 一些有关 Bot 的信息也可以在此定义和存储

    async def handle_event(self, event: Event):
        # 根据需要，对事件进行某些预处理，例如：
        # 检查事件是否和机器人有关操作，去除事件消息首尾的 @bot
        # 检查事件是否有回复消息，调用平台 API 获取原始消息的消息内容
        ...
        # 调用 handle_event 让 NoneBot 对事件进行处理
        await handle_event(self, event)

    @override
    async def send(
        self,
        event: Event,
        message: Union[str, Message, MessageSegment],
        **kwargs: Any,
    ) -> Any:
        # 根据平台实现 Bot 回复事件的方法

        # 将消息处理为平台所需的格式后，调用发送消息接口进行发送，例如：
        data = message_to_platform_data(message)
        await self.send_message(
            data=data,
            ...
        )

from typing_extensions import override

from nonebot.compat import model_dump
from nonebot.adapters import Event as BaseEvent

class Event(BaseEvent):

    @override
    def get_event_name(self) -> str:
        # 返回事件的名称，用于日志打印
        return "event name"

    @override
    def get_event_description(self) -> str:
        # 返回事件的描述，用于日志打印，请注意转义 loguru tag
        return escape_tag(repr(model_dump(self)))

    @override
    def get_message(self):
        # 获取事件消息的方法，根据事件具体实现，如果事件非消息类型事件，则抛出异常
        raise ValueError("Event has no message!")

    @override
    def get_user_id(self) -> str:
        # 获取用户 ID 的方法，根据事件具体实现，如果事件没有用户 ID，则抛出异常
        raise ValueError("Event has no context!")

    @override
    def get_session_id(self) -> str:
        # 获取事件会话 ID 的方法，根据事件具体实现，如果事件没有相关 ID，则抛出异常
        raise ValueError("Event has no context!")

    @override
    def is_tome(self) -> bool:
        # 判断事件是否和机器人有关
        return False

from .message import Message

class HeartbeatEvent(Event):
    """心跳时间，通常为元事件"""

    @override
    def get_type(self) -> str:
        return "meta_event"

class MessageEvent(Event):
    """消息事件"""
    message_id: str
    user_id: str

    @override
    def get_type(self) -> str:
        return "message"

    @override
    def get_message(self) -> Message:
        # 返回事件消息对应的 NoneBot Message 对象
        return self.message

    @override
    def get_user_id(self) -> str:
        return self.user_id

class JoinRoomEvent(Event):
    """加入房间事件，通常为通知事件"""
    user_id: str
    room_id: str

    @override
    def get_type(self) -> str:
        return "notice"

class ApplyAddFriendEvent(Event):
    """申请添加好友事件，通常为请求事件"""
    user_id: str

    @override
    def get_type(self) -> str:
        return "request"

from typing import Type, Iterable
from typing_extensions import override

from nonebot.utils import escape_tag

from nonebot.adapters import Message as BaseMessage
from nonebot.adapters import MessageSegment as BaseMessageSegment

class MessageSegment(BaseMessageSegment["Message"]):
    @classmethod
    @override
    def get_message_class(cls) -> Type["Message"]:
        # 返回适配器的 Message 类型本身
        return Message

    @override
    def __str__(self) -> str:
        # 返回该消息段的纯文本表现形式，通常在日志中展示
        return "text of MessageSegment"

    @override
    def is_text(self) -> bool:
        # 判断该消息段是否为纯文本
        return self.type == "text"


class Message(BaseMessage[MessageSegment]):
    @classmethod
    @override
    def get_segment_class(cls) -> Type[MessageSegment]:
        # 返回适配器的 MessageSegment 类型本身
        return MessageSegment

    @staticmethod
    @override
    def _construct(msg: str) -> Iterable[MessageSegment]:
        # 实现从字符串中构造消息数组，如无字符串嵌入格式可直接返回文本类型 MessageSegment
        ...

📦 nonebot-plugin-{your-plugin-name}
├── 📂 nonebot_plugin_{your_plugin_name}
│   ├── 📜 __init__.py
│   └── 📜 config.py
├── 📜 pyproject.toml
└── 📜 README.md

# 安装 PDM（若未安装）
curl -sSL https://pdm-project.org/install-pdm.py | python3 -

# 安装项目依赖（自动创建虚拟环境）
pdm sync

# 添加新依赖（如 httpx）
pdm add httpx

from nonebot import require

require("nonebot_plugin_apscheduler")

from nonebot_plugin_apscheduler import scheduler

from nonebot.plugin import PluginMetadata

from .config import Config

__plugin_meta__ = PluginMetadata(
    # 基本信息（必填）
    name="天气查询",  # 插件名称
    description="查询指定城市的实时天气与未来预报",  # 插件介绍
    usage="发送【天气 城市名】获取天气信息",  # 插件用法

    # 发布额外信息
    type="application",  # 插件分类
    # 发布必填，当前有效类型有：`library`（为其他插件编写提供功能），`application`（向机器人用户提供功能）。

    homepage="https://github.com/你的用户名/nonebot-plugin-weather",
    # 发布必填。

    config=Config,
    # 插件配置项类，如果有配置类则必须填写。

    supported_adapters={"~onebot.v11"},
    # 支持的适配器集合，其中 `~` 在此处代表前缀 `nonebot.adapters.`，其余适配器亦按此格式填写。
    # 若插件只使用了 NoneBot 基本抽象，应显式填写 None，否则应该列出插件支持的适配器。
)

from nonebot import require
from nonebot.plugin import PluginMetadata, inherit_supported_adapters

from .config import Config

require("nonebot_plugin_alconna")  # 必须先 require 才能被 inherit_supported_adapters 处理

__plugin_meta__ = PluginMetadata(
    name="天气查询",
    description="查询指定城市的实时天气与未来预报",
    usage="发送【天气 城市名】获取天气信息",
    type="application",
    homepage="https://github.com/你的用户名/nonebot-plugin-weather",
    config=Config,

    supported_adapters=inherit_supported_adapters("nonebot_plugin_alconna"),
    # 继承 nonebot_plugin_alconna 插件的适配器支持列表
)

from pathlib import Path
from nonebot import require
require("nonebot_plugin_localstore")

import nonebot_plugin_localstore as store

# 获取插件缓存文件（夹）路径
weather_cache_dir: Path = store.get_plugin_cache_dir()
weather_cache_file: Path = store.get_plugin_cache_file("cache.json")

# 获取插件配置文件（夹）路径
weather_config_dir: Path = store.get_plugin_config_dir()
weather_config_file: Path = store.get_plugin_config_file("config.toml")

# 获取插件数据文件（夹）路径
weather_data_dir: Path = store.get_plugin_data_dir()
weather_data_file: Path = store.get_plugin_data_file("resource-index.json")

from nonebot import Bot, on_command
from nonebot_plugin_alconna.uniseg import Image, UniMessage


test = on_command("test")

@test.handle()
async def handle_test():
    await test.send(await UniMessage(Image(path="path/to/img")).export())

from nonebot import Bot, on_command
from nonebot_plugin_alconna.uniseg import UniMessage


test = on_command("test")

@test.handle()
async def handle():
    receipt = await UniMessage.text("hello!").send(at_sender=True, reply_to=True)
    await receipt.recall(delay=1)

async def send(
    self,
    target: Event | Target | None = None,
    bot: Bot | None = None,
    fallback: bool | FallbackStrategy = FallbackStrategy.rollback,
    at_sender: str | bool = False,
    reply_to: str | bool | Reply | None = False,
    **kwargs: Any,
) -> Receipt:
    ...

from arclet.alconna import Alconna, Args
from nonebot_plugin_alconna import Match, AlconnaMatcher, on_alconna
from nonebot_plugin_alconna.uniseg import At,  UniMessage


test_cmd = on_alconna(Alconna("test", Args["target?", At]))

@test_cmd.handle()
async def tt_h(matcher: AlconnaMatcher, target: Match[At]):
    if target.available:
        matcher.set_path_arg("target", target.result)

@test_cmd.got_path("target", prompt="请输入目标")
async def tt(target: At):
    await test_cmd.send(UniMessage([target, "\ndone."]))

from nonebot_plugin_alconna.uniseg import UniMessage, Target, SupportScope
from nonebot import get_driver


driver = get_driver()

@driver.on_startup
async def on_startup():
    target = Target("xxxx", scope=SupportScope.qq_client)
    await UniMessage("Hello!").send(target=target)

async def handle():
    receipt = await UniMessage.text("hello!").send(at_sender=True, reply_to=True)
    await receipt.recall(delay=1)
    recept1 = await UniMessage.text("hello!").send(at_sender=True, reply_to=True)
    await recept1.edit("world!")

from nonebot_plugin_alconna.uniseg import UniMessage, At


msg = UniMessage("Hello")
msg1 = UniMessage(At("user", "124"))
msg2 = UniMessage(["Hello", At("user", "124")])

from nonebot_plugin_alconna.uniseg import UniMessage, At, Image


msg = UniMessage.text("Hello").at("124").image(path="/path/to/img")
assert msg == UniMessage(
    ["Hello", At("user", "124"), Image(path="/path/to/img")]
)

>>> from nonebot_plugin_alconna.uniseg import UniMessage
>>>  UniMessage.template("{:At(user, target)}").format(target="123")
UniMessage(At("user", "123"))
>>> UniMessage.template("{:At(type=user, target=id)}").format(id="123")
UniMessage(At("user", "123"))
>>> UniMessage.template("{:At(type=user, target=123)}").format()
UniMessage(At("user", "123"))

from arclet.alconna import Alconna, Args
from nonebot_plugin_alconna import At, Match, UniMessage, AlconnaMatcher, on_alconna


test_cmd = on_alconna(Alconna("test", Args["target?", At]))

@test_cmd.handle()
async def tt_h(matcher: AlconnaMatcher, target: Match[At]):
    if target.available:
        matcher.set_path_arg("target", target.result)

@test_cmd.got_path(
    "target",
    prompt=UniMessage.template("{:At(user, $event.get_user_id())} 请确认目标")
)
async def tt():
    await test_cmd.send(
      UniMessage.template("{:At(user, $event.get_user_id())} 已确认目标为 {target}")
    )

# 消息序列与消息段相加
UniMessage("text") + Text("text")
# 消息序列与字符串相加
UniMessage([Text("text")]) + "text"
# 消息序列与消息序列相加
UniMessage("text") + UniMessage([Text("text")])
# 字符串与消息序列相加
"text" + UniMessage([Text("text")])
# 消息段与消息段相加
Text("text") + Text("text")
# 消息段与字符串相加
Text("text") + "text"
# 消息段与消息序列相加
Text("text") + UniMessage([Text("text")])
# 字符串与消息段相加
"text" + Text("text")

msg = UniMessage([Text("text")])
# 自加
msg += "text"
msg += Text("text")
msg += UniMessage([Text("text")])
# 附加
msg.append(Text("text"))
# 扩展
msg.extend([Text("text")])

# 是否存在消息段
At("user", "1234") in message
# 是否存在指定类型的消息段
At in message

# 是否都为 "test"
message.only("test")
# 是否仅包含指定类型的消息段
message.only(Text)

# 提取消息纯文本字符串
assert UniMessage(
    [At("user", "1234"), "text"]
).extract_plain_text() == "text"

for segment in message:  # type: Segment
    ...

message = UniMessage(
    [
        Reply(...),
        "text1",
        At("user", "1234"),
        "text2"
    ]
)
# 索引
message[0] == Reply(...)
# 切片
message[0:2] == UniMessage([Reply(...), Text("text1")])
# 类型过滤
message[At] == Message([At("user", "1234")])
# 类型索引
message[At, 0] == At("user", "1234")
# 类型切片
message[Text, 0:2] == UniMessage([Text("text1"), Text("text2")])

message.include(Text, At)
message.exclude(Reply)

message.filter(lambda x: isinstance(x, At) and x.flag == "user")  # 仅保留 At("user", xxx) 的消息段

# 指定类型首个消息段索引
message.index(Text) == 1
# 指定类型消息段数量
message.count(Text) == 2

# 获取指定类型指定个数的消息段
message.get(Text, 1) == UniMessage([Text("test1")])

message = UniMessage(
    [
        Text("text1"),
        Image(url="url1")(
            Text("text2"),
        )
    ]
)
assert message.select(Text) == UniMessage(
    [
        Text("text1"),
        Text("text2")
    ]
)

# 转换消息段为另一类型的消息段，此时返回结果仍是 UniMessage
message.map(lambda x: Text(x.target))  # 转换为 Text 消息段
# 转换消息段为另一类型的数据，此时返回结果为 list[T]
message.map(lambda x: x.target)  # 转换为 list[str]

rule = {
    "text": True,
    "at": lambda attrs, children: Text(attrs["target"])
}
message.transform(rule)

type Fragment = Segment | Segment[];
type Render<T> = (attrs: dict, children: Segment[]) => T;
type Transformer = boolean | Fragment | Render;

msg = UniMessage.text("foo bar").at("1234").text("baz qux")
# 分割，返回分割结果，类型为 list[UniMessage]
parts = msg.split(" ")
# 替换，返回替换结果，类型为 UniMessage。新文本可以用 str 或 Text 来替换
new_msg = msg.replace("ba", "baaa")
# 前缀/后缀检查
msg.startswith("foo")  # True
msg.endswith("qux")  # True
# 去除前缀/后缀
msg1 = msg.removeprefix("foo")  # UniMessage([Text(" bar"), At("user", "1234"), Text("baz qux")])
msg2 = msg.removesuffix("qux")  # UniMessage([Text("foo bar"), At("user", "1234"), Text("baz ")])
# 去除空格
msg1 = msg1.lstrip()  # UniMessage([Text("bar"), At("user", "1234"), Text("baz qux")])
msg2 = msg2.rstrip()  # UniMessage([Text("foo bar"), At("user", "1234"), Text("baz")])

msg = UniMessage.text("Hello").image(url="url")
data = msg.dump()  # [{"type": "text", "text": "Hello"}, {"type": "image", "url": "url"}]

assert UniMessage.load(data) == msg

def dump(self, media_save_dir: str | Path | bool | None = None, json: bool = False) -> str | list[dict[str, Any]]: ...

@classmethod
def load(cls, data: str | list[dict[str, Any]]) -> UniMessage: ...

from nonebot_plugin_alconna.uniseg import UniMessage

message = (
    UniMessage.text("hello world")
    .at("1234567890")
    .image(url="https://example.com/image.png")
)

from nonebot_plugin_alconna import Text, At, Image, UniMessage

message = UniMessage(
    [
        Text("hello world"),
        At("user", "1234567890"),
        Image(url="https://example.com/image.png"),
    ]
)

from nonebot_plugin_alconna import Button, UniMessage

message = (
    UniMessage.text("hello world")
    .keyboard(
        Button("link1", url="https://example.com/1"),
        Button("link2", url="https://example.com/2"),
        Button("link3", url="https://example.com/3"),
        row=3,
    )
)

@matcher.handle()
async def _():
    message = UniMessage.text("hello world").image(url="https://example.com/image.png")
    await message.send()
    # 类似于 `matcher.finish`
    await message.finish()

@matcher.handle()
async def _():
    message = UniMessage.text("hello world").image(url="https://example.com/image.png")
    await message.send(at_sender=True)

@matcher.handle()
async def _():
    message = UniMessage.text("hello world").image(url="https://example.com/image.png")
    await message.send(reply_to=True)

from nonebot_plugin_alconna import message_recall, message_edit, message_reaction

@matcher.handle()
async def _():
    await message_edit(UniMessage.text("hello world"))
    await message_reaction("👍")
    await message_recall()

@matcher.handle()
async def _():
    message = UniMessage.text("hello world").image(url="https://example.com/image.png")
    receipt = await message.send()
    await receipt.edit(UniMessage.text("hello world!"))
    await receipt.reaction("👍")
    await receipt.recall(delay=5)  # 5秒后撤回

from nonebot_plugin_alconna import UniMsg

@matcher.handle()
async def _(msg: UniMsg):
    ...

ans1 = Image in message
ans2 = message.has(Image)
ans3 = message.only(Image)

imgs_1 = message[Image]
imgs_2 = message.get(Image)
imgs_3 = message.include(Image)
imgs_4 = message.select(Image)
imgs_5 = message.filter(lambda x: x.type == "image")
imgs_6 = message.tranform({"image": True})

urls = imgs.map(lambda x: x.url)

@matcher.handle()
async def _(msg: UniMsg):
    if msg.startswith("hello"):
        await matcher.finish("hello world")
    else:
        await matcher.finish("not hello world")

@matcher.handle()
async def _(msg: UniMsg):
    if msg.startswith("hello"):
        msg = msg.removeprefix("hello")
        await matcher.finish(msg)
    else:
        await matcher.finish("not hello world")

from nonebot_plugin_alconna import UniMsg

@matcher.handle()
async def _(msg: UniMsg):
    data: list[dict] = msg.dump()
    # 你可以将 data 存储到数据库或者 JSON 文件中

from nonebot_plugin_alconna import UniMessage

@matcher.handle()
async def _():
    data = [
        {"type": "text", "text": "hello world"},
        {"type": "image", "url": "https://example.com/image.png"},
    ]
    message = UniMessage.load(data)

from nonebot_plugin_alconna import Alconna, Args, Image, on_alconna

meme = on_alconna(Alconna("make_meme", Args["name", str]["img", Image]))

@meme.handle()
async def _(img: Image):
    ...

/* Detailed source-code truncated for AI context efficiency. */

from nonebot_plugin_alconna import Args, Image, Alconna, select
from nonebot_plugin_alconna.builtins.uniseg.market_face import MarketFace

# 表示这个指令需要的图片会在目标元素下进行搜索，将所有符合 Image 的元素选出来并将第一个作为结果
alc1 = Alconna("make_meme", Args["name", str]["img", select(Image).first])  # 也可以使用 select(Image).nth(0)

# 表示这个指令需要的图片要么直接是 Image 要么是在 MarketFace 元素内的 Image
alc2 = Alconna("make_meme", Args["name", str]["img", [Image, select(Image).from_(MarketFace)]])

from dataclasses import dataclass

from nonebot.adapters import Bot
from nonebot.adapters import MessageSegment as BaseMessageSegment
from nonebot.adapters.satori import Custom, Message, MessageSegment

from nonebot_plugin_alconna.uniseg.builder import MessageBuilder
from nonebot_plugin_alconna.uniseg.exporter import MessageExporter
from nonebot_plugin_alconna.uniseg import Segment, custom_handler, custom_register


@dataclass
class MarketFace(Segment):
    tabId: str
    faceId: str
    key: str


@custom_register(MarketFace, "chronocat:marketface")
def mfbuild(builder: MessageBuilder, seg: BaseMessageSegment):
    if not isinstance(seg, Custom):
        raise ValueError("MarketFace can only be built from Satori Message")
    return MarketFace(**seg.data)(*builder.generate(seg.children))


@custom_handler(MarketFace)
async def mfexport(exporter: MessageExporter, seg: MarketFace, bot: Bot, fallback: bool):
    if exporter.get_message_type() is Message:
        return MessageSegment("chronocat:marketface", seg.data)(await exporter.export(seg.children, bot, fallback))

class Target:
    id: str
    """目标id；若为群聊则为 group_id 或者 channel_id，若为私聊则为 user_id"""
    parent_id: str
    """父级id；若为频道则为 guild_id，其他情况下可能为空字符串（例如 Feishu 下可作为部门 id）"""
    channel: bool
    """是否为频道，仅当目标平台符合频道概念时"""
    private: bool
    """是否为私聊"""
    source: str
    """可能的事件id"""
    self_id: str | None
    """机器人id，若为 None 则 Bot 对象会随机选择"""
    selector: Callable[[Bot], Awaitable[bool]] | None
    """选择器，用于在多个 Bot 对象中选择特定 Bot"""
    extra: dict[str, Any]
    """额外信息，用于适配器扩展"""

from nonebot_plugin_alconna.uniseg import UniMessage, MsgTarget, Target, SupportScope


matcher = on_xxx(...)

@matcher.handle()
async def _(target: MsgTarget):
    # 将消息发送给当前事件的发送者
    await UniMessage("Hello!").send(target=target)
    # 主动发送消息给群号为 12345 的 QQ 群聊
    target1 = Target("12345", scope=SupportScope.qq_client)
    await UniMessage("Hello!").send(target=target1)

async def _():
    target = Target("12345", scope=SupportScope.qq_client)
    bot = await target.select()

from nonebot_plugin_alconna.uniseg import message_recall

matcher = on_xxx(...)

@matcher.handle()
async def _(msg_id: MsgId):
    await message_recall(msg_id)

async def message_recall(
    message_id: str | None = None,
    event: Event | None = None,
    bot: Bot | None = None,
    adapter: str | None = None
): ...

from nonebot_plugin_alconna.uniseg import UniMessage, message_edit

matcher = on_xxx(...)

@matcher.handle()
async def _():
    await message_edit(UniMessage.text("1234"))

async def message_edit(
    msg: UniMessage,
    message_id: str | None = None,
    event: Event | None = None,
    bot: Bot | None = None,
    adapter: str | None = None,
): ...

from nonebot_plugin_alconna.uniseg import message_reaction

matcher = on_xxx(...)

@matcher.handle()
async def _():
    await message_reaction("👍")

async def message_reaction(
    reaction: str | Emoji,
    message_id: str | None = None,
    event: Event | None = None,
    bot: Bot | None = None,
    adapter: str | None = None,
    delete: bool = False,
): ...

from nonebot_plugin_alconna.uniseg import at_me

matcher = on_xxx(..., rule=at_me())