Ctrlk
For the complete documentation index, see llms.txt. This page is also available as Markdown.

topics

Event Streams topics are account-level, durable message channels. A topic import step in a flow (adaptor type TopicImport) — or celigo topics publish — appends messages to a topic, and every topic export step (TopicExport) that subscribes to it reads them independently, from its own read position: one message reaches every listener, and a listener that falls behind resumes where it stopped. Messages stay on the topic until retentionMs elapses, or until purge-messages drops them early. Flow steps bind to a topic through topic._topicId.

Topics are registered on integrations the way lookup caches are — the integration's _registeredTopicIds: pass __integrationIds in the create body, or run integrations register-topics, and list --integration <id> shows what an integration has registered. Event Streams is a licensed feature; see Gotchas.

This command group manages the topics and their messages: the standard verbs plus dependencies and audit, the retained messages (messages, purge-messages), publishing (publish), and a listener's read position (offsets).

Added in celigo-cli 2026.9.6.

REST API: Topics

celigo topics <subcommand> [args] [flags]

Supports all global flags.

Subcommands

Subcommand
Purpose

list

List every topic, or the ones registered to one integration (--integration); filter by externalId, sort by name or lastModified.

get <id>

Fetch one topic by ID.

create

Create a topic from a JSON body (--file <path> or stdin).

update <id>

Full-replace PUT from a JSON body (--file <path> or stdin).

set <id> key=value …

Edit fields: name and description as one atomic PATCH; every other field through GET + modify + PUT. Supports key=file://<path>.

delete <id>

Delete a topic — it moves to the recycle bin.

dependencies <id> (alias used-by)

List the resources that depend on this topic.

audit <id>

Show the audit log (change history) for one topic.

messages <id>

List the messages currently retained on the topic, oldest first, without moving any listener's read position.

purge-messages <id>

Remove every retained message; the topic, its retention setting, and its subscriptions stay.

publish <id>

Append messages to the topic from a JSON body (--file <path> or stdin).

offsets <id> <exportId>

Report how far a topic export (listener) has read on the topic.

celigo topics list

List every topic on the account, sorted by name unless you pass --sort-by lastModified. With --integration <id> the command lists the topics registered to that integration instead — the _registeredTopicIds that integrations register-topics maintains. Topics in the recycle bin are not listed.

The endpoint rejects the include/exclude field projection, so rows are trimmed client-side: --fields changes what is printed, not what is transferred.

Signature

Arguments

None.

Flags

Flag
Type
Default
Description

--integration <id>

string

—

List only the topics registered to this integration. Cannot be combined with --external-id or --sort-by: the per-integration endpoint does not support them, and the CLI refuses the combination rather than let the server ignore the filter.

--external-id <id>

string

—

Only topics whose externalId equals this value — an exact match applied server-side. Sent as externalId.

--sort-by <field>

name | lastModified

name

Sort order. Sent as sort_by; any other value is rejected locally.

--fields <spec>

string

default

Fields to print per row. default prints _id, name, and the table columns; all prints complete documents; a comma-separated list selects specific fields (dot notation allowed). See field projection.

--limit <n>

integer

—

Return at most <n> rows, fetching a single page.

Default table columns: _id, name, retentionMs, externalId, lastModified.

Example

Corresponds to: GET /v1/topics — with externalId and sort_by as query parameters; with --integration, GET /v1/integrations/{_id}/topics instead.

celigo topics get

Fetch one topic by ID. A topic in the recycle bin is reported as not found — restore it first.

Signature

Arguments

Argument
Type
Required
Description

<id>

string

Yes

Topic _id.

Flags

Uses global flags only.

Example

Corresponds to: GET /v1/topics/{_id}

celigo topics create

Create a topic from a JSON body. Read the body from a file with -f, --file (recommended), or pipe it on stdin. Pass __integrationIds to register the topic on integrations in the same call, or register it later with integrations register-topics. Creating counts against the license's topics.topicsAllowed quota.

The topic's stream is provisioned lazily, on the first publish, so a new topic reports no messages until something is published to it. Requires full mode.

Signature

Arguments

None.

Flags

Flag
Type
Default
Description

-f, --file <path>

string

—

Read the JSON body from a file instead of stdin (--file - also means stdin).

Request body

Field
Required
Description

name

Yes

Display name, up to 200 characters.

description

No

Free text.

retentionMs

No

How long a message stays on the topic, in milliseconds: at least one hour (3600000), at most 90 days (7776000000), and no longer than your license allows. Seven days when omitted.

externalId

No

A caller-chosen key for finding the topic again without storing its _id — list --external-id <id>. Integration Apps use it to locate a topic across installs.

__integrationIds

No

Integrations to register the topic on in the same call. Each must exist in the account.

Example

Corresponds to: POST /v1/topics

celigo topics update

Full-replace PUT from a JSON body (--file <path> or stdin). Requires full mode.

⚠️ update replaces the whole topic. Fields left out of the body are cleared, not kept: description and externalId are removed, and retentionMs reverts to seven days. GET the current document, modify it, and PUT the complete object back — or use set, which round-trips the document for you. The topic's _id, its subscriptions, and its retained messages are unaffected; changing retentionMs re-configures the stream's retention window.

Signature

Arguments

Argument
Type
Required
Description

<id>

string

Yes

Topic _id.

Flags

Flag
Type
Default
Description

-f, --file <path>

string

—

Read the JSON body from a file instead of stdin (--file - also means stdin).

--force

boolean

false

Submit even if the body contains masked credential values (***) copied from a GET.

Example

Corresponds to: PUT /v1/topics/{_id}

celigo topics set

Edit individual fields with key=value assignments. name and description are the server's whole PATCH whitelist: an assignment that touches only those is sent as one atomic PATCH. Any other field — retentionMs, externalId — goes through GET → modify → PUT with the complete document, so the fields you do not name are preserved. Values are auto-parsed as JSON (retentionMs=3600000 is a number, externalId=null removes the field). Dot notation and key=file://<path> are supported.

set is blocked in read mode, and operate mode restricts it to an allowlist of operational fields (disabled, debugUntil, schedule.*, …) that a topic does not carry — so in practice set on a topic needs full mode. See Profiles & regions.

Signature

Arguments

Argument
Type
Required
Description

<id>

string

Yes

Topic _id.

<key=value>

string

Yes (≥1)

Field assignments, for example name="Order events" or retentionMs=86400000.

Flags

Uses global flags only.

Example

Corresponds to: PATCH /v1/topics/{_id} for name and description; otherwise GET /v1/topics/{_id} then PUT /v1/topics/{_id}.

celigo topics delete

Delete a topic. It moves to the recycle bin and is removed from _registeredTopicIds on every integration that had it registered; restoring it does not re-register it. Its retained messages are kept — readable again if the topic is restored, gone for good once it is purged. Prompts for confirmation unless -y is supplied. Requires full mode.

⚠️ Deletion is refused while a step still binds to the topic. An export or import that references it through topic._topicId — or a flow that still contains such a step — blocks the delete. Delete or repoint those first; dependencies lists them.

Signature

Arguments

Argument
Type
Required
Description

<id>

string

Yes

Topic _id.

Flags

Flag
Type
Default
Description

-y, --yes

boolean

false

Skip the confirmation prompt.

Example

Corresponds to: DELETE /v1/topics/{_id}. To restore the topic, use recycle-bin.

celigo topics dependencies

List the resources that depend on this topic, keyed by resource type: the exports and imports bound to it through topic._topicId, the flows that contain those steps, and the integrations that list it in _registeredTopicIds. Aliased as used-by. Run it before delete: the exports and imports listed here block the delete until they are removed or repointed, and an empty object means nothing references the topic. An unknown topic id returns an empty map rather than an error.

Signature

Arguments

Argument
Type
Required
Description

<id>

string

Yes

Topic _id.

Flags

Uses global flags only.

Example

Corresponds to: GET /v1/topics/{_id}/dependencies

celigo topics audit

Show the audit log (change history) for one topic. Besides the create, update, delete, and restore entries, every purge-messages is recorded as a purge.

Signature

Arguments

Argument
Type
Required
Description

<id>

string

Yes

Topic _id.

Flags

Uses global flags only.

Example

Corresponds to: GET /v1/topics/{_id}/audit

celigo topics messages

List the messages currently retained on a topic, oldest first, without moving any listener's read position — to see what producers are publishing, or to confirm a purge. Each entry is {timestamp, payload}: when the message was appended to the stream (the value --time-gt and --time-lte filter on) and the JSON object that was published. The list is empty when the topic has never received a message, when everything has aged out of retentionMs, or when the cursor is already at the end of the stream.

Pages are cursor-based. A page that is not the last carries a nextPageURL with the same filters plus an afterOffset cursor; pass that cursor to --after-offset for the next page rather than computing one yourself. Works in read mode.

Signature

Arguments

Argument
Type
Required
Description

<id>

string

Yes

Topic _id.

Flags

Flag
Type
Default
Description

--limit <n>

integer

—

Maximum messages per page, 1–100. Sent as limit.

--after-offset <cursor>

string

—

Continuation cursor from the previous page's nextPageURL. Sent as afterOffset.

--time-gt <ms>

epoch milliseconds

—

Only messages published strictly after this time. Sent as time_gt.

--time-lte <ms>

epoch milliseconds

—

Only messages published at or before this time. Sent as time_lte.

Example

Corresponds to: GET /v1/topics/{_id}/messages — with the flags above as query parameters.

celigo topics purge-messages

Remove every retained message from a topic. The topic itself, its retentionMs, and its subscriptions stay. Irreversible, and recorded in the topic's audit log as a purge. Prompts for confirmation unless -y is supplied. Runs in operate mode.

⚠️ Purging drops every retained message. A listener that had not yet read them never will. Confirm with messages before you purge, not after.

Signature

Arguments

Argument
Type
Required
Description

<id>

string

Yes

Topic _id.

Flags

Flag
Type
Default
Description

-y, --yes

boolean

false

Skip the confirmation prompt.

Example

Corresponds to: DELETE /v1/topics/{_id}/messages

celigo topics publish

Append messages to a topic so every subscribed topic export receives them on its next read. The body is {"messages": [ {...}, {...} ]} — each message any JSON object, at most 1 MiB (1,048,576 bytes) serialized. Each one becomes one record that listeners receive; an empty array is accepted and publishes nothing. The CLI checks that messages is an array before sending anything.

The HTTP status is 200 whenever the body was valid, so check responses[].statusCode per message: an oversized message is rejected individually (topic_publish_event_too_large) while the rest of the batch is still published, and an entry whose status is not 200 says why it was not appended. The first publish to a new topic provisions its stream and takes a few seconds longer. Runs in operate mode.

Signature

Arguments

Argument
Type
Required
Description

<id>

string

Yes

Topic _id.

Flags

Flag
Type
Default
Description

-f, --file <path>

string

—

Read the JSON body from a file instead of stdin (--file - also means stdin).

Request body

Example

Corresponds to: POST /v1/topics/{_id}/publish

celigo topics offsets

Report how far a topic export (listener) has read on this topic — where its next run resumes. The result is {hasCursor, offset}: hasCursor is true once the export has committed a read position on the topic, and offset is the zero-based position of the next message it will read (-1 when hasCursor is false). An export that never ran, or that does not subscribe to the topic, has no cursor. Works in read mode.

Signature

Arguments

Argument
Type
Required
Description

<id>

string

Yes

Topic _id.

<exportId>

string

Yes

The topic export's _id — an export with adaptorType: TopicExport whose topic._topicId is this topic.

Flags

Uses global flags only.

Example

Corresponds to: GET /v1/topics/{_id}/exports/{_exportId}/offsets

Gotchas

  • Event Streams is a licensed feature. list works on every account, but the message routes — messages, purge-messages, publish, offsets — need the Event Streams entitlement and fail with 403 topics_subscription_required without it; a topic write fails with the same code as a 422. The license also caps how many topics the account may hold (topics.topicsAllowed) and the longest retentionMs allowed (422 topics_retention_exceeds_entitlement above it).

  • The quota is enforced on create. When the account already holds its topics.topicsAllowed quota of topics, create fails with 422 topics_limit_exceeded.

  • update clears what it does not carry. description and externalId are removed and retentionMs reverts to seven days when the body omits them. Prefer set: name/description apply as one PATCH, and everything else round-trips the complete document.

  • set needs full mode in practice. It is blocked in read mode, and operate mode only allows operational fields (disabled, debugUntil, schedule.*, …) that a topic does not have. See Profiles & regions.

  • --integration excludes the other list filters. --external-id and --sort-by are main-list parameters; combined with --integration the CLI refuses them instead of letting the per-integration endpoint silently ignore them.

  • A new topic has no stream until the first publish. messages returns an empty list on a topic nothing has published to yet, and the first publish provisions the stream, so it takes a few seconds longer than the ones after it.

  • messages never advances a listener. It is a read of the stream; only a topic export's own run moves that export's cursor. Its --time-gt/--time-lte take epoch milliseconds, not ISO timestamps.

  • publish succeeds per message, not per call. A 200 says the body was valid. Read responses[].statusCode: a message over 1 MiB is rejected on its own while the rest of the batch is appended.

  • delete moves the topic to the recycle bin and un-registers it everywhere. It is refused while an export, import, or flow still binds to the topic through topic._topicId. Restoring the topic does not re-register it on its integrations — run integrations register-topics again — and purging it from the recycle bin also requires that any deleted steps bound to it have been purged.

  • Permission modes. list, get, dependencies, audit, messages, and offsets work in read mode. publish and purge-messages need operate. create, update, delete — and, in practice, set — need full.

Related

  • integrations — register-topics / deregister-topics maintain _registeredTopicIds; topics list --integration <id> shows the result.

  • exports — a topic export (adaptorType: TopicExport) is the listener: it subscribes through topic._topicId and reads from its own cursor (offsets).

  • imports — a topic import (adaptorType: TopicImport) is the publisher inside a flow; publish does the same from the command line.

  • lookup-caches — the other resource registered on integrations the same way (list --integration, register-lookup-caches).

  • recycle-bin — restore or permanently purge a deleted topic.

Previoustemplates
Nexttools

Last updated

Was this helpful?