emit_delta - AiByDNA Help Center

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