{"id":"8b4a495d-458d-4ee1-b086-fed3b3ae07ae","slug":"raven-clown-ark","kind":"mcp","name":"ark","summary":"Config-driven Kafka-to-HTTP callback bridge: consume, retry, dead-letter, and produce, with an MCP interface for AI agents","owner":"raven-clown","repoUrl":"https://github.com/raven-clown/ark","license":"Apache-2.0","licenseTier":"permissive","trust":"community","stars":1,"match":null,"fewStars":true,"format":"mcp","createdAt":"2026-09-18T17:49:55.000Z","addedAt":"2026-10-21T09:43:54.251Z","logoUrl":"https://avatars.githubusercontent.com/u/160258096?u=274da79ca50000f94797488308a3a98be1df6576&v=4","linkOnly":false,"collection":null,"skillPath":null,"categories":["communication"],"tags":["apache-kafka","kafka","circuit-breaker","dead-letter-queue","event-driven","golang","self-hosted","webhook"],"runtime":"binary","local":true,"cloudCapable":false,"cloudCapability":"unknown","cloudReadyOnPlatform":false,"needsApiKey":false,"reviewed":false,"syncedAt":"2026-10-26T07:31:12.604Z","sample":false,"url":"https://conduits.dev/tools/raven-clown-ark","language":"Go","transports":["stdio"],"install":{},"requirements":{"env":[],"upstreamMayCharge":false},"tools":null,"readme":"

\n \n \n \"ARK\"\n \n

\n\n

Connect any Kafka topic to any HTTP app.
No Kafka client code. Nothing lost.

\n\n

\n \"CI\"\n \"Go\n \"Apache\n \"MCP\n

\n\n

\n Quick start ·\n How it works ·\n Features ·\n First pipeline ·\n Reference ·\n Website\n

\n\n

\n \n \n \"Messages\n \n

\n\nARK sits between Kafka and an HTTP endpoint your app already has. It\nconsumes each message, checks it, calls your app, and produces the answer\nto another topic. When one path isn't enough, draw a flow: conditions,\nseveral apps, webhooks and topics, and a line of its own for every\nreject and failure. Retries, ordering, dead letters, a circuit breaker,\nscaling and an AI assistant come built in. One Go binary, one YAML file.\n\nYour app stays a plain web service. It never sees a consumer group, an\noffset or a rebalance.\n\n## Why ARK\n\n- **Every team writes the same Kafka glue.** Consume, retry, dead-letter,\n commit offsets carefully, survive rebalances, add metrics. ARK is that\n glue, done once and tested against a real broker.\n- **Small on purpose.** ARK does one job: *get Kafka messages to the apps\n that act on them, and the answers back*. Flows add branches and fan-out\n when you need them, but it is not a general orchestrator like Kafka\n Connect, NiFi or Camel: there are no connectors to install, no framework\n to learn and no extra cluster to run.\n- **Safe by default.** At-least-once delivery, in-order commits, a stable\n correlation ID for idempotency, and every failure must have somewhere to\n go (a dead-letter topic or a line you draw), so a message always ends up\n somewhere you can see it.\n- **Easy to operate.** Ask it questions in plain language over MCP, browse\n and retry dead letters from the API, and scale by adding nodes that\n coordinate through Kafka itself.\n\n## Quick start\n\nYou need Docker. This starts Kafka, ARK, and a tiny demo app that echoes\nwhat it receives.\n\n```bash\ngit clone https://github.com/raven-clown/ark.git && cd ark\ndocker compose up -d --build\n```\n\nThe demo pipeline (`bridge-engine/config.demo.yaml`) reads `orders.raw`,\ncalls the demo app, and writes the answer to `orders.processed`. It also\nhas a data rule: `order_id` must be a string and `amount` a number of at\nleast 0.\n\n**1. Send three orders:** a good one, one that breaks the rule, and one\nthe app fails on.\n\n```bash\nprintf '%s\\n' \\\n '{\"order_id\":\"A1\",\"amount\":10}' \\\n '{\"order_id\":\"A2\",\"amount\":-5}' \\\n '{\"order_id\":\"A3\",\"amount\":7,\"fail\":true}' \\\n| docker compose exec -T kafka /opt/kafka/bin/kafka-console-producer.sh \\\n --bootstrap-server localhost:9092 --topic orders.raw\n```\n\n**2. See where each one went.**\n\n```bash\ntopic() { docker compose exec -T kafka /opt/kafka/bin/kafka-console-consumer.sh \\\n --bootstrap-server localhost:9092 --from-beginning --timeout-ms 5000 \\\n --property print.headers=true --topic \"$1\" 2>/dev/null; }\n\ntopic orders.processed # A1, with the app's answer\ntopic orders.rejected # A2, X-Ark-Reason: data rules: amount is -5, below the minimum 0\ntopic orders.dlq # A3, X-Ark-Reason: callback failed 3 time(s) ... answered status 500\n```\n\n**3. Ask ARK about it.**\n\n```bash\ncurl -s localhost:8080/api/v1/pipelines -H 'Authorization: Bearer demo-admin-token'\ncurl -s localhost:8080/api/v1/pipelines/orders/dlq -H 'Authorization: Bearer demo-admin-token'\n```\n\n**4. Or ask in plain language.** Connect any MCP client to\n`http://localhost:8080/mcp` with the token `demo-mcp-token`. For example,\nwith Claude Code:\n\n```bash\nclaude mcp add --transport http ark http://localhost:8080/mcp \\\n --header \"Authorization: Bearer demo-mcp-token\"\n```\n\nThen ask *\"how is the orders pipeline doing?\"* or *\"is there any weird\ndata in orders?\"* in English, Thai, or Chinese. ARK answers with what is\nhappening, why, and what to do.\n\n**5. Open the console.** Go to [http://localhost:8080](http://localhost:8080)\nand sign in with `demo-admin-token`. You'll see the pipeline with data\nflowing through it; click it to see its health, tail it live, browse dead\nletters, or change its rules.\n\n> The demo tokens are for trying ARK on your machine. Set your own with\n> `ARK_API_ADMIN_TOKENS` and `ARK_MCP_ADMIN_TOKENS` before running it\n> anywhere else, and point `ARK_CONFIG` at your own config file.\n\n### Run it with docker run\n\nOne image serves the console, the REST API, MCP and HTTP sources on one\nport. No reverse proxy or config file is needed; everything can be set\nwith environment variables.\n\n```bash\ndocker run -d --name ark -p 8080:8080 -v ark-data:/data \\\n -e ARK_BROKERS=kafka:9092 \\\n -e ARK_ADMIN_USER=admin -e ARK_ADMIN_PASSWORD='change-me-please' \\\n ghcr.io/raven-clown/ark:0.12\n```\n\nOpen http://localhost:8080 and sign in. Pipelines, projects and accounts\nyou create in the console are kept in `/data/config.yaml`, so keep the\nvolume. To try it without signing in at all, use\n`-e ARK_AUTH_ANONYMOUS=admin` instead of the two admin variables.\n\n| Variable | Default | What it does |\n|---|---|---|\n| `ARK_BROKERS` | | Kafka brokers, comma-separated (needed when there's no config file) |\n| `ARK_PORT` / `ARK_LISTEN` | `8080` | Port, or a full address such as `127.0.0.1:9000` |\n| `ARK_CONFIG_FILE` | `/data/config.yaml` | Config file; created on the first change if missing |\n| `ARK_PUBLIC_URL` | the request's host | Address browsers use, for sign-in redirects behind a load balancer |\n| `ARK_TIMEZONE` | `UTC` | Display timezone |\n| `ARK_REPLICATION_FACTOR` | `3` | For topics ARK creates (use `1` on a single broker) |\n| `ARK_TLS_CERT`, `ARK_TLS_KEY` | | Serve HTTPS with this certificate |\n| `ARK_TLS_DOMAINS`, `ARK_TLS_EMAIL` | | Serve HTTPS with a Let's Encrypt certificate for these domains |\n\nEvery sign-in method below has variables of its own. Each release is\npublished for amd64 and arm64 as `ghcr.io/raven-clown/ark` (tagged\n`0.12.0`, `0.12` and `latest`), plus the engine and console on their own\nas `ark-bridge` and `ark-console`. The image also holds `demo-echo`\n(`--entrypoint demo-echo`).\n\n## ARK Console\n\nOne place to see and run everything, with no Grafana or Kafka tooling on\nthe side.\n\n

\n \"The\n

\n\n- **Pipeline canvas.** Every pipeline, topic and target, chained pipelines\n joined through their shared topic. Each dot on a line is 10 real\n messages (more on busy lines, shown under the canvas); click a line to\n watch what crosses it, or right-click a pipeline for everything you can\n do with it. In cluster mode the numbers cover every node.\n- **Everything about one pipeline.** Health with the reason and next step,\n throughput against the last hour, p99 latency, a live tail of every\n message (in, each callback attempt, where it went and why), dead letters\n and rejects with retry and discard, the config, and pause, restart,\n scale and delete.\n- **Flow designer.** Drag steps onto a board (call your app, condition,\n webhook, send to topic, data check, reject, dead letter, drop) and draw\n lines from any way out to the next step, n8n style: a condition splits\n a message down several branches, a call has its own lines for the\n answer, a 4xx and used-up retries. Ready-made app steps send to\n OpenSearch, Elasticsearch, NiFi, another Kafka cluster, Slack, Discord,\n Microsoft Teams or any HTTP API, each with its own logo and send\n patterns: index, index by id or update in OpenSearch and Elasticsearch,\n a Block Kit card, embed or Adaptive Card for chat, PUT or PATCH by id\n for an API. Topic and consumer group fields offer what the cluster already\n has, and the fallback dead-letter and reject topics can be switched off\n once every step has its own lines. Review the generated config before it\n is applied. Right-click a running pipeline to open it in the designer; a\n fixed-path pipeline opens as the flow that does the same.\n- **Visual rule builder.** Pick a field from real messages, an operator and\n a value, and see how many recent messages would match before you save.\n- **Metrics.** Throughput, lag and callback latency percentiles from the\n last 5 minutes to the last 24 hours, per pipeline or summed per\n tenant, lag and traffic per source partition, plus the node's own\n resources.\n- **Projects.** Group pipelines, decide how far AI may go with them, give\n agents their own MCP endpoints, and choose each project's AI model.\n- **Ask ARK.** Chat with an assistant backed by the model you choose. It\n works out what you mean first, asks back only when the tools can't find\n the answer, and uses the same tools as any MCP agent, within your\n permissions. It answers in the language you write in, or always in the\n console's language if you choose that in Settings.\n- **Settings.** Console language, Ask ARK's reply language, timezone,\n default AI model and every engine tuning value.\n- In English, Thai, and simplified and traditional Chinese.\n\n\n\n\n\n\n\n\n\n\n\n\n\n
\"A\"Building
\"Projects\"Asking
\"The
\n\nThe console is its own small service (`bridge-ui`): it serves the web app\nand forwards `/api/v1/` to an ARK engine, so the browser never talks to\nKafka. Sign in with any `ARK_API_*` token; what you can do follows its\nscope.\n\n## How it works\n\nEvery message takes the same path, and each step has one clear outcome:\n\n```mermaid\nflowchart LR\n K[(source topic)] --> D{data rules}\n D -- breaks a rule --> R[(reject topic)]\n D -- ok --> F{fast path rules}\n F -- matched --> O\n F -- no match --> C[call your app]\n C -- 2xx --> P{post callback rules}\n P --> O[(destination topic)]\n C -- 4xx --> R\n C -- 5xx, timeout --> X{retries left?}\n X -- yes --> C\n X -- no --> Q[(dead letter topic)]\n```\n\n1. **Consume.** Each pipeline is a Kafka consumer group. `workers` members\n share the source topic's partitions.\n2. **Check.** `data_rules` validate the message (types, ranges, formats,\n required fields, key and headers) before anything else runs.\n3. **Route early.** `fast_path_rules` can pass, reject, drop or\n dead-letter a message on a condition, without calling your app.\n4. **Call back.** ARK POSTs the message to your app with a stable\n `X-Correlation-ID`, keeping order per key while different keys run in\n parallel.\n5. **Route late.** `post_callback_rules` look at your app's answer and can\n send it to a different topic or webhook.\n6. **Produce and commit.** The answer goes to the destination topic, and\n only then is the offset committed, in order.\n\nThat fixed path covers most pipelines. For anything else, a pipeline can\nrun a [flow](#reference) instead: the same steps joined in any shape,\nwhere a condition can split a message down several branches, one step can\nfeed several apps, webhooks and topics at once, and a reject or failure\ncan lead anywhere, not only to a topic.\n\n### What happens when...\n\n| Situation | What ARK does |\n|---|---|\n| Your app answers 2xx | The answer is produced to `destination_topic` |\n| Your app answers 4xx | The message is bad, not the app: it goes to `reject_topic` with the reason, no retry |\n| Your app answers 408, 425 or 429 | \"Come back later\": ARK waits (honoring `Retry-After`) and tries again without using a retry |\n| Your app answers 5xx or times out | Retried with backoff, then dead-lettered with the reason |\n| Your app is down | The circuit breaker opens, messages wait in Kafka, and ARK resumes the moment the health check passes |\n| A message breaks a data rule | Rejected, dead-lettered, or tagged and let through, your choice |\n| ARK crashes mid-message | Nothing was committed past it, so it is delivered again with the same correlation ID |\n| A node joins or leaves the cluster | Workers move to the live nodes in place, without restarting pipelines |\n| A step in a flow rejects or fails | The message follows that step's own line (a webhook, another app, a topic, or all of them); the offset commits once every path is done |\n\n## Features\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n
\n\n**Delivery you can trust**
\nAt-least-once with in-order commits. A message that can't finish is\nretried in place, never skipped. Stable correlation IDs make your app\nidempotent in one line.\n\n		\n\n**Data rules**
\nDeclare what a valid message looks like: required fields, types, ranges,\nlengths, patterns, enums, formats (email, uuid, date-time, url, ipv4),\nsize, unknown fields, key and headers.\n\n
\n\n**Rules without code**
\nRoute before or after the callback with\n[expr](https://expr-lang.org) conditions: auto-approve small orders,\ndead-letter obvious fraud, send big orders to another topic.\n\n		\n\n**Dead letters you can act on**
\nBrowse, retry or discard DLQ and reject entries over the API, each with\nits reason and time. Redrive them on a schedule with a limit. What was\nhandled stays handled across restarts.\n\n
\n\n**Built-in AI assistant (MCP)**
\nAny MCP client can ask what is happening and why, explain an error, find\nweird data, suggest tuning, and draft a pipeline that is applied only\nafter you confirm a preview.\n\n		\n\n**Cluster without a new dependency**
\nRun more nodes and they act as one: shared config, cluster-wide pause, a\nleader that places workers by label, fast failover. Coordination runs on\nKafka itself. No etcd, no ZooKeeper.\n\n
\n\n**Protects your app**
\nPer-pipeline concurrency limits, a circuit breaker with health probes,\nbackpressure on 429, and load balancing over several backends\n(round robin, least in flight, sticky partition).\n\n		\n\n**Secure and observable**
\nViewer, operator and admin tokens for the API and separately for MCP,\nper-pipeline AI access, audit logs. Prometheus metrics for throughput,\nlatency, lag, oldest uncommitted age, breaker state and rule matches.\n\n
\n\n**Flows like a workflow**
\nDraw steps on a board, n8n style: conditions anywhere, fan-out to several\napps, webhooks and topics, and a line for every answer, reject and\nfailure. Loops are refused, and a message still commits exactly once.\n\n		\n\n**Talks to the rest of your stack**
\nReady-made steps for OpenSearch, Elasticsearch, NiFi, another Kafka\ncluster, Slack, Discord, Microsoft Teams and any HTTP API, each with send\npatterns that fit the product (index by id, cards, PUT by id), url and\nbody templates, and secrets read from the environment.\n\n
\n\n## Your first pipeline\n\nPoint ARK at your own topic and endpoint. This is a complete, working\nconfig:\n\n```yaml\nbrokers: [kafka:9092]\ntimezone: Asia/Bangkok # every time ARK shows you is ISO 8601 in this zone\n\npipelines:\n - name: order-processor\n source_topic: orders.raw\n destination_topic: orders.processed\n dead_letter_topic: orders.dlq # required: failures always land somewhere visible\n reject_topic: orders.rejected # optional: bad data goes here instead of the DLQ\n consumer_group: order-processor\n workers: 3 # 3 consumers sharing the topic's partitions\n\n target:\n url: http://order-app:8080/process\n health_check_url: http://order-app:8080/health\n\n retry:\n max_attempts: 3\n backoff_ms: 1000\n\n data_rules:\n on_violation: reject # or dead_letter, or tag\n fields:\n - {path: order_id, required: true, type: string}\n - {path: amount, required: true, type: number, min: 0}\n - {path: email, format: email}\n\n fast_path_rules:\n - name: auto-approve-small\n condition: \"data.amount < 100\"\n action: pass_through\n```\n\nRun it:\n\n```bash\ndocker run -v $PWD/config.yaml:/etc/bridge/config.yaml -p 8080:8080 \\\n -e ARK_API_ADMIN_TOKENS=change-me ark # built from bridge-engine/Dockerfile\n# or: cd bridge-engine && go run ./cmd/bridge -config config.yaml\n```\n\nARK creates the source, dead-letter and reject topics if they are missing,\nhot-reloads the file when you call\n`POST /api/v1/config/reload`, and refuses an invalid config before it\ntouches a running pipeline. Every option is documented in\n[`config.example.yaml`](bridge-engine/config.example.yaml).\n\nYour app receives a normal HTTP POST:\n\n```http\nPOST /process HTTP/1.1\nContent-Type: application/json\nX-Correlation-ID: 30176dd15e71f315051f562780bb548d\n\n{\"order_id\":\"A1\",\"amount\":10}\n```\n\nWhatever it returns with a 2xx becomes the message on\n`destination_topic`.\n\nNeed branches, more than one destination, or your own path for rejects\nand failures? Open the pipeline in the console's flow designer, or write\na `flow:` by hand (see *Flows* under [Reference](#reference)).\n\n## Reference\n\n
\nDelivery guarantees in detail\n\n- **Offsets commit only after the result is produced**, and in order. A\n message that can't be completed blocks the commits behind it on its\n partition until it is, so a crash can never skip it. Killing ARK\n mid-message means it is delivered again.\n- **`X-Correlation-ID`** is derived from topic, partition and offset, so a\n retry or a redelivery after a crash carries the same ID. Delivery is\n at-least-once; use the ID as an idempotency key.\n- **Ordering.** `ordering: per_key` (default) sends messages with the same\n key one at a time and in order, while different keys run in parallel.\n `per_partition` and `none` are also available.\n- **Status codes.** 4xx means the message is the problem: it goes to\n `reject_topic` (or the DLQ without one) with no retry. 408, 425 and 429\n mean \"later\": ARK waits, honoring `Retry-After`, without spending a\n retry. `target.reject_statuses` overrides which codes are rejects.\n- **When the app is down,** the circuit breaker opens after\n `circuit_breaker.failure_threshold` different messages fail in a row and\n messages wait in Kafka instead of flooding the DLQ. Retries of a message\n that already failed don't count again, so a few messages your app always\n fails on can't hold up the rest. With `health_check_url` set, ARK\n resumes as soon as the app answers.\n- **`on_exhausted: block`** keeps retrying forever instead of\n dead-lettering, for pipelines where order matters more than progress.\n- Every routed message carries `X-Ark-Reason`, `X-Ark-Pipeline` and\n `X-Ark-Failed-At` headers so you can see why it ended up there.\n\n
\n\n
\nBatched calls\n\nFor an app with a cost per request (a database round trip, a slow\nupstream), `target.batch_size` sends several messages in one POST. It is\noff unless you set it, and works on flow call steps too.\n\n```yaml\ntarget:\n url: http://order-app:8080/batch\n batch_size: 50 # 2 to 1000\n batch_linger_ms: 5 # longest wait for a batch to fill (default 5)\n```\n\nARK posts, with an `X-Ark-Batch-Size` header:\n\n```json\n{\"items\": [{\"id\": \"30176dd1...\", \"key\": \"A1\", \"value\": {\"order_id\": \"A1\", \"amount\": 10}}]}\n```\n\n`id` is the message's correlation ID and `value` is the message (as a\nstring when it isn't JSON). The app answers with a 2xx:\n\n```json\n{\"results\": [{\"id\": \"30176dd1...\", \"status\": 200, \"body\": {\"ok\": true}}]}\n```\n\nEach message is then handled exactly as if it had been sent alone: a 2xx\n`status` (the default) sends `body` to `destination_topic`, a reject\nstatus sends it to `reject_topic`, 408, 425 and 429 wait, and a 5xx or a\nmissing result is a failed attempt for that message only. A non-2xx for\nthe whole request applies to every message in it. Retries, rules,\nordering and commits stay per message.\n\nA batch fills only when enough messages are waiting, so keep\n`workers x max_in_flight` at least `batch_size`; config review warns\nwhen it isn't. Measured on the compose stack against an app that takes\n20 ms per request and handles 8 at a time (4 workers, `max_in_flight:\n50`, 20,000 messages): 386 msg/s one message per call, 3,118 msg/s with\n`batch_size: 50`.\n\n
\n\n
\nData rules\n\n```yaml\ndata_rules:\n on_violation: tag # reject (default) | dead_letter | tag\n allow_unknown_fields: false # flag top-level fields no rule mentions\n max_bytes: 65536\n key: {required: true, pattern: \"^[A-Z]{2}-\\\\d+$\"}\n headers:\n - {name: source, enum: [web, mobile, pos]}\n fields:\n - {path: order_id, required: true, type: string, max_length: 32}\n - {path: amount, type: number, min: 0, max: 1000000}\n - {path: currency, enum: [THB, USD, EUR]}\n - {path: customer.email, format: email}\n - {path: created_at, format: date-time}\n```\n\nTypes: `string`, `number`, `integer`, `boolean`, `object`, `array`,\n`null`. Formats: `email`, `uuid`, `date-time`, `date`, `url`, `ipv4`.\nWith `on_violation: tag` the message goes through with an\n`X-Ark-Violations` header, a safe way to start. Not sure what to write?\nAsk the assistant to `check_data` on a pipeline: it samples real\nmessages, reports odd fields, mixed types, outliers and bad keys, and\ndrafts the rules for you.\n\n
\n\n
\nFast path and post-callback rules\n\n```yaml\nfast_path_rules: # before the callback\n - name: auto-approve-small-orders\n condition: \"data.amount < 1000 && data.risk_score < 0.3\"\n action: pass_through # pass_through | reject | drop | dead_letter\n - name: obvious-fraud\n condition: \"data.risk_score > 0.9\"\n action: dead_letter\n\npost_callback_rules: # after the callback, on its answer\n - name: high-value-order-alert\n condition: \"response.status == 200 && response.body.amount > 10000\"\n action: transform_route\n destination_override: orders.high-value # or webhook_override: http://...\n```\n\nConditions use [expr](https://expr-lang.org) syntax (`nil`, not `null`).\n\n
\n\n
\nFlows: steps joined in any shape\n\nWhen a fixed path isn't enough, give a pipeline a `flow:` instead of a\ntarget and rules. Every step can lead to several others, conditions can\nsplit a message anywhere, and an app's answer, a reject status and\nused-up retries each go their own way:\n\n```yaml\nflow:\n start: [check]\n steps:\n - {id: check, type: data_check, rules: {fields: [{path: order_id, required: true}]}, next: [route], on_fail: [bad]}\n - id: route\n type: condition\n match: all # or first\n branches:\n - {when: \"data.amount > 1000\", next: [vip-topic, crm]}\n otherwise: [app]\n - {id: vip-topic, type: topic, topic: orders.vip, next: [app]}\n - {id: crm, type: webhook, url: \"http://crm:8080/notify\"}\n - {id: app, type: call, target: {url: \"http://order-app:8080/process\"}, next: [after], on_reject: [bad], on_failure: [later]}\n - id: after\n type: condition\n branches:\n - {when: \"response.body.total > 10000\", next: [big]}\n otherwise: [done]\n - {id: big, type: topic, topic: orders.big}\n - {id: done, type: topic, topic: orders.processed}\n - {id: later, type: topic, topic: orders.retry}\n - {id: bad, type: reject}\n```\n\nSteps: `call`, `condition`, `data_check`, `topic`, `webhook`, `database`,\n`reject`, `dead_letter`, `drop`. Conditions read `data` (the message at that step),\n`original`, `response` (`status` and `body` of the last call), `reason`,\n`key` and `headers`. Every step but `drop` can lead on to more steps,\nreject and dead letter included, so a reject can also go to a webhook or\nanother app. Call and webhook steps take `headers`, where `${NAME}` is\nread from the environment so secrets stay out of the file; a webhook can\nset `method` (`POST`, `PUT`, `PATCH`), a `body` template, or a chat\n`message` template (sent as `{\"text\": ...}`, or `message_field: content`\nfor Discord). Templates are Go templates over the same values, with\n`json` to quote one safely: `{\"id\": {{json .data.order_id}}}`. The `url`\ncan be a template too, after `http(s)://host/`, so the host never comes\nfrom a message: `http://opensearch:9200/orders/_doc/{{path .original.order_id}}`\n(`path` escapes the value, and a missing one takes the failed line). A topic\nstep can set `brokers` to write to another Kafka cluster. `app:`\n(`opensearch`, `elasticsearch`, `nifi`, `kafka`, `slack`, `discord`,\n`teams`, `http`) only changes how the console draws a step.\n\nThe pipeline's `dead_letter_topic` and `reject_topic` are optional for a\nflow, as long as every step that can fail or reject has its own line for\nit; when set, they catch the steps that don't. Call steps take the same\n`target`, `retry` and `circuit_breaker` options as the fixed path. Loops\nare refused; to go around again, send to a topic a pipeline reads. A\nmessage is committed once every path it took is done, and counted once.\n\nA `database` step writes the message as a row in PostgreSQL:\n\n```yaml\n- id: save\n type: database\n database:\n driver: postgres\n dsn_env: ARK_ORDERS_DB # env var with the connection string\n table: public.orders\n columns:\n id: \"{{.data.order_id}}\"\n amount: \"{{.data.amount}}\"\n raw: \"{{json .data}}\"\n upsert_on: [id] # needs a unique index on id\n on_failure: [dead]\n```\n\nEach value is a template sent as a query parameter (an empty one is\nNULL), and table and column names must be plain names, so a message can\nnever change the SQL. With `upsert_on`, a message delivered twice updates\nits row instead of adding another. A write retries `retry.max_attempts`\ntimes, then takes `on_failure` (or the dead-letter topic). If `dsn_env`\nisn't set on the server, the step's writes fail that way too; the rest of\nARK keeps running.\n\n
\n\n
\nHTTP sources\n\nAn app that can only POST (a payment provider's webhooks, a form, a\nscript) can send straight into Kafka through ARK:\n\n```yaml\nsources:\n - name: shop\n type: http\n topic: shop.orders\n token_env: ARK_SOURCE_SHOP_TOKEN # env var with the bearer token\n key: order_id # optional: JSON path used as the Kafka key\n max_body_bytes: 1048576\n partitions: 3 # used when ARK creates the topic\n data_rules: # optional\n fields:\n - {path: order_id, required: true, type: string}\n```\n\n```\nPOST /ingest/shop\nAuthorization: Bearer \n\n{\"order_id\": \"A1\", \"amount\": 10}\n```\n\nARK answers once Kafka has the message:\n\n| Status | Meaning |\n|---|---|\n| 202 | On the topic. Any pipeline reading it takes it from there. |\n| 401 | Missing or wrong token. |\n| 413 | Over `max_body_bytes`. |\n| 400 | Not a JSON object, when `key` or `data_rules` needs one. |\n| 422 | Broke a data rule; the body lists each violation. Nothing was written. |\n| 503 | Kafka didn't take it; send it again after `Retry-After`. |\n\nAn `X-Correlation-ID` on the request is kept on the message, and\n`X-Ark-Source` says which source it came from. Sources show on the\nconsole canvas beside their topic. `ark_source_requests_total` counts\nrequests by source and outcome.\n\n
\n\n
\nDead letters and redrive\n\n- `GET /api/v1/pipelines/{name}/dlq` lists recent entries with reason,\n time, correlation ID and redrive count.\n- `POST .../dlq/{id}/retry` sends the message through the pipeline again\n (rules and all); `POST .../dlq/{id}/discard` removes it from the list.\n- The same routes exist under `.../reject`.\n- What was retried or discarded is recorded in Kafka, so it stays handled\n after a restart and can't be retried twice.\n- `dead_letter_redrive: {after_seconds: 3600, max_times: 3}` retries dead\n letters on a schedule and leaves the rest for a human.\n\n
\n\n
\nSeveral backends for one pipeline\n\n```yaml\ntarget:\n mode: multi_url\n urls:\n - http://order-app-1:8080/process\n - http://order-app-2:8080/process\n health_check_urls:\n - http://order-app-1:8080/health\n - http://order-app-2:8080/health\n strategy: round_robin # or least_inflight, sticky_partition\n```\n\nEach URL is probed on its own and only healthy ones get traffic. If every\nURL is down, messages wait in Kafka. `sticky_partition` keeps a partition\non the same backend while it stays healthy.\n\n
\n\n
\nAI assistant over MCP\n\nSet `ARK_MCP_VIEWER_TOKENS`, `ARK_MCP_OPERATOR_TOKENS` or\n`ARK_MCP_ADMIN_TOKENS` (comma-separated) and ARK serves MCP at `/mcp` on\nthe API port. Any MCP client and any model can use it.\n\n| Ask | Tool the assistant uses |\n|---|---|\n| \"Hi, what can you do?\" | `get_help`, `get_overview` |\n| \"Why is orders slow?\" | `diagnose_pipeline`: what, why, what to do |\n| \"What does this error mean?\" | `explain_error`: where it comes from and the fix |\n| \"What happened at 3am?\" | `get_recent_events` |\n| \"How busy was tenant acme since 9am?\" | `get_metrics`: a pipeline or a tenant over a time range |\n| \"Any weird data coming in?\" | `check_data`, `test_message` |\n| \"How do I handle 2000 msg/s?\" | `recommend_tuning` |\n| \"Create a pipeline from A to B\" | `get_pipeline_schema`, `validate_pipeline_config`, `create_pipeline` |\n| \"Send big orders to OpenSearch too, and ping Slack when the app fails\" | `get_pipeline_schema` (its `flow_example_yaml`), `validate_pipeline_config`, `apply_pipeline_config` |\n\nEvery conversation starts with `interpret_request`, which works out what\nyou mean (Thai, English, simplified and traditional Chinese, plus your\nown words through `assistant.lexicon`) and which pipeline you are talking\nabout, and asks back when something is unclear.\n\nAccess is layered. A token's scope is a ceiling: `viewer` reads,\n`operator` also pauses, resumes, retries and discards, `admin` also\ncreates and changes pipelines. Each pipeline's `mcp_access`\n(`read_only`, `read_write`, `none`) is a second limit. Config changes are\ntwo steps: the first call returns a preview and a diff, and only a second\ncall with the confirm token applies it. Every write is audit-logged.\n\n
\n\n
\nProjects and AI access\n\n```yaml\nprojects:\n - name: commerce\n ai_access: operate # none | read_only | operate | configure\n mcp_endpoints:\n - name: ops # served at /mcp/commerce/ops\n access: operate\n tokens_env: ARK_MCP_COMMERCE_OPS_TOKENS\n - name: support\n access: read_only\n tokens_env: ARK_MCP_COMMERCE_SUPPORT_TOKENS\n tools: [get_overview, diagnose_pipeline, explain_error]\n assistant: {provider: anthropic, model: claude-opus-5, api_key_env: ANTHROPIC_API_KEY}\n\npipelines:\n - name: orders\n project: commerce\n # ...\n```\n\nA project's `ai_access` is a ceiling everywhere, the global `/mcp`\nincluded: `none` hides its pipelines from agents, `read_only` lets them\nlook but not touch, `operate` adds pause, resume, retry and discard, and\n`configure` adds creating and changing pipelines (always after a\nconfirmed preview). Each endpoint gets its own tokens from the\nenvironment variable it names, never from the file, and only sees its\nproject. Projects can be managed from the console.\n\n
\n\n
\nThe assistant, with any model\n\nSet a default model under `assistant.model`, or one per project:\n\n| Provider | Example |\n|---|---|\n| Anthropic | `{provider: anthropic, model: claude-opus-5}` (key from `ANTHROPIC_API_KEY`) |\n| OpenAI | `{provider: openai, model: gpt-5}` (key from `OPENAI_API_KEY`) |\n| Google Gemini | `{provider: gemini, model: gemini-2.5-pro}` (key from `GEMINI_API_KEY`) |\n| Anything OpenAI-compatible | `{provider: openai_compatible, model: qwen3, base_url: \"http://localhost:11434/v1\"}` (DeepSeek, Mistral, Groq, OpenRouter, Together, Azure OpenAI, Ollama, vLLM, LM Studio, ...) |\n\n`api_key_env` changes which variable holds the key. The console's Ask ARK\npanel (and `POST /api/v1/assistant/chat`) answers in two passes: first\nthe model restates what you mean, or asks one question back; then it\nlooks things up and acts through ARK's MCP tools. Its permissions are the\nlower of your API token's scope and the project's `ai_access`.\n\nIt replies in `reply_language` when the request sets one (`en`, `th`,\n`zh-Hans`, `zh-Hant`), and otherwise in the language the message is\nwritten in, told to the model explicitly because small local models drift\ninto English after reading English tool results; an answer that still\ncomes back in the wrong language is rewritten once. Times are quoted as\nthe tools give them, ISO 8601 in the engine's timezone. A question back to\nyou only goes out when ARK's own parser agrees something is missing that\nno tool can look up.\n\n
\n\n
\nTuning\n\nEvery internal limit and interval has a default and can be changed under\n`tuning:` in the config or on the console's Settings page: retry backoff\nand Retry-After caps, commit and batch timeouts, DLQ browser size,\ndiagnosis thresholds, metric history, event log size, live tail payload\nsize, confirm token and assistant session lifetimes, and cluster\ncatch-up. See [`config.example.yaml`](bridge-engine/config.example.yaml)\nfor the full list with defaults.\n\n
\n\n
\nRunning more than one ARK\n\nWithout anything special, more copies with the same `consumer_group`\nalready split the work. Turn on cluster mode to make them act as one:\n\n```yaml\ncluster:\n enabled: true\n name: prod # namespaces ARK's internal topics\n labels: {zone: dmz} # used by placement.node_selector\n```\n\n- **One config.** Pipeline config lives in a compacted Kafka topic. A\n reload on any node reaches every node, and an invalid config is\n refused before it spreads.\n- **One control surface.** Pause through any node and the pipeline pauses\n everywhere. `GET /api/v1/cluster/pipelines` returns cluster-wide\n numbers with a per-node breakdown, and any node can browse and retry\n any pipeline's DLQ.\n- **Placement.** An elected leader spreads each pipeline's workers across\n live nodes matching `placement.node_selector`, never more than the topic\n has partitions, and adds or removes workers in place.\n- **Failover.** A node that stops cleanly hands over in about a second; one\n that crashes is replaced after `node_timeout_seconds`.\n\n
\n\n
\nSigning in\n\nEvery way of signing in is optional, and any mix can be on at once. With\nnone, ARK answers only its own machine. Each one ends with one of three\naccess levels: **viewer** reads, **operator** can also pause, resume,\nrestart, retry and discard, and **admin** can also change config and\naccounts. The console's sign-in page shows the methods that are on.\n\n| Method | Turn it on with | Access from |\n|---|---|---|\n| No sign-in | `ARK_AUTH_ANONYMOUS=viewer` (or `operator`, `admin`); `ARK_AUTH_MCP_ANONYMOUS` for MCP | The level you set. The console shows \"Open access\". |\n| API tokens | `ARK_API_VIEWER_TOKENS`, `_OPERATOR_TOKENS`, `_ADMIN_TOKENS`; `ARK_MCP_*_TOKENS` for MCP | Which list the token is in |\n| Accounts | `ARK_ADMIN_USER` and `ARK_ADMIN_PASSWORD` for the first admin, or `ARK_AUTH_ACCOUNTS=true` | Set per account under Settings, Accounts |\n| LDAP / Active Directory | `ARK_LDAP_URL`, then `ARK_LDAP_USER_DN` (such as `{username}@corp.example.com`) or `ARK_LDAP_BIND_DN`, `ARK_LDAP_BIND_PASSWORD` and `ARK_LDAP_BASE_DN` | `ARK_LDAP_ADMIN_GROUPS`, `_OPERATOR_GROUPS`, `_VIEWER_GROUPS`, from `memberOf` or a search under `ARK_LDAP_GROUP_BASE_DN` |\n| HTTP Basic | `ARK_AUTH_BASIC=true`, with accounts or LDAP | As above, for scripts |\n| OpenID Connect | `ARK_OIDC_PRESET`, `ARK_OIDC_ISSUER`, `ARK_OIDC_CLIENT_ID`, `ARK_OIDC_CLIENT_SECRET` | `ARK_OIDC_ADMIN_GROUPS`, `_OPERATOR_GROUPS`, `_VIEWER_GROUPS`, read from `ARK_OIDC_ROLES_CLAIM` |\n| SAML 2.0 | `ARK_SAML_METADATA_URL` (or `_METADATA_FILE`), optional `ARK_SAML_CERT_FILE` and `_KEY_FILE` | `ARK_SAML_ADMIN_GROUPS` and so on, from `ARK_SAML_ROLES_CLAIM` |\n| GitHub, GitLab, Bitbucket | `ARK_GITHUB_CLIENT_ID` and `ARK_GITHUB_CLIENT_SECRET` (likewise `GITLAB`, `BITBUCKET`) | `ARK_GITHUB_ADMIN_GROUPS` and so on: `org` or `org/team` for GitHub, group paths for GitLab, workspaces for Bitbucket |\n| Other OAuth 2.0 | `ARK_OAUTH2_NAME`, `_CLIENT_ID`, `_CLIENT_SECRET`, `_AUTH_URL`, `_TOKEN_URL`, `_USERINFO_URL`, optional `_GROUPS_URL` | `ARK_OAUTH2_ADMIN_GROUPS` and so on |\n| JWT from your own system | `ARK_JWT_JWKS_URL`, `ARK_JWT_PUBLIC_KEY_FILE` or `ARK_JWT_SECRET`, plus `ARK_JWT_ISSUER`, `ARK_JWT_AUDIENCES` | `ARK_JWT_ADMIN_GROUPS` and so on, from `ARK_JWT_ROLES_CLAIM` |\n| A proxy that already signed people in | `ARK_HEADER_PRESET` (`azure-easy-auth`, `cloudflare-access`, `oauth2-proxy`) or `ARK_HEADER_USER`, and always `ARK_HEADER_TRUSTED_PROXIES` | `ARK_HEADER_ADMIN_GROUPS` and so on |\n| Client certificates | `ARK_CLIENT_CA`, with HTTPS on | `ARK_CLIENT_CERT_ADMIN_GROUPS` and so on, matched against the certificate's OUs |\n\nOpenID Connect presets fill in what each provider needs:\n\n| Preset | Issuer | Groups come from |\n|---|---|---|\n| `azure` | built from `ARK_OIDC_TENANT` | the `roles` claim (app roles) |\n| `adfs` | `https:///adfs` | the `group` claim; asks for `allatclaims` |\n| `google` | `https://accounts.google.com` | `hd`, the Google Workspace domain |\n| `keycloak` | `https:///realms/` | `groups` |\n| `okta` | `https://.okta.com` | `groups`; asks for the `groups` scope |\n| `auth0` | `https://.auth0.com/` | set `ARK_OIDC_ROLES_CLAIM` |\n| `cognito` | `https://cognito-idp..amazonaws.com/` | `cognito:groups` |\n\nRegister `https:///auth/oidc/callback` (OIDC),\n`https:///auth/oauth2//callback` (OAuth 2.0) or\n`https:///auth/saml/acs` with the provider; ARK's SAML metadata is at\n`/auth/saml/metadata`. With no `ARK_OIDC_VIEWER_GROUPS` (and the like),\nanyone the provider signs in gets in as a viewer.\n\nBrowser sign-ins end in an HttpOnly session cookie (12 hours, or\n`ARK_SESSION_HOURS`); changes made with it need the console's CSRF\nheader. Set `ARK_SESSION_SECRET` (32+ characters) so sessions survive a\nrestart and work on every node. A session ends as soon as its account is\ndeleted or its access changes. Every method can also be set under `auth:`\nin the config file, which `config.example.yaml` shows.\n\nOIDC, SAML and plain OAuth 2.0 were tested against Keycloak, LDAP against\nOpenLDAP, and the rest against ARK itself. The Azure, ADFS, Google, Okta,\nAuth0, Cognito, GitHub, GitLab and Bitbucket presets follow each\nprovider's documentation but haven't been tried against the real service.\n\n
\n\n
\nREST API\n\nEvery route except `/healthz`, `/metrics` and `/api/v1/auth/info` needs\nsomeone signed in (see *Signing in*). Viewers can read, operators can\nalso pause, resume, restart, retry and discard, and admins can also\nchange config (preview, confirm, scale, reload) and accounts. The\n`ARK_API_*` tokens are separate from the `ARK_MCP_*` ones.\n\n| Route | What it does |\n|---|---|\n| `GET /healthz` | Liveness |\n| `GET /metrics` | Prometheus metrics |\n| `GET /api/v1/auth/info` | Which sign-in methods are on (public) |\n| `POST /api/v1/auth/login` | Username and password; answers a session token and cookie |\n| `GET /auth/oidc/login`, `/auth/saml/login`, `/auth/oauth2//login`, `/auth/logout` | Browser sign-in and sign-out |\n| `GET`, `PUT`, `DELETE /api/v1/config/users[/]` | Accounts (admin) |\n| `POST /ingest/` | An HTTP source; its own bearer token, see *HTTP sources* |\n| `GET /api/v1/pipelines` | Status of every pipeline worker |\n| `GET /api/v1/pipelines/{name}` | Status of one pipeline |\n| `POST /api/v1/pipelines/{name}/pause`, `/resume` | Pause or resume |\n| `GET /api/v1/pipelines/{name}/dlq`, `/reject` | Recent entries |\n| `GET .../dlq/{id}`, `.../reject/{id}` | One entry |\n| `POST .../dlq/{id}/retry`, `/discard` | Retry or discard (same under `/reject`) |\n| `POST /api/v1/config/reload` | Re-read the config |\n| `GET /api/v1/cluster` | Members, leader, config versions |\n| `GET /api/v1/cluster/pipelines` | Cluster-wide pipeline numbers |\n| `GET /api/v1/overview` | Health of every pipeline and what needs attention |\n| `GET /api/v1/events` | What happened and why (`pipeline`, `since_minutes`, `kinds`, `limit`) |\n| `GET /api/v1/topology` | How pipelines, topics and targets connect |\n| `GET /api/v1/topics` | Kafka topics and which pipelines use them |\n| `GET /api/v1/consumer-groups` | Consumer groups the cluster knows about |\n| `GET /api/v1/pipelines/{name}/tail` | Live tail as Server-Sent Events (`stage`, `to`, `key`, `correlation_id`, `max_per_sec`) |\n| `GET /api/v1/pipelines/{name}/diagnosis`, `/tuning`, `/data-check` | What's wrong and why, sizing advice, odd data |\n| `POST /api/v1/pipelines/{name}/test-message` | What the pipeline would do with a message |\n| `POST /api/v1/pipelines/{name}/restart` | Restart its workers on this node |\n| `POST /api/v1/pipelines/{name}/scale` | Change `workers` (admin) |\n| `GET /api/v1/config/pipelines`, `/config/schema` | Current config as YAML, and the schema |\n| `POST /api/v1/config/validate` | Check a pipeline without applying it |\n| `POST /api/v1/config/preview`, `/config/confirm` | Create, change or delete a pipeline in two steps (admin) |\n| `GET /api/v1/pipelines/{name}/rules`, `POST .../rules/test`, `.../rules/preview` | Read rules, test a condition on recent messages, change rules (admin) |\n| `GET /api/v1/history?minutes=&pipeline=` or `&tenant=` | Rates, lag, latency percentiles and per-partition lag for a pipeline or a tenant's total; 5s points for the last hour, 1-minute points up to 24 hours |\n| `GET /api/v1/projects`, `PUT`/`DELETE /api/v1/config/projects/{name}` | Projects, their endpoints and models (changes need admin) |\n| `POST /api/v1/assistant/chat` | Ask the assistant (acts within your scope; optional `reply_language`) |\n| `GET`/`PUT /api/v1/config/settings` | Timezone, default model and tuning (changes need admin) |\n\n
\n\n
\nMetrics\n\n`ark_messages_processed_total`, `ark_messages_rejected_total`,\n`ark_messages_dead_lettered_total`, `ark_messages_failed_total`,\n`ark_messages_backpressured_total`, `ark_callback_duration_seconds`,\n`ark_data_rule_violations_total`, `ark_consumer_lag`,\n`ark_oldest_uncommitted_age_seconds`, `ark_circuit_breaker_open`,\n`ark_pipeline_paused`, `ark_worker_up`,\n`ark_last_activity_timestamp_seconds`, `ark_fast_path_rule_matches_total`,\n`ark_post_callback_rule_matches_total`. All labeled by pipeline and\ntenant.\n\n
\n\n## Roadmap\n\n- **More sources and sinks,** such as change data capture or other\n brokers, if there is demand for them.\n\nThe full plan, with the reasoning behind every decision, is in\n[PLAN.md](PLAN.md).\n\n## Repository\n\n| Path | What's there |\n|---|---|\n| [`bridge-engine/`](bridge-engine) | The Go engine: consumer, rules, callback client, producer, REST API, MCP server, cluster |\n| [`bridge-ui/`](bridge-ui) | The ARK Console: a React app and a small Go server |\n| [`brand/`](brand) | Logo and colors |\n| [`docs/`](docs) | The website |\n\n## Contributing\n\nIssues and pull requests are welcome. Read\n[CONTRIBUTING.md](CONTRIBUTING.md) for the dev setup and where ARK's scope\nis deliberately bounded, which is worth a look before proposing something\nbig.\n\n## License\n\n[Apache License 2.0](LICENSE)\n"}