事件基类（events/base_events.py:66）：

# events/base_events.py:66
class BaseEvent(BaseModel):
    """Base class for all events"""
    timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    type: str                                     # 事件类型标识（字符串）
    source_fingerprint: str | None = None         # 谁发的（实体指纹）
    source_type: str | None = None                # "agent"/"task"/"crew"/"memory"...
    fingerprint_metadata: dict[str, Any] | None = None
    task_id: str | None = None
    task_name: str | None = None
    agent_id: str | None = None
    agent_role: str | None = None

单例总线（events/event_bus.py:93）与模块级实例（events/event_bus.py:952）：

# events/event_bus.py:93
class CrewAIEventsBus:
    """Singleton event bus for handling events in CrewAI.
    Synchronous handlers execute in a thread pool executor ...
    Asynchronous handlers execute in a dedicated event loop running in a daemon thread ..."""
    _instance: Self | None = None
    _instance_lock: threading.RLock = threading.RLock()
    _sync_handlers: dict[type[BaseEvent], SyncHandlerSet]       # 事件类型 → 同步 handler 集
    _async_handlers: dict[type[BaseEvent], AsyncHandlerSet]     # 事件类型 → 异步 handler 集

    def __new__(cls) -> Self:                     # ★单例：永远返回同一个实例
        if cls._instance is None:
            with cls._instance_lock:
                if cls._instance is None:
                    ...
        return cls._instance

# events/event_bus.py:952  模块级唯一实例——大家 import 的就是它
crewai_event_bus: Final[CrewAIEventsBus] = CrewAIEventsBus()

注册用的装饰器（events/event_bus.py:245）：

# events/event_bus.py:245
def on(self, event_type: type[BaseEvent],
       depends_on: Depends | list[Depends] | None = None):
    """Decorator to register an event handler for a specific event type.
    Handlers can accept 2 or 3 arguments:
        - (source, event) — standard handler
        - (source, event, state: RuntimeState) — handler with runtime state"""
    def decorator(handler):
        deps = None
        if depends_on is not None:
            deps = [depends_on] if isinstance(depends_on, Depends) else depends_on
        self._register_handler(event_type, handler, dependencies=deps)   # ★存进字典
        return handler                     # 原样返回，不包装
    return decorator

用起来是这样：

from crewai.events import crewai_event_bus
from crewai.events.types.crew_events import CrewKickoffStartedEvent

@crewai_event_bus.on(CrewKickoffStartedEvent)      # 订阅"kickoff 开始"事件
def my_logger(source, event):                       # (发送者, 事件对象)
    print(f"[{event.timestamp}] crew {event.crew_name} 开跑了！")

分发的核心（events/event_bus.py:572，裁剪）：

# events/event_bus.py:572
def emit(self, source: Any, event: BaseEvent) -> Future[None] | None:
    self._prepare_event(source, event)          # 填公共字段（时间戳/序号等）
    event_type = type(event)
    with self._rwlock.r_locked():                # 读锁：查 handler 时不被注册打断
        sync_handlers = self._sync_handlers.get(event_type, frozenset())
        async_handlers = self._async_handlers.get(event_type, frozenset())
    if not sync_handlers and not async_handlers:
        return None                              # ★没人订阅 → 直接返回，零开销

    if sync_handlers:
        if event_type is LLMStreamChunkEvent:    # 流式 chunk：必须同步、保证顺序
            self._call_handlers(source, event, sync_handlers, state)
        else:
            ctx = contextvars.copy_context()     # ★复制上下文给线程（Day 08 见过的手法）
            sync_future = self._sync_executor.submit(
                ctx.run, self._call_handlers, source, event, sync_handlers, state)
            ...
    if async_handlers:                            # 异步 handler 丢到专用事件循环线程
        return self._track_future(asyncio.run_coroutine_threadsafe(
            self._acall_handlers(source, event, async_handlers, state), self._loop))
    return None

监听器基类（events/base_event_listener.py）：

# events/base_event_listener.py
class BaseEventListener(ABC):
    """Abstract base class for event listeners."""
    verbose: bool = False
    def __init__(self) -> None:
        super().__init__()
        self.setup_listeners(crewai_event_bus)      # ★构造时就自动注册所有监听
        crewai_event_bus.validate_dependencies()

    @abstractmethod
    def setup_listeners(self, crewai_event_bus: CrewAIEventsBus) -> None:
        """Setup event listeners on the event bus."""

框架自带的默认监听器就这么用（events/event_listener.py:161）：

# events/event_listener.py:161
class EventListener(BaseEventListener):
    def setup_listeners(self, crewai_event_bus):
        @crewai_event_bus.on(CrewKickoffStartedEvent)
        def on_crew_started(source, event):
            self.formatter.handle_crew_started(event.crew_name or "Crew", source.id)
            source._execution_span = self._telemetry.crew_execution_span(source, event.inputs)

        @crewai_event_bus.on(CrewKickoffCompletedEvent)
        def on_crew_completed(source, event):
            final = event.output.raw
            self._telemetry.end_crew(source, final)         # 遥测：结束 span
            self.formatter.handle_crew_status(event.crew_name or "Crew", source.id, "completed", final)
        # ... 还有 failed / train / task / agent / tool 等一大堆

事件按领域分类，crew 类事件如（events/types/crew_events.py:36）：

# events/types/crew_events.py:12
class CrewBaseEvent(BaseEvent):
    """Base class for crew events with fingerprint handling"""
    crew_name: str | None
    crew: Crew | None = None
    def _set_crew_fingerprint(self):
        if self.crew is not None and self.crew.fingerprint:
            self.source_fingerprint = self.crew.fingerprint.uuid_str
            self.source_type = "crew"

# events/types/crew_events.py:36
class CrewKickoffStartedEvent(CrewBaseEvent):
    inputs: dict[str, Any] | None
    type: Literal["crew_kickoff_started"] = "crew_kickoff_started"   # ★固定 type

class CrewKickoffCompletedEvent(CrewBaseEvent):
    output: Any
    type: Literal["crew_kickoff_completed"] = "crew_kickoff_completed"
    total_tokens: int = 0