Notifications

Notifications are the organization’s event feed. Long-running work emits one when it finishes - training complete, export ready, batch auto-annotation done - so an agent can poll a single feed instead of holding on to every run id it ever started.

The feed is scoped to the API key’s organization. An API key has no user identity of its own, so list returns the whole organization’s feed and mark_all_read clears all of it, rather than one person’s inbox.

Every notification carries a type slug, a human-readable title and message, and a metadata object whose keys depend on the type - a training event carries the run id, an export event carries the export id. Branch on type, then read metadata.

list

One page of notifications, newest first.

Source: Notifications.list

for notification in client.notifications.list(unread_only=True, limit=25):
    print(notification.type, notification.title)
    if notification.type == "training_complete":
        print(notification.metadata)
    client.notifications.mark_read(
        notification_id=notification.id,
    )

pictograph notifications list --unread --limit 25

curl -s "https://api.pictograph.io/api/v1/developer/notifications?unread_only=true&limit=25&offset=0" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns list[Notification]

class Notification(BaseModel):
    """A single row from the `notifications` table."""
    id: str
    organization_id: str
    user_id: str | None = None
    type: str
    title: str
    message: str | None = None
    metadata: dict[str, Any] | None = None
    read: bool = False
    created_at: datetime

The REST response wraps that list alongside the current unread_count and the limit / offset you sent.

unread_count

Just the number, without fetching the rows. Cheap enough to poll on a short interval.

Source: Notifications.unread_count

pending = client.notifications.unread_count()
if pending:
    print(pending, "new events")

pictograph notifications unread-count

curl -s "https://api.pictograph.io/api/v1/developer/notifications/unread-count" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns int

mark_read

Acknowledge one notification. Idempotent, so re-marking a read notification is harmless.

Source: Notifications.mark_read

client.notifications.mark_read(
    notification_id="1a6e4f28-9b30-4c57-8d21-6f3b0a5e7c94",
)

pictograph notifications read 1a6e4f28-9b30-4c57-8d21-6f3b0a5e7c94

curl -s -X POST "https://api.pictograph.io/api/v1/developer/notifications/1a6e4f28-9b30-4c57-8d21-6f3b0a5e7c94/read" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns None

An id belonging to another organization returns 404 rather than 403, so the feed cannot be used to probe what exists elsewhere.

mark_all_read

Clear the whole organization feed in one call, and get back how many rows that actually changed.

Source: Notifications.mark_all_read

marked = client.notifications.mark_all_read()
print(marked, "notifications marked read")

pictograph notifications read-all

curl -s -X POST "https://api.pictograph.io/api/v1/developer/notifications/read-all" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns int

Idempotent: a second call returns 0.

delete

Remove a notification from the feed permanently. Marking read is usually what you want; delete is for pruning noise you never want to see again.

Source: Notifications.delete

client.notifications.delete(
    notification_id="1a6e4f28-9b30-4c57-8d21-6f3b0a5e7c94",
)

pictograph notifications delete 1a6e4f28-9b30-4c57-8d21-6f3b0a5e7c94

curl -s -X DELETE "https://api.pictograph.io/api/v1/developer/notifications/1a6e4f28-9b30-4c57-8d21-6f3b0a5e7c94" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns None

Polling pattern

Poll the count, and only fetch rows when it moves. The feed is the cheapest way to follow work an agent started, because one request covers every job type at once.

import time

while True:
    if client.notifications.unread_count():
        for notification in client.notifications.list(unread_only=True):
            print(notification.type, notification.title, notification.metadata)
            client.notifications.mark_read(
                notification_id=notification.id,
            )
    time.sleep(30)

For push delivery instead of polling, register a webhook endpoint.

Common errors