ESC
Ask Kestra AI
ESC
Kestra AI

Hi! I'm your Kestra AI assistant.
Ask me anything about workflows.

EXAMPLE QUESTIONS

How to set up CI/CD for kestra flows?
How to run Python script?
How to run dbt?

AMQP Amqp1Publish

AMQP Amqp1Publish

Certified

Publish messages to an AMQP 1.0 address

Sends one or more messages to a queue or topic over AMQP 1.0, serializing the payload according to serdeType and waiting for the broker to accept each message. Works with Azure Service Bus, ActiveMQ Artemis, RabbitMQ 4 and other AMQP 1.0 brokers; use io.kestra.plugin.amqp.Publish for AMQP 0.9.1.

yaml
type: io.kestra.plugin.amqp.Amqp1Publish

Publish messages to an ActiveMQ Artemis queue.

yaml
id: amqp1_publish
namespace: company.team

tasks:
  - id: publish
    type: io.kestra.plugin.amqp.Amqp1Publish
    host: localhost
    port: 5672
    username: artemis
    password: "{{ secret('AMQP_PASSWORD') }}"
    address: orders
    from:
      - data: value-1
        applicationProperties:
          source: kestra
      - data: value-2
        durable: true

Publish JSON messages to an Azure Service Bus queue, authenticating with a shared access policy.

yaml
id: azure_service_bus_publish
namespace: company.team

tasks:
  - id: publish
    type: io.kestra.plugin.amqp.Amqp1Publish
    host: my-namespace.servicebus.windows.net
    ssl: true
    username: RootManageSharedAccessKey
    password: "{{ secret('SERVICE_BUS_SAS_KEY') }}"
    address: orders
    serdeType: JSON
    from:
      - data:
          orderId: 42
          status: CREATED
        contentType: application/json
        messageId: order-42
        subject: order-created
        applicationProperties:
          region: eu-west
Properties

Target address

Address to send to, in the broker's format: the queue or topic name for ActiveMQ Artemis and Azure Service Bus, or /queues/<queue> and /exchanges/<exchange>/<routing-key> for RabbitMQ 4.

Structured data items, either as a map, a list of map, a URI, or a JSON string.

Structured data items can be defined in the following ways:

  • A single item as a map (a document).
  • A list of items as a list of maps (a list of documents).
  • A URI, supported schemes are kestra for internal storage files, file for host local files, and nsfile for namespace files.
  • A JSON String that will then be serialized either as a single item or a list of items.
Definitions
applicationPropertiesobject

Application properties

User-defined key/value pairs (AMQP 0.9.1 headers, Azure Service Bus custom properties); values must be simple types such as strings, numbers or booleans.

contentEncodingstring

MIME content encoding of the message body

contentTypestring

MIME content type of the message body

correlationIdstring

Correlation identifier used to match replies to requests

creationTimestring
Formatdate-time

Time the message was created

dataobject

Message body

Serialized according to serdeType when publishing, deserialized the same way when consuming.

durableboolean

Whether the broker must persist the message

AMQP defaults to false when not set.

groupIdstring

Group the message belongs to (Azure Service Bus session ID)

messageAnnotationsobject

Message annotations

Broker-specific annotations, e.g. x-opt-partition-key sent to, or x-opt-enqueued-time received from, Azure Service Bus.

messageIdstring

Application message identifier

priorityinteger

Message priority, 0 to 255

AMQP defaults to 4 when not set.

replyTostring

Address replies should be sent to

subjectstring

Application-specific subject (Azure Service Bus label)

timeToLivestring
Formatduration

Time-to-live before the message expires

userIdstring

Identity of the user that produced the message

Broker host

Hostname or IP of the AMQP 1.0 broker, e.g. my-namespace.servicebus.windows.net for Azure Service Bus.

Assets this task consumes as inputs or produces as outputs, for lineage tracking and the asset graph (Enterprise Edition). A flow declaring this property on a task is rejected in the open-source edition.

Definitions
assetFailureBehaviorstring
Possible Values
IGNOREFAILWARN

Asset failure behavior

Behavior applied to the task state when a declared asset fails to render, emit, or be persisted (e.g. a lock conflict): FAIL escalates it to FAILED, WARN (default) warns it if it would otherwise succeed, IGNORE leaves the state untouched.

enableAutobooleanstring

Whether to auto-register assets referenced dynamically at runtime that are not statically declared in inputs or outputs.

inputsarray

The assets consumed as inputs.

id*string
Min length1
typestring
outputs

The assets produced as outputs.

id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*string
Min length1

Custom asset type

descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
DefaultPT30S

Connection timeout

Maximum time to open or close the connection and its links, and to wait for the broker to settle each published message. Defaults to PT30S.

Password

SASL PLAIN password; for Azure Service Bus use the shared access key.

Broker port

TCP port of the AMQP 1.0 listener; defaults to 5672, or 5671 when ssl is enabled.

DefaultSTRING
Possible Values
STRINGJSON

Serializer used for the message payload

Defaults to STRING.

Defaultfalse

Use TLS (AMQPS)

Encrypts the connection with TLS. The broker certificate and hostname are always verified, against the JVM's trusted certificate authorities unless sslCaCertificate is set. Required by Azure Service Bus. Defaults to false.

CA certificate for TLS

PEM-encoded certificate (or chain) of the authority that signed the broker certificate, for brokers using a private or self-signed CA. Only used when ssl is true; when set, only this CA is trusted. Example: {{ secret('AMQP_CA_CERT') }}.

Username

SASL PLAIN username; for Azure Service Bus use the shared access policy name (e.g. RootManageSharedAccessKey). SASL ANONYMOUS is used when not set.

Number of messages published

Unitrecords

Number of messages published to the AMQP 1.0 address

More Plugins in this Category

Twilio
Twilio

Tasks that integrate with the Twilio ecosystem.

BusinessData

9 tasks

3 blueprints

phpIPAM
phpIPAM

Interact with a self-hosted phpIPAM instance to manage IP Address Management (IPAM) resources: sections, subnets, IP addresses, VLANs, and VRFs.

Infrastructure

29 tasks

1 blueprints