AMQP RealtimeTrigger
CertifiedStream AMQP messages into real-time executions
AMQP RealtimeTrigger
Stream AMQP messages into real-time executions
Creates one execution per message with manual ACK by default until the trigger is stopped, using consumer tag Kestra and serde STRING by default. Use the batch Trigger for time/volume-based batching; deprecated url remains for compatibility—prefer host/port/virtualHost.
type: io.kestra.plugin.amqp.RealtimeTriggerExamples
Consume a message from a AMQP queue in real-time.
id: amqp
namespace: company.team
tasks:
- id: log
type: io.kestra.plugin.core.log.Log
message: "{{ trigger.data }}"
triggers:
- id: realtime_trigger
type: io.kestra.plugin.amqp.RealtimeTrigger
host: localhost
port: 5672
username: guest
password: "{{ secret('AMQP_PASSWORD') }}"
virtualHost: /my_vhost
queue: amqpTrigger.queue
Properties
host *Requiredstring
Broker host
Hostname or IP of the RabbitMQ broker; required unless using the deprecated url.
queue *Requiredstring
Queue name to consume
AMQP queue to read from; required and must already exist.
serdeType *Requiredstring
STRINGSTRINGJSONPayload serde format
Controls how message bodies are read and written; use STRING for raw text or JSON for structured data. Defaults to STRING.
allowConcurrent Non-dynamicboolean
falseSpecifies whether a trigger is allowed to start a new execution even if a previous run is still in progress.
autoAck booleanstring
falseAutomatic acknowledgment
When true, the broker acknowledges messages as soon as they are delivered. When false, the trigger ACKs after emitting the execution event.
consumerTag string
KestraConsumer tag
Client-supplied consumer tag used for tracing and cancellations; defaults to Kestra in tasks and triggers.
password string
Password
Password for the connection; required when the broker enforces authentication.
port string
5672Broker port
TCP port for AMQP connections; defaults to 5672.
stopAfter Non-dynamicarray
CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTEDList of execution states after which a trigger should be stopped (a.k.a. disabled).
username string
Username
Username for the connection; uses broker default (often guest) when not set.
virtualHost string
/Virtual host
Broker virtual host path; defaults to /.
when string
trueA condition that determines whether the trigger should run.
A Pebble expression evaluated at trigger time. The trigger fires only when the expression evaluates to a truthy value (true, a non-empty string, a non-zero number). Use this to gate trigger execution on dynamic runtime values such as execution labels, flow variables, or environment conditions.
Outputs
appId string
Identifier of the publishing application
contentEncoding string
MIME content encoding of the message body
contentType string
MIME content type of the message body
correlationId string
Correlation identifier used to match replies to requests
data object
Deserialized message body
deliveryMode integer
Delivery mode: 1 for non-persistent, 2 for persistent
expiration string
durationTime-to-live before the message expires
headers object
Arbitrary application-specific message headers
messageId string
Application message identifier
priority integer
Message priority, 0 to 9
replyTo string
Queue name replies should be sent to
timestamp string
date-timeTime the message was created
userId string
Identifier of the user that published the message
