快速开始
约 380 字大约 1 分钟
本页目标
运行一个不需要网络、Token 或配置文件的完整应用，观察一次事件分发并确认 进程没有遗留后台任务。
前置条件
已按安装准备仓库开发环境。
运行仓库示例
uv run examples/minimal_source_example.py预期输出：
ready该文件由测试直接执行，不只是语法片段。
完整代码
"""不依赖外部服务的最小完整示例."""
import asyncio
from dataclasses import dataclass
from butterbot.app import BotApp, Event, RuntimeConfig
from butterbot.core.data import BaseDataMixin
from butterbot.core.source import BaseSource
from butterbot.core.types import BaseType
class TickType(BaseType):
"""示例事件类型."""
ALL = "tick.all"
READY = "tick.ready"
@dataclass
class TickData(BaseDataMixin):
"""示例事件数据."""
message: str
class TickSource(BaseSource):
"""启动后发布一次事件，并负责回收自己的后台任务."""
supported_types = TickType
def __init__(self) -> None:
super().__init__()
self._task: asyncio.Task[None] | None = None
async def on_start(self) -> None:
self._task = asyncio.create_task(self._publish_once())
async def _publish_once(self) -> None:
event = Event(data=TickData(message="ready"), status=TickType.READY)
await self.ctx.bus.publish(self.uuid, event)
async def on_stop(self) -> None:
task = self._task
self._task = None
if task is None:
return
if not task.done():
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
async def main() -> None:
"""运行应用，收到事件后通过上下文管理器正常关闭."""
app = BotApp(RuntimeConfig())
source = app.add_source(TickSource)
received = asyncio.Event()
@app.subscribe(source.uuid, TickType.READY)
async def handle_tick(event: Event[TickData]) -> None:
print(event.data.message)
received.set()
async with app:
await asyncio.wait_for(received.wait(), timeout=1.0)
if __name__ == "__main__":
asyncio.run(main())发生了什么
RuntimeConfig()创建一个空的内存配置，不读取config.yaml。app.add_source(TickSource)只注册事件源，不立即启动。@app.subscribe(...)在启动前注册异步处理器。async with app调用app.start()，TickSource.on_start()创建自己拥有的任务。- Source 发布
Event[TickData]，EventBus创建并持有 Handler 任务。 - Handler 设置
received，主协程退出上下文。 app.close()停止 Source、排空 Handler，并关闭所有 API 实例。
验证正常退出
仓库测试会在 5 秒超时内以子进程运行该示例，并断言输出：
uv run pytest tests/test_examples.py -q如果示例超时，优先检查自定义 Source 是否取消并等待了自己创建的任务，以及 应用是否调用了 close()。