ERP Example Calls
This page provides copy-paste examples for the most common ERP integration calls. Use the cURL / Python selector on each example to view it in your preferred language. Replace the placeholder values with your actual endpoint and authentication token before running.
In every example below, replace:
YOUR_PICO_API_ENDPOINT— your Pico API GraphQL URL (e.g.https://yourcompany.picomes.com/graph/v2)YOUR_API_TOKEN— your Pico API authentication token (see Authentication)
The Python examples use the requests library (pip install requests). Selecting cURL or Python on any example switches every example on the page to that language.
1. List Operations
Operations are Pico's representation of your products and processes — they define what can be built or performed. When your ERP creates a work order to request that something be built, Pico represents that as an operation order. Query the available operations to discover what can be ordered. Deploying a product or process in Pico will create or update an operation.
You can receive these updates in real time three ways: (a) subscribe to operationsStream over WebSocket — see the Subscriptions Guide for client setup; (b) consume the same subscription as a Server-Sent Events (SSE) stream by POSTing the subscription query with Accept: text/event-stream (example below); or (c) register a webhook with webhookSubscribe — see section 3 for the webhook-style pattern (the same responseFragment approach works with any subscription, including operationsStream).
Use the returned operation id as the operationId when saving an operation order in section 2.
- cURL
- Python
curl -X POST YOUR_PICO_API_ENDPOINT \
-H "Content-Type: application/json" \
-H "x-pico-api-org: YOUR_API_TOKEN" \
-d '{
"query": "{ operations { id externalId name updatedAt } }"
}'
import requests
response = requests.post(
"YOUR_PICO_API_ENDPOINT",
headers={"x-pico-api-org": "YOUR_API_TOKEN"},
json={"query": "{ operations { id externalId name updatedAt } }"},
)
print(response.json())
SSE subscription example (real-time stream over HTTP):
- cURL
- Python
curl -N -X POST YOUR_PICO_API_ENDPOINT \
-H "Content-Type: application/json" \
-H "Accept: text/event-stream" \
-H "x-pico-api-org: YOUR_API_TOKEN" \
-d '{"query": "subscription { operationsStream(cursor: { initialValue: { updatedAt: \"2024-01-01T00:00:00Z\" } }, batchSize: 100) { id externalId name updatedAt } }"}'
import json
import requests
with requests.post(
"YOUR_PICO_API_ENDPOINT",
headers={
"x-pico-api-org": "YOUR_API_TOKEN",
"Accept": "text/event-stream",
},
json={"query": 'subscription { operationsStream(cursor: { initialValue: { updatedAt: "2024-01-01T00:00:00Z" } }, batchSize: 100) { id externalId name updatedAt } }'},
stream=True,
) as response:
for line in response.iter_lines(decode_unicode=True):
if line and line.startswith("data: "):
print(json.loads(line[6:]))
Expected response:
{
"data": {
"operations": [
{
"id": "op-abc-123",
"externalId": "PROD-101",
"name": "Main Assembly",
"updatedAt": "2025-01-15T10:00:00.000Z"
}
]
}
}
2. Save an Operation Order
Use the operationOrderSave mutation to create an operation order. ERPs often call this a "work order" — it tells Pico to build a specific product for a specific order (e.g. a sales order from your ERP).
Required fields:
operationId— the Pico operation ID (from section 1)externalOrderId— your ERP's order identifier for tracking (technically optional in the schema, but strongly recommended for ERP integrations)
- cURL
- Python
curl -X POST YOUR_PICO_API_ENDPOINT \
-H "Content-Type: application/json" \
-H "x-pico-api-org: YOUR_API_TOKEN" \
-d '{
"query": "mutation SaveOrder($input: OperationOrderSaveInput!) { operationOrderSave(input: $input) { message } }",
"variables": {
"input": {
"operationId": "your-operation-id",
"externalOrderId": "ERP-ORDER-123"
}
}
}'
import requests
response = requests.post(
"YOUR_PICO_API_ENDPOINT",
headers={"x-pico-api-org": "YOUR_API_TOKEN"},
json={
"query": "mutation SaveOrder($input: OperationOrderSaveInput!) { operationOrderSave(input: $input) { message } }",
"variables": {
"input": {
"operationId": "your-operation-id",
"externalOrderId": "ERP-ORDER-123",
}
},
},
)
print(response.json())
Expected response:
{
"data": {
"operationOrderSave": {
"message": "created"
}
}
}
If an operation order with the same externalOrderId and operationId already exists, the response message will be "updated" instead of "created".
3. Subscribe to Operation Order Completion Webhook
Use the webhookSubscribe mutation to register a webhook that receives a POST request every time an operation order completes.
Required fields:
subscriptionID— a URL-safe identifier you choose for this subscription (used to unsubscribe later)responseFragment— a GraphQL fragment that describes which fields to include in the webhook payload (must inherit from one of the subscription response types)webhookUrl— the URL where Pico willPOSTcompletion events
Optional fields:
webhookHeaders— custom headers to include on every callback (e.g. for authenticating with your endpoint)
- cURL
- Python
curl -X POST YOUR_PICO_API_ENDPOINT \
-H "Content-Type: application/json" \
-H "x-pico-api-org: YOUR_API_TOKEN" \
-d '{
"query": "mutation Subscribe($input: SubscribeRequestInput!) { webhookSubscribe(input: $input) { subscriptionId webhookUrl createdAt } }",
"variables": {
"input": {
"subscriptionID": "erp-order-completions",
"responseFragment": "fragment OrderComplete on OperationOrderComplete { externalOrderId operation { id externalId name } orderIndex at operationSummary { startedAt consumedSerials { id value } } endState { producedSerial } }",
"webhookUrl": "https://your-erp-system.com/api/pico-webhooks",
"webhookHeaders": {"X-API-Key": "your-erp-api-key"}
}
}
}'
import requests
response = requests.post(
"YOUR_PICO_API_ENDPOINT",
headers={"x-pico-api-org": "YOUR_API_TOKEN"},
json={
"query": "mutation Subscribe($input: SubscribeRequestInput!) { webhookSubscribe(input: $input) { subscriptionId webhookUrl createdAt } }",
"variables": {
"input": {
"subscriptionID": "erp-order-completions",
"responseFragment": "fragment OrderComplete on OperationOrderComplete { externalOrderId operation { id externalId name } orderIndex at operationSummary { startedAt consumedSerials { id value } } endState { producedSerial } }",
"webhookUrl": "https://your-erp-system.com/api/pico-webhooks",
"webhookHeaders": {"X-API-Key": "your-erp-api-key"},
}
},
},
)
print(response.json())
Expected response:
{
"data": {
"webhookSubscribe": {
"subscriptionId": "erp-order-completions",
"webhookUrl": "https://your-erp-system.com/api/pico-webhooks",
"createdAt": "2025-01-15T10:30:00.000Z"
}
}
}
Webhook Delivery Behavior
Successful delivery: Pico considers a webhook delivered when your endpoint returns a 2xx HTTP status code.
Failed delivery (non-2xx response): If your endpoint returns an HTTP status >= 400, Pico logs the failure. The current event delivery is skipped, but the subscription remains active and will attempt delivery for future events.
Connection retry: If the internal event stream disconnects, Pico automatically reconnects with a 5-second delay between attempts. Reconnection retries indefinitely until the subscription is explicitly unsubscribed.
Extended downtime: If your webhook endpoint is unreachable or consistently failing, events that occur during the outage will not be redelivered. The subscription continues to listen, and delivery resumes for new events once connectivity is restored.
4. Unsubscribe from a Webhook
Use the webhookUnsubscribe mutation with the subscriptionId you used when subscribing.
- cURL
- Python
curl -X POST YOUR_PICO_API_ENDPOINT \
-H "Content-Type: application/json" \
-H "x-pico-api-org: YOUR_API_TOKEN" \
-d '{
"query": "mutation Unsubscribe($id: String!) { webhookUnsubscribe(subscriptionId: $id) { success subscriptionId message } }",
"variables": {
"id": "erp-order-completions"
}
}'
import requests
response = requests.post(
"YOUR_PICO_API_ENDPOINT",
headers={"x-pico-api-org": "YOUR_API_TOKEN"},
json={
"query": "mutation Unsubscribe($id: String!) { webhookUnsubscribe(subscriptionId: $id) { success subscriptionId message } }",
"variables": {"id": "erp-order-completions"},
},
)
print(response.json())
Expected response:
{
"data": {
"webhookUnsubscribe": {
"success": true,
"subscriptionId": "erp-order-completions",
"message": "Subscription cancelled"
}
}
}
5. List Active Webhook Subscriptions
To verify which webhooks are currently active, query webhookSubscriptions:
- cURL
- Python
curl -X POST YOUR_PICO_API_ENDPOINT \
-H "Content-Type: application/json" \
-H "x-pico-api-org: YOUR_API_TOKEN" \
-d '{
"query": "{ webhookSubscriptions { subscriptionId responseFragment webhookUrl createdAt lastEventAt } }"
}'
import requests
response = requests.post(
"YOUR_PICO_API_ENDPOINT",
headers={"x-pico-api-org": "YOUR_API_TOKEN"},
json={"query": "{ webhookSubscriptions { subscriptionId responseFragment webhookUrl createdAt lastEventAt } }"},
)
print(response.json())
Expected response:
{
"data": {
"webhookSubscriptions": [
{
"subscriptionId": "erp-order-completions",
"responseFragment": "fragment OrderComplete on OperationOrderComplete { externalOrderId ... }",
"webhookUrl": "https://your-erp-system.com/api/pico-webhooks",
"createdAt": "2025-01-15T10:30:00.000Z",
"lastEventAt": "2025-01-15T14:22:00.000Z"
}
]
}
}
6. Example Webhook Payload
When an operation order completes, Pico sends a POST request to your webhookUrl with the fields specified in your responseFragment. Below is an example payload based on the subscription created in section 3:
{
"externalOrderId": "ERP-ORDER-123",
"operation": {
"id": "op-abc-123",
"externalId": "PROD-101",
"name": "Main Assembly"
},
"orderIndex": 0,
"at": "2025-01-15T14:22:00.000Z",
"operationSummary": {
"startedAt": "2025-01-15T14:00:00.000Z",
"consumedSerials": [
{
"id": "attr-serial-1",
"value": "SN-20250115-001"
},
{
"id": "attr-serial-2",
"value": "SN-20250115-042"
}
]
},
"endState": {
"producedSerial": "ASSY-20250115-007"
}
}
Payload Field Reference
| Field | Description |
|---|---|
externalOrderId | The ERP order identifier you provided when creating the operation order |
operation.id | Pico's internal operation identifier |
operation.externalId | The part number configured in Pico |
operation.name | Human-readable operation name |
orderIndex | 0-based index when multiple of the same operation are on the same order |
at | ISO 8601 timestamp of when the order was completed |
operationSummary.startedAt | When the first process in the operation began |
operationSummary.consumedSerials | Materials consumed during the build (resolve id via upfront operation mapping) |
endState.producedSerial | Serial number of the produced unit (if captured) |
See Also
- Work Order Integration Guide — End-to-end walkthrough of the ERP integration pattern
- Subscriptions Guide — Real-time event streaming with WebSocket subscriptions
- Authentication — How to obtain and use API tokens
- OperationOrderSave Mutation — Full mutation reference
- webhookSubscribe Mutation — Full webhook subscription reference
- webhookUnsubscribe Mutation — Full unsubscribe reference