Messages API: Send and Browse Broker Messages

The Messages API lets you interact with your brokers directly — without building a workflow. You can publish individual messages to Kafka topics or RabbitMQ exchanges, page through message history, open a live WebSocket stream for real-time observation, and fire one-off messages using the single-step executor. All HTTP endpoints are relative to http://localhost:8082.

POST /kafka/messages/send

Publish a single message to a Kafka topic. Devset routes the message through the named connector and returns once the broker acknowledges the write.

string

required

The name of the Kafka connector to use. Must match a connector saved via the Connectors API.

string

required

The Kafka topic to publish the message to.

string

The message key used for partition assignment. Omit to let Kafka assign a partition automatically.

string

required

The message value as a string. For JSON payloads, serialize your object to a JSON string before sending.

object

Optional key-value pairs to attach as Kafka message headers.

Request example

{
  "producerName": "local-kafka",
  "topic": "orders.events",
  "key": "order-123",
  "message": "{\"id\": \"order-123\", \"amount\": 99}",
  "headers": { "X-Source": "devset" }
}

Response — 200 OK This endpoint returns no response body.

GET /kafka/topics

List all topics available on a Kafka connection. Query parameters

string

required

The name of the Kafka connector to query.

Request example

GET /kafka/topics?connectionName=local-kafka

Response — 200 OK

["orders.events", "payments.events", "notifications.events"]

The response is a JSON array of topic name strings.

GET /kafka/messages

Fetch a page of messages from a Kafka topic, ordered newest-first. Results are cursor-paginated using beforeTimestamp. The maximum page size is 500 messages per request. Query parameters

string

required

The name of the Kafka connector to read from.

string

required

The topic to read messages from.

integer

Maximum number of messages to return. Defaults to 50. Hard limit is 500.

string

An ISO 8601 timestamp (e.g. 2024-01-15T10:30:00Z). When provided, only messages produced before this instant are returned. Use the timestamp of the oldest message in the current page to fetch the next page.

Request example

GET /kafka/messages?connectionName=local-kafka&topic=orders.events&limit=50&beforeTimestamp=2024-01-15T10:30:00Z

Response — 200 OK

[
  {
    "partition": 0,
    "offset": 42,
    "timestamp": "2024-01-15T10:29:55Z",
    "key": "order-123",
    "value": "{\"id\": \"order-123\", \"amount\": 99}",
    "headers": { "X-Source": "devset" }
  },
  {
    "partition": 0,
    "offset": 41,
    "timestamp": "2024-01-15T10:29:40Z",
    "key": "order-122",
    "value": "{\"id\": \"order-122\", \"amount\": 50}",
    "headers": {}
  }
]

The response is a JSON array of message records, newest first.

integer

The Kafka partition this message was stored in.

integer

The message offset within its partition.

string

ISO 8601 timestamp when the message was produced.

string | null

The message key, or null if the producer did not set one.

string

The raw message value as a string.

object

String key-value pairs attached as Kafka headers.

To paginate, take the timestamp of the last message in the current response and pass it as beforeTimestamp in your next request.

WS /ws/kafka/topic-stream

Open a WebSocket connection to stream live messages from a Kafka topic as they arrive. Devset forwards each incoming record to the client as a JSON text frame. The connection stays open until you close it. Connection URL

ws://localhost:8082/ws/kafka/topic-stream?connectionName=local-kafka&topic=orders.events&offsetMode=latest

Query parameters

string

required

The name of the Kafka connector to subscribe on.

string

required

The topic to stream.

string

Controls where consumption starts. Use latest to receive only new messages produced after the connection opens, or earliest to replay the topic from the beginning. Defaults to latest.

Streamed message frame Each WebSocket frame contains a single JSON message record in the same shape as the records returned by GET /kafka/messages:

{
  "partition": 0,
  "offset": 43,
  "timestamp": "2024-01-15T10:30:01Z",
  "key": "order-124",
  "value": "{\"id\": \"order-124\", \"amount\": 120}",
  "headers": {}
}

WebSocket connections are not authenticated. Keep the Devset engine accessible only on localhost or within a trusted network.

POST /rabbit/message/send

Publish a message to a RabbitMQ broker. You can target either a queue directly or an exchange with a routing key. Devset routes the message through the named producer connector.

string

required

The name of the RabbitMQ connector to use. Must match a connector saved via the Connectors API.

string

The RabbitMQ queue to publish to directly. Provide either queueName or exchange, not both.

string

The RabbitMQ exchange to publish to.

string

The routing key used to route the message to bound queues. Used with exchange.

string

required

The message payload as a string. For JSON payloads, serialize your object to a JSON string before sending.

Request example

{
  "producerName": "local-rabbit",
  "exchange": "orders",
  "routingKey": "order.created",
  "message": "{\"id\": \"order-123\", \"amount\": 99}"
}

Response — 200 OK This endpoint returns no response body.

GET /rabbit/broker-resources

List the queues and exchanges available on a RabbitMQ connection. This endpoint requires the RabbitMQ Management Plugin to be enabled on your broker. If the plugin is not available, the response returns "available": false instead of raising an error. Query parameters

string

required

The name of the RabbitMQ connector to query.

Request example

GET /rabbit/broker-resources?connectionName=local-rabbit

Response — 200 OK (plugin available)

{
  "available": true,
  "queues": ["orders.queue", "payments.queue"],
  "exchanges": ["orders", "payments", ""]
}

Response — 200 OK (plugin not enabled)

{
  "available": false,
  "queues": [],
  "exchanges": []
}

boolean

true if the RabbitMQ Management Plugin is reachable; false otherwise.

array

Names of all queues declared on the broker. Empty when available is false.

array

Names of all exchanges declared on the broker. Empty when available is false.

If available is false, enable the RabbitMQ Management Plugin by running rabbitmq-plugins enable rabbitmq_management on your broker host and restarting the broker.

POST /single-step/execute

Execute a single ad-hoc message production immediately, without creating or referencing a saved workflow. This is useful for quick one-off tests or exploratory development. Returns a run identifier you can use to look up events via the Engine API.

string

required

The name of the connector to use for this execution.

string

required

The broker type: kafka or rabbit.

string

The payload content type. Defaults to json.

string

The Kafka topic to publish to. Required when messageType is kafka.

string

The RabbitMQ exchange to publish to. Used when messageType is rabbit.

string

The RabbitMQ routing key. Used when messageType is rabbit.

string

A name label for this pipeline stage.

string

A name label for the event being produced.

object

The payload fields to set. Supports Devset DSL functions such as { "$fn": "uuid()" } for dynamic values.

object

Key-value pairs to attach as message headers.

integer

Number of times to execute. Defaults to 1.

object

Initial state map available to $ref and $path expressions in set.

Request example

{
  "producerName": "local-kafka",
  "topic": "orders.events",
  "messageType": "kafka",
  "stage": "order-created",
  "event": "order-created",
  "set": {
    "id": { "$fn": "uuid()" },
    "status": "READY"
  },
  "headers": {},
  "executions": 1
}

Response — 202 Accepted

{
  "historyId": "e1b2c3d4-0000-0000-0000-000000000001",
  "runId": "a3f1c2d4-89ab-4e12-b456-426614174000",
  "status": "PENDING",
  "executions": 1,
  "workflowId": "order-created"
}

string

Identifier of the persisted history entry for this execution.

string

The underlying engine run ID. Use it with GET /engine/runs/{runId}/events to retrieve the produced events.

string

Initial run status. Typically PENDING immediately after submission.

integer

Number of executions that were requested.

string

The workflow identifier associated with this execution.

GET /single-step/history

Retrieve the history of all single-step executions performed in the current session. Results are ordered most-recent-first. Response — 200 OK

[
  {
    "id": "e1b2c3d4-0000-0000-0000-000000000001",
    "createdAtEpochMillis": 1705316400000,
    "runId": "a3f1c2d4-89ab-4e12-b456-426614174000",
    "workflowId": "order-created",
    "messageType": "kafka",
    "producerName": "local-kafka",
    "topic": "orders.events",
    "stage": "order-created",
    "event": "order-created",
    "executions": 1,
    "set": {
      "id": { "$fn": "uuid()" },
      "status": "READY"
    },
    "headers": {}
  }
]

string

Unique identifier of this history entry.

integer

Creation timestamp in Unix epoch milliseconds.

string

The engine run ID associated with this execution.

string

The workflow identifier used for this execution.

string

The broker type used: kafka or rabbit.

string

The connector used for the execution.

string | null

The Kafka topic that was targeted, or null for RabbitMQ executions.

string | null

The RabbitMQ exchange that was targeted, or null for Kafka executions.

string | null

The RabbitMQ routing key used, or null for Kafka executions.

string

The stage name label used for this execution.

string

The event name label used for this execution.

integer

The number of executions that were requested.

object

The payload definition that was used, including any DSL expressions.

object

The message headers that were applied.

DSL Reference

API Reference

Messages API: Send and Browse Broker Messages

POST /kafka/messages/send

GET /kafka/topics

GET /kafka/messages

WS /ws/kafka/topic-stream

POST /rabbit/message/send

GET /rabbit/broker-resources

POST /single-step/execute

GET /single-step/history

​POST /kafka/messages/send

​GET /kafka/topics

​GET /kafka/messages

​WS /ws/kafka/topic-stream

​POST /rabbit/message/send

​GET /rabbit/broker-resources

​POST /single-step/execute

​GET /single-step/history

POST /kafka/messages/send

GET /kafka/topics

GET /kafka/messages

WS /ws/kafka/topic-stream

POST /rabbit/message/send

GET /rabbit/broker-resources

POST /single-step/execute

GET /single-step/history