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#

ApprovalManager(approval_config)

Pure configuration service providing strongly typed approval models.

GlobalApprovalConfig(global_mode, ...)

Global approval configuration integrating all capability-specific settings.

PythonExecutionApprovalConfig(enabled, mode)

Configuration model for Python code execution approval settings.

MemoryApprovalConfig(enabled)

Configuration model for memory operation approval settings.

ApprovalMode(value)

Enumeration of approval modes for capability-specific approval control.

PythonExecutionApprovalEvaluator(config)

Business logic evaluator for Python code execution approval decisions.

MemoryApprovalEvaluator(config)

Business logic evaluator for memory operation approval decisions.

ApprovalDecision(needs_approval, reasoning)

Structured result of an approval evaluation decision.

System Functions#

create_approval_type(capability_name[, ...])

Generate dynamic approval type identifier from capability and operation.

create_code_approval_interrupt(code, ...[, ...])

Create structured interrupt data for Python code execution approval.

create_plan_approval_interrupt(execution_plan)

Create structured interrupt data for execution plan approval with file-based storage support.

create_memory_approval_interrupt(content, ...)

Create structured interrupt data for memory operation approval.

get_approval_resume_data(state, ...)

Extract and validate approval resume data from agent state.

get_approval_manager()

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:
  1. Global mode settings (disabled, selective, all_capabilities)

  2. Capability-specific settings (python_execution, memory, etc.)

  3. 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:

PythonExecutionApprovalConfig

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:

MemoryApprovalConfig

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:

PythonExecutionApprovalEvaluator

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:

MemoryApprovalEvaluator

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:

GlobalApprovalConfig

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 configuration PythonExecutionApprovalEvaluator : Evaluator that uses this configuration ApprovalManager : Manager that provides instances of this configuration from_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:

PythonExecutionApprovalConfig

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 method ApprovalMode : Enum values validated by this method GlobalApprovalConfig.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:

MemoryApprovalConfig

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 enum PythonExecutionApprovalEvaluator : Evaluator that processes these modes GlobalApprovalConfig : 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 evaluator ApprovalDecision : Decision model returned by evaluation methods ApprovalManager : Manager that creates instances of this evaluator evaluate() : Main evaluation method of this class

Initialize 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:

ApprovalDecision

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 method ApprovalMode : Enum values processed by this evaluation logic PythonExecutionApprovalConfig : Configuration that controls evaluation framework.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:

ApprovalDecision

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 decision MemoryApprovalEvaluator : Evaluator class that returns this decision framework.approval.create_code_approval_interrupt() : Uses reasoning for user messages

Create 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 creation create_memory_approval_interrupt() : Uses this function for approval type creation create_plan_approval_interrupt() : Uses this function for approval type creation get_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 function create_approval_type() : Approval type generation used by this function get_approval_resume_data() : Function that processes the resume payload clear_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:

  1. Verify approval configuration exists in global config

  2. Validate configuration structure and required fields

  3. Ensure global_mode and capabilities sections are present

  4. Create and cache ApprovalManager instance

  5. Log configuration summary for audit trail

Returns:

Configured approval manager instance ready for use

Return type:

ApprovalManager

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 function GlobalApprovalConfig : Configuration model used for initialization configs.config.get_config_value() : Configuration source ApprovalManager.get_config_summary() : Configuration validation method

Warning

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