capsule-run 0.4.1

Secure WASM runtime to isolate and manage AI agent tasks

capsule-run

A secure, durable runtime for agentic workflows

Overview

Capsule is a runtime for coordinating AI agent tasks in isolated environments. It is designed to handle long-running workflows, large-scale processing, autonomous decision-making securely, or even multi-agent systems.

Each task runs inside its own WebAssembly sandbox, providing:

• Isolated execution: Each task runs isolated from your host system
• Resource limits: Set CPU, memory, and timeout limits per task
• Automatic retries: Handle failures without manual intervention
• Lifecycle tracking: Monitor which tasks are running, completed, or failed

Installation

pip install capsule-run

Quick Start

Create hello.py:

from capsule import task

@task(name="main", compute="LOW", ram="64MB")
def main() -> str:
    return "Hello from Capsule!"

Run it:

capsule run hello.py

[!TIP] Use --verbose to display real-time task execution details.

How It Works

Simply annotate your Python functions with the @task decorator:

from capsule import task

@task(name="analyze_data", compute="MEDIUM", ram="512MB", timeout="30s", max_retries=1)
def analyze_data(dataset: list) -> dict:
    """Process data in an isolated, resource-controlled environment."""
    return {"processed": len(dataset), "status": "complete"}

[!NOTE] The runtime requires a task named "main" as the entry point. Python can define the main task itself, but it's recommended to set it manually.

When you run capsule run main.py, your code is compiled into a WebAssembly module and executed in a dedicated sandbox.

Response Format

Every task returns a structured JSON envelope containing both the result and execution metadata:

{
  "success": true,
  "result": { "processed": 5, "status": "complete" },
  "error": null,
  "execution": {
    "task_name": "data_processor",
    "duration_ms": 1523,
    "retries": 0,
    "fuel_consumed": 45000
  }
}

Response fields:

• success — Boolean indicating whether the task completed successfully
• result — The actual return value from your task (json, string, null on failure etc..)
• error — Error details if the task failed ({ error_type: string, message: string })
• execution — Performance metrics:
  • task_name — Name of the executed task
  • duration_ms — Execution time in milliseconds
  • retries — Number of retry attempts that occurred
  • fuel_consumed — CPU resources used (see Compute Levels)

Documentation

Task Configuration Options

Parameter	Description	Type	Default	Example
name	Task identifier	str	function name	"process_data"
compute	CPU level: "LOW", "MEDIUM", "HIGH"	str	"MEDIUM"	"HIGH"
ram	Memory limit	str	unlimited	"512MB", "2GB"
timeout	Maximum execution time	str	unlimited	"30s", "5m"
max_retries	Retry attempts on failure	int	0	3
allowed_files	Folders accessible in the sandbox	list	[]	["./data", "./output"]

Compute Levels

• LOW: Minimal allocation for lightweight tasks
• MEDIUM: Balanced resources for typical workloads
• HIGH: Maximum fuel for compute-intensive operations
• CUSTOM: Specify exact fuel value (e.g., compute="1000000")

Project Configuration (Optional)

Create a capsule.toml file in your project root to set default options:

[workflow]
name = "My AI Workflow"
version = "1.0.0"
entrypoint = "src/main.py"  # Run `capsule run` without specifying a file

[tasks]
default_compute = "MEDIUM"
default_ram = "256MB"
default_timeout = "30s"

Task-level options always override these defaults.

HTTP Client

Standard requests library isn't compatible with WASM. Use Capsule's HTTP client:

from capsule import task
from capsule.http import get, post

@task(name="fetch", compute="MEDIUM", timeout="30s")
def main() -> dict:
    response = get("https://api.example.com/data")
    return {"status": response.status_code, "ok": response.ok()}

File Access

Tasks can read and write files within directories specified in allowed_files. Any attempt to access files outside these directories is not possible.

from capsule import task

@task(name="restricted_writer", allowed_files=["./output"])
def restricted_writer() -> None:
    with open("./output/result.txt", "w") as f:
        f.write("result")

@task(name="main")
def main() -> str:
    restricted_writer()

Compatibility

✅ Supported:

• Pure Python packages and standard library
• json, math, re, datetime, collections, etc.

⚠️ Not yet supported:

• Packages with C extensions (e.g numpy, pandas)

Links

• GitHub
• Issues