Glinx

The Universal Hardware to Agent Middleware

Glinx is a high-performance middleware framework that connects physical hardware to AI agents. It ingests sensor data from any protocol, enriches it with semantic meaning, and exposes clean MCP-compatible tools that any agent can reason over. Built for the era of physical AI and robotics with a hybrid C++/Python architecture for real-time IoT.

Wire once. Think forever.

Why Glinx?

IoT devices generate thousands of events per second. Python hits performance limits:

• Accelerometers: 100-1000 Hz
• IMUs: 400-1000 Hz
• Force sensors: 100+ Hz
• GIL bottleneck at high frequencies

Glinx solves this with a hybrid architecture:

• C++ core: Real-time sensor ingestion (<1ms latency, 10k+ msgs/sec)
• Python layer: Semantic enrichment, AI integration, MCP tools
• Zero-copy IPC: Shared memory between layers

Result: 1000x faster than pure Python, while maintaining Python's simplicity for AI/LLM work.

Quickstart

from glinx import Glinx

app = Glinx()

@app.sensor("room_temp", protocol="mqtt", broker="localhost", topics=["home/temp"])
def room_temp(raw):
    return {"temperature_c": raw["t"], "humidity_pct": raw["h"]}

@app.rule("room_temp", when="temperature_c > 40", priority="HIGH")
def overheat(event):
    return "Room temperature critically high"

app.serve()  # MCP server is live — any agent can call get_room_temp_status()

That's it. Your sensor is now an MCP tool.

Install

pip install glinx                  # core only
pip install glinx[mqtt]            # + MQTT driver (ESP32, IoT)
pip install glinx[serial]          # + Serial/UART driver (Arduino, Pi)
pip install glinx[mcp]             # + MCP server support
pip install glinx[all]             # everything

For development:

uv sync --extra dev
uv run pytest

How It Works

  Hardware (MQTT / Serial / BLE / CAN / Camera / ...)
      │
      ▼
  Glinx Ingestion Layer ─── protocol drivers normalize data
      │
      ▼
  Schema + Semantic Layer ── auto-infer types, enrich with meaning
      │
      ▼
  Event Filter Layer ─────── rules, anomaly detection, summaries
      │
      ▼
  Agent Bridge ───────────── MCP tools exposed to any LLM agent
      │
      ▼
  AI Agent (LangGraph / LangChain / AutoGen / custom)

Raw sensor data like {"p": 58.4, "temp": 31.5} becomes:

{
  "sensor": "left_fingertip",
  "sensor_type": "force_sensor",
  "location": "robot.hand.left.fingertip",
  "contact_pressure_kPa": 58.4,
  "surface_temperature_kPa": 31.5,
  "semantic_summary": "left_fingertip at robot.hand.left.fingertip: contact_pressure=58.4, surface_temperature=31.5"
}

The agent sees meaning, not bytes.

Examples

MQTT Sensor (ESP32, IoT devices)

from glinx import Glinx

app = Glinx()

@app.sensor(
    "ultrasonic",
    protocol="mqtt",
    broker="192.168.1.100",
    topics=["robot/distance"],
    fields={"dist_cm": "distance"},
    sensor_type="proximity",
    location="robot.front",
    unit="cm",
)
def ultrasonic(raw):
    return raw

@app.rule("ultrasonic", when="distance_cm < 30", priority="HIGH")
def collision_warning(event):
    return "Obstacle within collision range"

app.serve()

Serial Sensor (Arduino, Raspberry Pi)

from glinx import Glinx

app = Glinx()

@app.sensor(
    "imu",
    protocol="serial",
    port="COM3",          # or /dev/ttyUSB0 on Linux
    baudrate=115200,
    fields={"ax": "accel_x", "ay": "accel_y", "az": "accel_z"},
    sensor_type="imu",
    location="robot.base",
)
def imu(raw):
    return raw

@app.rule("imu", when="abs(accel_z) > 15", priority="HIGH")
def fall_detected(event):
    return "Fall detected based on Z-axis acceleration"

app.serve()

Mock Sensor (testing without hardware)

from glinx import Glinx

app = Glinx()

@app.sensor(
    "finger",
    protocol="mock",
    payloads=[{"p": 23.4, "temp": 31.1}, {"p": 58.4, "temp": 31.5}],
    fields={"p": "contact_pressure", "temp": "surface_temperature"},
    sensor_type="force_sensor",
    location="robot.hand.left.fingertip",
    unit="kPa",
)
def finger(raw):
    return raw

@app.rule("finger", when="contact_pressure_kPa > 50", priority="HIGH")
def grip_overload(event):
    return "Grip pressure exceeded safe threshold"

app.serve()

Event Callbacks (headless mode)

from glinx import Glinx

app = Glinx()

@app.sensor("temp", protocol="mock", payloads=[{"t": 55}])
def temp(raw):
    return {"temperature": raw["t"]}

@app.rule("temp", when="temperature > 50", priority="HIGH", label="hot")
def hot_rule(event):
    return "Too hot"

@app.on_event("hot")
def handle_hot(event):
    print(f"🔥 Alert: {event.description}")

app.run(interval=1.0)  # no MCP, just poll + fire callbacks

CLI

Run a Python app

glinx run demo:app                       # like uvicorn main:app

YAML config mode (power users)

glinx inspect-config --config glinx.yaml  # show detected sources/sensors
glinx print-tools --config glinx.yaml     # show generated MCP tool specs
glinx start --config glinx.yaml --serve-mcp  # start MCP server

YAML Configuration (Advanced)

For complex setups with many sensors, use glinx.yaml instead of decorators:

glinx:
  name: demo_robot
  agent_bridge: mcp

ingestion:
  sources:
    - id: base_imu
      protocol: mqtt
      broker: localhost
      topics: [robot/imu]
    - id: left_fingertip
      protocol: serial
      port: /dev/ttyUSB0
      baudrate: 115200

sensors:
  - id: base_imu
    type: imu
    location: robot.base
    fields:
      ax: accel_x
      ay: accel_y
      az: accel_z
  - id: left_fingertip
    type: force_sensor
    location: robot.hand.left.fingertip
    unit: kPa
    fields:
      p: contact_pressure
      temp: surface_temperature

event_rules:
  - sensor: left_fingertip
    condition: contact_pressure_kPa > 50
    priority: HIGH
    label: grip_overload

summary_windows:
  - sensors: [base_imu, left_fingertip]
    interval_seconds: 5
    label: periodic_status

glinx start --config glinx.yaml --serve-mcp

Supported Protocols

Key Features

• Decorator API — @app.sensor(), @app.rule(), like FastAPI for hardware
• Auto MCP tool generation — sensors become agent-callable tools automatically
• Semantic enrichment — raw fields get human-readable names, units, and summaries
• Event filtering — threshold rules, z-score anomaly detection, summary windows
• Config-driven or code-driven — decorators for simplicity, YAML for complex setups
• Protocol drivers — pluggable, community-extensible driver registry
• Async-native — built on asyncio for concurrent hardware ingestion

Architecture

src/glinx/
  app.py           Decorator-based Glinx API (primary interface)
  runtime.py       Orchestration layer
  config.py        YAML configuration models (Pydantic)
  models.py        Core message, event, and snapshot models
  schema.py        Schema inference engine
  semantic.py      Semantic enrichment logic
  events.py        Rule engine, anomaly detection, summary windows
  bus.py           Internal async pub/sub bus
  cli.py           CLI entrypoint (Typer)
  drivers/
    base.py        BaseDriver ABC + DriverRegistry
    mock.py        Mock driver for testing
    mqtt.py        MQTT driver (aiomqtt)
    serial.py      Serial/UART driver (pyserial-asyncio-fast)
  bridges/
    mcp.py         MCP bridge (FastMCP)
tests/
  test_app.py      Decorator API tests
  test_runtime.py  Runtime integration tests
  test_events.py   Event filter tests
  test_drivers.py  Driver tests
  test_bridge.py   MCP bridge tests

Architecture: Hybrid C++/Python

Glinx uses a hybrid architecture for maximum performance:

Hardware → C++ Core (real-time) → Python Bridge (AI/semantics) → AI Agents
           ↓ <1ms latency        ↓ Enrichment                   ↓ MCP Tools
           Lock-free buffers      Schema inference               LangGraph/etc
           10k+ msgs/sec          Event filtering

Why C++? Python hits GIL limits at high sensor frequencies:

• Accelerometers: 100-1000 Hz
• IMUs: 400-1000 Hz
• Multiple sensors: 1000s of events/second

The C++ core provides:

• ✅ < 1µs latency sensor-to-buffer
• ✅ 10,000+ msgs/sec sustained throughput
• ✅ Zero GIL contention - lock-free buffers
• ✅ Direct hardware access (I2C, SPI, Serial/UART)

See README_CPP.md for C++ core details and examples/benchmark_comparison.py for performance benchmarks.

Roadmap

v0.1 (current)

• ✅ Decorator-based API
• ✅ Mock, MQTT, Serial drivers (Python)
• ✅ C++ core with Serial, I2C, SPI drivers
• ✅ Lock-free ring buffers and IPC
• ✅ Schema inference + semantic enrichment
• ✅ Rule-based events + anomaly detection
• ✅ MCP bridge scaffold
• ✅ CLI with glinx run
• ✅ Python bindings (nanobind)
• ✅ Benchmarks and tests

v0.2

• CAN bus driver (C++)
• WebSocket driver (Python)
• BLE driver (C++)
• Camera modality handler

v0.3

• ROS2 topic bridge
• Audio modality (Whisper integration)
• LangGraph / LangChain bridges
• Web dashboard for live sensor monitoring

v1.0

• Auto-discovery for known sensor types
• Distributed multi-node runtime
• Edge deployment (Raspberry Pi, Jetson Nano)
• RTOS integration for hard real-time

Development

uv sync --extra dev
uv run pytest -v

License

MIT