Python SDK quick guide

Emit, consume, and run the control round-trip from Python, over the pydantic v2 types.

agenteventprotocol-sdk provides emit/consume/control helpers over the schema-generated pydantic v2 types (README). File paths below are relative to the python-sdk repository root.

It's a local, unpublished package: Private :: Do Not Upload in pyproject.toml until the v0.1 tag, so install it from a clone, not from PyPI.

This guide covers the three tasks the smoke test exercises: emit, consume, control round-trip.

Setup

Python ≥ 3.10; the only dependency is pydantic>=2 (pyproject.toml). From a clone's root, run against the package directly (the smoke test does this via sys.path.insert, tests/smoke.py) or install it editable: pip install -e .. Verify with bash tests/run-smoke.sh (tests/run-smoke.sh): it starts the vendored relay test fixture on an ephemeral port and runs tests/smoke.py, the source of every snippet below, via uv run.

Emit an event

Emitter builds valid envelopes for one producer identity. .session() returns a SessionEmitter that owns that session's (epoch, seq) counter (aep_sdk/emit.py):

python
from aep_sdk import Emitter, http_sink

session = Emitter("my-agent", "host-1", http_sink("http://127.0.0.1:8787"), epoch=1) \
    .session("s_001")
session.emit("session.started", {"client": {"name": "my-agent"}})

epoch is a constructor keyword you own; bump it whenever the process restarts without durable seq state, per AEP-0001 §7. http_sink() POSTs each Event to the relay's ingest endpoint. jsonl_sink(path) is the append-only file binding (aep_sdk/emit.py's http_sink/jsonl_sink). Built Events validate against the generated pydantic models; the smoke test does this explicitly:

python
from aep_sdk import AepEvent, ToolCompletedData

AepEvent.model_validate(built_event)
ToolCompletedData.model_validate(built_event["data"])

(aep_sdk.gen.aep_types is generated from schemas/, re-exported at the top of aep_sdk/__init__.py; never edit it directly.)

Consume / subscribe

subscribe() opens an SSE connection in a daemon thread, dedupes by id, and tracks (session, epoch, seq) resume positions (aep_sdk/consume.py; aep_sdk/aio.py mirrors the same surface for asyncio):

python
from aep_sdk import subscribe

sub = subscribe("http://127.0.0.1:8787", print, filter={"session": "s_001"})
# later:
positions = sub.positions()   # persist and pass back as from_=... to resume
sub.close()

The filter argument is the attr-match dialect. See Write a consumer for why persisting positions() and deduping by id are both required, not just one or the other.

Control round-trip

Control is transport-agnostic in this SDK: the standard library has no WebSocket client, and control commands require an authenticated duplex binding (AEP-0004 §4). ControlSender takes a transport_send callable you provide; you feed inbound Events to on_event() (aep_sdk/control.py). A bundled WS transport may ship later as an optional dependency; it is not in this package today. See README.

python
from aep_sdk import ControlSender, NackError

sender = ControlSender(my_ws_send, agent="phone", host="host-1")

try:
    ack = sender.send(
        "control.attention.respond", "s_001",
        subject=request_id, cause=request_id,
        data={"answer": {"option": "allow"}},
        ack_window_ms=5000, retries=1,
    )
    print("accepted", ack["id"])
except NackError as e:
    print(e.reason, e.detail)

Feed every inbound Event on your transport to sender.on_event(ev). It matches acks by cause (the command id) and resolves or raises the pending send() call. Retries reuse the same command id (AEP-0004 §2, item 3), so a retried send never double-executes on the target.

On the receiving side, ControlTarget is the target-side conformance helper: dedupe on (source, id), one ack decision per command with the recorded ack re-emitted byte-identically on a retransmit, unsupported nack for undeclared types (aep_sdk/control.py):

python
from aep_sdk import ControlTarget

target = ControlTarget(session_emitter, accepts=["control.attention.respond"])
target.handle(incoming_command, execute=lambda cmd: ...)  # acks, then runs execute() once

The full round-trip, including the dedupe-on-retry and ack-window-timeout paths, is exercised end-to-end in tests/smoke.py.

The three helper areas

graph TD
    E["Emitter / SessionEmitter\n(emit.py)"] --> S[http_sink or jsonl_sink]
    Sub["subscribe()\n(consume.py)"] --> SSE[SSE stream, daemon thread]
    CS["ControlSender / ControlTarget\n(control.py)"] --> WS[caller-supplied duplex]
    S --> R[relay]
    SSE --> R
    WS --> R

See also

On this page