Aegis Orchestrator
Reference

Python SDK

AegisClient, ManifestBuilder, and the bootstrap.py Dispatch Protocol types for Python agent authors.

Python SDK

Install the AEGIS Python SDK:

pip install aegis-sdk

The SDK consists of two distinct layers:

  • AegisClient — management plane: deploy agents, start executions, query status (runs on your machine or in CI)
  • bootstrap.py / Dispatch Protocol types — agent-side: the wire format exchanged between a custom bootstrap script and the orchestrator's /v1/dispatch-gateway

AegisClient

from aegis import AegisClient

client = AegisClient(
    base_url="https://your-aegis-node",
    api_key="your-keycloak-jwt",     # Optional; alternatively set AEGIS_API_KEY env var
)

Methods

deploy_agent(manifest)

Deploy an agent to the orchestrator.

from aegis import AegisClient, AgentManifest

async with AegisClient("https://your-aegis-node", api_key=TOKEN) as client:
    response = await client.deploy_agent(manifest)
    print(response.agent_id)   # "agt-uuid"

Parameters:

NameTypeDescription
manifestAgentManifestValidated agent manifest object

Returns: DeploymentResponse with agent_id: str.


execute_task(agent_id, task_input)

Start an agent execution.

from aegis.types import TaskInput

output = await client.execute_task(
    agent_id="agt-uuid",
    task_input=TaskInput(prompt="Write a primality check in Python"),
)
print(output.result)

Parameters:

NameTypeDescription
agent_idstrUUID of a deployed agent
task_inputTaskInputTask input including prompt and optional context

Returns: TaskOutput with result: str and logs: list[str].


get_agent_status(agent_id)

status = await client.get_agent_status("agt-uuid")
print(status.state)   # "deployed" | "paused" | "archived"

Returns: AgentStatus with state: str, agent_id: str, updated_at: str.


terminate_agent(agent_id)

await client.terminate_agent("agt-uuid")

Terminates and archives the agent. Running executions are cancelled.


AgentManifest

Construct an agent manifest programmatically:

from aegis import AgentManifest

manifest = AgentManifest(
    name="code-reviewer",
    version="1.0.0",
    runtime={
        "language": "python",
        "version": "3.11",
        "isolation": "docker",
    },
    task={
        "instruction": "Review pull requests and output structured feedback.",
    },
    security={
        "network": {
            "mode": "allow",
            "allowlist": ["api.github.com"],
        },
        "filesystem": {
            "read": ["/workspace"],
            "write": ["/workspace"],
        },
        "resources": {
            "cpu": 1000,
            "memory": "1Gi",
            "timeout": "300s",
        },
    },
)

Alternatively, load from a YAML file:

import yaml
from aegis import AgentManifest

with open("agent.yaml") as f:
    manifest = AgentManifest.model_validate(yaml.safe_load(f))

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"},
)
FieldTypeDescription
execution_idstrUUID from AEGIS_EXECUTION_ID env var
iteration_numberint1-indexed iteration counter
model_aliasstrLLM alias from AEGIS_MODEL_ALIAS env var
promptstrFully-rendered prompt for this iteration
messageslist[dict]Prior conversation history for continuation
agent_idstrOptional — from AEGIS_AGENT_ID env var

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,
)
FieldTypeDescription
execution_idstrSame as the originating GenerateMessage
dispatch_idstrUUID echoed from the DispatchMessage
exit_codeintProcess exit code. -1 for bootstrap-level errors.
stdoutstrCaptured stdout, tail-trimmed if truncated=True
stderrstrCaptured stderr
duration_msintWall-clock execution time
truncatedboolTrue when combined output exceeded max_output_bytes

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:

VariableDescription
AEGIS_AGENT_IDUUID of the deployed agent
AEGIS_EXECUTION_IDUUID of this execution instance
AEGIS_MODEL_ALIASLLM alias to use (e.g. "default", "fast")
AEGIS_ORCHESTRATOR_URLInternal URL for bootstrap.py callbacks

See Also

On this page