Python SDK

Install the AEGIS Python SDK:

pip install aegis-sdk

The SDK consists of two distinct layers:

• AegisClient — async HTTP client: manage agents, executions, workflows, volumes, credentials, secrets, SEAL tools, and more
• bootstrap.py / Dispatch Protocol types — agent-side: the wire format exchanged between a custom bootstrap script and the orchestrator's /v1/dispatch-gateway

AegisClient

Authentication

The client supports two authentication modes:

Service-to-service (OAuth2 client credentials):

from aegis import AegisClient

async with AegisClient(
    base_url="https://your-orchestrator.com",
    keycloak_url="https://auth.example.com",
    realm="aegis-system",
    client_id="your-client-id",
    client_secret="your-client-secret",
) as client:
    resp = await client.start_execution("agt-uuid", "Write a primality check in Python")
    print(resp.execution_id)

Bearer token passthrough (for user-scoped access):

from aegis import AegisClient

async with AegisClient(
    base_url="https://your-orchestrator.com",
    bearer_token="user-jwt-token-here",
) as client:
    agents = await client.list_agents()

Method Reference

Agent Management

deploy_agent(manifest)

Deploy a new agent from a manifest dictionary.

resp = await client.deploy_agent({
    "name": "code-reviewer",
    "language": "python",
    "version": "3.12",
    "instruction": "You are a code reviewer.",
})
print(resp["agent_id"])

Parameters:

Returns: Dict[str, Any] with agent_id.

list_agents()

List all deployed agents visible to the caller's tenant.

Returns: Dict[str, Any] with agents: List.

get_agent(agent_id)

Get details for a specific agent.

Returns: Dict[str, Any] — full agent record.

lookup_agent(name)

Look up an agent by its unique name.

Returns: Dict[str, Any] with agent_id, name, tenant_id.

update_agent(agent_id, payload)

Update an agent's configuration (partial update).

Returns: Dict[str, Any].

delete_agent(agent_id)

Delete a deployed agent.

Returns: Dict[str, Any].

execute_agent(agent_id, input, intent?, context_overrides?)

Execute an agent via the direct POST /v1/agents/{agent_id}/execute route.

Parameters:

Returns: Dict[str, Any] with execution_id.

list_agent_versions(agent_id)

List the version history of an agent.

Returns: Dict[str, Any] with versions: List.

update_agent_scope(agent_id, payload)

Update an agent's visibility scope (e.g., public, tenant-only).

Returns: Dict[str, Any].

stream_agent_events(agent_id)

Stream real-time SSE events for all executions of a given agent.

Returns: AsyncGenerator[ExecutionEvent, None].

Execution

start_execution(agent_id, input, intent?, context_overrides?)

Start an agent execution and receive an execution ID for streaming.

async with AegisClient(
    base_url="https://your-orchestrator.com",
    keycloak_url="https://auth.example.com",
    realm="aegis-system",
    client_id="my-client",
    client_secret="my-secret",
) as client:
    resp = await client.start_execution(
        agent_id="agt-uuid",
        input="Write a primality check in Python",
        intent="code-generation",
    )
    print(resp.execution_id)  # "exec-uuid"

Parameters:

Returns: StartExecutionResponse with execution_id: str.

stream_execution(execution_id, token?)

Stream real-time execution events via SSE. Returns an async generator.

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

Parameters:

Returns: AsyncGenerator[ExecutionEvent, None] — each event has event_type: str and data: Dict[str, Any].

get_execution(execution_id)

Get details of a specific execution.

Returns: Dict[str, Any] — full execution record including status, agent_id, timestamps.

list_executions()

List all executions visible to the caller.

Returns: Dict[str, Any] with executions: List.

cancel_execution(execution_id)

Cancel a running execution.

Returns: Dict[str, Any].

delete_execution(execution_id)

Delete an execution record.

Returns: Dict[str, Any].

get_execution_file(execution_id, path)

Download a file from an execution's workspace.

Parameters:

Returns: bytes — raw file content.

Human Approvals

list_pending_approvals()

List all pending human-in-the-loop approval requests.

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

Returns: List[PendingApproval] — each with id, execution_id, prompt, created_at, timeout_seconds.

get_pending_approval(approval_id)

Fetch a single pending approval by ID.

Returns: PendingApproval.

approve_request(approval_id, feedback?, approved_by?)

Approve a pending request, optionally with feedback.

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

Returns: ApprovalResponse with status: str.

reject_request(approval_id, reason, rejected_by?)

Reject a pending request with a mandatory reason.

Returns: ApprovalResponse with status: str.

Workflow Management

register_workflow(payload)

Register a new workflow definition.

resp = await client.register_workflow({
    "name": "data-pipeline",
    "definition": { ... },
})

Returns: Dict[str, Any].

list_workflows()

List all registered workflow definitions.

Returns: Dict[str, Any] with workflows: List.

get_workflow(name)

Get a workflow definition by name.

Returns: Dict[str, Any] — full workflow record.

delete_workflow(name)

Delete a workflow definition.

Returns: Dict[str, Any].

list_workflow_versions(name)

List version history for a workflow.

Returns: Dict[str, Any] with versions: List.

update_workflow_scope(name, payload)

Update a workflow's visibility scope.

Returns: Dict[str, Any].

run_workflow(name, payload)

Run a workflow by name with the given input payload.

Returns: Dict[str, Any] with execution details.

execute_temporal_workflow(payload)

Execute a Temporal workflow directly.

Returns: Dict[str, Any].

Workflow Execution Management

list_workflow_executions()

List all workflow executions.

Returns: Dict[str, Any] with executions: List.

get_workflow_execution(execution_id)

Get a workflow execution by ID.

Returns: Dict[str, Any] — full workflow execution record including final_output when available.

remove_workflow_execution(execution_id)

Remove a workflow execution record.

Returns: Dict[str, Any].

signal_workflow_execution(execution_id, payload)

Send a signal to a running workflow execution (e.g., to advance state or provide input).

Returns: Dict[str, Any].

cancel_workflow_execution(execution_id)

Cancel a running workflow execution.

Returns: Dict[str, Any].

SEAL

attest_seal(payload)

Obtain a security attestation token for SEAL tool invocation.

Returns: SealAttestationResponse with security_token: str.

invoke_seal(payload)

Invoke a SEAL tool with an attested payload.

Returns: Dict[str, Any] — tool-specific response.

list_seal_tools(security_context?)

List available SEAL tools, optionally filtered by security context.

Returns: SealToolsResponse with protocol, attestation_endpoint, invoke_endpoint, security_context, and tools: List[Any].

Dispatch Gateway

dispatch_gateway(payload)

Send a raw dispatch gateway request. Used by custom bootstrap scripts.

Returns: Dict[str, Any].

Stimulus

list_stimuli()

List all stimulus events.

Returns: Dict[str, Any] with stimuli: List.

get_stimulus(stimulus_id)

Get a stimulus event by ID.

Returns: Dict[str, Any].

ingest_stimulus(payload)

Ingest a stimulus event into the stimulus-response pipeline.

Returns: Dict[str, Any].

send_webhook(source, payload)

Send a webhook-style stimulus from an external source.

Returns: Dict[str, Any].

Volume Management

create_volume(payload)

Create a new storage volume.

vol = await client.create_volume({"name": "my-dataset"})
print(vol["id"])

Returns: Dict[str, Any] with volume details.

list_volumes()

List all volumes owned by the caller's tenant.

Returns: Dict[str, Any] with volumes: List.

get_volume(volume_id)

Get volume details by ID.

Returns: Dict[str, Any].

rename_volume(volume_id, payload)

Rename a volume.

Returns: Dict[str, Any].

delete_volume(volume_id)

Delete a volume and all its contents.

Returns: Dict[str, Any].

get_volume_quota()

Get the caller's storage quota and current usage.

Returns: Dict[str, Any] with quota limits and usage.

list_files(volume_id, path?)

List files and directories in a volume at the given path.

Returns: Dict[str, Any] with files: List.

download_file(volume_id, path)

Download a file from a volume.

Returns: bytes — raw file content.

upload_file(volume_id, file, path?)

Upload a file to a volume.

Returns: Dict[str, Any].

mkdir(volume_id, path)

Create a directory in a volume.

Returns: Dict[str, Any].

move_path(volume_id, payload)

Move or rename a file or directory within a volume.

Parameters:

Returns: Dict[str, Any].

delete_path(volume_id, path)

Delete a file or directory from a volume.

Returns: Dict[str, Any].

Credential Management

list_credentials()

List all stored credentials for the caller's tenant.

Returns: Dict[str, Any] with credentials: List.

get_credential(credential_id)

Get a credential by ID (metadata only; secrets are not returned).

Returns: Dict[str, Any].

store_api_key(payload)

Store an API key credential.

cred = await client.store_api_key({
    "name": "github-token",
    "provider": "github",
    "api_key": "ghp_xxxx",
})

Returns: Dict[str, Any] with credential ID.

revoke_credential(credential_id)

Revoke (delete) a credential.

Returns: Dict[str, Any].

rotate_credential(credential_id)

Rotate a credential's secret value.

Returns: Dict[str, Any].

oauth_initiate(payload)

Initiate an OAuth credential flow. Returns a URL for the user to authorize.

Returns: Dict[str, Any] with authorization URL.

oauth_callback(params)

Handle OAuth callback after user authorization.

Returns: Dict[str, Any].

device_poll(payload)

Poll for device authorization flow completion.

Returns: Dict[str, Any].

list_grants(credential_id)

List agent grants for a credential.

Returns: Dict[str, Any] with grants: List.

add_grant(credential_id, payload)

Grant an agent access to a credential.

Returns: Dict[str, Any].

revoke_grant(credential_id, grant_id)

Revoke an agent's access to a credential.

Returns: Dict[str, Any].

Secrets Management

list_secrets()

List all secrets in the caller's tenant namespace.

Returns: Dict[str, Any] with secrets: List.

get_secret(path)

Read a secret by its path.

Returns: Dict[str, Any] with secret data.

write_secret(path, payload)

Write or update a secret at the given path.

Returns: Dict[str, Any].

delete_secret(path)

Delete a secret.

Returns: Dict[str, Any].

API Key Management

list_api_keys()

List all API keys for the caller.

Returns: Dict[str, Any] with api_keys: List.

create_api_key(payload)

Create a new API key. The raw key value is only returned once at creation time.

key = await client.create_api_key({"name": "ci-pipeline"})
print(key["api_key"])  # Save this — it won't be shown again

Returns: Dict[str, Any] with id and api_key.

revoke_api_key(key_id)

Revoke an API key.

Returns: Dict[str, Any].

Colony Management

list_members()

List all members of the caller's colony (tenant team).

Returns: Dict[str, Any] with members: List.

invite_member(payload)

Invite a new member to the colony.

Returns: Dict[str, Any].

remove_member(user_id)

Remove a member from the colony.

Returns: Dict[str, Any].

update_role(payload)

Update a colony member's role.

Returns: Dict[str, Any].

get_saml_config()

Get the colony's SAML IdP configuration.

Returns: Dict[str, Any].

set_saml_config(payload)

Set or update the colony's SAML IdP configuration for SSO.

Returns: Dict[str, Any].

get_subscription()

Get the colony's subscription details.

Returns: Dict[str, Any].

Billing

All billing methods raise an AegisApiError with status 501 if STRIPE_SECRET_KEY is not configured on the orchestrator.

list_prices()

List all available pricing tiers with their Stripe price IDs.

resp = await client.list_prices()
for tier in resp.tiers:
    print(tier.tier, tier.monthly.amount if tier.monthly else None)

Returns: PricingResponse with tiers: List[TierPricing].

create_checkout_session(price_id, seat_price_id=None, seats=None)

Create a Stripe Checkout Session URL for subscribing to a plan.

resp = await client.create_checkout_session(
    price_id="price_xxx",
    seat_price_id="price_seat_xxx",
    seats=5,
)
# Redirect user to resp["url"]

Parameters:

Returns: Dict[str, Any] with url: str.

create_portal_session()

Create a Stripe Customer Portal session URL for managing payment methods and invoices.

resp = await client.create_portal_session()
# Redirect user to resp["url"]

Returns: Dict[str, Any] with url: str.

get_subscription_billing()

Get the current user's subscription billing details from Stripe.

sub = await client.get_subscription_billing()
print(sub["tier"], sub["status"])

Returns: Dict[str, Any] with tier, status, current_period_end, cancel_at_period_end, stripe_customer_id.

get_invoices()

List invoices for the current user.

resp = await client.get_invoices()
for inv in resp["invoices"]:
    print(inv["id"], inv["amount"], inv["status"])

Returns: Dict[str, Any] with invoices: List[Dict] containing id, amount, currency, status, created, pdf_url.

Cluster & Infrastructure

get_cluster_status()

Get the overall cluster status.

Returns: Dict[str, Any].

get_cluster_nodes()

List all nodes in the cluster with their status and capabilities.

Returns: Dict[str, Any] with nodes: List.

Swarm Coordination

list_swarms()

List all active swarms (multi-agent coordination groups).

Returns: Dict[str, Any] with swarms: List.

get_swarm(swarm_id)

Get details of a specific swarm including member agents and coordination state.

Returns: Dict[str, Any].

Observability

get_dashboard_summary()

Get an aggregated observability dashboard summary (execution counts, error rates, active agents).

Returns: Dict[str, Any].

list_security_incidents()

List recent security incidents.

Returns: Dict[str, Any] with incidents: List.

list_storage_violations()

List storage policy violations.

Returns: Dict[str, Any] with violations: List.

Cortex

list_cortex_patterns()

List learned execution patterns from the Cortex memory system.

Returns: Dict[str, Any] with patterns: List.

get_cortex_skills()

Get the Cortex skill inventory — a summary of capabilities the system has learned.

Returns: Dict[str, Any].

get_cortex_metrics()

Get Cortex performance metrics (memory utilization, recall accuracy, decay stats).

Returns: Dict[str, Any].

User

get_user_rate_limit_usage()

Get the current authenticated user's rate limit usage.

Returns: Dict[str, Any] with usage records.

Workflow Logs

get_workflow_execution_logs(execution_id, limit?, offset?)

Fetch paginated workflow execution logs.

Returns: WorkflowExecutionLogs with execution_id, events: List[Any], count, limit, offset.

stream_workflow_execution_logs(execution_id)

Stream workflow execution logs in real time via SSE.

Returns: AsyncGenerator[ExecutionEvent, None].

Admin: Tenant Management

create_tenant(slug, display_name, tier?)

Create a new tenant. Defaults to "enterprise" tier.

Returns: Tenant with slug, display_name, status, tier, keycloak_realm, openbao_namespace, quotas, created_at, updated_at.

list_tenants()

List all tenants.

Returns: List[Tenant].

suspend_tenant(slug) / delete_tenant(slug)

Suspend or delete a tenant by slug.

Returns: Dict[str, str].

Admin: Rate Limits

list_rate_limit_overrides(tenant_id?, user_id?)

List rate limit overrides, optionally filtered by tenant or user.

Returns: List[RateLimitOverride] with id, resource_type, bucket, limit_value, tenant_id, user_id, burst_value, created_at, updated_at.

create_rate_limit_override(payload)

Create a rate limit override.

Returns: RateLimitOverride.

delete_rate_limit_override(override_id)

Delete a rate limit override.

Returns: Dict[str, str].

get_rate_limit_usage(scope_type, scope_id)

Get rate limit usage records for a given scope.

Returns: List[UsageRecord] with scope_type, scope_id, resource_type, bucket, window_start, counter.

Health

health_live() / health_ready()

Liveness and readiness probes.

Returns: Dict[str, str].

Lifecycle

aclose()

Close the underlying HTTP client and release resources. Called automatically when using the async with context manager.

# Manual lifecycle
client = AegisClient(base_url="...", bearer_token="...")
try:
    await client.list_agents()
finally:
    await client.aclose()

SEAL Protocol

The aegis.seal subpackage provides low-level cryptographic primitives and client utilities for workloads that need to self-attest and call SEAL tools directly — without going through AegisClient. Useful for custom bootstrap scripts and server-side envelope verification.

Import paths

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

SEALClient

A high-level client that handles key generation, attestation, and tool invocation in one object.

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

Constructor:

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

Methods:

AttestationResult:

Ed25519Key

Low-level Ed25519 keypair for manual envelope construction.

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

Envelope utilities

For server-side workloads (e.g. a gateway plugin) that need to construct or verify SEAL envelopes manually.

create_seal_envelope(security_token, mcp_payload, private_key)

Construct a signed SEAL envelope from a security token, an MCP payload dict, and an Ed25519Key.

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

verify_seal_envelope(envelope, public_key_bytes, max_age_seconds?)

Server-side verification. Raises on invalid signature or expired timestamp.

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

create_canonical_message(security_token, payload, timestamp_unix)

Low-level helper that produces the canonical bytes that are signed.

from aegis.seal.envelope import create_canonical_message

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

Agent Manifests

The aegis.manifest module provides Pydantic models and a fluent builder for constructing agent manifest YAML files programmatically.

from aegis.manifest import AgentManifestBuilder, AgentManifest, ImagePullPolicy

AgentManifestBuilder

Fluent builder — the recommended API for creating manifests in code.

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")

Constructor:

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

Methods:

AgentManifest

The root manifest model. Serialises to the AEGIS agent manifest YAML format.

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

Class methods:

ImagePullPolicy

from aegis.manifest import ImagePullPolicy

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

Key manifest types

For the full field reference, see the Agent Manifest Reference.

Dispatch Protocol Types

When writing a custom bootstrap script (spec.advanced.bootstrap_path in the agent manifest), import these types to build and parse the protocol payloads in a type-safe way.

The default bootstrap script injected by the orchestrator does not import this module — it implements the same wire format using stdlib only. These classes are for custom bootstrap authors.

GenerateMessage

Sent by bootstrap to start an inner-loop iteration:

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"},
)

DispatchResultMessage

Sent by bootstrap after executing a dispatched command:

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,
)

FinalMessage (parse from orchestrator response)

The orchestrator responds with a FinalMessage when the inner loop completes:

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

DispatchMessage (parse from orchestrator response)

The orchestrator responds with a DispatchMessage when it wants bootstrap to run a command:

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

Environment Variables (inside agent containers)

The orchestrator injects these into every agent container:

See Also

• TypeScript SDK Reference
• Custom Runtime Agents — using a custom bootstrap.py
• REST API Reference — raw HTTP endpoints