FastAPI router and thread-safe job engine for streaming G-code to GRBL controllers. A library, not an application: mount the router in your own FastAPI app and run it with the ASGI server of your choice. One active job per serial port, live status, and a retained terminal result per port.
Part of the pygrbl family: pygrbl-build (generate G-code) → pygrbl-streamer (stream to GRBL) → pygrbl-server (expose over HTTP).
pip install pygrbl-serverDependencies: fastapi and pygrbl-streamer. Bring your own ASGI server.
from contextlib import asynccontextmanager
from fastapi import FastAPI
from pygrbl_server import JobRegistry, create_router
registry = JobRegistry() # streamer_factory(port, baudrate), defaults to GrblStreamer
@asynccontextmanager
async def lifespan(app: FastAPI):
yield
registry.shutdown()
app = FastAPI(lifespan=lifespan)
app.include_router(create_router(registry))uvicorn yourmodule:app --host 0.0.0.0 --port 8000Events are logged to the pygrbl_server logger (alarms and failures at
WARNING/ERROR). Inject a custom streamer_factory to swap the streamer backend.
| Endpoint | Description |
|---|---|
POST /stream |
Start a job: {"port": "...", "files": [...]}. 202 started, 409 port busy, 404 file missing. |
POST /stop?port= |
Cancel the active job. 404 if none. |
POST /pause?port=&paused= |
Feed hold (true, default) or resume (false). 404 if none. |
GET /status?port= |
Port state; omit port for all ports. |
Status is idle, working, paused, or the retained done / error of the
last finished job (kept until a new job starts on that port):
curl 'localhost:8000/status?port=/dev/ttyUSB0'
# {"port": "/dev/ttyUSB0", "status": "working", "progress": 42, "error": null}No authentication. Any client that can reach the port can control physical
machines capable of causing fires or injury. Run on a trusted LAN only. The
files in /stream are paths on the server's filesystem, not uploads. Auth,
TLS, and rate limiting belong in the app that mounts the router.
The library streams files and reports state; everything else lives in your layers:
- Machine setup (homing, air assist, preambles) belongs in your G-code.
- Delays between files:
M5+G4 P<seconds>in your G-code. - Notifications: poll
/statusor attach a handler to thepygrbl_serverlogger. - Runner, CLI, auth, CORS: your app's job — this ships as a router on purpose.