Method emit

lib/crewai/src/crewai/events/event_bus.py:572–647

Emit an event to all registered handlers. If handlers have dependencies (registered with depends_on), they execute in dependency order. Otherwise, handlers execute as before (sync in thread pool, async fire-and-forget). Stream chunk events always execute synchronous

(self, source: Any, event: BaseEvent)

        self._record_event(event)

    def emit(self, source: Any, event: BaseEvent) -> Future[None] | None:
        """Emit an event to all registered handlers.

        If handlers have dependencies (registered with depends_on), they execute
        in dependency order. Otherwise, handlers execute as before (sync in thread
        pool, async fire-and-forget).

        Stream chunk events always execute synchronously to preserve ordering.

        Args:
            source: The emitting object
            event: The event instance to emit

        Returns:
            Future that completes when handlers finish. Returns:
            - Future for sync-only handlers (ThreadPoolExecutor future)
            - Future for async handlers or mixed handlers (asyncio future)
            - Future for dependency-managed handlers (asyncio future)
            - None if no handlers or sync stream chunk events

        Example:
            >>> future = crewai_event_bus.emit(source, event)
            >>> if future:
            ...     await asyncio.wrap_future(future)  # In async test
            ...     # or future.result(timeout=5.0) in sync code
        """
        self._prepare_event(source, event)

        event_type = type(event)

        with self._rwlock.r_locked():
            if self._shutting_down:
                self._console.print(
                    "[CrewAIEventsBus] Warning: Attempted to emit event during shutdown. Ignoring."
                )
                return None
            has_dependencies = event_type in self._handler_dependencies
            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

        self._ensure_executor_initialized()
        self._has_pending_events = True

        state = self._runtime_state

        if has_dependencies:
            return self._track_future(
                asyncio.run_coroutine_threadsafe(
                    self._emit_with_dependencies(source, event, state),
                    self._loop,
                )
            )

        if sync_handlers:
            if event_type is LLMStreamChunkEvent: