Node execution context that bundles every service and state the executor needs. Attributes: tool_manager: Tool manager shared by executors function_manager: Function manager registry log_manager: Structured log manager memory_managers: Mapping of node_id to `
| 21 | |
| 22 | @dataclass |
| 23 | class ExecutionContext: |
| 24 | """Node execution context that bundles every service and state the executor needs. |
| 25 | |
| 26 | Attributes: |
| 27 | tool_manager: Tool manager shared by executors |
| 28 | function_manager: Function manager registry |
| 29 | log_manager: Structured log manager |
| 30 | memory_managers: Mapping of node_id to ``MemoryManager`` instances |
| 31 | thinking_managers: Mapping of node_id to ``ThinkingManagerBase`` instances |
| 32 | token_tracker: Token tracker used for accounting |
| 33 | global_state: Shared global state dictionary |
| 34 | """ |
| 35 | tool_manager: ToolManager |
| 36 | function_manager: FunctionManager |
| 37 | log_manager: LogManager |
| 38 | memory_managers: Dict[str, MemoryManager] = field(default_factory=dict) |
| 39 | thinking_managers: Dict[str, ThinkingManagerBase] = field(default_factory=dict) |
| 40 | token_tracker: Optional[TokenTracker] = None |
| 41 | global_state: Dict[str, Any] = field(default_factory=dict) |
| 42 | workspace_hook: Optional[Any] = None |
| 43 | human_prompt_service: Optional[HumanPromptService] = None |
| 44 | cancel_event: Optional[Any] = None |
| 45 | |
| 46 | def get_memory_manager(self, node_id: str) -> Optional[MemoryManager]: |
| 47 | """Return the memory manager for a given node.""" |
| 48 | return self.memory_managers.get(node_id) |
| 49 | |
| 50 | def get_thinking_manager(self, node_id: str) -> Optional[ThinkingManagerBase]: |
| 51 | """Return the thinking manager for a given node.""" |
| 52 | return self.thinking_managers.get(node_id) |
| 53 | |
| 54 | def get_token_tracker(self) -> Optional[TokenTracker]: |
| 55 | """Return the configured token tracker.""" |
| 56 | return self.token_tracker |
| 57 | |
| 58 | def get_human_prompt_service(self) -> Optional[HumanPromptService]: |
| 59 | """Return the interactive human prompt service.""" |
| 60 | return self.human_prompt_service |
| 61 | |
| 62 | |
| 63 | class NodeExecutor(ABC): |
no outgoing calls
no test coverage detected