POE Python Client SDK (poe-py / poe-client)

The official Python client SDK for the POE (Predictive Operational Edge) intelligence daemon.

Designed for downstream systems engineers, telemetry monitoring pipelines, robotics scripts, and automated control loops.

Features

• Type-Safe Primitive Extraction: Direct extraction of continuous Gauge, binary Gate, categorical Route, semantic Noul, discrete Choice, and ordinal Score heads.
• Transparent Backpressure Resilience: Automatic exponential backoff with full jitter on HTTP 429 OVERLOADED and 503 DEADLINE_EXCEEDED respecting Retry-After headers.
• Hardware Fallback Awareness: Immediate detection of neural graph fallbacks (response.is_fallback, fallback_triggered).
• Synchronous & Asynchronous: High-performance sync (PoeClient) and async (AsyncPoeClient) clients built on top of httpx.

Installation

pip install poe-edge

Quickstart

from poe_client import PoeClient
from poe_client.exceptions import PoeOverloadError

client = PoeClient(
    base_url="http://localhost:8080",
    max_retries=3,
    base_backoff_ms=50
)

try:
    response = client.evaluate(
        domain_id="drone_flight",
        context_log="Winds at 40 knots. Vision sensor degraded. GPS altitude spiking."
    )

    # 1. Dictionary-like or attribute access
    altitude = response.gauges["altitude"].predicted_value
    sigma = response.gauges["altitude"].uncertainty_sigma
    is_breached = response.gates["geofence_breach"].decision
    maneuver = response.routes["maneuver"].selected

    # 2. Direct helper access
    alt = response.gauge("altitude").predicted_value
    breached = response.gate("geofence_breach")  # returns bool
    man = response.route("maneuver")            # returns str

    # 3. Fallback inspection
    if response.gauges["altitude"].fallback_triggered:
        print("Warning: Altitude gauge executed safe fallback envelope!")
    if response.is_fallback:
        print("Degraded heads:", response.fallback_heads)

    print(f"Maneuver: {maneuver}, Altitude: {altitude}m (±{sigma})")

except PoeOverloadError as e:
    print(f"Edge daemon saturated after {e.retries} retries: {e}")

Async Usage

import asyncio
from poe_client import AsyncPoeClient

async def main():
    async with AsyncPoeClient("http://localhost:8080") as client:
        response = await client.evaluate("drone_flight", "Clear skies, nominal telemetry")
        print("Altitude:", response.gauge("altitude").predicted_value)

asyncio.run(main())