energyops-mcp

A hand-rolled implementation of the Model Context Protocol over JSON-RPC 2.0 — no MCP SDK — plus a non-trivial local MCP server for building energy analysis: anomaly detection, flexible-load schedule optimization, and battery/solar outage-autonomy simulation.

This package is fully independent: it does not require any other repository, any LLM API key, or any building's real data. Everything it analyzes is a reproducible synthetic dataset generated from a fixed seed.

What's in here

• energyops_mcp.protocol — the MCP transport and lifecycle layer, built directly against JSON-RPC 2.0 and the MCP 2025-06-18 spec:

  • jsonrpc.py: message parsing/serialization, standard error codes.

  • client.py: an asyncio MCP client — initialize, ping, tools/list (with pagination), tools/call, timeouts with notifications/cancelled, clean shutdown. Protocol errors (JsonRpcError) are kept structurally distinct from tool execution failures (CallToolResult(isError=True)).

  • server.py: a synchronous stdio MCP server base class (McpServer) — register tools with a decorator, it handles initialize/ping/tools/list/tools/call framing.

  • stdio_transport.py: child-process transport for the client side (newline-delimited JSON over stdin/stdout). Includes a fix for a real Windows issue — asyncio.create_subprocess_exec cannot launch a .cmd/.bat shim (like npx) directly; this is routed through cmd.exe /c automatically.

  • http_transport.py: a Streamable HTTP client transport (POST/GET to a single endpoint, Mcp-Session-Id handling, JSON and text/event-stream response modes), for talking to a remote MCP server over HTTPS.

• energyops_mcp.energy — the analytical engine, exposed as seven MCP tools by energy/server.py:

A demo building auto-seeds on first use: five independent consumption circuits (lighting, HVAC, critical services, a pump, and general equipment) plus a solar generation meter, eight weeks of 15-minute-resolution history and one target day, generated from a fixed random seed — fully reproducible, and never double-counting energy (consumption and generation are always summed separately). A known anomaly is deliberately injected into the target day so detect_anomalies has something real to find.

Related MCP server: bim2sim-mcp

Install

Requires Python 3.13+.

git clone https://github.com/iancumes/energyops-mcp.git
cd energyops-mcp
python -m venv .venv
# Windows: .venv\Scripts\activate      macOS/Linux: source .venv/bin/activate
pip install -e ".[dev]"

No API key, no environment variables, and no other repository are required for anything in this package.

Run the server

energyops-mcp-server --db-path energyops.sqlite3
# or:
python -m energyops_mcp.energy.server --db-path energyops.sqlite3

This starts a stdio MCP server: it speaks newline-delimited JSON-RPC on stdin/stdout and writes diagnostics to stderr, so it is meant to be launched by an MCP client/host as a child process, not run interactively. --import-dir (default data/imports) sets the only directory import_readings is allowed to read a CSV from.

Connecting from your own client

Any MCP client that speaks JSON-RPC 2.0 over stdio can talk to this server. Using the client included in this package:

import asyncio
from energyops_mcp.protocol.client import McpClient
from energyops_mcp.protocol.stdio_transport import StdioTransport

async def main():
    transport = StdioTransport("python", ["-m", "energyops_mcp.energy.server"])
    client = McpClient(transport, server_name="energyops-mcp")
    await client.start()
    await client.initialize()

    tools = await client.list_tools()
    print([t.name for t in tools])

    result = await client.call_tool("list_buildings", {})
    print(result.text())

    await client.aclose()

asyncio.run(main())

Example: importing readings

result = await client.call_tool("import_readings", {"csv_path": "data/imports/january.csv"})

The CSV must have exactly the header timestamp,meter_id,energy_kwh, with UTC ISO-8601 timestamps. Re-importing the same file is a no-op (rows are keyed by (meter_id, timestamp)); unknown meters and malformed rows are reported back, not silently dropped or fatal.

Example: detecting anomalies

result = await client.call_tool(
    "detect_anomalies",
    {"building_id": "demo-building", "target_date": "2026-03-02"},
)
print(result.structured_content["incidents"])

Example: optimizing a flexible load schedule

tariff_bands = [{"start_minute": 0, "end_minute": 24 * 60, "price_per_kwh": 0.2}]
result = await client.call_tool(
    "optimize_schedule",
    {"building_id": "demo-building", "date": "2026-03-02", "tariff_bands": tariff_bands},
)
print(result.structured_content["proposed"])

Design notes

• Protocol errors vs. tool failures are structurally distinct. A malformed message, an unknown method, or a transport failure raises JsonRpcError. A tool that runs but fails (bad input, no data for the period, an infeasible schedule) returns a normal result with isError: true instead — this is what lets a host tell "the protocol is broken" apart from "the tool reported a failure."

• The server-side I/O loop is deliberately synchronous, not asyncio. A local tool server talks to exactly one client over one pipe and processes one request at a time, so blocking line I/O is simpler and sidesteps a well-known fragility of piping a process's own stdin/stdout through asyncio on Windows. The client side is asyncio-based because a host application typically needs to manage the child subprocess concurrently with everything else it's doing (an LLM call, a UI) — which is the well-supported case on every platform via asyncio.create_subprocess_exec.

• Every reported figure comes from a real calculation, never from an LLM guessing. The MCP layer only decides which tool to call; energyops_mcp.energy does the arithmetic.

Testing

pip install -e ".[dev]"
pytest -v
ruff check src tests

The test suite spawns this server as a real subprocess and drives it over actual stdio pipes — not just mocks — including a hand-checkable acceptance scenario (a 10 kWh battery at 100% state of charge with a 20% reserve, ideal efficiency, a steady 2 kW load and no solar must yield exactly 4 hours of autonomy) and detection of the deliberately injected anomaly through the full protocol round trip.

License

MIT — see LICENSE.