Human Approval#
LangGraph-native approval system for production-ready human-in-the-loop workflows with configurable security policies.
Note
For implementation guides and examples, see Human Approval.
Core Components#
|
Pure configuration service providing strongly typed approval models. |
|
Global approval configuration integrating all capability-specific settings. |
|
Configuration model for Python code execution approval settings. |
|
Configuration model for memory operation approval settings. |
|
Enumeration of approval modes for capability-specific approval control. |
|
Business logic evaluator for Python code execution approval decisions. |
|
Business logic evaluator for memory operation approval decisions. |
|
Structured result of an approval evaluation decision. |
System Functions#
|
Generate dynamic approval type identifier from capability and operation. |
|
Create structured interrupt data for Python code execution approval. |
|
Create structured interrupt data for execution plan approval with file-based storage support. |
|
Create structured interrupt data for memory operation approval. |
|
Extract and validate approval resume data from agent state. |
Get global ApprovalManager instance using singleton pattern. |
Configuration Management#
- class framework.approval.ApprovalManager(approval_config)[source]#
Bases:
object
Pure configuration service providing strongly typed approval models.
Serves as the centralized configuration management system for all approval settings across the framework. This class implements a clean separation of concerns by handling only configuration loading, validation, and provision of typed configuration objects to capabilities.
The manager implements a hierarchical configuration system where global approval modes can override capability-specific settings, ensuring consistent security posture across the entire system while allowing for granular control when needed.
- Responsibilities:
Load and validate approval configuration from global config system
Apply global mode overrides to capability-specific settings
Provide strongly typed configuration objects with validation
Create configured evaluator instances for capabilities
Maintain audit trail through comprehensive logging
- Explicitly NOT responsible for:
Business logic implementation (delegated to evaluators)
Approval decision making (capability-specific in evaluators)
State management (stateless configuration service)
- Configuration Hierarchy:
Global mode settings (disabled, selective, all_capabilities)
Capability-specific settings (python_execution, memory, etc.)
Resolved effective configuration (global overrides applied)
- Parameters:
approval_config (dict) – Raw approval configuration dictionary from config.yml
Examples
Initialize with configuration:
>>> config_dict = { ... 'global_mode': 'selective', ... 'capabilities': { ... 'python_execution': {'enabled': True, 'mode': 'epics_writes'}, ... 'memory': {'enabled': False} ... } ... } >>> manager = ApprovalManager(config_dict)
Access resolved configuration:
>>> python_config = manager.get_python_execution_config() >>> print(f"Effective setting: {python_config.enabled}")
Create evaluators:
>>> evaluator = manager.get_python_execution_evaluator() >>> # Evaluator is configured with resolved settings
Note
The manager is designed to be instantiated once at application startup and reused throughout the application lifecycle.
Warning
Configuration validation failures will raise ValueError to prevent insecure default behavior in production environments.
Initialize with approval configuration.
- Parameters:
approval_config (dict) – Raw approval configuration from config.yml
- Raises:
ValueError – If configuration is invalid or missing required fields
- __init__(approval_config)[source]#
Initialize with approval configuration.
- Parameters:
approval_config (dict) – Raw approval configuration from config.yml
- Raises:
ValueError – If configuration is invalid or missing required fields
- get_python_execution_config()[source]#
Get strongly typed Python execution approval configuration.
Applies global mode overrides to capability-specific settings, ensuring consistent behavior across the approval system.
- Returns:
Configuration object with resolved approval settings
- Return type:
- get_memory_config()[source]#
Get strongly typed memory approval configuration.
Applies global mode overrides to capability-specific settings, ensuring consistent behavior across the approval system.
- Returns:
Configuration object with resolved approval settings
- Return type:
- get_python_execution_evaluator()[source]#
Get configured Python execution approval evaluator.
Creates a new evaluator instance with current configuration settings. The evaluator contains the business logic for making approval decisions.
- Returns:
Evaluator instance configured with current settings
- Return type:
- get_memory_evaluator()[source]#
Get configured memory approval evaluator.
Creates a new evaluator instance with current configuration settings. The evaluator contains the business logic for making approval decisions.
- Returns:
Evaluator instance configured with current settings
- Return type:
- get_config_summary()[source]#
Get configuration summary for debugging and monitoring.
Provides a structured view of current approval configuration settings for logging, debugging, and administrative review.
- Returns:
Dictionary containing configuration summary with keys: - ‘global_mode’: Current global approval mode - ‘python_execution’: Python execution approval settings - ‘memory’: Memory operation approval settings
- Return type:
dict
- class framework.approval.GlobalApprovalConfig(global_mode, python_execution, memory)[source]#
Bases:
object
Global approval configuration integrating all capability-specific settings.
Top-level configuration model that combines global approval mode settings with capability-specific configurations. This class implements the hierarchical configuration system where global modes can override individual capability settings for consistent system-wide approval behavior.
The global configuration supports three main modes: - disabled: All approvals disabled system-wide - selective: Use capability-specific settings - all_capabilities: All approvals enabled system-wide
- Parameters:
global_mode (str) – System-wide approval mode controlling all capabilities
python_execution (PythonExecutionApprovalConfig) – Python code execution approval configuration
memory (MemoryApprovalConfig) – Memory operation approval configuration
Examples
Create global configuration:
>>> python_config = PythonExecutionApprovalConfig( ... enabled=True, mode=ApprovalMode.EPICS_WRITES ... ) >>> memory_config = MemoryApprovalConfig(enabled=False) >>> global_config = GlobalApprovalConfig( ... global_mode="selective", ... python_execution=python_config, ... memory=memory_config ... )
Access capability configurations:
>>> print(f"Global mode: {global_config.global_mode}") >>> print(f"Python enabled: {global_config.python_execution.enabled}") >>> print(f"Memory enabled: {global_config.memory.enabled}")
Note
This is a frozen dataclass representing immutable configuration state. The configuration hierarchy allows global modes to override capability settings when applied by the ApprovalManager.
- global_mode: str#
- python_execution: PythonExecutionApprovalConfig#
- memory: MemoryApprovalConfig#
- classmethod from_dict(data)[source]#
Create global configuration instance from dictionary with comprehensive validation.
Factory method that creates a GlobalApprovalConfig instance from a complete configuration dictionary. Performs extensive validation of the global structure and delegates capability-specific validation to appropriate config classes.
The method applies security-first defaults for missing configuration sections and provides detailed error messages for configuration issues. All capability configurations are validated and instantiated as strongly-typed objects.
- Required Structure:
global_mode: One of ‘disabled’, ‘selective’, ‘all_capabilities’
capabilities: Dictionary containing capability-specific settings
- Parameters:
data (dict) – Complete approval configuration dictionary from config.yml
- Returns:
Validated global configuration instance
- Return type:
- Raises:
ValueError – If configuration structure is invalid or contains invalid values
Examples
Create from complete configuration:
>>> config_dict = { ... 'global_mode': 'selective', ... 'capabilities': { ... 'python_execution': {'enabled': True, 'mode': 'epics_writes'}, ... 'memory': {'enabled': False} ... } ... } >>> config = GlobalApprovalConfig.from_dict(config_dict) >>> print(f"Global mode: {config.global_mode}")
Create with missing sections (secure defaults):
>>> minimal_config = {'global_mode': 'selective', 'capabilities': {}} >>> config = GlobalApprovalConfig.from_dict(minimal_config) >>> # Missing capabilities will use secure defaults
Handle validation errors:
>>> try: ... config = GlobalApprovalConfig.from_dict({'global_mode': 'invalid'}) ... except ValueError as e: ... print(f"Configuration error: {e}")
Warning
Missing capability sections will be created with security-first defaults. This ensures the system remains secure even with incomplete configuration.
- __init__(global_mode, python_execution, memory)#
Configuration Models#
- class framework.approval.PythonExecutionApprovalConfig(enabled, mode)[source]#
Bases:
object
Configuration model for Python code execution approval settings.
Immutable dataclass that encapsulates all settings related to Python code execution approval. The configuration includes both a high-level enabled flag and a granular mode setting that controls when approval is required based on code characteristics.
The configuration supports multiple approval modes to balance security needs with operational efficiency. The most common mode is EPICS_WRITES, which requires approval only for code that can modify accelerator systems.
- Parameters:
enabled (bool) – Whether Python execution approval is enabled globally
mode (ApprovalMode) – Granular approval mode controlling when approval is required
Examples
Create configuration for EPICS writes approval:
>>> config = PythonExecutionApprovalConfig( ... enabled=True, ... mode=ApprovalMode.EPICS_WRITES ... ) >>> print(f"Approval enabled: {config.enabled}") >>> print(f"Mode: {config.mode.value}")
Create configuration with all code approval:
>>> config = PythonExecutionApprovalConfig( ... enabled=True, ... mode=ApprovalMode.ALL_CODE ... )
Note
This is a frozen dataclass - instances cannot be modified after creation. Create new instances for different configurations.
See also
ApprovalMode
: Enum values used by this configurationPythonExecutionApprovalEvaluator
: Evaluator that uses this configurationApprovalManager
: Manager that provides instances of this configurationfrom_dict()
: Factory method for creating instances from dictionaries- enabled: bool#
- mode: ApprovalMode#
- classmethod from_dict(data)[source]#
Create configuration instance from dictionary with comprehensive validation.
Factory method that creates a PythonExecutionApprovalConfig instance from a configuration dictionary, applying security-first defaults and comprehensive validation. The method provides helpful error messages for configuration issues.
- Default Behavior:
Missing ‘enabled’ field defaults to True (secure default)
Missing ‘mode’ field defaults to ‘all_code’ (most secure mode)
Invalid mode values raise ValueError with valid options
- Parameters:
data (dict) – Configuration dictionary containing approval settings
- Returns:
Validated configuration instance
- Return type:
- Raises:
ValueError – If data is not a dict or contains invalid mode values
Examples
Create from complete configuration:
>>> config_dict = {'enabled': True, 'mode': 'epics_writes'} >>> config = PythonExecutionApprovalConfig.from_dict(config_dict) >>> print(f"Enabled: {config.enabled}, Mode: {config.mode.value}")
Create with defaults (secure fallbacks):
>>> config = PythonExecutionApprovalConfig.from_dict({}) >>> print(f"Default enabled: {config.enabled}") # True >>> print(f"Default mode: {config.mode.value}") # 'all_code'
Handle validation errors:
>>> try: ... config = PythonExecutionApprovalConfig.from_dict({'mode': 'invalid'}) ... except ValueError as e: ... print(f"Validation error: {e}")
Warning
This method applies security-first defaults. Missing configuration will default to the most secure settings (approval required).
See also
PythonExecutionApprovalConfig
: Configuration class created by this methodApprovalMode
: Enum values validated by this methodGlobalApprovalConfig.from_dict
: Similar factory method for global config
- __init__(enabled, mode)#
- class framework.approval.MemoryApprovalConfig(enabled)[source]#
Bases:
object
Configuration model for memory operation approval settings.
Immutable dataclass that encapsulates settings related to memory operation approval. Currently supports simple enabled/disabled logic but is designed for extension with more sophisticated approval rules in the future.
The configuration controls whether memory operations (create, update, delete) require human approval before execution. This provides protection for sensitive user data and system memory state.
- Parameters:
enabled (bool) – Whether memory operation approval is required
Examples
Create memory approval configuration:
>>> config = MemoryApprovalConfig(enabled=True) >>> print(f"Memory approval enabled: {config.enabled}")
Create disabled configuration:
>>> config = MemoryApprovalConfig(enabled=False) >>> # Memory operations will not require approval
Note
This is a frozen dataclass - instances cannot be modified after creation. The design allows for future extensions with additional fields.
- enabled: bool#
- classmethod from_dict(data)[source]#
Create configuration instance from flexible input format with validation.
Factory method that creates a MemoryApprovalConfig instance from either a boolean value (for simple enabled/disabled) or a dictionary (for future extensibility). Applies security-first defaults when configuration is ambiguous.
- Supported Input Formats:
bool: Directly sets the enabled flag
dict: Extracts ‘enabled’ field with secure default
- Parameters:
data (Union[bool, dict]) – Configuration data as boolean or dictionary
- Returns:
Validated configuration instance
- Return type:
- Raises:
ValueError – If data is neither bool nor dict, or bool value is invalid
Examples
Create from boolean:
>>> config = MemoryApprovalConfig.from_dict(True) >>> print(f"Enabled: {config.enabled}")
Create from dictionary:
>>> config_dict = {'enabled': False} >>> config = MemoryApprovalConfig.from_dict(config_dict) >>> print(f"Enabled: {config.enabled}")
Create with secure default:
>>> config = MemoryApprovalConfig.from_dict({}) >>> print(f"Default enabled: {config.enabled}") # True
Handle invalid input:
>>> try: ... config = MemoryApprovalConfig.from_dict("invalid") ... except ValueError as e: ... print(f"Invalid input: {e}")
Note
When using dictionary format, missing ‘enabled’ field defaults to True for security. This ensures approval is required unless explicitly disabled.
- __init__(enabled)#
- class framework.approval.ApprovalMode(value)[source]#
Bases:
Enum
Enumeration of approval modes for capability-specific approval control.
Defines the available approval modes that control when human approval is required for various operations within the system. Each mode represents a different level of approval granularity, from completely disabled to requiring approval for all operations.
The modes are designed to provide flexible control over approval requirements while maintaining clear semantic meaning for each level of restriction.
- Available Modes:
DISABLED: No approval required for any operations EPICS_WRITES: Approval required only for operations that write to EPICS ALL_CODE: Approval required for all code execution operations
Examples
Use in configuration:
>>> mode = ApprovalMode.EPICS_WRITES >>> print(f"Mode value: {mode.value}") >>> print(f"Mode name: {mode.name}")
Validate mode from string:
>>> try: ... mode = ApprovalMode("epics_writes") ... print(f"Valid mode: {mode}") ... except ValueError: ... print("Invalid mode string")
Note
These modes are primarily used for Python execution approval but the pattern can be extended to other capabilities as needed.
See also
PythonExecutionApprovalConfig
: Configuration class that uses this enumPythonExecutionApprovalEvaluator
: Evaluator that processes these modesGlobalApprovalConfig
: Global configuration that can override mode settings- DISABLED = 'disabled'#
- EPICS_WRITES = 'epics_writes'#
- ALL_CODE = 'all_code'#
Business Logic Evaluators#
- class framework.approval.PythonExecutionApprovalEvaluator(config)[source]#
Bases:
object
Business logic evaluator for Python code execution approval decisions.
Implements capability-specific rules for determining when Python code execution requires human approval. The evaluator supports multiple approval modes ranging from disabled (no approval) to all_code (approval for everything).
The evaluation logic considers both the configured approval mode and the specific characteristics of the code being evaluated, such as EPICS operations. This provides granular control over approval requirements based on operational risk assessment.
- Supported Approval Modes:
DISABLED: No approval required regardless of code content
EPICS_WRITES: Approval required only for code with EPICS write operations
ALL_CODE: Approval required for all Python code execution
- Parameters:
config (PythonExecutionApprovalConfig) – Configuration object containing approval settings
Examples
Create evaluator with EPICS writes mode:
>>> config = PythonExecutionApprovalConfig( ... enabled=True, ... mode=ApprovalMode.EPICS_WRITES ... ) >>> evaluator = PythonExecutionApprovalEvaluator(config)
Evaluate code with EPICS writes:
>>> decision = evaluator.evaluate( ... has_epics_writes=True, ... has_epics_reads=False ... ) >>> print(f"Approval needed: {decision.needs_approval}") >>> print(f"Reason: {decision.reasoning}")
Note
The evaluator is stateless and can be reused for multiple evaluations with the same configuration settings.
See also
PythonExecutionApprovalConfig
: Configuration model used by this evaluatorApprovalDecision
: Decision model returned by evaluation methodsApprovalManager
: Manager that creates instances of this evaluatorevaluate()
: Main evaluation method of this classInitialize evaluator with Python execution approval configuration.
- Parameters:
config (PythonExecutionApprovalConfig) – Configuration object containing approval mode and settings
- __init__(config)[source]#
Initialize evaluator with Python execution approval configuration.
- Parameters:
config (PythonExecutionApprovalConfig) – Configuration object containing approval mode and settings
- evaluate(has_epics_writes, has_epics_reads)[source]#
Evaluate whether Python code execution requires human approval.
Applies configured approval rules to determine if the given code characteristics require human approval before execution. The evaluation considers both global settings and code-specific risk factors.
The evaluation logic follows this hierarchy: 1. Check if approval is globally disabled 2. Apply mode-specific rules (disabled, epics_writes, all_code) 3. Fall back to secure default (approval required) for unknown modes
- Parameters:
has_epics_writes (bool) – Whether code contains EPICS write operations
has_epics_reads (bool) – Whether code contains EPICS read operations
- Returns:
Decision object with approval requirement and reasoning
- Return type:
Examples
Evaluate read-only EPICS code:
>>> decision = evaluator.evaluate( ... has_epics_writes=False, ... has_epics_reads=True ... ) >>> # Result depends on configured mode
Evaluate code with EPICS writes:
>>> decision = evaluator.evaluate( ... has_epics_writes=True, ... has_epics_reads=True ... ) >>> # Will require approval in EPICS_WRITES or ALL_CODE modes
Evaluate pure Python code:
>>> decision = evaluator.evaluate( ... has_epics_writes=False, ... has_epics_reads=False ... ) >>> # Requires approval only in ALL_CODE mode
Note
Unknown approval modes default to requiring approval for security.
See also
ApprovalDecision
: Decision structure returned by this methodApprovalMode
: Enum values processed by this evaluation logicPythonExecutionApprovalConfig
: Configuration that controls evaluationframework.approval.create_code_approval_interrupt()
: Uses evaluation results
- class framework.approval.MemoryApprovalEvaluator(config)[source]#
Bases:
object
Business logic evaluator for memory operation approval decisions.
Implements approval rules for memory operations including creating, updating, and deleting stored memories. Currently supports simple enabled/disabled logic but is designed for extension with more sophisticated rules based on content sensitivity, user permissions, or operation types.
The evaluator provides a foundation for future enhancements such as: - Content-based approval (sensitive data detection) - User-specific approval requirements - Operation-type granularity (create vs update vs delete) - Memory size or complexity thresholds
- Parameters:
config (MemoryApprovalConfig) – Configuration object containing memory approval settings
Examples
Create evaluator with approval enabled:
>>> config = MemoryApprovalConfig(enabled=True) >>> evaluator = MemoryApprovalEvaluator(config)
Evaluate memory operation:
>>> decision = evaluator.evaluate(operation_type="create") >>> print(f"Approval required: {decision.needs_approval}") >>> print(f"Reasoning: {decision.reasoning}")
Note
This evaluator is designed for future extensibility. Additional evaluation parameters can be added without breaking existing usage.
Initialize evaluator with memory approval configuration.
- Parameters:
config (MemoryApprovalConfig) – Configuration object containing memory approval settings
- __init__(config)[source]#
Initialize evaluator with memory approval configuration.
- Parameters:
config (MemoryApprovalConfig) – Configuration object containing memory approval settings
- evaluate(operation_type='general')[source]#
Evaluate whether memory operation requires human approval.
Determines approval requirements for memory operations based on configuration settings. Currently implements simple enabled/disabled logic but the interface supports future extensions for operation-specific or content-based approval rules.
- Parameters:
operation_type (str) – Type of memory operation for future rule extensions
- Returns:
Decision object with approval requirement and reasoning
- Return type:
Examples
Evaluate general memory operation:
>>> decision = evaluator.evaluate() >>> print(f"Needs approval: {decision.needs_approval}")
Evaluate specific operation type:
>>> decision = evaluator.evaluate(operation_type="delete") >>> # Currently same logic, but ready for future extensions
Note
The operation_type parameter is reserved for future functionality where different operation types may have different approval rules.
- class framework.approval.ApprovalDecision(needs_approval, reasoning)[source]#
Bases:
NamedTuple
Structured result of an approval evaluation decision.
Represents the outcome of an approval evaluation with both the decision and the reasoning behind it. This structured approach ensures consistent decision reporting across all evaluators and provides clear audit trails for approval decisions.
- Parameters:
needs_approval (bool) – Whether human approval is required for the operation
reasoning (str) – Human-readable explanation of the decision logic
Examples
Approval required decision:
>>> decision = ApprovalDecision( ... needs_approval=True, ... reasoning="Code contains EPICS write operations" ... ) >>> print(f"Decision: {decision.needs_approval}") >>> print(f"Reason: {decision.reasoning}")
No approval needed:
>>> decision = ApprovalDecision( ... needs_approval=False, ... reasoning="Python execution approval is disabled" ... )
Note
The reasoning field is crucial for logging, debugging, and providing clear feedback to users about why approval was or wasn’t required.
See also
PythonExecutionApprovalEvaluator
: Evaluator class that returns this decisionMemoryApprovalEvaluator
: Evaluator class that returns this decisionframework.approval.create_code_approval_interrupt()
: Uses reasoning for user messagesCreate new instance of ApprovalDecision(needs_approval, reasoning)
- needs_approval: bool#
Alias for field number 0
- reasoning: str#
Alias for field number 1
Approval System Functions#
- framework.approval.create_approval_type(capability_name, operation_type=None)[source]#
Generate dynamic approval type identifier from capability and operation.
Creates unique approval type identifiers that replace the hard-coded ApprovalType enum with a flexible system. This enables any capability to request approval without requiring framework modifications, while maintaining clear identification and supporting operation-level granularity within capabilities.
The generated identifiers follow a consistent naming pattern that ensures uniqueness and readability for logging, debugging, and user interfaces.
- Parameters:
capability_name (str) – Name of the capability requesting approval
operation_type (str, optional) – Optional specific operation type for granular control
- Returns:
Unique string identifier for the approval type
- Return type:
str
Examples
Basic capability approval:
>>> approval_type = create_approval_type("python") >>> print(approval_type) "python"
Operation-specific approval:
>>> approval_type = create_approval_type("memory", "save") >>> print(approval_type) "memory_save"
Complex capability with operation:
>>> approval_type = create_approval_type("data_analysis", "execute_query") >>> print(approval_type) "data_analysis_execute_query"
See also
create_code_approval_interrupt()
: Uses this function for approval type creationcreate_memory_approval_interrupt()
: Uses this function for approval type creationcreate_plan_approval_interrupt()
: Uses this function for approval type creationget_approval_resume_data()
: Uses approval types for state management
- framework.approval.create_code_approval_interrupt(code, analysis_details, execution_mode, safety_concerns, notebook_path=None, notebook_link=None, execution_request=None, expected_results=None, execution_folder_path=None, step_objective='Execute Python code')[source]#
Create structured interrupt data for Python code execution approval.
Generates LangGraph-compatible interrupt data for Python code that requires human approval before execution. The interrupt provides comprehensive context including code analysis, safety assessment, execution environment details, and clear approval instructions.
The function supports multiple execution modes and integrates with Jupyter notebooks for code review. Safety concerns and analysis details are presented to help users make informed approval decisions.
- param code:
Python code requiring approval before execution
- type code:
str
- param analysis_details:
Results from code analysis including safety assessment
- type analysis_details:
Dict[str, Any]
- param execution_mode:
Mode of execution (readonly, simulation, write)
- type execution_mode:
str
- param safety_concerns:
List of identified safety concerns or risks
- type safety_concerns:
List[str]
- param notebook_path:
File system path to Jupyter notebook for review
- type notebook_path:
Path, optional
- param notebook_link:
Web link to notebook interface for code review
- type notebook_link:
str, optional
- param execution_request:
Complete execution request data for context
- type execution_request:
Any, optional
- param expected_results:
Anticipated results or outputs from code execution
- type expected_results:
Dict[str, Any], optional
- param execution_folder_path:
Directory path where code will be executed
- type execution_folder_path:
Path, optional
- param step_objective:
High-level objective description for user context
- type step_objective:
str
- return:
Dictionary containing user_message and resume_payload for LangGraph
- rtype:
Dict[str, Any]
- Examples:
Basic code approval:
>>> interrupt_data = create_code_approval_interrupt( ... code="import pandas as pd
- df = pd.read_csv(‘data.csv’)”,
… analysis_details={‘safety_level’: ‘low’, ‘file_operations’: [‘read’]}, … execution_mode=’readonly’, … safety_concerns=[], … step_objective=”Load and analyze data” … ) >>> ‘yes’ in interrupt_data[‘user_message’] True
Code with safety concerns:
>>> interrupt_data = create_code_approval_interrupt( ... code="os.system('rm -rf /')", ... analysis_details={'safety_level': 'critical'}, ... execution_mode='write', ... safety_concerns=['System command execution', 'File deletion'], ... notebook_link="http://localhost:8888/notebooks/review.ipynb" ... )
Warning
This function is used for security-critical approval decisions. Ensure analysis_details and safety_concerns are thoroughly populated.
- Return type:
Dict[str, Any]
- framework.approval.create_plan_approval_interrupt(execution_plan, plan_file_path=None, pending_plans_dir=None)[source]#
Create structured interrupt data for execution plan approval with file-based storage support.
Generates LangGraph-compatible interrupt data that presents execution plans to users for approval. The interrupt includes formatted step details, clear approval instructions, and structured payload data for seamless resume operations after user approval.
The function supports file-based execution plan storage for enhanced human-in-the-loop workflows, particularly for Open WebUI integration.
The generated user message provides a comprehensive view of planned operations with step-by-step breakdown, making it easy for users to understand and evaluate the proposed execution plan.
- Parameters:
execution_plan (ExecutionPlan) – Execution plan object containing steps and configuration
plan_file_path (str, optional) – Optional file path where the execution plan was saved
pending_plans_dir (str, optional) – Optional directory path for pending plan files
- Returns:
Dictionary containing user_message and resume_payload for LangGraph
- Return type:
Dict[str, Any]
Examples
Basic plan approval:
>>> from framework.base.planning import ExecutionPlan >>> plan = ExecutionPlan(steps=[ ... {'task_objective': 'Load data', 'capability': 'data_loader'}, ... {'task_objective': 'Analyze trends', 'capability': 'data_analysis'} ... ]) >>> interrupt_data = create_plan_approval_interrupt(plan) >>> print(interrupt_data['user_message']) # Contains formatted approval request
Note
The interrupt data follows LangGraph’s standard structure with user_message for display and resume_payload for execution continuation.
See also
framework.base.planning.ExecutionPlan
: Input structure for this functioncreate_approval_type()
: Approval type generation used by this functionget_approval_resume_data()
: Function that processes the resume payloadclear_approval_state()
: State cleanup after approval processing
- framework.approval.create_memory_approval_interrupt(content, operation_type, user_id, existing_memory='', step_objective='Save content to memory')[source]#
Create structured interrupt data for memory operation approval.
Generates LangGraph-compatible interrupt data for memory operations that require human approval. The interrupt presents the memory content clearly formatted for user review, along with operation context and clear approval instructions.
This function supports all memory operations (create, update, delete) and provides appropriate context for each operation type. The structured payload enables seamless resume after user approval.
- Parameters:
content (str) – Memory content to be saved, updated, or referenced for deletion
operation_type (str) – Type of memory operation being requested
user_id (str) – Unique identifier for the user requesting the operation
existing_memory (str, optional) – Current memory content when updating existing memories
step_objective (str) – High-level objective description for user context
- Returns:
Dictionary containing user_message and resume_payload for LangGraph
- Return type:
Dict[str, Any]
Examples
Create new memory:
>>> interrupt_data = create_memory_approval_interrupt( ... content="User prefers morning meetings", ... operation_type="create", ... user_id="user123", ... step_objective="Save user preference" ... ) >>> print('yes' in interrupt_data['user_message']) # Shows approval options
Update existing memory:
>>> interrupt_data = create_memory_approval_interrupt( ... content="Updated preference: afternoon meetings", ... operation_type="update", ... user_id="user123", ... existing_memory="User prefers morning meetings", ... step_objective="Update user preference" ... )
Note
The content is displayed in a code block format for clear readability and to preserve formatting of structured data.
- framework.approval.get_approval_resume_data(state, expected_approval_type)[source]#
Extract and validate approval resume data from agent state.
Provides standardized, type-safe access to approval state with comprehensive validation. This function serves as the single source of truth for checking approval resume state, ensuring all capabilities handle approval consistently.
The function performs extensive validation to detect invalid or inconsistent approval states, preventing security vulnerabilities from malformed approval data. It distinguishes between normal execution, approved resumes, and rejected operations.
- Parameters:
state (AgentState) – Current agent state containing approval information
expected_approval_type (str) – Expected approval type for validation
- Returns:
Tuple containing resume status and payload data - has_approval_resume: True if this is resuming from approval - approved_payload: Payload data if approved, None if rejected/normal
- Return type:
tuple[bool, Optional[Dict[str, Any]]]
- Raises:
ValueError – If approval state structure is invalid or inconsistent
Examples
Normal execution (no approval state):
>>> has_resume, payload = get_approval_resume_data(state, "python") >>> print(f"Resume: {has_resume}, Payload: {payload}") Resume: False, Payload: None
Approved resume:
>>> # After user approves code execution >>> has_resume, payload = get_approval_resume_data(state, "python_executor") >>> if has_resume and payload: ... print(f"Executing approved code: {payload['code'][:50]}...")
Rejected operation:
>>> # After user rejects approval >>> has_resume, payload = get_approval_resume_data(state, "python_executor") >>> if has_resume and not payload: ... print("Operation was rejected by user")
Note
All capabilities should use this function instead of direct state access to ensure consistent approval handling and proper validation.
Utility Functions#
- framework.approval.get_approval_manager()[source]#
Get global ApprovalManager instance using singleton pattern.
Provides access to the system-wide approval manager instance, initializing it from the configuration system on first access. The function performs extensive validation to ensure security-critical approval configuration is present and valid before creating the manager instance.
The singleton pattern ensures consistent configuration across all capabilities and prevents multiple initialization of security-critical settings. Configuration is loaded once at startup and cached for the application lifetime.
Validation Process:
Verify approval configuration exists in global config
Validate configuration structure and required fields
Ensure global_mode and capabilities sections are present
Create and cache ApprovalManager instance
Log configuration summary for audit trail
- Returns:
Configured approval manager instance ready for use
- Return type:
- Raises:
ValueError – If approval configuration is missing, has invalid structure, or fails validation checks
KeyError – If required configuration sections are missing
Examples
Get manager instance:
>>> manager = get_approval_manager() >>> print(f"Manager ready with mode: {manager.config.global_mode}")
Handle configuration errors:
>>> try: ... manager = get_approval_manager() ... except ValueError as e: ... print(f"Configuration error: {e}") ... # Fix config.yml and restart application
See also
ApprovalManager
: Manager class created by this functionGlobalApprovalConfig
: Configuration model used for initializationconfigs.config.get_config_value()
: Configuration sourceApprovalManager.get_config_summary()
: Configuration validation methodWarning
This function will cause immediate application failure if approval configuration is missing or invalid. This is intentional to prevent insecure operation in production environments.
Note
The manager instance is cached globally. Subsequent calls return the same instance without re-reading configuration files.
See also
- Human Approval
Complete implementation guide and examples
framework.services.python_executor.PythonExecutorService
Service that integrates with approval system