airalogy-engine (Python)

Airalogy protocol execution sandbox for Python. Run protocol packages (parse, assign, validate) inside a secure BoxLite sandbox, and execute AIMD workflow transition assignments across protocol Records.

Installation

pip install airalogy-engine

Sandbox Image

The engine runs protocol code in a BoxLite sandbox. You can use either a remote Docker image or a local OCI rootfs directory.

Remote Image

from airalogy_engine import AiralogyEngine

engine = AiralogyEngine(
    protocol_path="/path/to/your/protocol",
    image="numbcoder/airalogy-engine:latest",
)
result = await engine.parse_protocol()

Local OCI Rootfs (Recommended)

Build and export the image locally for faster, offline execution:

cd packages/runtime/airalogy-engine-image
docker build -t airalogy-engine:latest .
docker save airalogy-engine:latest -o airalogy-engine-image.tar
mkdir airalogy-engine-image
tar -xf airalogy-engine-image.tar -C airalogy-engine-image

Then use rootfs_path:

from airalogy_engine import AiralogyEngine

engine = AiralogyEngine(
    protocol_path="/path/to/your/protocol",
    rootfs_path="./airalogy-engine-image",
)
result = await engine.parse_protocol()

If neither image nor rootfs_path is provided, the engine falls back to the default remote image numbcoder/airalogy-engine:latest.

Usage

import asyncio
from airalogy_engine import AiralogyEngine

async def main():
    protocol_path = "/path/to/your/protocol"
    rootfs_path = "/path/to/airalogy-engine-image"  # or use image="..." instead
    engine = AiralogyEngine(
        protocol_path=protocol_path,
        rootfs_path=rootfs_path,
        boxlite_home="/tmp/airalogy-engine-worker-1",
    )

    # 1. Parse the protocol
    result = await engine.parse_protocol(env_vars={"API_KEY": "xxx"})
    print(result["data"]["meta_data"])
    print(result["data"]["json_schema"])

    # 2. Assign a variable
    result = await engine.assign_variable(
        var_name="duration",
        dependent_data={"seconds": 3600},
        env_vars={"API_KEY": "xxx"},
    )
    print(result["data"])

    # 3. Validate variables
    result = await engine.validate_variables(
        variables={"seconds": 60, "duration": "PT1M"},
    )
    print(result["data"])

    # 4. Import records from a file inside the protocol directory
    result = await engine.import_records(input_filename="records.json")
    print(result["data"]["records"])

    await engine.close()

asyncio.run(main())

You can also use the engine as an async context manager:

async with AiralogyEngine(
    protocol_path=protocol_path,
    rootfs_path=rootfs_path,
    boxlite_home="/tmp/worker-1",
) as engine:
    result = await engine.parse_protocol()

Workflow Usage

AiralogyWorkflowEngine executes fenced workflow definitions from a workflow.aimd file. It resolves transition.inputs, runs workflow-level Python assigners, exposes outputs under ${transition_id.outputs.key}, and applies transition.assign into target Record drafts. It does not persist Records or create Record versions; callers should save returned Record drafts through their platform or database layer.

import asyncio
from airalogy_engine import AiralogyWorkflowEngine

async def main():
    engine = AiralogyWorkflowEngine(
        workflow_path="/path/to/workflow.aimd",
        rootfs_path="/path/to/airalogy-engine-image",
    )
    result = await engine.run(
        records={
            "measurement": {"data": {"var": {"raw_data": [1, 2, 3]}}},
            "literature_review": {"data": {"var": {"summary": "known background"}}},
        },
    )
    print(result["data"]["records"]["analysis"])
    print(result["data"]["transition_outputs"])
    await engine.close()

asyncio.run(main())

For local tests or trusted scripts, pass assigner_runtime="local" to execute workflow assigners in the host Python process instead of BoxLite.

API

All engine methods are async and return a dict with success, message, and data keys.

Engine parameters:

• protocol_path: Protocol package directory. It must contain protocol.aimd and is mounted writable at /home/airalogy/protocols/protocol inside the sandbox.
• boxlite_home: BoxLite runtime home directory. Use a distinct value for each OS process when running multiple workers.
• image: Remote Docker image name (e.g., "numbcoder/airalogy-engine:0.1").
• rootfs_path: Path to a local OCI rootfs directory (overrides image).
• timeout: Execution timeout in seconds (default: 300). The sandboxed process will be killed once it times out.
• memory_mib: Memory limit in MiB (default: 512).
• cpus: CPU limit (default: 1).
• auto_stop: Stop the box after each command when True (default). Set to False to keep one running box until stop() or close().

Concurrency

Use one AiralogyEngine instance per protocol and worker process. Concurrent async operations through one engine run on its current box:

engine = AiralogyEngine(
    protocol_path=protocol_path,
    rootfs_path=rootfs_path,
    boxlite_home="/tmp/worker-1",
    auto_stop=False,
)

results = await asyncio.gather(
    engine.parse_protocol(),
    engine.validate_variables({"seconds": 60, "duration": "PT1M"}),
)

await engine.stop()

BoxLite locks each runtime home per OS process. Two independent processes must not share the same boxlite_home or default ~/.boxlite; give each process a distinct directory, for example /tmp/airalogy-worker-1 and /tmp/airalogy-worker-2.

Testing

cd python
uv sync

# Default: local OCI rootfs mode
uv run pytest tests/ -v

# Custom rootfs path
uv run pytest tests/ -v --sandbox-mode=rootfs --rootfs-path=../../runtime/airalogy-engine-image/airalogy-engine-image

# Remote image mode
uv run pytest tests/ -v --sandbox-mode=image --sandbox-image=numbcoder/airalogy-engine:latest