Core API

This section documents the core components of AIECS.

Core Interfaces

class aiecs.core.interface.execution_interface.IToolProvider

Bases: ABC

Tool provider interface - Domain layer abstraction

abstract get_tool(tool_name)

Get tool instance

Parameters:

tool_name (str)

Return type:

Any

abstract has_tool(tool_name)

Check if tool exists

Parameters:

tool_name (str)

Return type:

bool

class aiecs.core.interface.execution_interface.IToolExecutor

Bases: ABC

Tool executor interface - Domain layer abstraction

abstract execute(tool, operation_name, **params)

Execute tool operation synchronously

Parameters:

• tool (Any)

• operation_name (str)

Return type:

Any

abstract async execute_async(tool, operation_name, **params)

Execute tool operation asynchronously

Parameters:

• tool (Any)

• operation_name (str)

Return type:

Any

class aiecs.core.interface.execution_interface.ICacheProvider

Bases: ABC

Cache provider interface - Domain layer abstraction

abstract generate_cache_key(operation_type, user_id, task_id, args, kwargs)

Generate cache key

Parameters:

• operation_type (str)

• user_id (str)

• task_id (str)

• args (tuple)

• kwargs (Dict[str, Any])

Return type:

str

abstract get_from_cache(cache_key)

Get data from cache

Parameters:

cache_key (str)

Return type:

Any | None

abstract add_to_cache(cache_key, value)

Add data to cache

Parameters:

• cache_key (str)

• value (Any)

Return type:

None

class aiecs.core.interface.execution_interface.IOperationExecutor

Bases: ABC

Operation executor interface - Domain layer abstraction

abstract async execute_operation(operation_spec, params)

Execute single operation

Parameters:

• operation_spec (str)

• params (Dict[str, Any])

Return type:

Any

abstract async batch_execute_operations(operations)

Batch execute operations

Parameters:

operations (List[Dict[str, Any]])

Return type:

List[Any]

abstract async execute_operations_sequence(operations, user_id, task_id, stop_on_failure=False, save_callback=None)

Execute operations sequence sequentially

Parameters:

• operations (List[Dict[str, Any]])

• user_id (str)

• task_id (str)

• stop_on_failure (bool)

• save_callback (Callable | None)

Return type:

List[TaskStepResult]

abstract async execute_parallel_operations(operations)

Execute operations in parallel

Parameters:

operations (List[Dict[str, Any]])

Return type:

List[TaskStepResult]

class aiecs.core.interface.execution_interface.ExecutionInterface

Bases: ABC

Unified execution interface that defines standard methods for service and tool execution. Supports plugin-based execution engines, allowing future introduction of new executors without modifying upper-level code.

abstract async execute_operation(operation_spec, params)

Execute a single operation (e.g., tool operation or service subtask).

Parameters:

• operation_spec (str) – Operation specification, format as ‘tool_name.operation_name’ or other identifier

• params (Dict[str, Any]) – Operation parameters

Returns:

Operation result

Return type:

Any

abstract async execute_task(task_name, input_data, context)

Execute a single task (e.g., service task).

Parameters:

• task_name (str) – Task name

• input_data (Dict[str, Any]) – Input data

• context (Dict[str, Any]) – Context information

Returns:

Task result

Return type:

Any

abstract async batch_execute_operations(operations)

Batch execute multiple operations.

Parameters:

operations (List[Dict[str, Any]]) – List of operations, each containing ‘operation’ and ‘params’

Returns:

List of operation results

Return type:

List[Any]

abstract async batch_execute_tasks(tasks)

Batch execute multiple tasks.

Parameters:

tasks (List[Dict[str, Any]]) – List of tasks, each containing ‘task_name’, ‘input_data’, ‘context’

Returns:

List of task results

Return type:

List[Any]

register_executor(executor_type, executor_instance)

Register new executor type, supporting plugin-based extension.

Parameters:

• executor_type (str) – Executor type identifier

• executor_instance (Any) – Executor instance

Return type:

None

Storage interfaces for the middleware architecture.

This module defines the core storage abstractions following the same pattern as other core interfaces, enabling dependency inversion and clean architecture.

class aiecs.core.interface.storage_interface.ISessionStorage

Bases: ABC

Session storage interface - Domain layer abstraction

abstract async create_session(session_id, user_id, metadata=None)

Create a new session.

Parameters:

• session_id (str)

• user_id (str)

• metadata (Dict[str, Any] | None)

Return type:

Dict[str, Any]

abstract async get_session(session_id)

Get session by ID.

Parameters:

session_id (str)

Return type:

Dict[str, Any] | None

abstract async update_session(session_id, updates=None, increment_requests=False, add_processing_time=0.0, mark_error=False)

Update session with activity and metrics.

Parameters:

• session_id (str)

• updates (Dict[str, Any] | None)

• increment_requests (bool)

• add_processing_time (float)

• mark_error (bool)

Return type:

bool

abstract async end_session(session_id, status='completed')

End a session and update metrics.

Parameters:

• session_id (str)

• status (str)

Return type:

bool

class aiecs.core.interface.storage_interface.IConversationStorage

Bases: ABC

Conversation storage interface - Domain layer abstraction

abstract async add_conversation_message(session_id, role, content, metadata=None)

Add message to conversation history.

Parameters:

• session_id (str)

• role (str)

• content (str)

• metadata (Dict[str, Any] | None)

Return type:

bool

abstract async get_conversation_history(session_id, limit=50)

Get conversation history for a session.

Parameters:

• session_id (str)

• limit (int)

Return type:

List[Dict[str, Any]]

class aiecs.core.interface.storage_interface.ICheckpointStorage

Bases: ABC

Checkpoint storage interface - Domain layer abstraction

abstract async store_checkpoint(thread_id, checkpoint_id, checkpoint_data, metadata=None)

Store checkpoint data.

Parameters:

• thread_id (str)

• checkpoint_id (str)

• checkpoint_data (Dict[str, Any])

• metadata (Dict[str, Any] | None)

Return type:

bool

abstract async get_checkpoint(thread_id, checkpoint_id=None)

Get checkpoint data. If checkpoint_id is None, get the latest.

Parameters:

• thread_id (str)

• checkpoint_id (str | None)

Return type:

Dict[str, Any] | None

abstract async list_checkpoints(thread_id, limit=10)

List checkpoints for a thread, ordered by creation time.

Parameters:

• thread_id (str)

• limit (int)

Return type:

List[Dict[str, Any]]

class aiecs.core.interface.storage_interface.ITaskContextStorage

Bases: ABC

Task context storage interface - Domain layer abstraction

abstract async get_task_context(session_id)

Get TaskContext for a session.

Parameters:

session_id (str)

Return type:

Any | None

abstract async store_task_context(session_id, context)

Store TaskContext for a session.

Parameters:

• session_id (str)

• context (Any)

Return type:

bool

class aiecs.core.interface.storage_interface.IStorageBackend

Bases: ISessionStorage, IConversationStorage, ICheckpointStorage, ITaskContextStorage

Unified storage backend interface - Domain layer abstraction

This interface combines all storage capabilities and follows the same pattern as other core interfaces in the middleware architecture.

abstract async initialize()

Initialize the storage backend.

Return type:

bool

abstract async close()

Close the storage backend.

abstract async health_check()

Perform health check.

Return type:

Dict[str, Any]

abstract async get_metrics()

Get comprehensive metrics.

Return type:

Dict[str, Any]

abstract async cleanup_expired_sessions(max_idle_hours=24)

Clean up expired sessions and associated data.

Parameters:

max_idle_hours (int)

Return type:

int

class aiecs.core.interface.storage_interface.IPermanentStorageBackend

Bases: ABC

Permanent storage backend interface for disk-based cold archive.

Used for dual-write alongside Redis (hot cache). Append-only semantics, optimized for analytics and long-term retention. Typical implementation: ClickHouse, PostgreSQL, etc.

All methods are fire-and-forget: failures should not block the primary Redis write path. Implementations should handle errors internally.

abstract async append_session_event(session_id, user_id, event_type, payload, created_at=None)

Append session create/update/end event for audit/analytics.

Parameters:

• session_id (str)

• user_id (str)

• event_type (str)

• payload (Dict[str, Any])

• created_at (str | None)

Return type:

bool

abstract async append_conversation_message(session_id, role, content, metadata=None, created_at=None)

Append conversation message (append-only).

Parameters:

• session_id (str)

• role (str)

• content (str)

• metadata (Dict[str, Any] | None)

• created_at (str | None)

Return type:

bool

abstract async append_checkpoint(thread_id, checkpoint_id, checkpoint_data, metadata=None, created_at=None)

Append checkpoint data.

Parameters:

• thread_id (str)

• checkpoint_id (str)

• checkpoint_data (Dict[str, Any])

• metadata (Dict[str, Any] | None)

• created_at (str | None)

Return type:

bool

abstract async append_checkpoint_writes(thread_id, checkpoint_id, task_id, writes_data, created_at=None)

Append checkpoint writes.

Parameters:

• thread_id (str)

• checkpoint_id (str)

• task_id (str)

• writes_data (List[tuple])

• created_at (str | None)

Return type:

bool

abstract async append_conversation_session(session_key, session_data, created_at=None)

Append conversation session metadata.

Parameters:

• session_key (str)

• session_data (Dict[str, Any])

• created_at (str | None)

Return type:

bool

abstract async append_task_context_snapshot(session_id, context_data, created_at=None)

Append task context snapshot (versioned).

Parameters:

• session_id (str)

• context_data (Dict[str, Any])

• created_at (str | None)

Return type:

bool

abstract async initialize()

Initialize the permanent storage backend.

Return type:

bool

abstract async close()

Close the permanent storage backend.

Return type:

None

class aiecs.core.interface.storage_interface.ICheckpointerBackend

Bases: ABC

Checkpointer backend interface for LangGraph integration.

This interface defines the minimal contract needed by BaseServiceCheckpointer to work with any storage backend, following dependency inversion principle.

abstract async put_checkpoint(thread_id, checkpoint_id, checkpoint_data, metadata=None)

Store a checkpoint for LangGraph workflows.

Parameters:

• thread_id (str)

• checkpoint_id (str)

• checkpoint_data (Dict[str, Any])

• metadata (Dict[str, Any] | None)

Return type:

bool

abstract async get_checkpoint(thread_id, checkpoint_id=None)

Retrieve a checkpoint for LangGraph workflows.

Parameters:

• thread_id (str)

• checkpoint_id (str | None)

Return type:

Dict[str, Any] | None

abstract async list_checkpoints(thread_id, limit=10)

List checkpoints for LangGraph workflows.

Parameters:

• thread_id (str)

• limit (int)

Return type:

List[Dict[str, Any]]

abstract async put_writes(thread_id, checkpoint_id, task_id, writes_data)

Store intermediate writes for a checkpoint.

Parameters:

• thread_id (str)

• checkpoint_id (str)

• task_id (str)

• writes_data (List[tuple])

Return type:

bool

abstract async get_writes(thread_id, checkpoint_id)

Get intermediate writes for a checkpoint.

Parameters:

• thread_id (str)

• checkpoint_id (str)

Return type:

List[tuple]