Session¶
aigise.session.aigise_session ¶
AigiseSession: Unified session management for AigiseAgent Framework
This module provides the primary session management architecture that consolidates all session-specific managers (config, agents, sandboxes) under a unified interface.
Each AigiseSession instance represents a single session and manages all resources for that session without relying on global singletons.
logger = logging.getLogger(__name__) module-attribute ¶
AigiseConfig dataclass ¶
Complete SecAgentFramework configuration.
copy() -> AigiseConfig ¶
Create a deep copy of this configuration.
create_default() -> AigiseConfig classmethod ¶
Create a default configuration from TOML file with environment variable overrides.
from_toml(config_path: Optional[str] = None) -> AigiseConfig classmethod ¶
Create configuration from TOML file with template variable expansion.
get_llm_config(model_name: str) ¶
Get LLM configuration for a specific model.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model_name | str | Name of the model configuration to get | required |
Returns:
| Type | Description |
|---|---|
| ModelConfig for the specified model, or None if not found |
get_sandbox_config(sandbox_type: str) ¶
Get sandbox configuration for a specific type.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sandbox_type | str | Type of sandbox configuration to get | required |
Returns:
| Type | Description |
|---|---|
| ContainerConfig for the specified sandbox type, or None if not found |
save_to_toml(toml_path: str) -> None ¶
Save configuration to TOML file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
toml_path | str | Path to save TOML file | required |
AigiseSession ¶
Unified session manager for AigiseAgent Framework.
Each instance manages all resources for a specific session, including: - Configuration management (TOML loading, env overrides) - Agent lifecycle management (creation, persistence, cleanup) - Sandbox management (Docker containers, resource isolation) - Agent ensemble management (thread-safe tools, agent discovery)
This replaces the previous singleton-based architecture with a clear session-bound resource management model.
__init__(aigise_session_id: str, config_path: Optional[str] = None) ¶
Initialize AigiseSession for a specific session.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
aigise_session_id | str | Unique identifier for this session | required |
config_path | Optional[str] | Optional path to TOML configuration file | None |
cleanup() -> None ¶
Cleanup all resources for this session.
get_session_info() -> Dict ¶
Get comprehensive information about this session.
Returns:
| Type | Description |
|---|---|
Dict | Dictionary containing session information |
load_config_from_toml(toml_path: str) -> None ¶
Load configuration from TOML file for this session.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
toml_path | str | Path to TOML configuration file | required |
save_config_to_toml(toml_path: str) -> None ¶
Save current configuration to TOML file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
toml_path | str | Path to save TOML file | required |
update_config_from_env() -> None ¶
Update configuration from environment variables.
AigiseSessionRegistry ¶
Global registry for managing AigiseSession instances.
This is the only global singleton in the new architecture, responsible for: - Creating and tracking session managers - Preventing duplicate sessions - Coordinating session cleanup - Managing global signal handlers for graceful shutdown
cleanup_all_sessions() -> None classmethod ¶
Cleanup all active sessions.
This should be called during application shutdown to ensure all resources are properly cleaned up.
get_aigise_session(aigise_session_id: str, config_path: Optional[str] = None) -> AigiseSession classmethod ¶
Get or create a session manager for the given session ID.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
aigise_session_id | str | Unique session identifier | required |
Returns:
| Type | Description |
|---|---|
AigiseSession | AigiseSession instance for the session |
list_sessions() -> list[str] classmethod ¶
Get list of all active session IDs.
Returns:
| Type | Description |
|---|---|
list[str] | List of active session IDs |
remove_session(aigise_session_id: str) -> bool classmethod ¶
Remove and cleanup a session.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
aigise_session_id | str | Session ID to remove | required |
Returns:
| Type | Description |
|---|---|
bool | True if removed, False if not found |
cleanup_aigise_session(aigise_session_id: str) -> bool ¶
Cleanup and remove an AigiseSession.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
aigise_session_id | str | Session ID to cleanup | required |
Returns:
| Type | Description |
|---|---|
bool | True if cleaned up, False if not found |
Example
cleanup_aigise_session("user_123_task_456")
get_aigise_session(aigise_session_id: str, config_path: Optional[str] = None) -> AigiseSession ¶
Get or create an AigiseSession for the given session ID.
aigise.session.aigise_sandbox_manager ¶
AigiseSandboxManager: Session-specific sandbox management
This module provides session-bound sandbox management, replacing the global SandboxManager with session-isolated sandbox handling.
logger = logging.getLogger(__name__) module-attribute ¶
AigiseConfig dataclass ¶
Complete SecAgentFramework configuration.
copy() -> AigiseConfig ¶
Create a deep copy of this configuration.
create_default() -> AigiseConfig classmethod ¶
Create a default configuration from TOML file with environment variable overrides.
from_toml(config_path: Optional[str] = None) -> AigiseConfig classmethod ¶
Create configuration from TOML file with template variable expansion.
get_llm_config(model_name: str) ¶
Get LLM configuration for a specific model.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model_name | str | Name of the model configuration to get | required |
Returns:
| Type | Description |
|---|---|
| ModelConfig for the specified model, or None if not found |
get_sandbox_config(sandbox_type: str) ¶
Get sandbox configuration for a specific type.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sandbox_type | str | Type of sandbox configuration to get | required |
Returns:
| Type | Description |
|---|---|
| ContainerConfig for the specified sandbox type, or None if not found |
save_to_toml(toml_path: str) -> None ¶
Save configuration to TOML file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
toml_path | str | Path to save TOML file | required |
AigiseSandboxManager ¶
Session-specific sandbox manager.
Each AigiseSession gets its own AigiseSandboxManager instance, ensuring complete sandbox isolation between sessions.
config: AigiseConfig property ¶
Get latest config from session dynamically.
__init__(session) ¶
Initialize AigiseSandboxManager.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
session | AigiseSession instance (stores reference, not copied) | required |
attach_sandbox(sandbox_type: str, *, container_id: Optional[str] = None, pod_name: Optional[str] = None, container_name: Optional[str] = None) -> None async ¶
Attach to an existing container/Pod and register it to this session, then call ensure_ready.
- native (Docker): requires container_id
- k8s: requires pod_name + container_name
cache_sandboxes(cache_dir: Optional[str] = None) -> Dict[str, Any] ¶
Cache current sandbox states and shared volume content.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
cache_dir | Optional[str] | Directory to store cache files (default: ./sandbox_cache/{task_name}) | None |
Returns:
| Type | Description |
|---|---|
Dict[str, Any] | Dictionary with cache results including backup paths and cached images |
cleanup() -> None ¶
Cleanup all sandboxes for this session.
get_sandbox(sandbox_type: str) -> BaseSandbox ¶
Get the sandbox instance for the given sandbox type.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sandbox_type | str | Type of sandbox to get or create | required |
Returns:
| Type | Description |
|---|---|
BaseSandbox | BaseSandbox instance for the session and type |
get_session_statistics() -> Dict ¶
Get statistics for this session's sandboxes.
Returns:
| Type | Description |
|---|---|
Dict | Dictionary with session statistics |
get_shared_volume() -> Optional[str] ¶
Get the shared volume ID for this session.
Returns:
| Type | Description |
|---|---|
Optional[str] | Volume ID or None if no shared volume exists |
initialize_all_sandboxes(*, continue_on_error: bool = False) -> None async ¶
initialize_shared_volumes(*, tools_top_roots: set[str] | None = None, enabled_skills: Any = None) -> None ¶
Initialize shared volumes (scripts/shared-data/tools).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tools_top_roots | set[str] | None | Optional set of top-level bash_tools roots to stage into the tools volume/PVC. If None, stage all tools. | None |
enabled_skills | Any | enabled_skills setting from the root agent (None, "all", or List[str]). Stored for sandbox initializers to conditionally run skill dependency installers. | None |
launch_all_sandboxes(sandbox_types: Optional[Set[str]] = None) -> None async ¶
Launch configured sandbox instances based on backend type.
This method should be called during session initialization. If sandboxes already exist, this method will skip to avoid conflicts.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sandbox_types | Optional[Set[str]] | Optional set of sandbox types to launch. If None, launches all configured sandboxes. If provided, only launches sandboxes of the specified types. Use collect_sandbox_dependencies() to get this from an agent. | None |
Example::
# Launch only required sandboxes
from aigise.toolbox.decorators import collect_sandbox_dependencies
deps = collect_sandbox_dependencies(root_agent) # {'main', 'gdb_mcp'}
await session.sandboxes.launch_all_sandboxes(sandbox_types=deps)
# Or launch all configured sandboxes
await session.sandboxes.launch_all_sandboxes()
list_sandboxes() -> Dict[str, BaseSandbox] ¶
List all sandboxes for this session.
Returns:
| Type | Description |
|---|---|
Dict[str, BaseSandbox] | Dictionary mapping sandbox types to sandbox instances |
load_sandbox_caches_to_config() -> list[str] ¶
Load cached sandbox images and update sandbox configurations.
This method looks for cached images with the naming pattern: {normalized_task_name}sandbox:cached
For each found cached image, it updates the corresponding sandbox configuration to use the cached image instead of the original.
Returns:
| Type | Description |
|---|---|
list[str] | List of sandbox types that don't have cached images available |
remove_sandbox(sandbox_type: str) -> bool ¶
Remove and cleanup a specific sandbox.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sandbox_type | str | Type of sandbox to remove | required |
Returns:
| Type | Description |
|---|---|
bool | True if removed, False if not found |
BaseSandbox ¶
Bases: ABC
Base class for all sandbox implementations.
async_initialize() -> None async ¶
Initialize the sandbox.
cache_sandboxes(sandbox_instances: dict, shared_volume_id: str, cache_dir: str, task_name: str) -> dict abstractmethod classmethod ¶
Cache sandbox states and shared volume content.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sandbox_instances | dict | Dictionary mapping sandbox types to sandbox instances | required |
shared_volume_id | str | Shared volume identifier to backup | required |
cache_dir | str | Directory to store cache files | required |
task_name | str | Task name for cache naming | required |
Returns:
| Type | Description |
|---|---|
dict | Dictionary with cache results including backup paths and cached images |
copy_file_from_container(src_path: str, dst_path: str) abstractmethod ¶
Copy a file from the container to local filesystem.
copy_file_to_container(local_path: str, container_path: str) abstractmethod ¶
Copy a file from local filesystem to the container.
create_shared_volume(volume_name_prefix: str, init_data_path: Path = None, tools_top_roots: set[str] | None = None) -> tuple[str, str, str] abstractmethod classmethod ¶
Create and initialize three shared volumes.
Creates three volumes: 1. Read-only volume with sandbox scripts (mapped to /sandbox_scripts) 2. Read-write volume with user data (mapped to /shared) 3. Read-write volume with bash tools (mapped to /bash_tools)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
volume_name_prefix | str | Prefix for volume names (e.g., session_id) | required |
init_data_path | Path | Path to initial data to copy into the rw volume (optional) | None |
tools_top_roots | set[str] | None | Optional set of top-level bash_tools roots to stage. If None, stage all bash tools. | None |
Returns:
| Type | Description |
|---|---|
tuple[str, str, str] | Tuple of (scripts_volume_id, data_volume_id, tools_volume_id) |
create_single_sandbox(session_id: str, sandbox_type: str, container_config) -> Exception abstractmethod async classmethod ¶
Create a single sandbox instance asynchronously.
delete_shared_volumes(scripts_volume_id: str = None, data_volume_id: str = None, tools_volume_id: str = None) -> None abstractmethod classmethod ¶
Delete shared volumes.
ensure_ready() -> None async ¶
Ensure the sandbox is ready.
extract_file_from_container(filepath: str) -> str abstractmethod ¶
Extract file content from the container.
extract_file_from_container_bytes(filepath: str) -> bytes abstractmethod ¶
Extract file content from the container.
get_work_dir() abstractmethod ¶
Get the current working directory in the container.
launch_all_sandboxes(session_id: str, sandbox_configs: dict, shared_volume_id: str = None, scripts_volume_id: str = None, tools_volume_id: str = None) -> dict abstractmethod async classmethod ¶
Launch all sandbox instances for a session.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
session_id | str | Session identifier | required |
sandbox_configs | dict | Dictionary of sandbox_type -> ContainerConfig | required |
shared_volume_id | str | Optional shared volume to mount to all sandboxes | None |
scripts_volume_id | str | Optional scripts volume to mount to all sandboxes | None |
tools_volume_id | str | Optional tools volume to mount to all sandboxes | None |
Returns: Dictionary mapping sandbox_type to sandbox instance or connection info
run_command_in_container(command: str | list[str], timeout: int | None = None) -> tuple[str, int] abstractmethod ¶
Run a command inside the container.
wait_for_ready_or_error() -> bool async ¶
Wait for a specific sandbox to be ready or error.
SandboxState ¶
Bases: Enum
Sandbox initialization states.
can_pull_image(image_name: str) -> bool ¶
Try to pull Docker image and return success status.
create_sandbox_class(backend_class: Type[BaseSandbox], initializer_class: Type) -> Type[BaseSandbox] ¶
Create a sandbox class by combining a backend with a initializer.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
backend_class | Type[BaseSandbox] | The backend sandbox class (e.g., NativeDockerSandbox) | required |
initializer_class | Type | Initializer class to add functionality | required |
Returns:
| Type | Description |
|---|---|
Type[BaseSandbox] | A new class that combines the backend and initializer |
get_backend_class(backend_type: str, config=None) -> Type[BaseSandbox] ¶
Get the backend class for a given backend type.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
backend_type | str | The type of backend needed (e.g., 'native', 'k8s') | required |
config | Optional config to inject into backend (for remotedocker) | None |
Returns:
| Type | Description |
|---|---|
Type[BaseSandbox] | The backend class |
Raises:
| Type | Description |
|---|---|
ValueError | If backend type is not supported |
get_initializer_class(sandbox_type: str) -> Type ¶
Get the initializer class for a given sandbox type.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sandbox_type | str | The type of sandbox functionality needed | required |
Returns:
| Type | Description |
|---|---|
Type | The initializer class, or the base SandboxInitializer if not found |
image_exists_locally(image_name: str) -> bool ¶
Check if Docker image exists locally.
aigise.session.aigise_dynamic_agent_manager ¶
DynamicAgentManager: Session-specific agent lifecycle management
This module provides session-bound agent management, replacing the global DynamicAgentManager with session-isolated agent handling.
logger = logging.getLogger(__name__) module-attribute ¶
AgentMetadata dataclass ¶
Metadata for dynamically created agents.
AgentStatus ¶
Bases: Enum
Agent lifecycle status.
AigiseAgent ¶
Bases: LlmAgent
update_enabled_skills(enabled_skills: Optional[Union[List[str], str]]) -> None ¶
Update enabled_skills and regenerate system prompt with new bash tools.
This method: 1. Updates the _enabled_skills attribute 2. Removes the old bash tools section from instruction 3. Generates new tool prompt based on new enabled_skills 4. Appends the new tool prompt to instruction
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
enabled_skills | Optional[Union[List[str], str]] | New enabled_skills value (None, "all", or List[str]) | required |
DynamicAgentManager ¶
Session-specific manager for dynamic agent creation, lifecycle, and persistence.
Each AigiseSession gets its own DynamicAgentManager instance, ensuring complete agent isolation between sessions.
config property ¶
Get latest config from session dynamically.
__init__(session) ¶
Initialize DynamicAgentManager.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
session | AigiseSession instance (stores reference, not copied) | required |
cleanup() -> None ¶
Cleanup all agents and resources for this session.
create_agent(config: Dict[str, Any], creator: Optional[str] = None, persist: bool = True) -> tuple[str, AigiseAgent] async ¶
Create a new agent dynamically for this session.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config | Dict[str, Any] | Agent configuration dictionary | required |
creator | Optional[str] | Optional creator identifier | None |
persist | bool | Whether to persist agent metadata | True |
Returns:
| Type | Description |
|---|---|
tuple[str, AigiseAgent] | Tuple of (agent_id, agent_instance) |
get_agent(agent_id: str) -> Optional[BaseAgent] ¶
Get an agent by ID for this session.
get_agent_metadata(agent_id: str) -> Optional[AgentMetadata] ¶
Get agent metadata by ID for this session.
get_session_statistics() -> Dict ¶
Get statistics for this session's agents.
Returns:
| Type | Description |
|---|---|
Dict | Dictionary with session statistics |
list_agents(status: Optional[AgentStatus] = None, creator: Optional[str] = None) -> List[AgentMetadata] ¶
List agents with optional filtering for this session.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
status | Optional[AgentStatus] | Optional status filter | None |
creator | Optional[str] | Optional creator filter | None |
Returns:
| Type | Description |
|---|---|
List[AgentMetadata] | List of agent metadata matching the filters |
remove_agent(agent_id: str, cascade: bool = False) -> bool async ¶
Remove an agent from this session.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
agent_id | str | ID of the agent to remove | required |
cascade | bool | Whether to remove child agents as well | False |
Returns:
| Type | Description |
|---|---|
bool | True if removed successfully, False if not found |
update_agent_status(agent_id: str, status: AgentStatus) -> bool async ¶
Update agent status and trigger lifecycle hooks.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
agent_id | str | ID of the agent to update | required |
status | AgentStatus | New status to set | required |
Returns:
| Type | Description |
|---|---|
bool | True if updated successfully, False if agent not found |