emit_delta

Report an event to Aether without ever breaking your agent.

Sends one Delta (an event) to Aether and returns an acknowledgement. Aether stores the Delta; it does not decide anything inline. Decisions come later, when the orchestrator requests a Trigger Plan.

emit_delta fails open: on any failure it returns None instead of raising, so reporting to Aether can never stop your agent.

​

When to use it

Use emit_delta for routine reporting, where a lost Delta must not interrupt the work.

Use emit_delta_strict when you need to know why a Delta was rejected.

​

Signature

async def emit_delta(self, delta: DeltaCreate) -> DeltaAck | None

Argument	Type	Description
delta	DeltaCreate	The event to report.

​

DeltaCreate

Field	Type	Required	Description
event	str	Yes	Event name, e.g. "scan.completed".
workspace_revision	str	Yes	Revision of the workspace the event happened on.
payload	dict	No	Extra data. Must be JSON-serializable. Default: {}.
occurred_at	str	No	ISO 8601 timestamp. Default: now (UTC).

The caller’s identity comes from the API key. There is no system id in DeltaCreate.

​

Returns

DeltaAck, or None on failure.

Field	Type	Description
id	str	Id of the stored Delta.
received_at	str	When Aether received it.

​

Returns None when

• Aether is unreachable or does not answer in time.
• The API key is missing, invalid, revoked or expired (401).
• The environment is inactive (409 environment_inactive).
• The payload does not match the event’s contract (422 contract_violation).
• The rate limit is exceeded (429).
• The payload is not JSON-serializable.
• Aether returns a malformed response.

​

Example

import asyncio

from aether_sdk import AetherClient, DeltaCreate


async def main() -> None:
    client = AetherClient()

    ack = await client.emit_delta(
        DeltaCreate(
            event="scan.completed",
            workspace_revision="abc123",
            payload={"verification_status": "verified"},
        )
    )
    if ack is None:
        return  # Aether unavailable or Delta rejected: carry on

    print(ack.id, ack.received_at)


asyncio.run(main())

Building the payload from your own dataclasses? Pass it through to_jsonable first: payload=to_jsonable(my_result).