# AWS (SQS & SNS) (/connectors/aws)



Point your AWS SQS / SNS application at KubeMQ by changing only the endpoint URL. The
**AWS connector** is a built-in, wire-protocol bridge inside kubemq-server that speaks the
genuine AWS SQS and SNS HTTP protocols on a dedicated second listener — any standard,
unmodified AWS SDK (boto3, `aws-sdk-go-v2`, the AWS SDK for Java/JS/.NET/Ruby/Rust) talks
to KubeMQ with no LocalStack, no library swap, and no KubeMQ SDK.

## What is the AWS connector [#what-is-the-aws-connector]

The connector is **one binary with two service surfaces** that map onto two distinct
KubeMQ models:

* **SQS → KubeMQ Queue.** Every SQS queue maps onto a native KubeMQ **Queue** channel
  `sqs.{name}`. AWS producers and native gRPC/REST consumers share the same messages on
  that channel. A FIFO group fans onto its own channel `sqs.{name}.fifo.g.{enc(group)}`.
* **SNS → virtual fan-out.** SNS topics are **virtual** — a registry replicated across
  cluster nodes, with no native channel. At publish time a topic fans out to every
  confirmed subscription: subscribed SQS queues (a batch send) and HTTP/HTTPS webhooks (a
  delivery engine).

Because SQS is point-to-point and SNS is publish/subscribe — neither is request/reply —
there is **no RPC**: no Commands, no Queries, no gRPC responder anywhere. The connector
exposes the queue and fan-out surfaces only.

<Callout type="warn">
  The AWS connector is &#x2A;*opt-in (disabled by default)** — enabling it opens a **new HTTP
  listener on port 4566** that is not bound until you set `CONNECTORS_AWS_ENABLE=true`. Unlike
  the other wire-protocol connectors, a stock server does **not** serve AWS until you turn it
  on. See [Getting started](/connectors/aws/tutorials/getting-started).
</Callout>

## How it works [#how-it-works]

An AWS SDK client sends a signed SQS or SNS request to the connector's endpoint. The
connector detects the protocol, verifies the SigV4 signature shape, and dispatches: SQS
operations land on the KubeMQ Queue channel `sqs.{name}` through the message broker; SNS
publishes resolve the virtual topic registry and fan out to the subscribed targets.

<Mermaid
  chart="`
graph LR
APP[&#x22;AWS SDK client<br/>(SQS / SNS)&#x22;]
CONN[&#x22;AWS connector<br/>:4566&#x22;]
Q{{&#x22;Queue channel<br/>sqs.orders&#x22;}}
REG[&#x22;SNS registry<br/>(virtual, replicated)&#x22;]
BROKER[&#x22;Message Broker&#x22;]
SUB[&#x22;Consumer<br/>(SQS / gRPC / REST)&#x22;]
HOOK[&#x22;HTTP/HTTPS webhook&#x22;]

APP -- &#x22;SendMessage&#x22; --> CONN
APP -- &#x22;Publish&#x22; --> CONN
CONN -- &#x22;(Queue, sqs.orders)&#x22; --> Q
CONN -- &#x22;resolve fan-out&#x22; --> REG
REG -. &#x22;to SQS subs&#x22; .-> Q
REG -. &#x22;to webhooks&#x22; .-> HOOK
Q --> BROKER
BROKER -. &#x22;deliver&#x22; .-> SUB

class APP,SUB client
class CONN,REG connector
class Q,BROKER broker
`"
/>

*SQS requests map onto the KubeMQ Queue channel `sqs.{name}` through the message broker; SNS publishes resolve the virtual topic registry and fan out to subscribed SQS queues and HTTP/HTTPS webhooks.*

## Ports & protocol surface [#ports--protocol-surface]

| Port   | Transport          | Protocol                 | Notes                                                                                                                                                                                                  |
| ------ | ------------------ | ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `4566` | Plain HTTP (SigV4) | AWS SQS JSON + SNS Query | A dedicated second listener (the LocalStack convention). **Bound only when the connector is enabled** (`CONNECTORS_AWS_ENABLE=true`), and must differ from the gRPC/REST/HTTP ports.                   |
| —      | HTTPS              | AWS SQS JSON + SNS Query | TLS is provided by the server-wide `Security` block — there is **no AWS-specific TLS option**. SigV4 over plain HTTP is unencrypted on the wire; production deployments should use the HTTPS listener. |

The listener accepts both `POST /` and `GET /` on a single AWS-style endpoint — there are
no per-route REST paths. SQS uses the AWS **JSON protocol** (`X-Amz-Target: AmazonSQS.{Op}`)
with a Query-protocol fallback; SNS uses the AWS **Query protocol** (form body / GET query →
XML). See [Architecture](/connectors/aws/concepts/architecture) for the dispatch detail.

## Send a message [#send-a-message]

The example below runs the full SQS round-trip — `CreateQueue` → `GetQueueUrl` →
`SendMessage` → `ReceiveMessage` → `DeleteMessage` — over a stock AWS SDK. The only change
versus a real-AWS app is the **endpoint override**: each client points at `KUBEMQ_AWS_URL`
(default `http://localhost:4566`). **Dummy credentials are still required** so the SDK forms
a valid SigV4 signature; the connector's default accept-any mode checks the signature shape,
not its value.

<Tabs groupId="language" items="['Go','Python','Java','JavaScript','C#','Ruby','Rust']">
  <Tab value="Go">
    ```go
    package main

    import (
    	"context"
    	"fmt"
    	"log"
    	"os"
    	"time"

    	"github.com/aws/aws-sdk-go-v2/aws"
    	"github.com/aws/aws-sdk-go-v2/config"
    	"github.com/aws/aws-sdk-go-v2/credentials"
    	"github.com/aws/aws-sdk-go-v2/service/sqs"
    )

    func awsURL() string {
    	if v := os.Getenv("KUBEMQ_AWS_URL"); v != "" {
    		return v
    	}
    	return "http://localhost:4566"
    }

    func main() {
    	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
    	defer cancel()

    	// Dummy static credentials are required so the SDK forms a valid SigV4
    	// request; the connector's accept-any mode does not verify their value.
    	cfg, err := config.LoadDefaultConfig(ctx,
    		config.WithRegion("us-east-1"),
    		config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("test", "test", "")),
    	)
    	if err != nil {
    		log.Fatalf("load config: %v", err)
    	}

    	// Override ONLY the endpoint URL — everything else is a normal AWS SDK app.
    	client := sqs.NewFromConfig(cfg, func(o *sqs.Options) {
    		o.BaseEndpoint = aws.String(awsURL())
    	})

    	// CreateQueue "orders" maps to KubeMQ Queue channel "sqs.orders".
    	created, err := client.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String("orders")})
    	if err != nil {
    		log.Fatalf("CreateQueue: %v", err)
    	}
    	queueURL := aws.ToString(created.QueueUrl)

    	if _, err := client.SendMessage(ctx, &sqs.SendMessageInput{
    		QueueUrl:    aws.String(queueURL),
    		MessageBody: aws.String("hello from the AWS SDK"),
    	}); err != nil {
    		log.Fatalf("SendMessage: %v", err)
    	}

    	recv, err := client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
    		QueueUrl:            aws.String(queueURL),
    		MaxNumberOfMessages: 1,
    		WaitTimeSeconds:     5,
    	})
    	if err != nil || len(recv.Messages) != 1 {
    		log.Fatalf("ReceiveMessage: %v (got %d)", err, len(recv.Messages))
    	}
    	msg := recv.Messages[0]
    	fmt.Printf("received: %q\n", aws.ToString(msg.Body))

    	// DeleteMessage acks the message off the queue by its receipt handle.
    	if _, err := client.DeleteMessage(ctx, &sqs.DeleteMessageInput{
    		QueueUrl:      aws.String(queueURL),
    		ReceiptHandle: msg.ReceiptHandle,
    	}); err != nil {
    		log.Fatalf("DeleteMessage: %v", err)
    	}
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python
    import os

    import boto3

    QUEUE = "orders"


    def make_sqs():
        # Override ONLY the endpoint URL; dummy credentials are still required so
        # boto3 forms a valid SigV4 request (accept-any mode checks shape only).
        return boto3.client(
            "sqs",
            endpoint_url=os.environ.get("KUBEMQ_AWS_URL", "http://localhost:4566"),
            region_name="us-east-1",
            aws_access_key_id="test",
            aws_secret_access_key="test",
        )


    def main() -> None:
        sqs = make_sqs()

        # CreateQueue "orders" maps to KubeMQ Queue channel "sqs.orders".
        queue_url = sqs.create_queue(QueueName=QUEUE)["QueueUrl"]

        sqs.send_message(QueueUrl=queue_url, MessageBody="hello from boto3")

        recv = sqs.receive_message(QueueUrl=queue_url, MaxNumberOfMessages=1, WaitTimeSeconds=5)
        msg = recv["Messages"][0]
        print(f"received: {msg['Body']!r}")

        # DeleteMessage acks the message off the queue by its receipt handle.
        sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=msg["ReceiptHandle"])


    if __name__ == "__main__":
        main()
    ```
  </Tab>

  <Tab value="Java">
    ```java
    import java.net.URI;

    import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
    import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
    import software.amazon.awssdk.regions.Region;
    import software.amazon.awssdk.services.sqs.SqsClient;
    import software.amazon.awssdk.services.sqs.model.Message;

    public final class Main {
        public static void main(String[] args) {
            String url = System.getenv().getOrDefault("KUBEMQ_AWS_URL", "http://localhost:4566");

            // endpointOverride is the only change versus a real-AWS app; dummy
            // credentials are still required to form a valid SigV4 request.
            try (SqsClient sqs = SqsClient.builder()
                    .endpointOverride(URI.create(url))
                    .region(Region.US_EAST_1)
                    .credentialsProvider(StaticCredentialsProvider.create(
                            AwsBasicCredentials.create("test", "test")))
                    .build()) {

                // CreateQueue "orders" maps to KubeMQ Queue channel "sqs.orders".
                String queueUrl = sqs.createQueue(b -> b.queueName("orders")).queueUrl();

                sqs.sendMessage(b -> b.queueUrl(queueUrl).messageBody("hello from the AWS SDK for Java"));

                Message msg = sqs.receiveMessage(b -> b
                        .queueUrl(queueUrl)
                        .maxNumberOfMessages(1)
                        .waitTimeSeconds(5))
                        .messages().get(0);
                System.out.printf("received: %s%n", msg.body());

                // DeleteMessage acks the message off the queue by its receipt handle.
                sqs.deleteMessage(b -> b.queueUrl(queueUrl).receiptHandle(msg.receiptHandle()));
            }
        }
    }
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    import {
      SQSClient,
      CreateQueueCommand,
      SendMessageCommand,
      ReceiveMessageCommand,
      DeleteMessageCommand,
    } from "@aws-sdk/client-sqs";

    const QUEUE = "orders";

    // Override ONLY the endpoint; dummy credentials are still required so the SDK
    // forms a valid SigV4 request (accept-any mode checks the signature shape).
    const sqs = new SQSClient({
      endpoint: process.env["KUBEMQ_AWS_URL"] ?? "http://localhost:4566",
      region: "us-east-1",
      credentials: { accessKeyId: "test", secretAccessKey: "test" },
    });

    async function main(): Promise<void> {
      // CreateQueue "orders" maps to KubeMQ Queue channel "sqs.orders".
      const created = await sqs.send(new CreateQueueCommand({ QueueName: QUEUE }));
      const queueUrl = created.QueueUrl!;

      await sqs.send(new SendMessageCommand({ QueueUrl: queueUrl, MessageBody: "hello from the AWS SDK v3" }));

      const recv = await sqs.send(
        new ReceiveMessageCommand({ QueueUrl: queueUrl, MaxNumberOfMessages: 1, WaitTimeSeconds: 5 }),
      );
      const msg = recv.Messages![0];
      console.log(`received: ${msg.Body}`);

      // DeleteMessage acks the message off the queue by its receipt handle.
      await sqs.send(new DeleteMessageCommand({ QueueUrl: queueUrl, ReceiptHandle: msg.ReceiptHandle! }));
    }

    main().catch((err) => {
      console.error(err);
      process.exit(1);
    });
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    using Amazon.Runtime;
    using Amazon.SQS;
    using Amazon.SQS.Model;

    var url = Environment.GetEnvironmentVariable("KUBEMQ_AWS_URL") ?? "http://localhost:4566";

    // ServiceURL carries the full http://host:port; dummy credentials are still
    // required to form a valid SigV4 request (accept-any mode checks shape only).
    var config = new AmazonSQSConfig { ServiceURL = url, AuthenticationRegion = "us-east-1" };
    using var sqs = new AmazonSQSClient(new BasicAWSCredentials("test", "test"), config);

    const string queueName = "orders";

    // CreateQueue "orders" maps to KubeMQ Queue channel "sqs.orders".
    var created = await sqs.CreateQueueAsync(new CreateQueueRequest { QueueName = queueName });
    var queueUrl = created.QueueUrl;

    await sqs.SendMessageAsync(new SendMessageRequest
    {
        QueueUrl = queueUrl,
        MessageBody = "hello from AWSSDK.NET",
    });

    var recv = await sqs.ReceiveMessageAsync(new ReceiveMessageRequest
    {
        QueueUrl = queueUrl,
        MaxNumberOfMessages = 1,
        WaitTimeSeconds = 5,
    });
    var msg = recv.Messages[0];
    Console.WriteLine($"received: {msg.Body}");

    // DeleteMessage acks the message off the queue by its receipt handle.
    await sqs.DeleteMessageAsync(new DeleteMessageRequest
    {
        QueueUrl = queueUrl,
        ReceiptHandle = msg.ReceiptHandle,
    });
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    # frozen_string_literal: true

    require "aws-sdk-sqs"

    # The Ruby SQS plugin rewrites the request endpoint to the full QueueUrl path,
    # which the single-endpoint connector rejects — remove it so requests stay on
    # the configured base endpoint (as boto3 and aws-sdk-go-v2 do).
    Aws::SQS::Client.remove_plugin(Aws::SQS::Plugins::QueueUrls)

    url = ENV.fetch("KUBEMQ_AWS_URL", "http://localhost:4566")

    # Override ONLY the endpoint; dummy credentials are still required so the SDK
    # forms a valid SigV4 request (accept-any mode checks the signature shape).
    sqs = Aws::SQS::Client.new(
      endpoint: url,
      region: "us-east-1",
      credentials: Aws::Credentials.new("test", "test")
    )

    # CreateQueue "orders" maps to KubeMQ Queue channel "sqs.orders".
    queue_url = sqs.create_queue(queue_name: "orders").queue_url

    sqs.send_message(queue_url: queue_url, message_body: "hello from aws-sdk-ruby")

    recv = sqs.receive_message(queue_url: queue_url, max_number_of_messages: 1, wait_time_seconds: 5)
    msg = recv.messages.first
    puts "received: #{msg.body.inspect}"

    # DeleteMessage acks the message off the queue by its receipt handle.
    sqs.delete_message(queue_url: queue_url, receipt_handle: msg.receipt_handle)
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    use aws_config::BehaviorVersion;
    use aws_sdk_sqs::config::Credentials;
    use aws_sdk_sqs::config::Region;
    use std::error::Error;

    fn aws_url() -> String {
        std::env::var("KUBEMQ_AWS_URL").unwrap_or_else(|_| "http://localhost:4566".to_string())
    }

    #[tokio::main]
    async fn main() -> Result<(), Box<dyn Error>> {
        // Override ONLY the endpoint; dummy credentials are still required so the
        // SDK forms a valid SigV4 request (accept-any mode checks shape only).
        let creds = Credentials::new("test", "test", None, None, "kubemq-aws");
        let conf = aws_config::defaults(BehaviorVersion::latest())
            .region(Region::new("us-east-1"))
            .credentials_provider(creds)
            .endpoint_url(aws_url())
            .load()
            .await;
        let sqs = aws_sdk_sqs::Client::new(&conf);

        // CreateQueue "orders" maps to KubeMQ Queue channel "sqs.orders".
        sqs.create_queue().queue_name("orders").send().await?;
        let url = sqs
            .get_queue_url()
            .queue_name("orders")
            .send()
            .await?
            .queue_url
            .ok_or("GetQueueUrl returned no URL")?;

        sqs.send_message()
            .queue_url(&url)
            .message_body("hello from aws-sdk-rust")
            .send()
            .await?;

        let received = sqs
            .receive_message()
            .queue_url(&url)
            .max_number_of_messages(1)
            .wait_time_seconds(5)
            .send()
            .await?;
        let msg = &received.messages()[0];
        println!("received: {}", msg.body().unwrap_or_default());

        // DeleteMessage acks the message off the queue by its receipt handle.
        let handle = msg.receipt_handle().ok_or("no receipt handle")?;
        sqs.delete_message().queue_url(&url).receipt_handle(handle).send().await?;
        Ok(())
    }
    ```
  </Tab>
</Tabs>

## Supported languages [#supported-languages]

The connector speaks the genuine AWS SQS and SNS wire protocols, so any standard AWS SDK
works — you only override the endpoint URL. There is no KubeMQ SDK, no proto bindings, and
no published package; the examples pin one native AWS SDK per language.

| Language                | AWS SDK / client library                                                                                                                           | Endpoint override                            |
| ----------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------- |
| Go                      | [`aws-sdk-go-v2`](https://github.com/aws/aws-sdk-go-v2) (`service/sqs`, `service/sns`)                                                             | `config.WithBaseEndpoint` / `o.BaseEndpoint` |
| Python                  | [`boto3`](https://boto3.amazonaws.com/v1/documentation/api/latest/index.html) (`client('sqs')`, `client('sns')`)                                   | `endpoint_url=` per client                   |
| Java                    | [AWS SDK for Java v2](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/home.html) (`sqs`, `sns`)                                    | `.endpointOverride(URI.create(...))`         |
| JavaScript / TypeScript | [AWS SDK v3](https://docs.aws.amazon.com/sdk-for-javascript/v3/developer-guide/welcome.html) (`@aws-sdk/client-sqs`, `@aws-sdk/client-sns`)        | `{ endpoint }`                               |
| C# / .NET               | [AWS SDK for .NET](https://docs.aws.amazon.com/sdk-for-net/latest/developer-guide/welcome.html) (`AWSSDK.SQS`, `AWSSDK.SimpleNotificationService`) | `ServiceURL`                                 |
| Ruby                    | [AWS SDK for Ruby v3](https://docs.aws.amazon.com/sdk-for-ruby/v3/developer-guide/welcome.html) (`aws-sdk-sqs`, `aws-sdk-sns`)                     | `endpoint:` per client                       |
| Rust                    | [AWS SDK for Rust](https://docs.aws.amazon.com/sdk-for-rust/latest/dg/welcome.html) (`aws-sdk-sqs`, `aws-sdk-sns`)                                 | `.endpoint_url(...)`                         |

<Callout type="info">
  Only `aws-sdk-go-v2` is proven by the KubeMQ server's integration tests; the other six SDKs
  are wire-compatible and the connector's example suite is their proof. The Ruby SQS client
  needs its `QueueUrls` plugin removed (shown above) so requests stay on the configured base
  endpoint. See [Connections endpoint](/connectors/aws/reference/connections-endpoint).
</Callout>

## Next steps [#next-steps]

<Cards>
  <Card title="Getting started" href="/connectors/aws/tutorials/getting-started" description="Enable the connector, point your AWS SDK at port 4566, and run an SQS round-trip in minutes." />

  <Card title="Configuration" href="/connectors/aws/concepts/configuration" description="The opt-in enable variable, the ten CONNECTORS_AWS_* settings, and accept-any vs static credentials." />

  <Card title="SQS queues" href="/connectors/aws/how-to/sqs-queues" description="Send, receive, visibility, long-poll, and FIFO over the KubeMQ Queue channel sqs.{name}." />

  <Card title="SNS fan-out" href="/connectors/aws/how-to/fan-out" description="Virtual SNS topics fanning out to subscribed SQS queues and HTTP/HTTPS webhooks." />
</Cards>