mq-bridge-app

io.github.marcomq由 GitHub 账号 marcomq 发布v1.0.2更新于 Oct 24, 2026

Move data between 15+ databases, queues and files at high throughput, without it entering context

概览

AI 生成的概览

mq-bridge 的 MCP 入口,提供零代码 ETL 引擎,在 15 种以上数据库、消息队列和文件之间搬运数据,且数据不进入上下文。

可以做什么
  • 运行已配置的路由,在消息中间件、数据库、文件和 HTTP 端点之间复制数据
  • 把 Postgres 或 MongoDB 的变更流式传输到 Kafka、文件或数据仓库
  • 读写 SQL 数据库、MongoDB、ClickHouse、对象存储以及 CSV/JSONL 文件
  • 在路由中应用转换、重试、死信队列、去重和限流
  • 调用兼容 OpenAI 的 LLM 与向量接口,对消息进行分类、摘要或向量化
适用场景
适合让助手搭建或驱动系统之间的大批量数据搬运,例如消息中间件互通、数据库入库、CDC 管道或文件格式转换,而不必让记录经过模型。它是数据搬运工具,不适合做窗口、连接等有状态流处理。
运行要求
以本地桌面进程通过 stdio 运行,提供 OCI 镜像和 Cargo 包两种分发方式。每条路由需要源端与目标系统的连接信息(中间件地址、数据库凭据、对象存储、HTTP 端点),通常通过 YAML 配置或环境变量提供。清单未声明任何认证或环境变量。
安装前请注意
路由会写入目标端,可能插入、upsert 或发布数据,配置错误可能覆盖或重复写入记录。配置中会包含数据库、中间件和云存储的凭据,避免将其暴露在共享上下文中。LLM 与向量接口会把消息内容发送给第三方 API。README 说明代码在大量 AI 辅助下编写。

安装

在 SourceWeft 中

  1. 打开 控制台中的 mq-bridge-app,将其添加到工作区。
  2. 为需要使用其工具的对话启用该服务。

Desktop only,通过 STDIO。 STDIO 服务会启动本地进程,因此需要 SourceWeft 桌面宿主。

其他 MCP 客户端

参照 仓库 中的启动说明。

README

[mq-bridge]

mqb · mq-bridge

crossing streams · library

One fast Rust engine to move and reshape data between brokers, databases, files and HTTP. Try it in the CLI, ship it in your Rust, Python or Node.js service.

[Crates.io] [Docs.rs] [Book] [Benchmark] [Linux] [Windows] [macOS] [Supply chain] [Security] [License]

🖥️ mq-bridge-app · 📊 ETL benchmarks · 🏗️ Architecture · ⚙️ Config · 📚 Middleware & endpoint reference

What is mq-bridge?

mq-bridge is an asynchronous Rust library that moves messages and data between systems: Kafka, NATS, RabbitMQ, MQTT, MongoDB, PostgreSQL/MySQL/SQLite, Postgres CDC, ClickHouse, Redis Streams, HTTP, WebSocket, gRPC, ZeroMQ, AWS SQS/SNS, IBM MQ, cloud object storage, files, and in-memory channels. Every endpoint sits behind the same batch-shaped receive_batch / send_batch traits, so any source can feed any sink.

A route connects one input to one output and shapes the data on the way: transform, filter, switch, fan out, enrich, retry, rate-limit, deduplicate, or answer a request, all by configuration. The engine is fast: a typing transform runs at ~1.6M rows/s, ~19.5x Redpanda Connect on the same job. Structural endpoints such as switch, fanout and sequence nest freely inside a route, and wrong delivery combinations fail when the route starts, not in production. Where configuration isn't enough, you add a handler in Rust, Python or Node.js. Batching, ack/nack, commit ordering and broker I/O stay in the Tokio-based core, so your code works with plain CanonicalMessages and doesn't handle Kafka offsets or AMQP nacks.

Try it in the CLI, ship it in your service: mq-bridge-app runs a route from the command line (mqb), as a server, desktop app or MCP server; the same config runs as a library inside your Rust, Python or Node.js service when you need lower latency or want the pipeline in your production code. No daemon or platform to operate.

Bridge between ecosystems: two plugins add over 100 more inputs and outputs: Redpanda Connect's components and faucet's connectors. They load only when a route uses them and run in the same process, so one route can read with one framework, process with a second and write with a third, under one ack chain and with mq-bridge's retry, DLQ and deduplication around all of it. See Combine ecosystems in one route.

Use cases

ProblemGuide with a runnable example
A RAG or semantic-search index drifts from its Postgres sourceKeep a vector index in sync with Postgres: CDC → embeddings → Qdrant
Search results show what MongoDB no longer hasKeep a search index in sync with MongoDB: change stream → Meilisearch
The warehouse is a day behind and misses every change in betweenStream Postgres changes into a warehouse: CDC → Bloblang → Snowflake
Free-text messages need routing by what they sayClassify and route messages with an LLM
Events need a database record and per-key totals to act onEnrich events and add per-key totals in Postgres
A service must update its database and publish an event atomicallyTransactional outbox with Postgres and Kafka
Tests for Kafka, NATS or RabbitMQ consumers need a running brokerTest message handlers without a broker

Quick Start

Install the library for your language:

LanguagePackageInstall
Rustmq-bridgecargo add mq-bridge --features kafka,nats,yaml
Pythonmq-bridge (PyPI)pip install mq-bridge
Node.jsmq-bridge (npm)npm install mq-bridge
C / C++mq_bridge.hcargo build --release -p mq-bridge-c (build from source)

Describe a route — here Kafka → NATS with retries — in routes.yaml:

yaml
kafka_to_nats:  input:    kafka: { url: "localhost:9092", topic: "orders", group_id: "bridge" }  output:    middlewares:      - retry: { max_attempts: 5 }    nats: { url: "nats://localhost:4222", subject: "orders.processed" }

Run it:

python
from mq_bridge import RouteRoute.from_file("routes.yaml", "kafka_to_nats").run()  # blocks; use start() to keep going
js
import { Route } from "mq-bridge";Route.fromFile("routes.yaml", "kafka_to_nats").start();
rust
mq_bridge::deploy_file("routes.yaml").await?; // every route in the file, in the background

Attach a handler (with_handler / withHandler) when you need business logic in between. No code at all? mq-bridge-app runs the same engine from the command line or a desktop UI:

bash
mqb copy 'kafka://localhost:9092?topic=orders' 'nats://localhost:4222?subject=orders.processed'

Capabilities at a glance

Every row works from the Rust crate, the Python and Node.js packages, and the zero-code app; the Cargo feature is only needed when you build the Rust crate yourself. The linked book documents the library as well as the app.

CapabilityHow (endpoint / middleware · Cargo feature)Docs
Message brokers: Kafka, NATS (JetStream), RabbitMQ (AMQP), MQTT, Redis Streams, AWS SQS/SNS, IBM MQ, ZeroMQkafka, nats, amqp, mqtt, redis_streams, aws, ibmmq, zeromq endpoints · features kafka, nats, amqp, mqtt, redis-streams, aws, ibm-mq, zeromqConnectors
Change Data Capture (CDC): stream Postgres or MongoDB changes to Kafka, files, a warehouse, …postgres_cdc (logical replication, pgoutput) · postgres-cdc; mongodb with consume: capture_all / capture_new (change streams) · mongodbPostgres CDC tutorial · MongoDB
SQL databases: PostgreSQL, MySQL, MariaDB, SQLite as source (cursor polling) or sink (insert / upsert)sqlx endpoint · sqlxPostgreSQL / MySQL / SQLite · Upserts
MongoDB as source or sinkmongodb endpoint · mongodbMongoDB
ClickHouse bulk insert and resumable readsclickhouse endpoint · clickhouseClickHouse
Search engines and other bulk document APIs (Meilisearch, Typesense, Elasticsearch, Qdrant, PostgREST / Supabase)http_bulk endpoint · http-bulkHTTP bulk
LLMs and embeddings over an OpenAI-compatible API (OpenAI, Ollama, vLLM, …): classify, summarise or embed each messageopenai_chat, openai_embeddings endpoints · http-bulkOpenAI-compatible APIs · LLM cookbook
Object storage: S3 (and S3-compatible, e.g. R2), GCS, Azure Blob, or a local directory, as Parquet, JSONL or CSVobject_store endpoint with format: parquet | raw | json | csv · object-store, parquetObject storage
Files: CSV and JSONL, read and writefile endpoint · built inFile
Analytics / warehouses: Snowflake, BigQuery, Databricks / Spark, Athena / Trino, DuckDB via Parquet on S3/GCS/Azure. These are not native connectors: mq-bridge writes the Parquet files, the warehouse loads or queries themobject_store with format: parquet · parquetSnowflake · BigQuery · Databricks / Spark · Athena / Trino · DuckDB
HTTP, gRPC, WebSocket as server or client, including request/replyhttp, grpc, websocket endpoints · http, grpc, websocketHTTP · gRPC · Request / reply
Schema validation and type coercion (JSON Schema subset, field mapping)transform middleware · built inTransform & schema mapping
Reliability: retries, dead-letter queue, deduplication, rate limitingretry, dlq, deduplication, limiter middleware · dedup for deduplicationMiddleware reference · Delivery guarantees
Broker-free tests: run a route or handler against in-memory channels, no Dockermemory endpoint · built inEmbed the library
ETL without codemq-bridge-app: mqb copy <from> <to>, YAML routes, desktop UI, MCP serverQuick start: mqb copy
More systems through plugins: Apache Pulsar, Meilisearch, Redpanda Connect's inputs, outputs and processors, and faucet's sources and sinks (MySQL / SQL Server / Oracle CDC, Snowflake, BigQuery, Iceberg, Delta, GraphQL APIs, …)native plugins · pluginPlugin endpoints · Connect plugin: inputs, outputs, processors · faucet plugin

The Connect plugin links only Redpanda Connect components free of Redpanda Community License code (bundled code is Apache-2.0 and MIT), so some upstream components, such as its Kafka, AWS and Snowflake ones, are not included; see what is not included. Native endpoints and the faucet plugin cover most of that gap.

Combine ecosystems in one route

No single framework has every connector and processor. mq-bridge combines three in one route:

Parts fromWhat they bringCount
mq-bridge, nativeKafka, NATS, RabbitMQ, MQTT, IBM MQ, Postgres and MongoDB CDC, SQL, object storage, paged HTTP APIs (http_bulk), HTTP / gRPC / WebSocket servers, …; transform, switch, fanout, aggregate, lookup, retry, DLQ, dedup20+ transports
Connect pluginRedpanda Connect's components; Bloblang processors as middleware on any endpoint56 inputs, 68 outputs, 86 processors
faucet pluginfaucet's connectors: MySQL / SQL Server / Oracle CDC, Snowflake, BigQuery, Iceberg, Delta, GraphQL, Singer taps; exactly-once into supporting sinks41 sources, 34 sinks

Plugins load only when a route names them and run in-process. Acks, commits, retry, dlq and deduplication work across all parts, and plugin endpoints nest inside switch, fanout and sequence. Where a system has both, prefer the native endpoint.

ScenarioNativeConnectfaucet
Postgres into Snowflake, backfill then changes, exactly oncepostgres_cdc, retry, dlqconnect_mappingsnowflake sink
MySQL, SQL Server or Oracle changes into Kafkakafka output, transformconnect_mappingmysql-cdc, mssql-cdc, oracle-cdc
Sensor data: live statistics, alerts to a time-series DB, all to the lakemqtt, aggregate, switch, fanoutquestdb outputiceberg or delta sink

Install the Connect plugin with brew, conda, pip or npm (install), the faucet plugin with pip install mq-bridge-faucet or npm install mq-bridge-faucet (README).

For AI agents: llms.txt is a curated index of the documentation, and npx skills add marcomq/mq-bridge installs a skill that teaches Claude Code, Codex and other agents to drive mqb.

Benchmarks

Like-for-like ETL comparisons, measured through the zero-code mq-bridge-app; every number, baseline and command is in the ETL benchmark harness:

Scenariomq-bridgeCompared with
CSV → JSONL, 1M mixed-type rows (~116 MiB)3,134,796 rows/s, ~29 MiB RAMMeltano (tap-csv → target-jsonl): 9,771 rows/s, ~444 MiB RAM — mq-bridge ~321x faster
DuckDB, all cores: 2,109,704 rows/s — mq-bridge ~1.5x faster, ~18x less memory
Kafka → file, 1M rows, no transform878,105 rows/s, ~160 MiB RAMSea Streamer, mimalloc build: 492,465 rows/s, ~894 MiB RAM — mq-bridge ~1.8x faster (also ~1.8x vs. its default-allocator build)
CSV → JSONL, 1M rows, typing transform1,594,896 rows/sRedpanda Connect 4.112.0, Bloblang mapping: 81,893 rows/s — mq-bridge ~19.5x faster, outputs asserted identical
Kafka → JSONL, 1M rows, projection766,280 rows/sRedpanda Connect 4.112.0: 65,741 rows/s — mq-bridge ~11.7x faster, under half the memory, identical sink bytes

The CSV row reuses an existing, independently published 1M-row CSV → JSONL workload, on a byte-identical fixture: 7 mixed-type columns, seed 42, 121,981,421 bytes, SHA-256 a84894e0…0c45b221 (full definition). Whole-process wall-clock, no transformation, Meltano run on the same machine and file.

All figures: mq-bridge 0.4.20, except the Redpanda Connect rows, which were measured together with their native mq-bridge figures in one later session (scenario 10). DuckDB is a throughput ceiling for the conversion itself, not an ETL tool.

External benchmark: both the Rust library (mq-bridge) and the Python binding (mq-bridge-py) are entries on the third-party http-arena.com leaderboard, which compares HTTP frameworks by requests per second (live, so rankings shift over time). It measures mq-bridge's HTTP (and WebSocket) serving path, which is one endpoint among many, not broker or ETL throughput; those are covered under Benchmarks. See the Python analysis notes for the local comparison harness.

Why mq-bridge

What you get from one engine, whichever language you call it from:

  • 20+ native transports, 100+ more via plugin, one API: every connector listed above, plus dir_spool (a crash-safe directory FIFO queue), behind the same receive_batch / send_batch shape.
  • Redpanda Connect reach: the Connect plugin adds Redpanda Connect's production-proven components — 56 inputs and 68 outputs as endpoints, plus 86 processors, most of which also run as middleware on any endpoint — while mq-bridge keeps routing, batching, retries, DLQ and deduplication.
  • faucet reach: the faucet plugin adds faucet's 41 sources and 34 sinks as one faucet endpoint: MySQL / SQL Server / Oracle CDC, Snowflake, BigQuery, Iceberg, Delta, GraphQL APIs and Singer taps, with commit-token exactly-once into sinks that support it.
  • The best of three frameworks in one route: native endpoints, Connect and faucet parts mix freely in one route and one process; see Combine ecosystems in one route.
  • Change Data Capture: stream row-level changes from Postgres (logical replication / pgoutput) and MongoDB (change streams) as flat rows with an operation marker.
  • Restart-safe delivery: batch-aware ack/nack with commit sequencing for cumulative-ack brokers; the integration suite shows no data loss during in-flight broker restarts, including a Postgres CDC restart-safety test.
  • Reliability middleware, not a framework: retries, dead-letter queues, deduplication, rate limiting, and cookie/session persistence wrap any endpoint.
  • Polyglot on one engine: the same Rust core ships as native Python and Node.js bindings; routing, batching, and broker I/O stay in Rust.
  • TLS everywhere, one config shape: a single TlsConfig block (CA bundle, client cert/key for mTLS, insecure-skip) is reused across transports.
  • Self-hosted, no daemon: generate config in the optional UI, paste it into your code, run it in-process. No hosted control plane, no separate scheduler.

Prefer not to write code? mq-bridge-app runs the exact same engine as a standalone, zero-code ETL service configured entirely by YAML or environment variables. It ships a Postman-style UI to build, send, and inspect messages against a route, and can import Postman collections and AsyncAPI documents to scaffold routes and endpoints. Everything about the app — install, CLI, UI, cookbook, tuning — is in the 📖 book.

When to use mq-bridge

  • Hybrid Messaging: Connect systems speaking different protocols (e.g., MQTT to Kafka) without writing a custom adapter for every pair.
  • Batch-heavy Pipelines: Increase throughput by moving messages in batches while keeping per-message ack/nack decisions.
  • Infrastructure Abstraction: Write business logic against CanonicalMessages and swap the underlying transport later.
  • Resilient Pipelines: Apply retry, DLQ, deduplication, limiter, and cookie/session behavior consistently around endpoints.
  • Database Integration: Combine databases with message brokers, for example by ingesting messages into SQL/MongoDB or forwarding outbox rows to a broker.
  • Sidecar / Gateway: Run the bridge beside another service to ingest, filter, and route messages before they reach the core application.

When NOT to use mq-bridge

  • Stateful Stream Processing: For windowing, joins, or complex aggregations over time, dedicated stream processing engines are more suitable.
  • Domain Aggregate Management: If you need a framework to manage the lifecycle, versioning, and replay of domain aggregates (Event Sourcing), use a specialized library. mq-bridge handles the bus, not the entity.
  • Protocol-Specific Power Features: mq-bridge intentionally exposes a common subset: publish/consume, pub/sub where possible, request-reply where possible, batching, middleware, and ack/nack handling. If your application depends on highly specific broker features, using that broker's native client directly may be better.

Documentation

The 📖 mq-bridge book covers the library and the app. Connectors, middleware and tuning use the same settings in both.

I want to…Read
Embed the library, write handlers, publish from codeQuick start: library · Embed the library · Bindings API · docs.rs
Move data without codeQuick start: mqb copy · Desktop UI · CLI · MCP server
Configure a connectorConnectors, with YAML and URL examples for each
Know what each backend supports (subscriber mode, request-reply, nack, CDC vs. polling)Overview & capabilities · Capability tables (acknowledgement, commit order, replies, TLS)
Add retries, DLQ, dedup, transform, routingCookbook · Middleware & structural endpoints
Understand delivery guarantees and idempotent sinksDelivery guarantees
Build request/reply or CQRS-style flowsRequest / reply · Embed the library: CQRS
Tune throughputPerformance tuning
Write my own endpoint, middleware or pluginEXTENDING.md · PLUGINS.md
Understand the internalsARCHITECTURE.md · CONFIGURATION.md

Status

mq-bridge was created in 2025. Its reliability behavior is exercised by an automated integration and performance suite across supported endpoints, in each of the queue and subscriber modes that endpoint supports:

  • All endpoints showed no data loss during in-flight broker restarts; MQTT publish confirmation was hardened until a chaos test drove in-flight loss to zero.
  • Postgres CDC has a restart-safety test: an un-acked, in-flight batch is redelivered after a database restart with no loss and no gap.

It is used as rust library in armature and as python lib in omniload.

Stability and Rust version

Since 1.0 the Rust API follows semantic versioning: the crate root and the modules models, traits, errors, outcomes, extensions, plugin (the SDK), type_handler, publisher, route and checkpoint. The implementation modules endpoints, middleware, support and test_utils may change in a minor release, and so may the major version of a re-exported driver crate (mq_bridge::sqlx, mq_bridge::rdkafka, ...).

A minor release may add a configuration field, a variant of a configuration or error enum, and raise the minimum Rust version. Code that writes config structs with ..Default::default() and gives a match on these enums a _ arm compiles unchanged across such releases.

The minimum Rust version is 1.89. The features aws, sqlx and postgres-cdc, and so full, need 1.94.1; zeromq-omq needs 1.93.

Running Tests

The project includes integration and performance tests. Most backend tests require Docker.

To run the performance benchmarks for all supported backends:

sh
cargo test --test integration_test --release -- --ignored --nocapture --test-threads=1

To run the criterion benchmarks:

sh
cargo bench --features "full"

Criterion numbers vary with the machine they run on; for comparable throughput figures use the integration performance test above, or see the ETL benchmarks.

Contributing

Contributions are welcome. See CONTRIBUTING.md for setup notes, code style, and pull request guidelines.

AI Disclaimer

This library was written with a lot of AI assistance. The core started as my own code; many endpoints and docs were expanded with Gemini, CodeRabbit, Claude, and Codex, which made adding transports much faster once the endpoint traits were stable.

Generated code can look plausible while missing important details, so I review every commit manually, clean up and refactor the output, and rely on the integration suite to catch regressions. I trust the current code as much as if I had written it myself. Given the large feature set there may still be open issues; the current focus is testing and documentation.

License

mq-bridge is dual-licensed under either the MIT license or the Apache License, Version 2.0, at your option.

Binary distributions carry their applicable third-party terms in THIRD_PARTY_LICENSES.txt. Their current maintenance process is documented in docs/THIRD_PARTY_LICENSES.md.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted for inclusion in mq-bridge by you, as defined in the Apache-2.0 license, shall be dual licensed as above, without any additional terms or conditions.

来源:README.md,提交 136e88c

工具

0
SourceWeft 尚未在沙箱中启动过该软件包,因此其工具未知。

版本历史

13
  1. v1.0.2最新Oct 20, 2026
  2. v1.0.1Oct 16, 2026
  3. v1.0.0Oct 13, 2026
  4. v0.4.20Oct 9, 2026
  5. v0.4.19Oct 4, 2026
  6. v0.4.18Oct 3, 2026
  7. v0.4.17Oct 2, 2026
  8. v0.4.16Sep 29, 2026
  9. v0.4.15Sep 26, 2026
  10. v0.4.14Sep 25, 2026
  11. v0.4.13Sep 21, 2026
  12. v0.4.12Sep 19, 2026
  13. v0.4.11Sep 16, 2026

评分

已满足适用检查项的 50%

该评分衡量此收录条目的完整程度,依据注册中心条目、端点与公开来源计算得出。它不是对服务器本身的评价,也不表示服务器安全或当前可用:可连通性是带日期的检查结果,评分也不包含安装量或用户评价。

  • 近 30 天内可连通25 分

    请保持端点能够响应 MCP 握手;SourceWeft 每 30 天探测一次。

  • 已声明认证方式10 分

    请在注册中心条目中声明凭据请求头,或保持端点能够响应探测,以便判定其认证方式。

  • 已列出工具15 分

    请让服务器在无凭据时响应 tools/list,或发布可在沙箱中无凭据启动的软件包。

  • 描述必需15 分
  • 图标10 分
  • 文档15 分
  • 许可证10 分