Quickstart
Go from an API token to a confirmed hardware operation: list your products, send
open, and see the result arrive over the realtime connection.
Prerequisites
- A product and a Gateway on the same lot, paired with your account, and an API subscription for the product. See What you need.
- Your account's Super Admin email and password.
- For the code: Node.js 24+ with
socket.io-client(the Node examples are TypeScript, which Node 24 runs directly:node open.ts), or Python 3 withhttpxandpython-socketio.
1. Create an API token (once)
Log in as the Super Admin with POST /v2/auth/login, then
create a token with POST /v2/api-tokens. Do this by
hand, once: logging in signs that user out of the Connect app.
USER_TOKEN=$(curl -s -X POST https://api.parklio.com/v2/auth/login \
-H "Content-Type: application/json" \
-d '{"email": "admin@example.com", "password": "<your-password>"}' | jq -r .data.token)
curl -s -X POST https://api.parklio.com/v2/api-tokens \
-H "Authorization: Bearer $USER_TOKEN" \
-H "Content-Type: application/json" \
-d '{"name": "quickstart"}' | jq -r .data.token
These commands use jq to pick fields out of the JSON. The last
one prints your API token. It's shown only once: store it, and export it
for the next steps.
export PARKLIO_API_TOKEN="<your-api-token>"
More in Authentication.
2. List your products
- curl
- Node.js
- Python
curl -s https://api.parklio.com/v2/products \
-H "Authorization: Bearer $PARKLIO_API_TOKEN"
const res = await fetch('https://api.parklio.com/v2/products', {
headers: { Authorization: `Bearer ${process.env.PARKLIO_API_TOKEN}` },
});
const { data } = await res.json();
for (const p of data) console.log(p.id, p.name, p.product_type_id, p.status);
import os
import httpx
res = httpx.get(
"https://api.parklio.com/v2/products",
headers={"Authorization": f"Bearer {os.environ['PARKLIO_API_TOKEN']}"},
)
for p in res.json()["data"]:
print(p["id"], p["name"], p["product_type_id"], p["status"])
{
"success": true,
"data": [
{
"id": 123,
"device_id": "B1A2B3C",
"name": "Spot 12",
"lot_id": 4,
"zone_id": 9,
"product_type_id": 2,
"status": "online",
"power_type": "battery",
"battery_status": 87,
"rssi": -71,
"barrier": {
"state": "closed",
"is_autoclose": 1,
"is_car_on_top": 0,
"autoclose_time": 30
},
"product_error": null
}
],
"meta": { "total": 1, "per_page": 10, "current_page": 1, "last_page": 1 }
}
Pick a product and check two things:
statusisonline. A Gateway can see it, so a command can reach it.- The type block. Each product has exactly one:
barrier,gate,chain,bollardorbrain, with its currentstate. See Products and operations.
export PRODUCT_ID=123
3. Connect, send open, and wait for the result
Results arrive only over the realtime connection, so connect first, then send the
command, then wait for the event carrying your event_id.
On a barrier, open lowers the arm so the spot can be used. On a gate, chain,
bollard or brain, open opens the passage.
- curl
- Node.js
- Python
curl can't hold the realtime connection, so this sends the command only. Use the Node or Python version to see the result.
curl -s -X POST "https://api.parklio.com/v2/products/$PRODUCT_ID/state" \
-H "Authorization: Bearer $PARKLIO_API_TOKEN" \
-H "Content-Type: application/json" \
-d '{"operation": "open"}'
{ "success": true, "data": { "event_id": "a1b2c3d4" } }
import { io } from 'socket.io-client';
const token = process.env.PARKLIO_API_TOKEN;
const productId = process.env.PRODUCT_ID;
interface OperationEvent {
event_id?: string;
type?: number | 'Timeout';
message?: string;
data: Record<string, any>;
}
type Result = { ok: boolean; event: OperationEvent };
// 1. Connect first. `connect` fires once the server has accepted the token.
const socket = io('https://api.parklio.com/v2/realtime', {
path: '/socketio',
query: { token },
});
await new Promise((resolve, reject) => {
socket.once('connect', () => resolve(undefined));
socket.once('connect_error', reject);
});
// 2. Collect results by event_id. One can arrive before the HTTP response does.
const results = new Map<string, Result>();
const waiters = new Map<string, (result: Result) => void>();
const settle = (ok: boolean) => (raw: string) => {
const event: OperationEvent = JSON.parse(raw); // payloads are JSON strings
if (!event.event_id) return; // a status update nobody requested
const waiter = waiters.get(event.event_id);
if (waiter) waiter({ ok, event });
else results.set(event.event_id, { ok, event });
};
socket.on('lot_change', settle(true));
socket.on('gateway_error', settle(false));
// 3. Send the command.
const res = await fetch(`https://api.parklio.com/v2/products/${productId}/state`, {
method: 'POST',
headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json' },
body: JSON.stringify({ operation: 'open' }),
});
const body = await res.json();
if (!body.success) throw new Error(JSON.stringify(body.errors));
const eventId: string = body.data.event_id;
console.log('Sent, event_id', eventId);
// 4. Wait for the matching event.
const { ok, event } = await new Promise<Result>((resolve) => {
const early = results.get(eventId);
if (early) resolve(early);
else waiters.set(eventId, resolve);
});
console.log(ok ? `Done: ${JSON.stringify(event.data.barrier ?? event.data)}` : `Failed: ${event.message ?? event.type}`);
socket.close();
import json
import os
import threading
import httpx
import socketio
TOKEN = os.environ["PARKLIO_API_TOKEN"]
PRODUCT_ID = os.environ["PRODUCT_ID"]
NS = "/v2/realtime"
sio = socketio.Client()
results = {}
event_id = None
done = threading.Event()
lock = threading.Lock()
def settle(ok):
def handler(raw):
global event_id
event = json.loads(raw) # payloads are JSON strings
if not event.get("event_id"):
return # a status update nobody requested
with lock:
results[event["event_id"]] = (ok, event)
if event["event_id"] == event_id:
done.set()
return handler
sio.on("lot_change", settle(True), namespace=NS)
sio.on("gateway_error", settle(False), namespace=NS)
# 1. Connect first. connect() returns once the server has accepted the token.
sio.connect(
f"https://api.parklio.com?token={TOKEN}",
namespaces=[NS],
socketio_path="socketio",
wait_timeout=10,
)
# 2. Send the command.
res = httpx.post(
f"https://api.parklio.com/v2/products/{PRODUCT_ID}/state",
headers={"Authorization": f"Bearer {TOKEN}"},
json={"operation": "open"},
)
body = res.json()
if not body["success"]:
raise SystemExit(body["errors"])
with lock:
event_id = body["data"]["event_id"]
if event_id in results: # it arrived before the HTTP response
done.set()
print("Sent, event_id", event_id)
# 3. Wait for the matching event.
done.wait(timeout=180)
ok, event = results.get(event_id, (False, {"type": "no result"}))
print("Done" if ok else f"Failed: {event.get('message', event.get('type'))}")
sio.disconnect()
Verify it worked
- The script prints the
event_idfrom the response, then Done once alot_changewith the sameevent_idarrives. - The type block in that event shows the new state, for example
"state": "opened". - On site, the product has moved: a barrier's arm is down, a gate is open.
Common errors
| Code | Where | Meaning | Fix |
|---|---|---|---|
1 | HTTP 401 | The token is missing, invalid or revoked. | Check the Authorization: Bearer <token> header. |
30 | HTTP 403 | The product has no API subscription that includes remote control. | Subscribe it; see What you need. |
214 | HTTP 403 | An earlier operation on this product hasn't finished. | Wait for its result, then try again. |
704 | HTTP 403 | No Gateway has seen the product recently. | Check the Gateway is online, on the same lot and in range. |
gateway_error | realtime | The command reached the Gateway but failed, for example the arm is jammed or a car is on top. type: "Timeout" means no Gateway answered. | See the type table in Events. |
Nothing arrives at all? Make sure you connected before sending the command, with the same API token. See Troubleshooting.
Next
- Asynchronous operations: timeouts, retries and
wait_for_completion. - Products and operations: which operations each product supports.
- Events: everything else you can listen to.