Agent Integration Examples

GitHub-ready example repository content. Publishing to GitHub still requires an external account or token.

JSON file map · README.md · Distribution kit

# Netzhandwerker Energy Hub Agent Examples

Copy-paste starter code for agents that call the Netzhandwerker Energy Research Hub.

The API is pay-per-call with x402. First call paid endpoints without a payment proof to receive HTTP 402 terms. Your wallet/client signs the payment, then retries with `X-Payment`.

Do not log or publish full payment proofs.

## Endpoints

- Discovery: https://energy.netzhandwerker.de/.well-known/x402.json
- OpenAPI: https://energy.netzhandwerker.de/openapi.json
- MCP: https://energy.netzhandwerker.de/mcp
- Cheapest paid smoke test: GET https://energy.netzhandwerker.de/demo/paid-sample
- Main decision endpoint: POST https://energy.netzhandwerker.de/energy/decision
- Demand sensor: POST https://energy.netzhandwerker.de/demand/submit

## Run

```bash
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
python basic_x402_probe.py
python mcp_client.py
```

## Files

- `basic_x402_probe.py` probes a paid endpoint and prints the 402 terms.
- `crewai_energy_tool.py` wraps the API as a CrewAI tool.
- `langgraph_energy_node.py` wraps the API as a LangGraph node.
- `autogen_energy_tool.py` provides an AutoGen-compatible async tool function.
- `mcp_client.py` lists MCP tools and calls one paid tool.
- `demand_submit.py` submits a missing endpoint request without payment.

Payment is intentionally left as a placeholder. Use your x402 wallet/client to create `X-Payment`.

requests>=2.32.0
crewai>=0.130.0
langgraph>=0.4.0
typing-extensions>=4.12.0

ENERGY_HUB_BASE_URL=https://energy.netzhandwerker.de
# X_PAYMENT is optional and should be produced by your x402 wallet/client.
X_PAYMENT=

import os
import requests

BASE = os.getenv("ENERGY_HUB_BASE_URL", "https://energy.netzhandwerker.de")
BODY = {"goal": "cheapest_ev_charging", "location": "DE"}

headers = {"Content-Type": "application/json"}
if os.getenv("X_PAYMENT"):
    headers["X-Payment"] = os.environ["X_PAYMENT"]

r = requests.post(f"{BASE}/energy/decision", json=BODY, headers=headers, timeout=20)
print("status:", r.status_code)
print(r.text[:2000])

if r.status_code == 402:
    print("Create X-Payment with your x402 wallet/client, then retry the same request.")

import os
import requests
from crewai.tools import tool

BASE = os.getenv("ENERGY_HUB_BASE_URL", "https://energy.netzhandwerker.de")

@tool("energy_hub_decision")
def energy_hub_decision(goal: str, location: str = "DE") -> str:
    """Return German/EU energy decision context from the Energy Research Hub."""
    r = requests.post(f"{BASE}/energy/decision", json={"goal": goal, "location": location}, timeout=20)
    if r.status_code == 402:
        return "Payment required. Send the 402 terms to your x402 wallet/client and retry with X-Payment."
    r.raise_for_status()
    return r.text

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

BASE = os.getenv("ENERGY_HUB_BASE_URL", "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}/energy/decision", json=body, timeout=20)
    if r.status_code == 402:
        return {"result": "HTTP 402 received. Pay with x402 and 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 os
import requests

BASE = os.getenv("ENERGY_HUB_BASE_URL", "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."""
    r = requests.post(f"{BASE}/energy/decision", json={"goal": goal, "location": location}, 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 os
import requests

BASE = os.getenv("ENERGY_HUB_BASE_URL", "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": "energy_decision",
        "arguments": {"goal": "cheapest_ev_charging", "location": "DE"}
    }
}
print(requests.post(f"{BASE}/mcp", json=call_tool, timeout=20).json())

import os
import requests

BASE = os.getenv("ENERGY_HUB_BASE_URL", "https://energy.netzhandwerker.de")
body = {
    "need": "15-minute electricity prices for a specific EU market",
    "desired_format": "json",
    "max_budget_usdc": 0.01,
    "recurring": True,
    "frequency": "hourly"
}
r = requests.post(f"{BASE}/demand/submit", json=body, timeout=20)
print(r.status_code, r.text)