chore: apply ruff linting fixes to events module

fix: apply ruff linting to events

Author: greysonlalonde

src/crewai/events/base_events.py

@@ -1,5 +1,6 @@
 from datetime import datetime, timezone
-from typing import Any, Dict, Optional
+from typing import Any
+
 from pydantic import BaseModel, Field
 
 from crewai.utilities.serialization import to_serializable
@@ -10,11 +11,11 @@ class BaseEvent(BaseModel):
 
     timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
     type: str
-    source_fingerprint: Optional[str] = None  # UUID string of the source entity
-    source_type: Optional[str] = (
+    source_fingerprint: str | None = None  # UUID string of the source entity
+    source_type: str | None = (
         None  # "agent", "task", "crew", "memory", "entity_memory", "short_term_memory", "long_term_memory", "external_memory"
     )
-    fingerprint_metadata: Optional[Dict[str, Any]] = None  # Any relevant metadata
+    fingerprint_metadata: dict[str, Any] | None = None  # Any relevant metadata
 
     def to_json(self, exclude: set[str] | None = None):
         """
@@ -28,13 +29,13 @@ def to_json(self, exclude: set[str] | None = None):
         """
         return to_serializable(self, exclude=exclude)
 
-    def _set_task_params(self, data: Dict[str, Any]):
+    def _set_task_params(self, data: dict[str, Any]):
         if "from_task" in data and (task := data["from_task"]):
             self.task_id = task.id
             self.task_name = task.name or task.description
             self.from_task = None
 
-    def _set_agent_params(self, data: Dict[str, Any]):
+    def _set_agent_params(self, data: dict[str, Any]):
         task = data.get("from_task", None)
         agent = task.agent if task else data.get("from_agent", None)
 

src/crewai/events/event_bus.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
 
 import threading
+from collections.abc import Callable
 from contextlib import contextmanager
-from typing import Any, Callable, Dict, List, Type, TypeVar, cast
+from typing import Any, TypeVar, cast
 
 from blinker import Signal
 
@@ -25,17 +26,17 @@ def __new__(cls):
         if cls._instance is None:
             with cls._lock:
                 if cls._instance is None:  # prevent race condition
-                    cls._instance = super(CrewAIEventsBus, cls).__new__(cls)
+                    cls._instance = super().__new__(cls)
                     cls._instance._initialize()
         return cls._instance
 
     def _initialize(self) -> None:
         """Initialize the event bus internal state"""
         self._signal = Signal("crewai_event_bus")
-        self._handlers: Dict[Type[BaseEvent], List[Callable]] = {}
+        self._handlers: dict[type[BaseEvent], list[Callable]] = {}
 
     def on(
-        self, event_type: Type[EventT]
+        self, event_type: type[EventT]
     ) -> Callable[[Callable[[Any, EventT], None]], Callable[[Any, EventT], None]]:
         """
         Decorator to register an event handler for a specific event type.
@@ -61,6 +62,18 @@ def decorator(
 
         return decorator
 
+    @staticmethod
+    def _call_handler(
+        handler: Callable, source: Any, event: BaseEvent, event_type: type
+    ) -> None:
+        """Call a single handler with error handling."""
+        try:
+            handler(source, event)
+        except Exception as e:
+            print(
+                f"[EventBus Error] Handler '{handler.__name__}' failed for event '{event_type.__name__}': {e}"
+            )
+
     def emit(self, source: Any, event: BaseEvent) -> None:
         """
         Emit an event to all registered handlers
@@ -72,17 +85,12 @@ def emit(self, source: Any, event: BaseEvent) -> None:
         for event_type, handlers in self._handlers.items():
             if isinstance(event, event_type):
                 for handler in handlers:
-                    try:
-                        handler(source, event)
-                    except Exception as e:
-                        print(
-                            f"[EventBus Error] Handler '{handler.__name__}' failed for event '{event_type.__name__}': {e}"
-                        )
+                    self._call_handler(handler, source, event, event_type)
 
         self._signal.send(source, event=event)
 
     def register_handler(
-        self, event_type: Type[EventTypes], handler: Callable[[Any, EventTypes], None]
+        self, event_type: type[EventTypes], handler: Callable[[Any, EventTypes], None]
     ) -> None:
         """Register an event handler for a specific event type"""
         if event_type not in self._handlers:

src/crewai/events/event_listener.py

@@ -1,15 +1,30 @@
 from __future__ import annotations
 
 from io import StringIO
-from typing import Any, Dict
+from typing import Any
 
 from pydantic import Field, PrivateAttr
-from crewai.llm import LLM
-from crewai.task import Task
-from crewai.telemetry.telemetry import Telemetry
-from crewai.utilities import Logger
-from crewai.utilities.constants import EMITTER_COLOR
+
 from crewai.events.base_event_listener import BaseEventListener
+from crewai.events.types.agent_events import (
+    AgentExecutionCompletedEvent,
+    AgentExecutionStartedEvent,
+    LiteAgentExecutionCompletedEvent,
+    LiteAgentExecutionErrorEvent,
+    LiteAgentExecutionStartedEvent,
+)
+from crewai.events.types.crew_events import (
+    CrewKickoffCompletedEvent,
+    CrewKickoffFailedEvent,
+    CrewKickoffStartedEvent,
+    CrewTestCompletedEvent,
+    CrewTestFailedEvent,
+    CrewTestResultEvent,
+    CrewTestStartedEvent,
+    CrewTrainCompletedEvent,
+    CrewTrainFailedEvent,
+    CrewTrainStartedEvent,
+)
 from crewai.events.types.knowledge_events import (
     KnowledgeQueryCompletedEvent,
     KnowledgeQueryFailedEvent,
@@ -25,34 +40,21 @@
     LLMStreamChunkEvent,
 )
 from crewai.events.types.llm_guardrail_events import (
-    LLMGuardrailStartedEvent,
     LLMGuardrailCompletedEvent,
-)
-from crewai.events.utils.console_formatter import ConsoleFormatter
-
-from crewai.events.types.agent_events import (
-    AgentExecutionCompletedEvent,
-    AgentExecutionStartedEvent,
-    LiteAgentExecutionCompletedEvent,
-    LiteAgentExecutionErrorEvent,
-    LiteAgentExecutionStartedEvent,
+    LLMGuardrailStartedEvent,
 )
 from crewai.events.types.logging_events import (
-    AgentLogsStartedEvent,
     AgentLogsExecutionEvent,
+    AgentLogsStartedEvent,
 )
-from crewai.events.types.crew_events import (
-    CrewKickoffCompletedEvent,
-    CrewKickoffFailedEvent,
-    CrewKickoffStartedEvent,
-    CrewTestCompletedEvent,
-    CrewTestFailedEvent,
-    CrewTestResultEvent,
-    CrewTestStartedEvent,
-    CrewTrainCompletedEvent,
-    CrewTrainFailedEvent,
-    CrewTrainStartedEvent,
-)
+from crewai.events.utils.console_formatter import ConsoleFormatter
+from crewai.llm import LLM
+from crewai.task import Task
+from crewai.telemetry.telemetry import Telemetry
+from crewai.utilities import Logger
+from crewai.utilities.constants import EMITTER_COLOR
+
+from .listeners.memory_listener import MemoryListener
 from .types.flow_events import (
     FlowCreatedEvent,
     FlowFinishedEvent,
@@ -61,26 +63,24 @@
     MethodExecutionFinishedEvent,
     MethodExecutionStartedEvent,
 )
+from .types.reasoning_events import (
+    AgentReasoningCompletedEvent,
+    AgentReasoningFailedEvent,
+    AgentReasoningStartedEvent,
+)
 from .types.task_events import TaskCompletedEvent, TaskFailedEvent, TaskStartedEvent
 from .types.tool_usage_events import (
     ToolUsageErrorEvent,
     ToolUsageFinishedEvent,
     ToolUsageStartedEvent,
 )
-from .types.reasoning_events import (
-    AgentReasoningStartedEvent,
-    AgentReasoningCompletedEvent,
-    AgentReasoningFailedEvent,
-)
-
-from .listeners.memory_listener import MemoryListener
 
 
 class EventListener(BaseEventListener):
     _instance = None
     _telemetry: Telemetry = PrivateAttr(default_factory=lambda: Telemetry())
     logger = Logger(verbose=True, default_color=EMITTER_COLOR)
-    execution_spans: Dict[Task, Any] = Field(default_factory=dict)
+    execution_spans: dict[Task, Any] = Field(default_factory=dict)
     next_chunk = 0
     text_stream = StringIO()
     knowledge_retrieval_in_progress = False