pip install aegis-sdk

from aegis import AegisClient

async with AegisClient(
    base_url="https://your-aegis-node",
    api_key="your-keycloak-jwt",     # Optional; alternatively set AEGIS_API_KEY env var
) as client:
    resp = await client.start_execution("agt-uuid", "Write a primality check in Python")
    print(resp.execution_id)

async with AegisClient("https://your-aegis-node", api_key=TOKEN) as client:
    resp = await client.start_execution(
        agent_id="agt-uuid",
        input="Write a primality check in Python",
    )
    print(resp.execution_id)  # "exec-uuid"

async with AegisClient("https://your-aegis-node", api_key=TOKEN) as client:
    resp = await client.start_execution("agt-uuid", "Solve the task")

    async for event in client.stream_execution(resp.execution_id):
        print(f"[{event.event_type}] {event.data}")

approvals = await client.list_pending_approvals()
for a in approvals:
    print(f"{a.id}: {a.prompt} (timeout: {a.timeout_seconds}s)")

resp = await client.approve_request(
    approval_id="approval-uuid",
    feedback="Looks good, proceed.",
    approved_by="jeshua@100monkeys.ai",
)
print(resp.status)  # "approved"

from aegis.seal import SEALClient, Ed25519Key
from aegis.seal.envelope import create_seal_envelope, verify_seal_envelope, create_canonical_message

from aegis.seal import SEALClient

client = SEALClient(
    gateway_url="https://your-aegis-node",
    workload_id="my-workload",
    security_scope="default",
)

try:
    result = await client.attest()
    print(result.security_token)

    output = await client.call_tool("read_file", {"path": "/workspace/main.py"})
    print(output)
finally:
    client.erase()   # zeroes the in-memory private key

SEALClient(gateway_url: str, workload_id: str, security_scope: str)

from aegis.seal import Ed25519Key

key = Ed25519Key.generate()
signature_b64 = key.sign_base64(b"my message")
pub_key_b64 = key.get_public_key_base64()
key.erase()  # zero the private key when done

from aegis.seal.envelope import create_seal_envelope
from aegis.seal import Ed25519Key

key = Ed25519Key.generate()
envelope = create_seal_envelope(
    security_token="<token>",
    mcp_payload={"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {...}},
    private_key=key,
)
# envelope: Dict[str, Any] — pass to AegisClient.invoke_seal() or the raw SEAL endpoint

from aegis.seal.envelope import verify_seal_envelope

payload = verify_seal_envelope(
    envelope=envelope_dict,
    public_key_bytes=bytes.fromhex(registered_public_key_hex),
    max_age_seconds=30,   # default: 30
)
# Returns the inner MCP payload Dict[str, Any] on success

from aegis.seal.envelope import create_canonical_message

msg_bytes = create_canonical_message(
    security_token="<token>",
    payload={"jsonrpc": "2.0", ...},
    timestamp_unix=1743600000,
)

from aegis.manifest import AgentManifestBuilder, AgentManifest, ImagePullPolicy

from aegis.manifest import AgentManifestBuilder, ImagePullPolicy

manifest = (
    AgentManifestBuilder(name="my-agent", language="python", version="3.12")
    .with_description("Analyses Python repos for security issues")
    .with_instruction("You are a security auditor. Analyse the code and report CVEs.")
    .with_execution_mode("iterative", max_iterations=15)
    .with_image_pull_policy(ImagePullPolicy.IF_NOT_PRESENT)
    .with_network_allow(["api.github.com", "pypi.org"])
    .with_tool("read_file")
    .with_tool("run_command")
    .with_env("LOG_LEVEL", "debug")
    .build()
)

manifest.to_yaml_file("./my-agent.yaml")

AgentManifestBuilder(
    name: str,
    language: Optional[str] = None,
    version: Optional[str] = None,
    image: Optional[str] = None,
)

class AgentManifest(BaseModel):
    apiVersion: str      # "100monkeys.ai/v1"
    kind: str            # "Agent"
    metadata: ManifestMetadata
    spec: AgentSpec

from aegis.manifest import ImagePullPolicy

ImagePullPolicy.ALWAYS          # "Always"
ImagePullPolicy.IF_NOT_PRESENT  # "IfNotPresent"
ImagePullPolicy.NEVER           # "Never"

from aegis.bootstrap import GenerateMessage

msg = GenerateMessage(
    execution_id="exec-uuid",
    iteration_number=1,
    model_alias="default",
    prompt="Task: Write a primality check\n\nInput: in Python",
    messages=[],
)

import json, httpx
response = httpx.post(
    f"{orchestrator_url}/v1/dispatch-gateway",
    content=msg.model_dump_json(),
    headers={"Content-Type": "application/json"},
)

from aegis.bootstrap import DispatchResultMessage

result = DispatchResultMessage(
    execution_id="exec-uuid",
    dispatch_id="dispatch-uuid",    # echo from DispatchMessage
    exit_code=0,
    stdout="All tests passed.\n",
    stderr="",
    duration_ms=1243,
    truncated=False,
)

from aegis.bootstrap import FinalMessage
import json, httpx

response = httpx.post(orchestrator_url + "/v1/dispatch-gateway", ...)
data = response.json()

if data["type"] == "final":
    final = FinalMessage.model_validate(data)
    print(final.content)                 # LLM's final text output
    print(final.tool_calls_executed)     # Number of tools invoked

from aegis.bootstrap import DispatchMessage

if data["type"] == "dispatch":
    cmd = DispatchMessage.model_validate(data)
    # cmd.action == "exec"
    # cmd.command == "python"
    # cmd.args == ["-m", "pytest", "test_prime.py"]
    # cmd.cwd == "/workspace"
    # cmd.timeout_secs == 60