npm install @openplait/core @openplait/adapter-sdk \
  @openplait/runtime @openplait/adapter-clickhouse

For Tempo, replace the last package with @openplait/adapter-tempo. You can install multiple adapters in the same app.

pip install openplait

From the repo (Poetry):

cd python
poetry install

Python is alpha: query parse/validate and runtime registration work today; ClickHouse execute parity with Node is still landing.

import { ClickHouseAdapter } from "@openplait/adapter-clickhouse";
import { DatasourceRegistry, OpenPlaitRuntime } from "@openplait/runtime";

const config = {
  url: process.env.CLICKHOUSE_URL!,
  username: process.env.CLICKHOUSE_USER,
  password: process.env.CLICKHOUSE_PASSWORD,
  database: "observability",
};

const registry = new DatasourceRegistry().register({
  name: "primary",
  kind: "ClickHouseDatasource",
  scope: "workspace",
  config,
  adapter: new ClickHouseAdapter(config),
});

export const runtime = new OpenPlaitRuntime(registry, {
  defaultTimeoutMs: 30_000,
});

import os
from openplait.adapters import ClickHouseAdapter, ClickHouseConfig
from openplait.runtime import DatasourceRegistry, OpenPlaitRuntime
from openplait.runtime.registry import RuntimeDatasourceRegistration

config = ClickHouseConfig(
    url=os.environ["CLICKHOUSE_URL"],
    username=os.environ.get("CLICKHOUSE_USER"),
    password=os.environ.get("CLICKHOUSE_PASSWORD"),
    database="observability",
)

registry = DatasourceRegistry().register(
    RuntimeDatasourceRegistration(
        name="primary",
        kind="ClickHouseDatasource",
        adapter=ClickHouseAdapter(config),
        config=config,
    )
)

runtime = OpenPlaitRuntime(registry, default_timeout_ms=30_000)

import {
  OPENPLAIT_API_VERSION,
  type SemanticQuery,
} from "@openplait/core";

export const query: SemanticQuery = {
  apiVersion: OPENPLAIT_API_VERSION,
  kind: "Query",
  metadata: { name: "requests-by-service" },
  spec: {
    mode: "semantic",
    datasource: { kind: "ClickHouseDatasource", name: "primary" },
    input: { signal: "traces", entity: "otel.spans" },
    timeRange: {
      field: "timestamp",
      from: "${__from}",
      to: "${__to}",
    },
    select: [
      { field: "service.name", as: "service" },
      { aggregate: { function: "count" }, as: "requests" },
    ],
    groupBy: [{ field: "service.name" }],
    orderBy: [{ field: "requests", direction: "desc" }],
    limit: 20,
  },
};

from openplait import OPENPLAIT_API_VERSION
from openplait.core import parse_query, validate_query

query = parse_query(
    {
        "apiVersion": OPENPLAIT_API_VERSION,
        "kind": "Query",
        "metadata": {"name": "requests-by-service"},
        "spec": {
            "mode": "semantic",
            "datasource": {"kind": "ClickHouseDatasource", "name": "primary"},
            "input": {"signal": "traces", "entity": "otel.spans"},
            "timeRange": {
                "field": "timestamp",
                "from": "${__from}",
                "to": "${__to}",
            },
            "select": [
                {"field": "service.name", "as": "service"},
                {"aggregate": {"function": "count"}, "as": "requests"},
            ],
            "groupBy": [{"field": "service.name"}],
            "orderBy": [{"field": "requests", "direction": "desc"}],
            "limit": 20,
        },
    }
)
assert validate_query(query).valid

const response = await runtime.execute({
  queries: [query],
  variables: {
    __from: "2026-08-05T00:00:00.000Z",
    __to: "2026-08-05T01:00:00.000Z",
  },
  audit: {
    requestId: "dashboard-7b9b4d",
    actorId: "user-42",
    tenantId: "workspace-a",
  },
});

for (const frame of response.result.frames) {
  console.log(frame.name, frame.length, frame.fields);
}

import asyncio

async def main() -> None:
    # Full ClickHouse execute is still scaffolding on Python —
    # validate + register today; prefer @openplait/* for production reads.
    result = validate_query(query)
    assert result.valid
    print(query.metadata.name, query.spec.datasource.name)

asyncio.run(main())

When Python adapter execute lands, the call shape mirrors TypeScript:

result = await runtime.execute(
    query,
    variables={
        "__from": "2026-08-05T00:00:00.000Z",
        "__to": "2026-08-05T01:00:00.000Z",
    },
)
for frame in result.frames:
    print(frame.name, frame.length, frame.fields)