edon 2.0.1

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 decisions, enforcing policies, and adapting behavior based on physiological or environmental state.

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

---

Installation

pip install edon

# With gRPC transport support
pip install edon[grpc]

---

Quick Start — CAV Engine

The CAV Engine takes 4-second windows of wearable sensor data and outputs a continuous autonomy state (restorative → balanced → focus → overload) and a CAV score (0–1).

import os
from edon import EdonClient

# Set your endpoint and token
client = EdonClient(
    base_url=os.environ["EDON_BASE_URL"],   # e.g. 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
    "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
    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"])   # control scales included in gRPC response

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

client.close()

---

Environment Variables

Variable	Default	Description
EDON_BASE_URL	http://127.0.0.1:8000	Gateway base URL
EDON_API_TOKEN	—	API token (X-EDON-TOKEN)

---

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 provides policy enforcement, audit logging, and multi-tenant governance for AI agents. Authenticate with X-EDON-TOKEN:

import requests

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

headers = {"X-EDON-TOKEN": TOKEN}

# Evaluate an agent action against your active policy
resp = requests.post(f"{GATEWAY}/v1/action/evaluate", headers=headers, json={
    "agent_id": "my-agent",
    "action":   "send_email",
    "context":  {"recipient": "[e-mail]"},
})
print(resp.json())   # {"verdict": "ALLOW", "policy": "default", ...}

# List recent decisions
decisions = requests.get(f"{GATEWAY}/v1/decisions", headers=headers).json()

Get your API token at edoncore.com.

---

License

MIT