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

fromglinximportGlinxapp=Glinx()
@app.sensor("room_temp", protocol="mqtt", broker="localhost", topics=["home/temp"])defroom_temp(raw):
return {"temperature_c": raw["t"], "humidity_pct": raw["h"]}
@app.rule("room_temp", when="temperature_c > 40", priority="HIGH")defoverheat(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)

fromglinximportGlinxapp=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",)defultrasonic(raw):
returnraw@app.rule("ultrasonic", when="distance_cm < 30", priority="HIGH")defcollision_warning(event):
return"Obstacle within collision range"app.serve()

Serial Sensor (Arduino, Raspberry Pi)

fromglinximportGlinxapp=Glinx()
@app.sensor("imu",protocol="serial",port="COM3", # or /dev/ttyUSB0 on Linuxbaudrate=115200,fields={"ax": "accel_x", "ay": "accel_y", "az": "accel_z"},sensor_type="imu",location="robot.base",)defimu(raw):
returnraw@app.rule("imu", when="abs(accel_z) > 15", priority="HIGH")deffall_detected(event):
return"Fall detected based on Z-axis acceleration"app.serve()

Mock Sensor (testing without hardware)

fromglinximportGlinxapp=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",)deffinger(raw):
returnraw@app.rule("finger", when="contact_pressure_kPa > 50", priority="HIGH")defgrip_overload(event):
return"Grip pressure exceeded safe threshold"app.serve()

Event Callbacks (headless mode)

fromglinximportGlinxapp=Glinx()
@app.sensor("temp", protocol="mock", payloads=[{"t": 55}])deftemp(raw):
return {"temperature": raw["t"]}
@app.rule("temp", when="temperature > 50", priority="HIGH", label="hot")defhot_rule(event):
return"Too hot"@app.on_event("hot")defhandle_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_robotagent_bridge: mcpingestion:
sources:
- id: base_imuprotocol: mqttbroker: localhosttopics: [robot/imu]
- id: left_fingertipprotocol: serialport: /dev/ttyUSB0baudrate: 115200sensors:
- id: base_imutype: imulocation: robot.basefields:
ax: accel_xay: accel_yaz: accel_z
- id: left_fingertiptype: force_sensorlocation: robot.hand.left.fingertipunit: kPafields:
p: contact_pressuretemp: surface_temperatureevent_rules:
- sensor: left_fingertipcondition: contact_pressure_kPa > 50priority: HIGHlabel: grip_overloadsummary_windows:
- sensors: [base_imu, left_fingertip]interval_seconds: 5label: periodic_status

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

Supported Protocols

Protocol	Status	Platform
Mock (testing)	✅ Built-in	pip install glinx
MQTT	✅ Implemented	pip install glinx[mqtt]
Serial / UART	✅ C++ Driver	pip install glinx[serial]
I2C	✅ C++ Driver	Linux (Raspberry Pi, etc.)
SPI	✅ C++ Driver	Linux (Raspberry Pi, etc.)
CAN Bus	✅ C++ Driver	Linux (SocketCAN)
WebSocket	🔜 Planned	—
BLE	🔜 Planned	—
ROS2	🔜 Planned	—
Camera (OpenCV)	🔜 Planned	—

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