Energy Trading Signals for AI Agents

What agents get

• Buy/sell context for agent reasoning
• Battery arbitrage window candidates
• Daily market summary and price forecast

Recommended endpoints

Framework copy-paste examples

The snippets intentionally leave PAYMENT-SIGNATURE to your wallet/client, which creates the signed proof after reading the 402 response. X-Payment is accepted as a compatibility alias.

curl -i -X POST https://energy.netzhandwerker.de/signals/buy-sell \
  -H "Content-Type: application/json" \
  -d '{"asset":"electricity_spot","region":"DE","horizon_hours":24,"risk_mode":"conservative"}'

# On HTTP 402, let your x402 wallet/client create PAYMENT-SIGNATURE, then retry.
# Do not log or share the full payment proof.

import requests

BASE = "https://energy.netzhandwerker.de"
BODY = {
         "asset": "electricity_spot",
         "region": "DE",
         "horizon_hours": 24,
         "risk_mode": "conservative"
       }

r = requests.post(f"{BASE}/signals/buy-sell", json=BODY, timeout=20)
if r.status_code == 402:
    terms = r.json()
    print("payment required:", terms)
    # signed_payment = your_x402_client.sign(terms)
    # r = requests.post(f"{BASE}/signals/buy-sell", json=BODY, headers={"PAYMENT-SIGNATURE": signed_payment}, timeout=20)

print(r.status_code, r.text[:500])

import requests
from crewai.tools import tool

BASE = "https://energy.netzhandwerker.de"

@tool("energy_hub_decision")
def energy_hub_decision(goal: str, location: str = "DE") -> str:
    """Call Netzhandwerker Energy Research Hub and return structured energy context."""
    body = {"goal": goal, "location": location}
    r = requests.post(f"{BASE}/signals/buy-sell", json=body, timeout=20)
    if r.status_code == 402:
        return "Payment required. Pass the 402 terms to your x402 wallet and retry with PAYMENT-SIGNATURE."
    r.raise_for_status()
    return r.text

import requests
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, END

BASE = "https://energy.netzhandwerker.de"

class State(TypedDict):
    goal: str
    location: str
    result: str

def call_energy_hub(state: State):
    body = {"goal": state["goal"], "location": state.get("location", "DE")}
    r = requests.post(f"{BASE}/signals/buy-sell", json=body, timeout=20)
    if r.status_code == 402:
        return {"result": "HTTP 402 received. Pay with x402, then retry the same call."}
    r.raise_for_status()
    return {"result": r.text}

graph = StateGraph(State)
graph.add_node("energy_hub", call_energy_hub)
graph.set_entry_point("energy_hub")
graph.add_edge("energy_hub", END)
app = graph.compile()

import requests

BASE = "https://energy.netzhandwerker.de"

async def energy_hub_tool(goal: str, location: str = "DE") -> str:
    """Register this function as a callable tool in your AutoGen agent/runtime."""
    body = {"goal": goal, "location": location}
    r = requests.post(f"{BASE}/signals/buy-sell", json=body, timeout=20)
    if r.status_code == 402:
        return "Payment required. Forward the 402 body to an x402-capable wallet client."
    r.raise_for_status()
    return r.text

import requests

BASE = "https://energy.netzhandwerker.de"

list_tools = {"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}
print(requests.post(f"{BASE}/mcp", json=list_tools, timeout=20).json())

call_tool = {
    "jsonrpc": "2.0",
    "id": 2,
    "method": "tools/call",
    "params": {
        "name": "buy_sell_signal",
        "arguments": {
                       "asset": "electricity_spot",
                       "region": "DE",
                       "horizon_hours": 24,
                       "risk_mode": "conservative"
                     }
    }
}
print(requests.post(f"{BASE}/mcp", json=call_tool, timeout=20).json())

import requests

BASE = "https://energy.netzhandwerker.de"
body = {
         "need": "Trading-grade German/EU power signal with budget and refresh interval",
         "desired_endpoint": "/requested/example",
         "desired_format": "json",
         "max_budget_usdc": 0.01,
         "recurring": true,
         "frequency": "daily",
         "use_case": "Trading, arbitrage and portfolio agents"
       }

r = requests.post(f"{BASE}/demand/submit", json=body, timeout=20)
print(r.status_code, r.json())

# Optional: pay the 0.001 USDC priority demand terms if your agent wants priority listing.

MCP tool names

energy_decision cheapest_window ev_charging_plan heating_plan flexibility_window battery_arbitrage buy_sell_signal today_summary submit_data_request

Honest scope: the API sells current German/EU energy intelligence and decision support. If your agent needs a missing endpoint, submit the need; implementation is not guaranteed.