mkio

Config-driven microservice framework for Python. Define your schema, services, and data flows in a TOML file — zero coding required for standard configurations.

A single TCP port serves HTTP and WebSocket, backed by an embedded SQLite database. Designed for restricted environments where runtime downloads aren't possible — everything installs via pip.

Contents

• Quick Start
• Features
• Authentication & Access Control
• Programmatic API
• Service Types
• WebSocket Protocol
• Client Libraries
• Expression Language
• Performance
• CLI Tools
• Using mkio from a Claude-Based Project
• Schema Migration
• License

Quick Start

pip install mkio

Create server.toml:

port = 8080

[tables.orders]
columns = { id = "TEXT PRIMARY KEY", symbol = "TEXT NOT NULL", qty = "INTEGER", status = "TEXT DEFAULT 'pending'" }

[services.add_order]
protocol = "transaction"
table = "orders"
op_type = "insert"
fields = ["id", "symbol", "qty"]

[services.all_orders]
protocol = "query"
primary_table = "orders"
filterable = ["status", "symbol"]

[static]
"/" = "./static"

Run:

mkio serve

Or programmatically:

from mkio import serve
serve("server.toml")
serve({...})  # or pass a dict

For programmatic control (custom routes, non-blocking lifecycle), see Programmatic API.

Features

• Single port — HTTP pages and WebSocket messages on one port
• Config-driven — define tables, transactions, and live data services in TOML
• Transaction services — insert, update, delete, upsert across multiple tables atomically
• SubPub — topic-based single-row subscription with live push, server-side where filtering and publish formatting, expression-based defaults for missing topics
• Stream — append-only ring buffer with cursor-based reconnection
• Query — snapshot + change feed from SQLite
• ReqRep — one-shot request-reply with parameterized SQL and/or expression evaluation, returning scalar values, single records, or result sets
• Expression language — one safe, extensible language for filters and formatters (qty > 100 && status == 'pending'), implemented identically in Python and JavaScript
• Schema migration — automatic detection of safe/destructive changes with interactive confirmation
• Write batching — hundreds of writes committed in a single SQLite transaction for high throughput
• Reconnection recovery — stream services use ref-based cursor reconnection persisted across server restarts via _mkio_ref column; subpub and query always replay a full snapshot
• Field projection — subscribers can request specific fields per subscription, reducing payload size. Framework fields (_mkio_ref, _mkio_row, _mkio_topic, _mkio_exists) are always preserved through projection
• Client libraries — Python and JavaScript clients with auto-reconnect and ref tracking
• Graceful shutdown — drains pending writes, checkpoints WAL, clean close
• Service monitoring — tap into any service's inbound/outbound message flow via CLI or WebSocket
• Service discovery — GET /api/services list and GET /api/services/<name> detail endpoints, mkio services CLI
• Connection identity — built-in _mkio reqrep service reports server name, version, framework version, protocol version, services, tables, config hash, and uptime — lets clients verify they're connected to the correct session. Also supports table schema introspection (columns, types, primary keys, defaults)
• Config endpoint — /config path serves TOML files as JSON (request foo.json, server reads foo.toml and returns JSON); falls back to literal .json files; other extensions served as-is
• CLI tools — send transactions, subscribe to live data, monitor traffic, inspect services
• Authentication & access control — table-driven roles (_mkio_rights) with config-driven per-service/per-op access. Supports SQL pre-check conditions (when), built-in username/password auth (_mkio_users with bcrypt), or custom auth via app.on_auth() (JWT, LDAP, etc.)
• Programmatic API — create_app() returns a controllable server handle with async start()/stop() lifecycle, custom HTTP routes, custom service registration, lifecycle hooks, and a data facade (execute/query/subscribe) for server-side interaction without WebSocket

Authentication & Access Control

Enable auth by adding the _mkio_rights table to your config. When it exists, all services are locked by default — each service must declare its access.

Setup

# Auth tables
[tables._mkio_users]
columns = { username = "TEXT PRIMARY KEY", password = "TEXT NOT NULL", role = "TEXT NOT NULL" }

[tables._mkio_rights]
columns = { role = "TEXT", right = "TEXT" }

# Service access control
[services.prices]
protocol = "subpub"
primary_table = "prices"
topic = "symbol"
access = "open"               # anyone, no auth needed

[services.positions]
protocol = "query"
primary_table = "positions"
access = "market"             # role must have "market" right

[services.orders]
protocol = "transaction"
access = "auth"               # any logged-in user (service-level default)

[services.orders.ops.place]
table = "orders"
op_type = "insert"
fields = ["symbol", "qty"]

[services.orders.ops.place.access]
trade_limited = "accounts WHERE username = :user AND balance >= :qty * :price"
trade_full = true

access values: "open" (no auth), "auth" (any logged-in user), "right_name" (role must have this right), or a dict mapping rights to SQL pre-check conditions.

Rights table

The _mkio_rights table maps roles to rights. Adding a new role is a table insert — no config change:

role       right
trader     market
trader     trade_limited
senior     market
senior     trade_full
admin      admin

Auth protocol

Clients authenticate after connecting:

{"type": "auth", "data": {"username": "alice", "password": "secret"}}
→ {"type": "auth", "ok": true, "user": "alice", "role": "trader"}

SQL pre-checks (when)

Dict-form access maps rights to SQL conditions. The condition is executed as SELECT 1 FROM <condition> LIMIT 1 — if it returns a row, access is granted. Bind parameters available: :user, :role, :topic (subpub), submitted data fields (:symbol, :qty, etc.), and any extra columns from the user's _mkio_users row.

For transaction ops, pre-checks run inside the same SAVEPOINT as the write — ACID-safe.

Custom auth

For JWT, LDAP, or other auth backends, use on_auth() instead of _mkio_users:

app = create_app(config)

async def my_auth(data):
    user = verify_jwt(data["token"])
    return {"user": user.name, "role": user.role}

app.on_auth(my_auth)
app.run()

Rights enforcement still uses the _mkio_rights table regardless of auth method.

Monitor access

When auth is enabled, monitoring is disabled by default. Set the top-level monitor_access key to control who can use mkio monitor:

monitor_access = "admin"    # role must have "admin" right

monitor_access accepts the same values as service access: "open" (no auth needed), "auth" (any logged-in user), a right name, or a dict with SQL pre-checks. Without this key, all monitor requests return "monitoring disabled".

CLI authentication

All WebSocket-based CLI commands accept --username for authentication. The password is read from the MKIO_PASSWORD environment variable, or prompted interactively if not set:

mkio monitor 8080 --username alice
mkio send 8080 orders --op new '{"symbol":"AAPL"}' --username alice
mkio subpub 8080 prices AAPL --username bob
mkio query 8080 all_orders --username bob
mkio stream 8080 audit_feed --username bob
mkio reqrep 8080 lookup name=test --username bob
mkio schema 8080 orders --username bob
mkio check 8080 --username bob

For scripted/CI use, set the MKIO_PASSWORD environment variable:

MKIO_PASSWORD=secret mkio send 8080 orders --op new '{"symbol":"AAPL"}' --username alice

Bootstrap

mkio adduser alice trader              # prompts for password (or reads MKIO_PASSWORD)
mkio adduser admin1 admin my.toml      # use a specific config file
mkio hashpass                          # generate a hashed password for seed files

Install bcrypt for strong password hashing: pip install mkio[auth]. Without it, PBKDF2 with per-password random salt is used as a fallback.

Programmatic API

For applications that embed mkio (e.g., adding custom HTTP routes, registering custom services, or controlling server lifecycle), use create_app() instead of serve().

create_app(config, *, routes=None) -> MkioApp

Creates a server instance from a TOML file path or config dict.

import asyncio
from aiohttp import web
from mkio import create_app

async def health(request: web.Request) -> web.Response:
    return web.json_response({"ok": True})

app = create_app("server.toml", routes=[
    ("GET", "/health", health),
    ("POST", "/api/custom", my_handler),
])

# Blocking (like serve())
app.run()

# Or async for non-blocking control
async def main():
    await app.start()       # binds port, begins serving
    # ... do other async work ...
    await app.wait()        # blocks until stop() or signal

asyncio.run(main())

MkioApp methods

Supported HTTP methods for routes: GET, POST, PUT, DELETE, PATCH, HEAD, OPTIONS.

Properties marked Unstable expose internal types that may change across versions. The facade methods (execute, query, subscribe) are the stable interface for the same operations.

Custom services

Register a Service subclass to add custom behavior reachable over WebSocket:

from mkio import create_app, Service

class AuditService(Service):
    async def start(self):
        self._q = self.bus.subscribe(["orders"])
        import asyncio
        asyncio.create_task(self._watch())

    async def _watch(self):
        while True:
            event = await self._q.get()
            print(f"order changed: {event.row}")

app = create_app("server.toml")
app.add_service("audit", AuditService)
app.run()

Custom services go through the same lifecycle as config-driven services — start() is called at server startup, monitors work, and WS dispatch routes messages to on_subscribe() / on_message().

Data facade

Write data, read data, and react to changes without going through WebSocket:

app = create_app(config)
await app.start()

# Write through the normal write path (WriteBatcher → ChangeBus → subscribers)
result = await app.execute("orders", {"symbol": "AAPL", "qty": 100}, op="new")

# Read
rows = await app.query("SELECT * FROM orders WHERE symbol = ?", ("AAPL",))

# React to changes
from mkio import ChangeEvent

async def on_change(event: ChangeEvent):
    print(f"{event.table} {event.op}: {event.row}")

unsub = await app.subscribe(["orders"], on_change)
# later: unsub()

Lifecycle hooks

Register callbacks for server and connection events:

app = create_app("server.toml")

async def setup():
    print("Server ready")

async def teardown():
    print("Server stopping")

async def on_connect(ws):
    print(f"Client connected")

async def on_disconnect(ws):
    print(f"Client disconnected")

app.on_startup(setup)
app.on_shutdown(teardown)
app.on_connect(on_connect)
app.on_disconnect(on_disconnect)
app.run()

Multiple callbacks can be registered for each hook — they are called in registration order.

get_default_config() -> dict

Returns the default scaffold config (what mkio init writes) as a Python dict. Modify it before passing to create_app():

from mkio import create_app, get_default_config

cfg = get_default_config()
cfg["port"] = 9090
cfg["db_path"] = ":memory:"
cfg["tables"]["widgets"] = {
    "columns": {"id": "INTEGER PRIMARY KEY", "name": "TEXT"},
}
cfg["services"]["widgets"] = {
    "protocol": "subpub",
    "primary_table": "widgets",
    "topic": "id",
}
app = create_app(cfg)
app.run()

init(directory, *, no_static=False) -> list[Path]

Programmatic equivalent of mkio init. Creates project files and returns the list of paths created.

from mkio import init

created = init("./my-project")
# [Path('my-project/server.toml'), Path('my-project/static/index.html')]

created = init("./api-only", no_static=True)
# [Path('api-only/server.toml')]

Raises FileExistsError if server.toml already exists in the target directory.

Service Types

Transaction

Execute INSERT, UPDATE, DELETE, or UPSERT operations. Supports multi-table atomic transactions with named ops and cross-op bind references.

[services.orders]
protocol = "transaction"

[services.orders.ops]
new = [
    { table = "orders", op_type = "insert", fields = ["side", "symbol", "qty", "price"] },
    { table = "audit_log", op_type = "insert", defaults = { event = "new" }, bind = { order_id = "$0.id", status = "$0.status" } },
]
accept = [
    { table = "orders", op_type = "update", key = ["id"], fields = ["status"], defaults = { status = "accepted" } },
    { table = "audit_log", op_type = "insert", defaults = { event = "accepted" }, bind = { order_id = "$0.id", status = "$0.status" } },
]

Clients select a named set by sending "op": "new" (or "accept", etc.) in the transaction message. For a service with only one workflow, ops may instead be a plain list — clients then omit the op field.

Bind references ($N.field) pull values from a prior op's RETURNING row, where N is the zero-based index of an earlier op in the same op set. Only insert, update, and upsert ops produce RETURNING rows that can be bound against. Op-level defaults provide static values the client doesn't need to send — here, event and status are set automatically per operation.

SubPub

Subscribe by topic (the topic column value) to get a single-row snapshot, then receive live updates as data changes. Every published row includes three framework fields: _mkio_exists (whether the topic was found), _mkio_topic (the subscribed topic value), and _mkio_ref (the timestamp ref of the last write, or null for not-found topics). Supports server-side where filtering (rows that don't match are never cached or published; once cached, a row that stops matching is frozen at its last matching state — no eviction, no notification), publish formatting with expressions, configurable defaults (expression strings) for topics that don't exist yet, and custom sql for computed topics or JOINs.

[services.last_trade]
protocol = "subpub"
primary_table = "orders"
topic = "symbol"
where = "status == 'filled'"
change_log_size = 10000

[services.last_trade.defaults]
price = "0"
time = "''"

[services.last_trade.publish]
symbol = "symbol"
price = "IF(side == 'Buy', price, -price)"

Use sql with a computed column when the topic doesn't map 1:1 to an existing column:

[services.last_trade_by_side]
protocol = "subpub"
primary_table = "orders"
topic = "topic_key"
sql = "SELECT *, symbol || ':' || side AS topic_key FROM orders"
where = "status == 'filled'"

Clients subscribe with topic: "AAPL:Buy". The topic must name a column in the sql result set.

Stream

Append-only data with ring buffer and ref-based cursor reconnection. Supports forward and backward pagination through the buffer.

[services.audit_feed]
protocol = "stream"
primary_table = "audit_log"
buffer_size = 10000

Query

Snapshot from SQLite with change feed. Every published row includes _mkio_row (primary key identifier) and _mkio_ref (timestamp ref of the last write).

[services.all_orders]
protocol = "query"
primary_table = "orders"
filterable = ["status"]

ReqRep

One-shot request-reply: the client sends a request with data, the server evaluates configured SQL and/or expressions, and returns a reply. No subscriptions or change feeds — pure request-reply. Supports three reply shapes determined by config:

# Scalar from expression
[services.tax]
protocol = "reqrep"
reply = "ROUND(qty * price * rate, 2)"

# Single record from expressions
[services.invoice]
protocol = "reqrep"
reply = { subtotal = "qty * price", tax = "ROUND(qty * price * 0.08, 2)" }

# SQL result set with params manipulation
[services.search]
protocol = "reqrep"
params = { symbol = "UPPER(symbol)" }
sql = "SELECT * FROM prices WHERE symbol = :symbol"

# SQL rows with per-row transform
[services.holdings]
protocol = "reqrep"
sql = "SELECT p.*, pr.price FROM positions p JOIN prices pr ON p.symbol = pr.symbol WHERE p.account = :account"
reply = { symbol = "symbol", qty = "qty", market_value = "ROUND(qty * price, 2)" }

Connection Identity (_mkio)

Every mkio server automatically registers a built-in _mkio reqrep service (no config required). Clients can verify they're connected to the correct server by sending a request:

{"type": "request", "service": "_mkio", "reqid": "hello"}

Reply:

{
  "type": "reply", "service": "_mkio", "reqid": "hello",
  "row": {
    "name": "order-book-dev",
    "version": "2.1.0",
    "mkio": "0.2.0",
    "protocol": "1.0",
    "expr": "1",
    "services": {"orders": "transaction", "last_trade": "subpub", "all_orders": "query"},
    "tables": ["orders", "audit_log"],
    "config_hash": "a3f7c2b1",
    "uptime": 3621.4,
    "started": "20260517 08:12:03.000000000000"
  }
}

Set name and version in your config to identify the application:

name = "order-book-dev"
version = "2.1.0"
port = 8080

From the CLI: mkio reqrep 8080 _mkio. From the browser console: mkio.reqrep("_mkio").

Version Compatibility

Clients can check whether they're compatible with the server by sending expected version(s) in the request data. The server replies with a compatible boolean (AND of all checks) and a compatibility dict with per-version results. version, protocol, and mkio use semantic versioning (caret ^ convention); expr must match exactly.

{"type": "request", "service": "_mkio", "reqid": "v1",
 "data": {"version": "2.0.0", "protocol": "1.0", "mkio": "0.2.0", "expr": "1"}}

Reply:

{
  "type": "reply", "service": "_mkio", "reqid": "v1",
  "row": {
    "name": "order-book-dev", "version": "2.3.0", "mkio": "0.2.0", "protocol": "1.0", "expr": "1",
    "compatible": true,
    "compatibility": {"version": true, "protocol": true, "mkio": true, "expr": true},
    ...
  }
}

If any version is incompatible, compatible is false and the failing key(s) show false in compatibility. When no version expectations are sent, neither field appears (backward compatible).

From the CLI: mkio check 8080 version=2.0.0 protocol=1.0 expr=1. From the browser console: mkio.check({version: "2.0.0", protocol: "1.0", expr: "1"}). The CLI exits with code 0 if compatible, 1 if not.

When authentication is enabled, _mkio still answers before login — with identity and compatibility fields only (name, version, mkio, protocol, expr, compatible, compatibility), so mkio check and client verification can pick a server without credentials and without learning its service or table names. Any authenticated user, whatever their rights, gets the full reply and the schema query.

Error replies to requests always echo reqid (and service), including access denials, so clients can correlate them; both clients also settle their oldest pending request if an older server sends an error without one.

The _mkio service is hidden from /api/services and error hints. Using the wrong protocol (e.g., mkio subpub 8080 _mkio) returns a nack with a hint suggesting the correct command.

Table Schema

Clients can query the schema of any table by sending data with a "table" key. The server returns column definitions from the live database:

{"type": "request", "service": "_mkio", "reqid": "s1", "data": {"table": "orders"}}

Reply:

{
  "type": "reply", "service": "_mkio", "reqid": "s1",
  "row": {
    "table": "orders",
    "columns": [
      {"name": "id", "type": "TEXT", "notnull": false, "pk": true, "dflt_value": null},
      {"name": "symbol", "type": "TEXT", "notnull": true, "pk": false, "dflt_value": null},
      {"name": "qty", "type": "INTEGER", "notnull": false, "pk": false, "dflt_value": null},
      {"name": "status", "type": "TEXT", "notnull": false, "pk": false, "dflt_value": "'pending'"},
      {"name": "_mkio_ref", "type": "TEXT", "notnull": false, "pk": false, "dflt_value": "''"}
    ]
  }
}

Unknown tables return an error listing available tables. From the CLI: mkio schema 8080 orders. From the browser console: mkio.schema("orders").

WebSocket Protocol

Connect to /ws (general) or /ws/{service_name} (per-service).

// Authenticate (before sending other messages to auth-protected services)
{"type": "auth", "data": {"username": "alice", "password": "secret"}}
// → {"type": "auth", "ok": true, "user": "alice", "role": "trader"}

// Transaction
{"service": "add_order", "ref": "...", "data": {"id": "1", "symbol": "AAPL", "qty": 100}}

// Named op transaction
{"service": "orders", "ref": "...", "op": "new", "data": {"side": "Buy", "symbol": "AAPL", "qty": 100, "price": 150}}

// Transaction with txnid (echoed back on result/error for async correlation)
{"service": "orders", "ref": "...", "op": "new", "txnid": "req-42", "data": {"side": "Buy", "symbol": "AAPL", "qty": 100, "price": 150}}

// Subscribe (subpub — topic required, protocol required; string or array)
{"service": "last_trade", "type": "subscribe", "protocol": "subpub", "topic": "AAPL"}
{"service": "last_trade", "type": "subscribe", "protocol": "subpub", "topic": ["AAPL", "MSFT", "GOOG"]}

// Subscribe (query — with filter)
{"service": "all_orders", "type": "subscribe", "protocol": "query", "filter": "status == 'pending'"}

// Subscribe with subid (echoed on every snapshot and update for this subscription)
{"service": "all_orders", "type": "subscribe", "protocol": "query", "subid": "my-sub-1"}

// Subscribe with field projection (receive only specified columns)
{"service": "all_orders", "type": "subscribe", "protocol": "query", "fields": ["symbol", "qty"]}

// Subscribe with pagination (server sends at most N rows per snapshot message)
{"service": "all_orders", "type": "subscribe", "protocol": "query", "maxcount": 50, "subid": "q1"}
// → {"type": "snapshot", "service": "all_orders", "subid": "q1", "rows": [...], "hasmore": true}

// Request next page (subid required to identify the subscription)
{"service": "all_orders", "type": "getmore", "subid": "q1"}
// → {"type": "snapshot", "service": "all_orders", "subid": "q1", "rows": [...], "hasmore": false}
// Once hasmore is false, live updates begin flowing

// Stream (ref resumes from that point; omit ref to start from beginning of buffer)
{"service": "audit_feed", "type": "subscribe", "protocol": "stream", "ref": "20260404 15:30:45.123456000000"}

// Stream with pagination (stateless — no getmore, just re-subscribe with returned ref)
{"service": "audit_feed", "type": "subscribe", "protocol": "stream", "maxcount": 100}
// → {"type": "snapshot", "service": "audit_feed", "ref": "<last-row-ref>", "rows": [...], "hasmore": true}
// Next page: subscribe again with ref from previous response
{"service": "audit_feed", "type": "subscribe", "protocol": "stream", "ref": "<last-row-ref>", "maxcount": 100}
// Once hasmore is false, subscribe without maxcount to go live

// Stream backward pagination (rows before a ref, newest N closest to the ref)
{"service": "audit_feed", "type": "subscribe", "protocol": "stream", "ref": "<ref>", "before": true, "maxcount": 20}
// → {"type": "snapshot", "ref": "<earliest-row-ref>", "rows": [...], "hasmore": true}
// Next page backward: subscribe again with returned ref and before: true
{"service": "audit_feed", "type": "subscribe", "protocol": "stream", "ref": "<earliest-row-ref>", "before": true, "maxcount": 20}

// ReqRep — one-shot request-reply (reqid echoed on reply for correlation)
{"service": "tax", "type": "request", "reqid": "r1", "data": {"qty": 10, "price": 99.95, "rate": 0.08}}
// → {"type": "reply", "service": "tax", "reqid": "r1", "value": 79.96}

{"service": "search", "type": "request", "reqid": "r2", "data": {"symbol": "AAPL"}}
// → {"type": "reply", "service": "search", "reqid": "r2", "rows": [{"symbol": "AAPL", ...}]}

Client Libraries

Python

from mkio.client import MkioClient

async with MkioClient("ws://localhost:8080/ws") as client:
    # Authenticate (credentials stored for auto-re-auth on reconnect)
    await client.auth({"username": "alice", "password": "secret"})

    result = await client.send("add_order", {"id": "1", "symbol": "AAPL", "qty": 100})

    async for msg in client.subscribe("last_trade", "subpub", topic="AAPL"):
        print(msg)  # single row with _mkio_exists, _mkio_topic, _mkio_ref

    async for msg in client.subscribe("last_trade", "subpub", topic=["AAPL", "MSFT"]):
        print(msg)  # snapshot with one row per topic, then individual updates

    async for msg in client.subscribe("all_orders", "query", filter="status == 'pending'"):
        print(msg)

    # Paginated query (client auto-sends getmore until snapshot complete)
    async for msg in client.subscribe("all_orders", "query", maxcount=50):
        print(msg)

    # Stream — backward pagination (last 20 rows before a ref)
    async for msg in client.subscribe("audit_feed", "stream", before=True, ref=some_ref, maxcount=20):
        print(msg)

    # ReqRep — one-shot request-reply (auto-generates reqid)
    result = await client.request("tax", {"qty": 10, "price": 99.95, "rate": 0.08})
    print(result)  # {"type": "reply", "value": 79.96, ...}

JavaScript

Auto-served at /mkio.js — no CDN or bundler needed.

<script src="/mkio.js"></script>
<script>
const client = new MkioClient("ws://localhost:8080/ws");
await client.connect();

// Authenticate (credentials stored for auto-re-auth on reconnect)
await client.auth({username: "alice", password: "secret"});

client.subscribe("last_trade", "subpub", {
    topic: "AAPL",
    onSnapshot: (rows) => renderTrade(rows[0]),
    onUpdate: (op, row) => renderTrade(row),
    onNack: (message) => console.error("Subscription rejected:", message),
});

client.subscribe("all_orders", "query", {
    filter: "status == 'pending'",
    onSnapshot: (rows) => renderTable(rows),
    onUpdate: (op, row) => updateRow(op, row),
});

// Paginated query (client auto-sends getmore; onSnapshot fires once with all rows)
client.subscribe("all_orders", "query", {
    maxcount: 50,
    onSnapshot: (rows) => renderTable(rows),
    onUpdate: (op, row) => updateRow(op, row),
});
</script>

Compatibility: Runs in all evergreen browsers (Chrome, Edge, Firefox, Safari) with no polyfills. Also works in Node.js ≥22, where WebSocket, TextDecoder, and performance are available as globals. On Node 18–21, assign a WebSocket polyfill to globalThis before importing:

globalThis.WebSocket = require("ws");
const { MkioClient } = require("./mkio.js");

The file uses CommonJS module.exports; load it via require(...) in Node, or <script src="/mkio.js"> in the browser.

Expressions in the browser

The expression language is also served, as an ES module, at /mkio-expr.js:

<script src="/mkio.js"></script>
<script type="module" src="/mkio-expr.js"></script>
<script>
// after the module has loaded: globalThis.mkioExpr, also reachable as mkio.expr
const pred = mkioExpr.compileFilter("status == 'pending' && qty > 100");
const label = mkioExpr.compileTemplate("${symbol}: ${NUM(qty * price, digits: 2, group: TRUE)}");
rows.filter((r) => pred(r)).map((r) => label.call(r));
</script>

or import { compile, compileTemplate } from "/mkio-expr.js" from your own module. With it loaded, mkio.monitor({filter: "direction == 'in'"}) accepts expression strings.

Debugging from the browser console

Once /mkio.js is loaded, a mkio object is available in DevTools with methods that mirror the mkio CLI (the <url> argument is dropped since the page already holds the connection):

mkio.help()                                // show help
mkio.auth({username: "alice", password: "secret"})  // authenticate
mkio.services()                            // list every service on the server
mkio.services("orders")                    // detail for one service
mkio.monitor()                             // log every frame to/from any service
mkio.monitor("orders")                     // filter to one service (call again to add more)
mkio.monitor({filter: e => e.direction === "in"})  // filter with a function
mkio.monitor({filter: "direction == 'in'"})       // ...or an expression (needs /mkio-expr.js)
mkio.monitor("off")                        // stop
mkio.send("orders", {side:"Buy",...}, {op:"new"})
mkio.subpub("last_trade", "AAPL")
mkio.subpub("last_trade", ["AAPL","MSFT","GOOG"])
mkio.subpub("last_trade", "AAPL", {fields:["bid","ask"], subid:"p1"})
mkio.stream("audit_feed")                 // ref auto-generated
mkio.stream("audit_feed", {ref:"...", filter:"qty > 100"})
mkio.stream("audit_feed", {before: true, ref:"...", maxcount: 20})  // backward
mkio.query("all_orders", {filter:"status == 'pending'"})
mkio.query("all_orders", {maxcount: 50})     // paginated snapshot
mkio.query("all_orders", {snapshotOnly: true})
mkio.query("all_orders", {updateOnly: true, fields:["id","status"]})
mkio.reqrep("tax", {qty: 10, price: 99.95, rate: 0.08})
mkio.reqrep("search", {symbol: "AAPL"})
mkio.schema("orders")                   // table schema (columns, types, keys)

All subscribe methods return a MkioSubscription with .stop(). Nack responses are logged to the console by default. Console commands auto-generate subid (subscriptions) and txnid (sends) with a _mkio_ prefix so they never intercept messages meant for the application.

mkio.monitor(...) only taps this tab's traffic. For traffic across all connected clients use the CLI's server-side mkio monitor instead.

Expression Language

One small, safe expression language is used everywhere mkio evaluates something per row or per request: client filters, server-side where, publish, defaults, reqrep params and reply, and the CLI's --filter. The same language ships as a JavaScript module (/mkio-expr.js), so a browser UI evaluates the identical grammar; both implementations run the shared conformance fixtures in tests/expr_cases.json, and the _mkio identity reply carries expr (the language version, currently "1").

qty * price > 1000 && status == 'open'
IF(side == 'Buy', price, -price)
qty * price |> (sub -> NUM(sub * 0.001, digits: 2))
items |> (xs -> SUM(MAP(xs, i -> i.qty * i.price)))
meta.region ?? 'n/a'

Syntax

Operators, lowest to highest precedence:

Semantics: falsy values are NULL, FALSE, 0, '', [], {}; everything else is truthy. &&, ||, ?? and the IF/CASE/TRY/LET functions evaluate only what they need, so qty > 0 && price / qty > 10 never divides by zero. Errors (unknown field, division by zero, type mismatch, bad index type) abort the expression with a message and position; TRY(expr, fallback) catches them. There are no statements, loops, assignments, or side effects — name intermediate results with LET(name, value, ..., body) or a pipe.

Scope and strictness. Bare names resolve in the scope the host supplies — the row, for server filters and formatters. The server compiles in strict mode: an unknown root name is an error listing the available fields. Hosts may choose a lenient environment where unknown names are NULL (a UI over heterogeneous rows, say). Missing map keys, out-of-range indexes, and indexing into NULL yield NULL in both modes, so meta.region ?? 'n/a' always works. Keys beginning with __ (and constructor / prototype) are never accessible.

Templates

compile_template handles strings with embedded expressions: "Order #${id} for ${UPPER(symbol)}". A template that is exactly one ${...} returns the expression's raw value (type preserved); any other template returns a string, with NULL rendering as empty. Write $${ for a literal ${.

Standard library

All libraries are enabled by default. Function names are shown upper-case by convention; upper(s) is the same function.

core

Control flow, coercion, and safe lookup. IF, CASE, TRY, and LET are lazy — only the branches they take are evaluated.

math

Numeric helpers. ROUND rounds half away from zero on the decimal representation (ROUND(2.675, 2) is 2.68).

string

REPLACE and SPLIT take literal strings; only MATCHES takes a regular expression. Lengths and positions count characters (code points).

format

Numbers to display strings. Rounding follows ROUND; no locale is involved.

time

The numeric unit is seconds since the Unix epoch. EPOCH accepts numbers, mkio ref strings ("20260828 12:34:56.123456789000"), and ISO-8601. Formats understand %Y %m %d %H %M %S %f %z %% (%f = microseconds); time zones are 'UTC' (default), 'local', or a fixed offset such as '+09:00'.

collection

Arrays and maps; the higher-order functions take lambdas.

Extending the language

Applications register functions, bundles of functions, and value types. Registered names are callable from every expression the server evaluates, and appear in the functions list of the service detail API.

from mkio import expr

# A plain function — arguments arrive evaluated
expr.register_function("MASK_PAN", lambda s: "****" + s[-4:], doc="Mask all but the last 4 digits")

# A library: metadata drives docs, named arguments, and the numeric-field analysis
expr.register_library("risk", {
    "VAR":    (lambda pos, conf=0.99: value_at_risk(pos, conf), {"numeric": True, "params": ("pos", "conf")}),
    "BUCKET": (lambda x, edges: sum(1 for e in edges if x >= e), {"numeric": True}),
})

# A lazy function receives unevaluated arguments (Arg thunks with .value(), .eval(scope), .name)
def when(ctx, cond, then):
    return then.value() if expr.truthy(cond.value()) else None
expr.register_function("WHEN", when, lazy=True)

# A host value type: teach the operators about it
from decimal import Decimal
expr.register_type("decimal", is_instance=lambda v: isinstance(v, Decimal),
                   add=lambda a, b: a + b, to_string=str,
                   compare=lambda a, b: (a > b) - (a < b))
# A `concat=` hook lets a type survive "text ${x}" template joining (mkui's
# rich text uses it); without one the type renders through to_string.

# Use the engine directly, with a custom environment
env = expr.Env(libraries=["core", "math", "risk"], strict=True)
alert = expr.compile("VAR(positions, conf: 0.95) > limit", env)
alert({"positions": pos, "limit": 1e6})
msg = expr.compile_template("VaR ${NUM(VAR(positions), digits: 0, group: TRUE)}", env)

Static analysis is available for validation and dependency tracking: expr.field_refs(ast), expr.function_refs(ast), expr.numeric_fields(ast) (fields used in numeric contexts — driven by each function's numeric flag). expr.parse(src) returns the AST; expr.compile(src, env) validates function names against the environment, so a typo fails at config load, not at request time.

The JavaScript module exposes the same API with camelCase names — compile, compileTemplate, compileFilter, registerFunction, registerLibrary, registerType, Env, fieldRefs, … — as ES-module exports and as globalThis.mkioExpr. Lazy JS functions are called as fn(ctx, args, kwargs).

Performance

• Write batching — collects writes over a 2ms window, commits as single SQLite transaction with per-request SAVEPOINTs
• WAL mode — dual connections (write + read) for concurrent reads during writes
• Zero-copy fan-out — change events serialized once, same bytes sent to all subscribers
• Optional acceleration — pip install mkio[fast] for orjson (5-10x JSON) and uvloop (2-4x I/O)

CLI Tools

The URL argument defaults to localhost, http://, and port 80, so you can use shorthand:

mkio services 8080                      # same as http://localhost:8080
mkio services localhost:8080            # same as http://localhost:8080
mkio services myhost                    # same as http://myhost:80
mkio services https://prod.example.com  # uses port 443, wss for WebSocket

List and inspect services

mkio services localhost:8080                # List all services
mkio services localhost:8080 orders         # Show detail for one service

Detail view shows fields, types, required/optional, auto-generated columns, and example commands.

Send transactions

mkio send localhost:8080 orders --op new '{"side":"Buy","symbol":"AAPL","qty":100,"price":150}'
mkio send localhost:8080 orders --op new orders.json    # From JSON file
mkio send localhost:8080 orders --op new orders.csv     # From CSV file
mkio send localhost:8080 orders mixed.csv               # CSV with per-row op column

Subscribe to live data

Each listener service type has its own command with only the relevant options:

# SubPub — topic-based snapshot + live updates
mkio subpub localhost:8080 last_trade AAPL
mkio subpub localhost:8080 last_trade AAPL MSFT GOOG
mkio subpub localhost:8080 last_trade AAPL --fields symbol,price

# Stream — ring buffer with cursor reconnect (ref defaults to now)
mkio stream localhost:8080 audit_feed
mkio stream localhost:8080 audit_feed --ref "20260404 15:30:45.123456000000"
mkio stream localhost:8080 audit_feed --fields event,order_id
mkio stream localhost:8080 audit_feed --maxcount 100    # page forward from beginning of buffer
mkio stream localhost:8080 audit_feed --before --ref "<ref>" --maxcount 20  # page backward

# Query — snapshot + live updates
mkio query localhost:8080 all_orders
mkio query localhost:8080 all_orders --filter "status == 'pending'"
mkio query localhost:8080 all_orders --fields symbol,qty --snapshotOnly

# ReqRep — one-shot request-reply
mkio reqrep localhost:8080 tax '{"qty": 10, "price": 99.95, "rate": 0.08}'
mkio reqrep localhost:8080 search symbol=AAPL

Monitor traffic

Tap into inbound and outbound message flow in real time. Monitor a single service or all services at once:

mkio monitor localhost:8080                 # Monitor all services
mkio monitor localhost:8080 orders          # Monitor one service
mkio monitor localhost:8080 --filter "direction == 'in'"    # Inbound only
mkio monitor localhost:8080 --filter "service == 'orders'"  # Filter by service
mkio monitor localhost:8080 --filter "direction == 'out' && CONTAINS(['snapshot', 'update'], message.type)"

[2026-04-04 15:30:45.123456 -0400] >> IN  subscribe
{ "type": "subscribe", "service": "last_trade", "protocol": "subpub" }

[2026-04-04 15:30:45.125789 -0400] << OUT snapshot
{ "type": "snapshot", "rows": [...] }

The --filter flag accepts any expression from the expression language, evaluated against each monitor envelope (direction, service, message).

The monitor protocol is a native framework feature — any mkio application supports it.

Inspect table schema

mkio schema localhost:8080 orders          # Show columns, types, keys, defaults

Schema management

mkio dbupdate                       # Apply safe schema changes
mkio dbupdate --allow-risky         # Include potentially destructive changes
mkio dbupdate --allow-destructive   # Include all changes
mkio dbupdate custom.toml           # Use a specific config file

Manage users

mkio adduser alice trader               # Add user, prompts for password
mkio adduser admin1 admin custom.toml   # Use a specific config file
mkio hashpass                           # Generate a hashed password for seed files
MKIO_PASSWORD=secret mkio adduser bob viewer  # Non-interactive (CI/scripts)

Initialize a project

mkio init                           # Create server.toml + static/ in current directory
mkio init ./my-project              # Create in a specific directory
mkio init --no-static               # Config only, no static/index.html

Error handling

All CLI commands show clean error messages instead of Python tracebacks. Common scenarios:

• Server not running — Error: could not connect to ... Is the mkio server running?
• Port already in use — Error: address already in use ... Stop the other process or change the port in server.toml
• Invalid TOML — Error: invalid TOML in server.toml ...
• Invalid config — Error: invalid config: ...

Use --traceback (or MKIO_TRACEBACK=1) to show the full Python traceback for debugging.

Config Endpoint

The [config] section maps routes to directories, with automatic TOML-to-JSON conversion. This keeps [static] strictly for static assets.

[config]
"/config" = "./configs"
"/settings" = "./settings"

Behavior:

• GET /config/app.json — reads ./configs/app.toml, parses it, and serves as application/json
• If no .toml file exists, falls back to serving ./configs/app.json directly
• GET /config/style.css — serves the file as-is (no conversion for non-.json extensions)
• Subdirectories are supported: GET /config/sub/db.json reads ./configs/sub/db.toml
• Multiple routes map to independent directories
• Path traversal is blocked

Config Validation

mkio validates your TOML config at load time and fails fast with clear error messages:

• Table references — primary_table, watch_tables, and op table fields must reference tables defined in [tables]
• Column references — op fields, key, defaults, bind columns, filterable, and subpub topic are checked against table schemas
• Protocol validation — service protocol must be a known type (transaction, subpub, stream, query, reqrep)
• Required fields — missing protocol, primary_table, topic, ops, or key (for update/delete/upsert) are caught immediately
• Bind references — forward references and out-of-bounds op indices in $N.field binds are rejected
• Typo detection — unknown config keys produce warnings with "did you mean?" suggestions

Runtime error messages include context to help debugging:

• Unknown service/op errors list available options
• Missing transaction fields show the op name and list provided fields
• Expression errors list available fields
• Requests to unknown services return nack (not generic errors), with the service name echoed back

Table Seeding

Tables can be populated with initial data from a file when first created. Add seed to any table config:

[tables.products]
columns = { id = "TEXT PRIMARY KEY", name = "TEXT NOT NULL", price = "REAL" }
seed = "data/products.csv"

Supported formats: .csv, .json (array of objects), .jsonl (one JSON object per line).

Seed data is loaded only when the table is first created — not on every restart. If a user deletes rows, they stay deleted. Each seeded row gets its own unique _mkio_ref timestamp.

Path resolution:

Error handling: Seed errors are fatal. Missing files, bad format, or unknown column names prevent the server from starting — same behavior as other config errors.

Schema Migration

When the config schema changes, mkio detects and classifies each difference:

By default, mkio serve refuses to start if the database schema differs from config. Use mkio dbupdate to apply changes explicitly:

mkio dbupdate                       # Apply safe changes only
mkio dbupdate --allow-risky         # Also apply potentially destructive changes
mkio dbupdate --allow-destructive   # Apply all changes (including data loss)

For automatic migration on startup, set auto_migrate in config:

auto_migrate = "safe"          # Apply safe changes on startup (same as true)
auto_migrate = "risky"         # Also apply potentially destructive
auto_migrate = "destructive"   # Apply all changes on startup

License

Apache-2.0