edon 2.0.3

Python SDK for the EDON CAV Engine — adaptive state engine for physical AI and autonomous systems

EDON Python SDK

EDON is a governance and adaptive state engine for physical AI and autonomous systems. It sits between your AI agent and the world — logging every decision, enforcing policies, and adapting behavior based on real-time physiological or environmental state.

This SDK provides a Python client for the EDON CAV Engine (Cognitive Autonomy Vector): reads wearable sensor data and outputs a normalized autonomy state used to scale robot or agent behavior in real time.

Get your API token at edoncore.com.

---

Installation

pip install edon

# With gRPC transport support
pip install edon[grpc]

---

Quick Start

import os
from edon import EdonClient

client = EdonClient(
    base_url=os.environ["EDON_BASE_URL"],   # https://edon-gateway.fly.dev
    api_key=os.environ["EDON_API_TOKEN"],
)

# 4-second sensor window (240 samples at 60 Hz)
window = {
    "EDA":   [0.1] * 240,   # Electrodermal activity (µS)
    "TEMP":  [36.5] * 240,  # Skin temperature (°C)
    "BVP":   [0.5] * 240,   # Blood volume pulse
    "ACC_x": [0.0] * 240,   # Accelerometer X
    "ACC_y": [0.0] * 240,   # Accelerometer Y
    "ACC_z": [1.0] * 240,   # Accelerometer Z
    "temp_c":     22.0,     # Ambient temperature (°C)
    "humidity":   50.0,     # Ambient humidity (%)
    "aqi":        35,       # Air quality index
    "local_hour": 14,       # Local hour 0–23
}

result = client.cav(window)
print(result["state"])               # "balanced"
print(result["cav_smooth"])          # 0.54
print(result["parts"]["p_stress"])   # 0.31

---

Robot / Agent Control Loop

from edon import EdonClient
import time

client = EdonClient()

while True:
    window = collect_sensor_window()  # your wearable integration (4s of data)
    result = client.cav(window)
    state  = result["state"]

    # Scale robot behavior based on operator physiological state
    scales = {
        "restorative": dict(speed=0.7, torque=0.7, safety=0.95),
        "balanced":    dict(speed=1.0, torque=1.0, safety=0.85),
        "focus":       dict(speed=1.2, torque=1.1, safety=0.80),
        "overload":    dict(speed=0.4, torque=0.4, safety=1.00),
    }.get(state, dict(speed=0.8, torque=0.8, safety=0.90))

    apply_scales(scales)
    time.sleep(4.0)

---

Batch Processing

windows = [window1, window2, window3]
results = client.cav_batch(windows)
for r in results:
    print(r["state"], r["cav_smooth"])

---

gRPC Transport

from edon import EdonClient, TransportType

client = EdonClient(
    transport=TransportType.GRPC,
    grpc_host="localhost",
    grpc_port=50051,
)

result = client.cav(window)
print(result["controls"]["speed"])

# Server-push streaming
for update in client.stream(window):
    print(update["state"], update["cav_smooth"])

client.close()

---

Environment Variables

Variable	Description
EDON_BASE_URL	Your gateway URL. Use https://edon-gateway.fly.dev for the hosted service, or http://127.0.0.1:8000 for a local deployment.
EDON_API_TOKEN	Your API token from edoncore.com. Sent as X-EDON-TOKEN on every request.

---

API Reference

Method	Description
cav(window)	Compute CAV from a sensor window
cav_batch(windows)	Batch CAV (up to 5 windows, REST only)
classify(window)	Returns just the state string
stream(window)	Server-push streaming (gRPC only)
health()	Health check
close()	Close gRPC channel

---

Gateway API (Governance)

The Gateway enforces policies, logs every agent decision to a tamper-evident audit trail, and provides multi-tenant governance. Authenticate with X-EDON-TOKEN:

import requests
from datetime import datetime, timezone

GATEWAY = "https://edon-gateway.fly.dev"
TOKEN   = "your-edon-token"

headers = {"X-EDON-TOKEN": TOKEN, "Content-Type": "application/json"}

# Submit an agent action for governance evaluation
resp = requests.post(f"{GATEWAY}/v1/action", headers=headers, json={
    "agent_id":      "my-agent-001",
    "action_type":   "email.send",           # format: "tool.operation"
    "action_payload": {
        "recipients": ["[e-mail]"],
        "subject":    "Weekly report",
    },
    "timestamp": datetime.now(timezone.utc).isoformat(),
    "context":   {"user_confirmed": False},
})

result = resp.json()
print(result["decision"])        # "ALLOW", "BLOCK", "DEGRADE", "HUMAN_REQUIRED"
print(result["decision_reason"]) # human-readable explanation
print(result["action_id"])       # use this to look up the audit record

# Pull decision timeseries (last 7 days)
stats = requests.get(f"{GATEWAY}/timeseries?days=7", headers=headers).json()

# Top block reasons
reasons = requests.get(f"{GATEWAY}/block-reasons", headers=headers).json()

Get your API token at edoncore.com.

---

License

MIT