EnergyChain is the Python developer infrastructure through which applications can read, normalize, remember, analyze and explain energy system observations.

The problem it solves

Energy APIs are fragmented across devices and providers. An application may need to understand watts versus watt-hours, battery state of charge, observation timestamps, equipment IDs, data trustworthiness, and which actions a user is permitted to take. EnergyChain supplies common developer contracts instead of forcing every app to reconstruct them.

What you can build

• Battery assistants: ask how much backup time remains under specific loads.
• Solar intelligence: combine weather irradiance and explicit PV assumptions to estimate generation.
• Energy analytics: summarize consumption, costs, statistical outliers and baseline forecasts.
• Fleet applications: observe multiple authorized sites without silently discarding missing readings.
• Advisory automation: construct repeatable, permission-checked workflows and planning tools.

Choose your path

1. New to EnergyChain? Install and complete the first ten-minute project.

2. Integrating data? Map CSV records, Home Assistant entities, or a custom connector.

3. Building AI apps? Study the tools, structured memory, and bounded agent runtime.

4. Operating an application? Start with permissions, data freshness, secrets and failure modes.

Five design principles

• Measured is not estimated. A source timestamp and quality label accompany readings.
• Tools are contracts. Deterministic calculations should not be delegated to an LLM.
• Memory is structured. Time-series records belong in data stores, not prompt strings.
• Authorization precedes execution. LLM output never grants permissions.
• Simulator first. Read-only integration precedes physical action.

Scope boundaries

This SDK is not a national grid controller, SCADA replacement or a packaged end-user dashboard. It provides reusable developer infrastructure for applications built around energy data and intelligent decision support. Production services must supply their own authenticated entry points, tenant isolation, operational monitoring and deployment infrastructure.

Requirements

Python 3.10+ and a Python package installer. The core SDK uses only Python standard-library runtime dependencies; you do not need a paid model account to run the simulator and deterministic tools.

Install the official release

python -m pip install --upgrade energychain
# Or pin the current alpha release:
python -m pip install energychain==0.3.0

The official PyPI package is energychain; v0.3.0 is published. For isolated development use a virtual environment: python -m venv .venv.

Optional packages

python -m pip install 'energychain[models]==0.3.0'
python -m pip install 'energychain[documents]==0.3.0'
# Both optional feature sets:
python -m pip install 'energychain[all]==0.3.0'

[models] includes jsonschema for structured model responses. [documents] adds PDF, DOCX and XLSX parsers. The core package remains dependency-light; provider credentials are supplied by each developer.

Verify the installation

python -c "import energychain; print(energychain.__version__)"
# Expected: 0.3.0

from energychain import EnergyChain, ModelGateway, EnergyKnowledgeBase
chain = EnergyChain.demo()
user = chain.principal("developer", allowed_sites={"demo-site"})
print([a.name for a in chain.tools.invoke("discover_assets", user, "demo-site")])

The demonstration uses simulated energy data. The local principal() convenience method is not production authentication.

Develop from source (optional)

git clone https://github.com/energychainai/energychain.git
cd energychain
python -m pip install -e '.[dev]'
python -m pytest -q

Only use the official organization repository. You do not need a development branch or private GitHub access to install the published PyPI package.

Build an energy assistant without hardware

Start from the deterministic hybrid-energy simulator. It supplies four assets representing solar, storage, office loads and backup generation.

from energychain import EnergyChain

chain = EnergyChain.demo()
developer = chain.principal(
    "quickstart-dev",
    allowed_sites={"demo-site"},
)

assets = chain.tools.invoke(
    "discover_assets", developer, "demo-site"
)
print([(asset.id, asset.kind.value) for asset in assets])

battery = chain.tools.invoke(
    "get_battery_state",
    developer,
    "demo-site",
    arguments={"asset_id": "demo-battery"},
)
print("Battery SOC:", battery["state_of_charge"].value)

estimate = chain.tools.invoke(
    "estimate_battery_runtime",
    developer,
    "demo-site",
    arguments={
        "battery_id": "demo-battery",
        "load_w": 900,
        "reserve_pct": 20,
    },
)
print("Estimated hours:", estimate["hours"])
print("Data quality:", estimate["quality"])

Understand the result

The runtime tool uses battery capacity, charge, reserve, an assumed efficiency and a fixed load. It returns an estimate, not a promise that equipment can run for that duration. In the simulator, source readings are labelled simulated.

Try a second tool

cost = chain.tools.invoke(
    "estimate_energy_cost",
    developer,
    "demo-site",
    arguments={
        "energy_wh": 2000,
        "price_per_kwh": 100,
    },
)
print(cost["cost"])  # 200.0 in the chosen currency

What just happened?

1. The connector registry located the registered site and asset.

2. The tool registry enforced site and permission rules.

3. The energy calculator applied unit-correct arithmetic.

4. The memory and tracer could record observations and execution events.

Production warning

chain.principal(...) is provided for local demonstrations only. Do not let untrusted callers self-assign site access or privileges in a hosted application.

Foundations → Integration → Intelligence → Operations

Use the documentation as a layered curriculum rather than a random reference catalog.

Foundation

Complete Installation, First application, Energy data model, and Tools API. You should be able to explain which objects represent a site, asset, reading and tariff.

Build with data

Complete Connector framework, CSV, Home Assistant, and Weather and solar. Focus on provenance, validation and freshness rather than integration count.

Build intelligence

Study Energy Memory, Agents and models, Energy Intelligence, Fleets, Workflows, and Skills. Keep statistics and equipment constraints deterministic.

Deploy responsibly

Read Security and actions, Testing and evaluation, Architecture, Operations, and Troubleshooting. An application based on a tested SDK is not automatically a field-certified energy control system.

Documentation conventions

• Working denotes code implemented on the published v0.3.0 SDK release.
• Experimental denotes a narrow API without production guarantees.
• Planned denotes architecture or roadmap only.
• Never supported by v0.3.0 denotes capabilities intentionally absent, such as unrestricted physical dispatch.

Python examples show public SDK imports where possible and modules where necessary. Time-series examples use timezone-aware ISO 8601 timestamps.

Reference architecture

Your application (identity, routes, operator UX)
  └── EnergyChain SDK
      ├── Energy Core: sites, assets, topology, readings
      ├── ConnectorRegistry: CSV / simulator / HA adapter
      ├── ToolRegistry: site-aware permissions and callable tools
      ├── Memory: readings / events / notes / preferences / episodes
      ├── EnergyContextBuilder: bounded scoped retrieval
      ├── FleetManager: authorized multi-site read views
      ├── EnergyAgent + BoundedEnergyAgent
      ├── EnergySkillRegistry: reusable instructions
      ├── Intelligence: forecasts / economics / optimization
      ├── TelemetryIngestor + in-process event bus
      ├── EnergyWorkflow + SQLite checkpoints
      └── ActionService: prepare first, dispatch disabled

The 15-component inventory

| --- | --- |

Important separation

The SDK is embedded inside the developer's application. It is not an HTTP service, identity server or production worker fleet. The host application owns authentication, API boundaries, authorization issuance, secrets, uptime and tenancy.

Dependency direction

Connector observations flow through normalized energy objects. Tools enforce permission boundaries and invoke domain calculators. Agents consume tools through a restricted contract, never through arbitrary Python execution. Structured memory stores evidence; prompt context is a bounded derivative of that evidence.

Deployment profile

The recommended v0.3.0 deployment is read-only, single-trust-domain integration with a known data source and explicit failure handling. See Operations and Security and actions for a realistic production integration checklist.

The common language of energy

The public API includes EnergySite, EnergyAsset, EnergyReading, EnergyEvent, Tariff, EnergyPlan, Metric, AssetKind and Capability. The published v0.3.0 SDK also includes EnergyRelationship, EnergyTopology, EnergyConstraint, EnergyActionSpec and require_fresh.

from datetime import datetime, timezone
from energychain import (
    EnergySite, EnergyAsset, EnergyReading, AssetKind, Metric
)

site = EnergySite("home-01", "Home energy")
solar = EnergyAsset("roof-pv", "home-01", "Solar PV", AssetKind.SOLAR)
reading = EnergyReading(
    asset_id="roof-pv",
    metric=Metric.POWER_W,
    value=1850.0,
    timestamp=datetime.now(timezone.utc),
    source="verified-export",
    quality="measured",
)
print(reading.unit)  # W

Physical quantity ≠ unit

| --- | --- | --- |

Do not add unrelated power observations across generators, batteries and loads without defining a consistent boundary and sign convention.

Provenance and freshness

Every reading has a timestamp, a provider source and quality. Measured comes from a sensor or data export; estimated comes from an algorithm; simulated comes from test fixtures. A provider's last_updated does not mean your application received the value in real time.

Use require_fresh when decisions need a bounded data age. Treat stale or missing values as explicit failures instead of silently carrying forward the last good reading.

Topology and constraints

EnergyTopology supports declared relationships across registered assets. EnergyConstraint checks bounded numeric values. Neither is a substitute for a certified electrical power-flow solver.

Recommended contract

Store stable, provider-independent identifiers. Always specify your site's timezone and tariff currency in the host domain model. Keep conversion logic at connector boundaries and retain original source details for troubleshooting.

Connector contract

All connectors derive from EnergyConnector and expose a consistent interface:

• sites() -> list[EnergySite]
• assets(site_id) -> list[EnergyAsset]
• read(asset_id, metric) -> EnergyReading
• history(asset_id, metric, start, end) -> list[EnergyReading]

ConnectorRegistry rejects conflicting site and asset identifiers across registered sources and routes observations to their owning connector.

Register a connector

from energychain import EnergyChain, HybridSiteSimulator

sdk = EnergyChain()
sdk.add_connector("north", HybridSiteSimulator(site_id="north-site"))
sdk.add_connector("south", HybridSiteSimulator(site_id="south-site"))
print([site.id for site in sdk.connectors.sites()])

Connector design checklist

1. Define stable asset identities, typed metrics and canonical units.

2. Validate the provider's timestamp, entity ID, numeric value and status.

3. Never report a cached value as live without a freshness check.

4. Map errors into explicit failures; avoid returning plausible substitute readings.

5. Declare unsupported methods with UnsupportedCapability.

6. Never infer authorization from the fact that a device is reachable.

7. Require TLS, authenticated providers and secrets supplied by your host app.

8. Make credential rotation, rate limits and provider outages visible to operators.

Included adapters

| --- | --- | --- |

Extending the SDK

Subclass the connector, implement the four methods and add it through sdk.add_connector(). Do not import proprietary API credentials into the package or publish customer data as tests.

Integration maturity

The existence of the adapter class proves only software contract implementation. Vendor certification, manufacturer-specific accuracy and actual pilot validation require testing on real approved installations.

Prepare your data

The CSV adapter accepts required fields asset_id,metric,value,timestamp,source with an optional quality field.

asset_id,metric,value,timestamp,source,quality
office-load,power_w,1200,2026-10-26T10:00:00Z,meter-export,measured
office-load,power_w,900,2026-10-26T11:00:00Z,meter-export,measured

Load the export

from pathlib import Path
from energychain import (
    EnergyChain, CSVConnector, EnergySite,
    EnergyAsset, AssetKind
)

site = EnergySite("office", "Office")
load = EnergyAsset("office-load", "office", "Main load", AssetKind.LOAD)
connector = CSVConnector.from_csv(
    Path("readings.csv"),
    site=site,
    assets=[load],
)
sdk = EnergyChain()
sdk.add_connector("office-csv", connector)
user = sdk.principal("local-user", allowed_sites={"office"})

latest = sdk.tools.invoke(
    "get_asset_state", user, "office",
    arguments={"asset_id": "office-load", "metric": "power_w"},
)
print(latest.value, latest.timestamp, latest.quality)

Semantics

The latest row is the most recently recorded measurement. It is not proof of a live connection. CSV readings require offset-aware timestamps; invalid values or unknown metric names are rejected.

Production considerations

For untrusted data feeds, add upload validation and file size limits outside the SDK, identify the author of every export, reconcile duplicate records and establish file retention policies.

Next

Move to Tools API to retrieve history and derive consumption summaries.

Intended use

Use a Home Assistant installation to expose battery, solar or load sensor readings to a developer application through EnergyChain's read-only connector.

Example: map a load sensor

import os
from energychain import (
    EnergyChain, EnergySite, EnergyAsset,
    AssetKind, Metric, HAEntityMapping,
    HomeAssistantConnector
)

site = EnergySite("home", "My home")
load = EnergyAsset("load", "home", "House load", AssetKind.LOAD)

adapter = HomeAssistantConnector(
    site=site,
    mappings=[
        HAEntityMapping(
            asset=load,
            entity_id="sensor.house_power",
            metric=Metric.POWER_W,
            source_unit="W",
        )
    ],
    base_url="https://ha.example.org",
    token_provider=lambda: os.environ["HA_TOKEN"],
)
sdk = EnergyChain()
sdk.add_connector("home-assistant", adapter)

Read a mapped state

user = sdk.principal("local-user", allowed_sites={"home"})
reading = sdk.tools.invoke(
    "get_asset_state", user, "home",
    arguments={"asset_id": "load", "metric": "power_w"},
)
print(reading.value, reading.unit, reading.timestamp)

The underlying read-only endpoint is Home Assistant GET /api/states/{entity_id}. The adapter checks that the returned entity and source unit match the declared mapping, rejects nonnumeric values and converts kW to W where configured.

Important limitations

• Historical recorder queries are not implemented by this adapter.
• The adapter does not dispatch smart-device commands.
• HA_TOKEN must be managed by your application or secret manager.
• The SDK does not implement credential rotation, provider reconnection or tenant-aware secret storage.
• A changed sensor entity, unavailable state or stale timestamp must cause an explicit application-level response.

Operational recommendation

Begin with one site and one read-only metric. Compare results with the Home Assistant dashboard and, where relevant, equipment measurements before declaring accuracy.

Weather is an input to energy intelligence

The optional OpenMeteoIrradiance source returns hourly forecast points. Weather irradiance is measured in W/m²; PV generation is an estimated electrical output conditioned on installed capacity and a performance ratio.

from energychain import OpenMeteoIrradiance, estimate_pv_power

source = OpenMeteoIrradiance()
points = source.hourly(7.25, 5.19, days=2)
for point in points[:3]:
    print(point.timestamp, point.irradiance_w_m2)

estimate_w = estimate_pv_power(
    irradiance_w_m2=780,
    rated_capacity_w=5000,
    performance_ratio=0.78,
)
print(estimate_w)

Modeling assumptions

The helper is an idealized conversion, not a validated rooftop PV digital twin. Roof tilt, shading, ambient and module temperature, inverter efficiency, clipping and degradation are not comprehensively modeled.

External data governance

Verify the provider's current commercial use rights, API terms, allowed polling frequency, attribution and rate limits before running a monetized service. A forecast can be late, unavailable or revised; treat each case explicitly.

Advanced route

Compare the estimated generation to actual inverter measurements on a historical holdout. Save model inputs, parameters and outcomes separately so operators can understand forecast drift.

All intelligent behaviour starts with a tool contract

Call registered tools using:

result = sdk.tools.invoke(
    "tool_name",
    principal,
    "site-id",
    arguments={"parameter": "value"},
)

Authorization is checked before the tool handler runs. A caller-supplied site_id must match the principal's authorization and the asset's registered site. Tool arguments do not confer new rights.

Built-in tools in expanded v0.3.0

| --- | --- |

Discover available tools

from energychain import EnergyChain
sdk = EnergyChain.demo()
for tool in sdk.tools.describe():
    print(tool)

Calculate a power balance

user = sdk.principal("dev", allowed_sites={"demo-site"})
balance = sdk.tools.invoke(
    "calculate_energy_balance", user, "demo-site",
    arguments={
        "solar_w": 1800,
        "load_w": 2200,
        "generator_w": 0,
    },
)
print(balance["deficit_w"])  # 400

Error handling

Expect ValidationError, AssetNotFound, PermissionDenied, ApprovalRequired and UnsupportedCapability rather than swallowing provider failures. Distinguish missing data from legitimate numeric zero.

Extensibility

ToolRegistry.register(ToolSpec(...)) supports application-specific tools. Treat every custom tool as a security boundary; no text-model response should be able to register a privileged tool.

Why energy memory is different from a chatbot history

A production energy app cannot rely on a conversation transcript to understand state of charge, recurring outages or energy cost. EnergyChain represents typed readings and episodes separately from text prompts.

The memory architecture

Observed readings ────> InMemoryEnergyMemory / SQLiteEnergyMemory
Energy events ────────> Event history
User preferences ─────> Keyed structured preferences
Operational episodes ─> InMemoryEpisodeStore / SQLiteEpisodeStore
                      └── EnergyContextBuilder
                           └── Bounded context for agents and tools

Persist preferences and readings

from energychain import SQLiteEnergyMemory

with SQLiteEnergyMemory("energy.db") as memory:
    memory.set_preference("demo-site", "reserve_pct", 25)
    print(memory.preferences("demo-site"))

The SQLite backend also supports add_reading, history, add_event, events, remember and recall. Its semantics are intended for locally trusted environments, not arbitrary tenants sharing one database file.

Episodic memory

EnergyEpisode records an event and evidence-backed outcome under a specific site. SQLiteEpisodeStore is a separate durable store with site-scoped retrieval and deletion support.

Use episodes to ask what happened previously and what followed, not to fabricate missing sensor data.

Scoped context retrieval

EnergyContextBuilder composes readings, notes, episodes and preferences into bounded results. Its output should never replace the source-of-truth history table.

Seven memory concepts

1. Working context: current bounded task state.

2. Asset memory: persisted equipment descriptions and metadata.

3. Time-series memory: historical measured/estimated readings.

4. Episodic memory: events, explanations and outcomes.

5. Preference memory: site-specific user rules.

6. Semantic knowledge: manuals, documents and retrieval — not yet a production vector-RAG implementation.

7. Procedural memory: reusable tool workflows and policies.

Advanced cautions

• Do not put every numeric reading in a vector database.
• Enforce ownership at the storage layer, not just in prompts.
• Establish retention, deletion, encryption, backups and audit rules in the host service.
• Handle chronological ordering and offset-aware timestamps.
• Distinguish prior observations from proposed future actions.
• SQLite is not a distributed, tenant-isolated managed time-series service.

Two runtime styles

EnergyAgent accepts a tool explicitly chosen by the developer. BoundedEnergyAgent can accept a restricted, structured tool plan from an injected text model.

Both use the same permission-checking ToolRegistry. Neither grants hardware privileges to an LLM.

Model-agnostic explicit agent

from energychain import EnergyChain, CallableTextModel

sdk = EnergyChain.demo()
principal = sdk.principal("dev", allowed_sites={"demo-site"})

model = CallableTextModel(
    lambda system, user, context: "Energy reading: " + context[:100]
)
agent = sdk.agent(model)
response = agent.run(
    "How is the battery doing?",
    principal=principal,
    site_id="demo-site",
    tool_name="get_battery_state",
    tool_arguments={"asset_id": "demo-battery"},
)
print(response.answer)

The callback shown is a simple local function to demonstrate the interface—not an LLM integration.

Bounded planner

sdk.planning_agent(model, max_calls=4, allowed_tools={...}) creates a planner that accepts only a strict finite JSON plan of allowed read/analysis tools. The SDK rejects fabricated tool names, attempts to call privileged execution tools, malformed plans and disallowed scopes.

Why bounded planning matters

A free-form LLM response is not a safe executable instruction. The system requires schema checks, allowlists, call limits and per-tool authorization.

Prompt injection considerations

Treat source text, manual content, free-form notes, agent summaries and external tool responses as untrusted. The bounded planner restricts tool access, but does not fully solve prompt injection in a larger application.

Production responsibilities

Your host service must provide user identity, model credentials, budget/rate limits, prompt/result logging policy and escalation rules. Human-readable explanations should identify assumptions and timestamps instead of fabricating source evidence.

What is not shipped

Multi-agent swarms, autonomously scheduled jobs, arbitrary recursive tool execution and field-certified agent-directed equipment control remain out of v0.3.0 scope.

A skill is a reusable procedure, not an independent model

EnergySkill stores a name, an instruction, and the set of tools appropriate for an energy task. EnergySkillRegistry validates that referenced tool names exist.

from energychain import EnergyChain

sdk = EnergyChain.demo()
for skill in sdk.skills.list():
    print(skill.name, skill.tools)

Built-in concept areas

The standard registry contains reusable battery, cost and fleet skill instructions. These help developers express domain-specific reasoning consistently while continuing to enforce tool-level permissions.

Authoring principles

• Make the task objective explicit.
• Declare exact tools and expected types.
• Never imply missing data or successful hardware actuation.
• Require the agent to state uncertainty and provenance.
• Preserve a separation between instructions and verified calculations.
• Keep parameterized calculations in Python functions, not natural-language guesses.

Advanced patterns

Compose skills with an allowlisted agent and a context builder. A skill can suggest reading battery SOC, estimating runtime and explaining reserve assumptions, but should never implicitly authorize altering equipment settings.

Not yet included

A hosted skills marketplace, verified third-party skill signatures, complex prompt versioning and a distributed agent skills runtime are future work.

Deterministic first

EnergyChain includes intentionally understandable numerical methods, not an undisclosed “AI optimization” black box.

Battery runtime

The approximate model is:

usable_Wh = capacity_Wh × max(SOC% - reserve%, 0) / 100 × efficiency
runtime_h = usable_Wh / load_W

Assumptions include constant load, efficiency and no future charging. It is not a manufacturer-certified safety calculation.

Demand forecasting

simple_average_forecast computes an arithmetic mean of same-metric observations, repeating that value for the requested horizon. It is a baseline, not a trained forecasting model.

Statistical anomaly screening

detect_outliers flags z-score outliers in consistent metric readings. Rare anomalies, seasonal patterns and changing operating regimes require additional validated models.

Power balance

from energychain import energy_balance, estimate_pv_power

balance = energy_balance(
    solar_w=2300,
    load_w=3100,
    generator_w=0,
)
print(balance.deficit_w)  # 800

pv_w = estimate_pv_power(
    irradiance_w_m2=700,
    rated_capacity_w=4000,
    performance_ratio=0.78,
)
print(pv_w)

Advisory load scheduling

optimize_load_schedule chooses among bounded load-start options across declared planning slots to minimize a simplified grid-cost objective while respecting site power constraints.

The result is advisory, not a physical schedule dispatched to equipment. Check slot timestamps, tariffs, base loads, flexible load durations and unit interpretation.

Evaluation

evaluate_forecast(actual, forecast) exposes MAE, RMSE, bias and MAPE-style diagnostics. chronological_mean_backtest uses a chronological training/test split instead of random temporal leakage.

Model governance

For real deployments, test seasonal holdouts, sensor dropout, tariff shifts, negative values, battery operating envelopes, equipment failures and out-of-distribution conditions. Do not make savings claims until independently measured against a credible baseline.

Fleets are groups of sites, not a superuser privilege

A Fleet identifies a set of registered sites. Fleet operations require the requesting principal to be authorized for all included sites before retrieving data.

from energychain import EnergyChain, HybridSiteSimulator, Fleet

sdk = EnergyChain.demo()
sdk.add_connector(
    "second",
    HybridSiteSimulator(site_id="site-two"),
)
sdk.fleets.register(
    Fleet("pilot-fleet", frozenset({"demo-site", "site-two"}))
)
principal = sdk.principal(
    "pilot-operator",
    allowed_sites={"demo-site", "site-two"},
)
snapshot = sdk.tools.invoke(
    "get_fleet_snapshot",
    principal,
    "demo-site",
    arguments={
        "fleet_id": "pilot-fleet",
        "metric": "power_w",
        "max_age_seconds": 3600,
    },
)
print(snapshot["observations"])
print(snapshot["failures"])

Data-quality behaviour

A fleet snapshot has a timestamp, metric, observations and a list of explicit site/asset failures. It deliberately does not compute an aggregate “fleet net power” across heterogeneous solar, loads, generators and batteries.

Scale and boundaries

The built-in manager is an in-process structure. A production fleet service should add tenancy, durable enrollment, site ownership verification, pagination, live telemetry pipelines and independent operations audit.

Common use cases

• Compare site data freshness and coverage.
• Build dashboards across multiple approved sites.
• Identify missing observations for operator follow-up.
• Perform read-only fleet-wide performance summaries.

A fleet is a convenient developer model; it does not imply privileged access to physical devices.

Two workflow mechanisms

EnergyWorkflow runs an ordered list of tool steps in a single process. SQLiteWorkflowCheckpoints records and reuses read/analysis workflow results to reduce accidental duplicated work after restarts.

Sequential pipeline

from energychain import EnergyChain, EnergyWorkflow, WorkflowStep

sdk = EnergyChain.demo()
operator = sdk.principal("developer", allowed_sites={"demo-site"})

flow = EnergyWorkflow(
    sdk.tools,
    [
        WorkflowStep("discover_assets"),
        WorkflowStep(
            "get_battery_state",
            {"asset_id": "demo-battery"},
        ),
    ],
)
result = flow.run(operator, "demo-site")
print(result.succeeded, len(result.outputs))

Checkpoint semantics

SQLite checkpoints bind a workflow run ID to its principal, site and plan. Reusing a run ID with a different scope/plan is rejected rather than silently replaying an unrelated result.

Safety and reliability

• Read/analysis steps are appropriate for the current checkpoint mechanism.
• Persistent read checkpoints are not an exactly-once physical action guarantee.
• Handler failure stops the in-process workflow.
• A production task queue needs durability, retry policies, idempotency keys, poison-message handling and operations visibility.

Event-driven patterns

The local event bus can notify subscribed callbacks about an event. It is not a durable Kafka/MQTT substitute and should not be treated as a high-availability control plane.

A practical reference application

A scheduled host job can fetch observations from an approved connector, ingest validated readings, use tools to estimate reserve capacity, and create an operator-facing advisory notification. The host remains responsible for scheduling, network retries, delivery and authenticated recipients.

Ingestion principles

TelemetryIngestor writes normalized readings to the chosen memory backend after validating their site/asset association. Optional source event identifiers allow duplicate detection.

Event processing pipeline

External observation
  -> Provider adapter
  -> EnergyReading (source / metric / quality / timestamp)
  -> TelemetryIngestor
  -> Memory
  -> Optional EnergyEventBus listener
  -> Application-specific operator notification

Avoid subtle errors

• A duplicate delivery should not silently inflate integrated energy.
• A sensor returning 0 W is different from a sensor not reporting.
• Event timestamps from multiple timezones require normalization.
• A provider connection may remain available while its readings are stale.
• “Last written to memory” is not proof of “last physically observed”.

In-process bus

The EnergyEventBus supports subscribe(listener), unsubscribe(id), publish(event) and scoped recent event retrieval. A failing subscriber must not be mistaken for a successful external delivery.

Operational scaling limits

In v0.3.0 the event bus is in-memory and bounded. There is no durable broker, distributed deduplication table, cross-process fanout, regulated-retention audit service or automated incident queue.

Design a production ingestion layer

Use an authenticated ingestion API, schema validation, source signatures where available, backpressure, retries, quotas, durable event storage, observability and bounded retention. Export measurements to a dedicated time-series store for scale.

Foundational rule

An AI model can explain or recommend; it must never become the authority to control hardware.

EnergyChain separates permissions into READ, ANALYZE, PREPARE and EXECUTE. Tool scopes and asset-site ownership are checked independently of any model-produced text.

Principals and permissions

from energychain import (
    Principal, Permission, SafetyPolicy
)

operator = Principal(
    "operator-123",
    frozenset({"my-site"}),
    frozenset({Permission.READ, Permission.ANALYZE}),
)
policy = SafetyPolicy()
policy.authorize(operator, "my-site", Permission.READ)

Prepare is not dispatch

EnergyActionSpec declares an action's permitted parameters. ActionService supports bounded preparation, but physical dispatch is disabled by default and does not imply that a device accepted or completed a command.

Signed approvals

SignedApprovalAuthority can issue a short-lived signature bound to the principal, site, action name and exact arguments, with a one-time nonce.

Critical deployment warning: Its single-use memory is local to an authority process. A real distributed approval service needs shared atomic replay protection, independent issuer identity, robust authorization controls and key management.

Four outcomes to distinguish

1. Requested: a caller proposed an action.

2. Approved: a trusted authority authorized that exact action.

3. Accepted: a device/API acknowledged the command.

4. Verified: independent telemetry confirms the expected physical state.

The current SDK should not claim stages 3–4 for live equipment.

Security checklist

• Never expose sdk.principal() as a public identity endpoint.
• Keep approving and acting roles separate.
• Bind approval to every action argument, not merely the action name.
• Treat tool outputs and source documents as untrusted.
• Require electrical safety constraints and manufacturer-specific interlocks.
• Store audit evidence in an append-only or independently managed system.
• Prevent SSRF and credential leakage in custom connectors.
• Enforce service quotas and request deadlines.
• Threat-model time-of-check/time-of-use races and concurrent approvals.

Not in v0.3.0

Field-certified physical dispatch, distributed execution authorization, hardware interlocks, fail-safe device feedback loops and guaranteed command rollback.

Test pyramid

1. Unit tests: object validation, arithmetic, safety and memory invariants.

2. Adapter tests: mock provider payloads, faults, units and timestamps.

3. Integration tests: real approved data sources under read-only access.

4. Operations tests: reconnects, overload, retention, credential rotation and outages.

5. Field tests: measurement accuracy, hardware safety and independent action verification.

Run locally

python -m pip install -e '.[dev]'
python -m ruff check --isolated --select E4,E7,E9,F src tests scripts examples
python -m pytest -q
python -m build
python -m twine check --strict dist/*

Current engineering evidence

The published EnergyChain SDK includes automated Python tests, wheel-building and installation checks. Consult the current GitHub Actions runs for exact test counts and supported environments; previous development-branch results are not a current release certification.

Forecast evaluation

Use evaluate_forecast to compare paired actual and predicted numbers. For chronological data, use temporal train/test separation to avoid look-ahead leakage.

Acceptance gates for a pilot

• Operator-authorized connection and repeatable credentials.
• Correct asset IDs, units and measured time stamps.
• Data provenance and missing/stale telemetry handling.
• Isolation between independent operators or sites.
• Real-world accuracy measurements and baseline comparisons.
• Monitoring of failed provider calls, request latency and quota usage.
• Verified data rights, retention and privacy procedures.
• Clear escalation for any suggested changes to physical systems.

Understand what is deployable

An EnergyChain SDK is not itself a complete production service. The host app still needs HTTP APIs, authentication, organizational tenancy, observability infrastructure, secrets, scheduled workers and operational support.

A recommended read-only application topology

Web / mobile app
     |
API gateway + real authentication + rate limiting
     |
Application backend (EnergyChain SDK)
     |-- Tenant / site authorization service
     |-- Connector registry -> approved provider APIs
     |-- Tool registry -> calculators / forecasts
     |-- Scoped context and memory
     |-- Fleet data and advisory workflow
     |
Durable database / time-series store
Metrics + traces + alerting service
Secret manager + rotation and audit

Host responsibilities

| --- | --- | --- |

Operational checklist

• Separate staging and production environments.
• Pin versions, validate supply-chain provenance and scan dependencies.
• Keep API tokens in a secret manager and rotate them.
• Enforce TLS for all external endpoints; limit egress.
• Implement health checks, structured logs, tracing and SLIs/SLOs.
• Plan backups, tested restoration and data retention.
• Isolate tenants in databases and authorization code.
• Add a change-control process before enabling anything capable of physical impact.
• Maintain equipment/vendor support plans and escalation procedures.

Reference integration goal

Start with a read-only deployment connected to one authorized real energy data source. Measure freshness, ingestion reliability, unit accuracy and operator usefulness. Expand to more sites only after those signals meet documented thresholds.

Not included

EnergyChain Cloud, hosted multi-tenant services, production device command runtime and certified dispatch controls are not currently shipped.

Version and API scope

This page documents the released Python SDK v0.3.0 (alpha), not a versioned hosted REST API. The website's registration, demo and anonymous-analytics endpoints are website infrastructure—not a public energy-control or managed inference API. There is currently no hosted EnergyChain developer API or managed production runtime.

Core public imports

Basic invocation

from energychain import EnergyChain
chain = EnergyChain.demo()
principal = chain.principal("developer", allowed_sites={"demo-site"})
assets = chain.tools.invoke("discover_assets", principal, "demo-site")
runtime = chain.tools.invoke(
    "estimate_battery_runtime", principal, "demo-site",
    arguments={"battery_id": "demo-battery", "load_w": 900, "reserve_pct": 20},
)
print(assets, runtime)

New v0.3.0 API surfaces

Model Gateway: ModelGateway(provider, model, timeout=..., max_retries=...), gateway.complete(system=..., user=..., context=..., json_schema=...), gateway.as_text_model().

Document Intelligence: EnergyKnowledgeBase(path, tenant_id=..., allowed_root=...), index.ingest_file(path, principal=..., site_id=...), index.search(query, principal=..., site_id=..., limit=...), index.answer(query, model=..., principal=..., site_id=...).

Model Gateway guide · Document Intelligence guide

Production responsibilities

Production callers must authenticate principals and validate site-level authorization at ingress; apply model quotas; check data freshness; isolate untrusted document parsing; and require explicitly approved workflows for any future physical actuation. External provider integrations are opt-in and not field-certified by EnergyChain.

Complete release API: public package exports and reference examples.

Import errors

Symptom: ModuleNotFoundError: energychain

Activate the virtual environment used to install the SDK. Confirm Python and pip refer to the same interpreter:

python -m pip show energychain
python -c "import energychain; print(energychain.__version__)"

Missing deep-v0.3.0 APIs

If Fleet, HomeAssistantConnector or BoundedEnergyAgent cannot be imported, check that you installed the published v0.3.0 SDK rather than the earlier alpha source or an unrelated PyPI package.

Forbidden site access

Symptom: PermissionDenied

The principal's allowed site list does not include the requested site, or the asset belongs to a different registered site. Do not fix this by assigning wildcard access to unknown callers; use the trusted identity provider and tenant authorization service.

Missing data or stale readings

An observation may be old even when successfully returned. Check reading.timestamp, reading.quality, reading.source and the configured freshness threshold. Treat missing data as failure, not zero.

Home Assistant failure

Verify the HTTPS origin, valid token, exact entity ID, current unit attribute and whether the provider returns a numeric state. This connector intentionally does not expose historical recorder data.

Forecast looks constant

The included demand forecast is an arithmetic mean baseline. A flat horizon is expected. Train and validate an external forecasting model for stronger predictive accuracy.

Unapproved action

Symptom: ApprovalRequired

Execution requires exact trusted authorization. An old unsigned Approval object cannot substitute for a verified signed approval under the strengthened safety design. Do not disable execution approval checks to bypass this.

CI and release

Run local pytest, ruff, python -m build and twine check. Confirm correct distribution metadata and clean-wheel imports. An open draft pull request is not a published stable release.

Need help?

Start with the architecture, security model or Pull Request #2.

Current public release: 0.3.0

Released on PyPI: energychain 0.3.0, MIT-licensed, Python 3.10+, published from the official EnergyChain organization. These documentation pages reflect the published package, not an unmerged development branch.

Available for developer exploration

• Typed energy models, deterministic analysis, CSV/simulator connectors and read-only Home Assistant/solar-weather adapters.
• Permission-scoped tools, read/analysis agents and workflows; local memory, fleets, telemetry ingestion, planning and approval primitives.
• v0.3.0 Model Gateway: six selectable provider families and optional JSON Schema output validation.
• v0.3.0 Document Intelligence: local scoped PDF, DOCX, TXT, Markdown, CSV and XLSX indexing, search and citation metadata.
• Opt-in voluntary SDK activity reporting example (not embedded background tracking).

Not shipped as a managed service

• No hosted developer API, managed inference, account-based SDK usage service or EnergyChain Cloud.
• No vendor-certified physical device control, production dispatch runtime or independently verified energy savings.
• No autonomous unbounded multi-agent execution or production-grade shared multi-tenant identity service.
• Provider integrations and model endpoints require developer credentials, testing and authorization.

Next engineering milestones

Wider energy vendor connector validation, hardened production deployment patterns, optional managed infrastructure, developer account management and field-tested safety controls are future work—not advertised as currently delivered.

Included in SDK v0.3.0

ModelGateway is a synchronous adapter around six model-provider families. It does not host inference, provision credentials, guarantee particular model availability, or give an LLM permission to operate energy hardware.

Install and call a model

python -m pip install 'energychain[models]==0.3.0'
# Set your provider key securely in your server environment:
export OPENAI_API_KEY='YOUR_API_KEY'

from energychain import ModelGateway
gateway = ModelGateway("openai", "gpt-4o-mini", timeout=30, max_retries=2)
result = gateway.complete(
    system="Only explain equipment information. Do not authorize controls.",
    user="What is the documented maximum battery charge current?",
    context="Unverified datasheet: maximum charge current 40 A.",
)
print(result.text, result.usage.input_tokens, result.duration_ms)

Developers supply API credentials and choose their own provider and supported model. Provider capabilities and pricing can vary. An existing EnergyAgent can accept model=gateway.as_text_model().

Schema-validated structured output

schema = {
    "type": "object",
    "properties": {
        "equipment": {"type": "string"},
        "rating_kw": {"type": "number"},
    },
    "required": ["equipment", "rating_kw"],
    "additionalProperties": False,
}
result = gateway.complete(
    system="Extract stated values only; return JSON.",
    user="Extract the inverter rating.",
    context="Inverter nominal output 5 kW.",
    json_schema=schema,
)
print(result.json_value)

Responses are locally validated against the supplied schema. Each provider may implement a different structured-output subset, and live model/account compatibility must be tested independently.

Model Gateway security boundaries

• Keep credentials on a trusted server and outside browser clients, telemetry and logs.
• Official providers use host-locked endpoints; compatible endpoints require HTTPS and reject private IPs and internal hostnames; Ollama is loopback HTTP only.
• Transport limits prompts, request and response size, timeouts, redirects and retries; external request charges can still apply.
• No streaming, hosted inference, tool-call execution or energy equipment dispatch is provided by the Model Gateway.
• External responses and retrieved text remain untrusted; tool permissions are enforced separately.

Complete implementation guide: Model Gateway reference on GitHub.

Supported sources and indexing

EnergyKnowledgeBase extracts text and source provenance from PDF, DOCX, TXT, Markdown, CSV and XLSX documents. It chunks content into a tenant-bound SQLite search index, retaining file SHA-256 fingerprints and source page, row, sheet or section references where available.

Install document parsers

python -m pip install 'energychain[documents]==0.3.0'

The documents extra installs pypdf, python-docx and openpyxl. Keyword search, SQLite indexing, TXT, Markdown and CSV do not need extra runtime dependencies.

Ingest and search by tenant and site

from energychain import EnergyKnowledgeBase, Principal, Permission
index = EnergyKnowledgeBase(
    "tenant-knowledge.sqlite3",
    tenant_id="utility-customer-acme",
    allowed_root="/srv/acme/manuals",
)
operator = Principal(
    "operator-123",
    frozenset({"site-17"}),
    frozenset({Permission.READ, Permission.ANALYZE, Permission.PREPARE}),
)
stored = index.ingest_file(
    "/srv/acme/manuals/battery.pdf",
    principal=operator,
    site_id="site-17",
)
for match in index.search(
    "battery reserve current", principal=operator, site_id="site-17", limit=5
):
    print(match.citation, match.source_sha256, match.energy_facts, match.text)

The tenant ID must come from trusted backend authentication, not a user request or LLM output. PREPARE permissions govern ingestion and deletion; READ permissions govern search and answers.

Optional model answers and embeddings

# After configuring a ModelGateway as "gateway":
answer = index.answer(
    "Which current is explicitly documented?",
    model=gateway, principal=operator, site_id="site-17",
)
print(answer.answer, answer.sources)

A developer-provided embedding callback may be used; it can send document contents to an external service. Search scores rank relevance, not truth. Citations point to retrieved passages, not independent verification of model responses.

Security and operational limits

• Authenticate tenant, site and user permissions at your own API boundary; use allowed_root and reject symlinks or traversal.
• Run PDF and Office parsing in an isolated, resource-constrained process; built-in parser limits are not a sandbox.
• Scanned PDFs require external OCR; password-protected PDFs and unsupported formats are refused.
• Manage database encryption, retention, deletion, backups and data residency yourself.
• Retrieved documents are untrusted evidence and never grant physical device control authority.

Full documentation: Document Intelligence on GitHub.

No telemetry by default

Installing, importing or using energychain does not transmit usage information. The Python SDK remains local-first and works offline. The EnergyChain Console separately displays public PyPI download counts, which do not indicate unique installations or active users.

Voluntary activity example

The repository includes an example that transmits a single optional activity ping, only when run explicitly with the --opt-in flag:

python -m pip install energychain
# From the official EnergyChain repository checkout:
python examples/opt_in_usage_telemetry.py --opt-in

The example sends a randomly generated installation UUID, SDK version and explicit consent flag to https://energychain.app/api/sdk-telemetry. The server hashes the UUID before storing private aggregate activity. It does not intentionally collect energy readings, prompts, API keys, customer records, or application files.

What counts mean

• PyPI downloads: download requests, not unique individuals or active SDK users.
• Opt-in instances: self-reported installation identifiers, not verified accounts.
• Hosted API requests: not available until a managed developer runtime launches.

No persistent background reporting or runtime hooks are installed. Full guide: optional SDK telemetry and privacy.