内容
# Airbyte MCP 服务器（airbyte-mcp）
用于 [Airbyte 公共 API](https://reference.airbyte.com/reference/getting-started) 的 MCP 服务器。 使用官方的 [MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk) （FastMCP）构建。
允许任何与 MCP 兼容的客户端（Cursor、Claude Desktop、Claude Code、MCP Inspector 等）通过自然语言与您的 Airbyte 实例交互。
## 功能
- 36 种工具，涵盖工作区、源、目标、连接、作业、刷新/清除、作业日志、标签、流和连接器定义
- 核心资源的读取和写入操作（创建、更新、删除）
- 通过内部配置 API（自我管理）进行作业诊断：详细的失败原因、按流统计和结构化日志
- 通过 `airbyte_get_cloud_sync_logs` 进行云全文同步日志（仅限 Airbyte Cloud；与官方复制 MCP 保持一致）
- 具有内存缓存和透明 401 重试的自动令牌交换
- 摘要工具的 Markdown 和 JSON 响应格式；日志工具仅限 JSON
- 所有列表工具均支持分页（limit/offset）
- 通过 **stdio** （本地）运行 — 与 Cursor、Claude Desktop、Claude Code 和 Docker 配合使用
- 与自我管理的 Airbyte（abctl）和 Airbyte Cloud 配合使用
## 可用工具
| 工具 | 描述 |
|---|---|
| **健康检查** | |
| `airbyte_health_check` | Ping Airbyte API |
| **工作区** | |
| `airbyte_list_workspaces` | 列出工作区，支持分页 |
| `airbyte_get_workspace` | 按 ID 获取工作区详情 |
| **源** | |
| `airbyte_list_sources` | 列出源连接器（按工作区过滤） |
| `airbyte_get_source` | 按 ID 获取源详情 |
| `airbyte_create_source` | 创建新的源连接器 |
| `airbyte_update_source` | 更新现有的源 |
| **目标** | |
| `airbyte_list_destinations` | 列出目标连接器（按工作区过滤） |
| `airbyte_get_destination` | 按 ID 获取目标详情 |
| `airbyte_create_destination` | 创建新的目标连接器 |
| `airbyte_update_destination` | 更新现有的目标 |
| **连接** | |
| `airbyte_list_connections` | 列出连接/管道（按工作区过滤） |
| `airbyte_get_connection` | 获取连接详情，包括流配置 |
| `airbyte_create_connection` | 创建新的连接（管道） |
| `airbyte_update_connection` | 更新现有的连接 |
| **作业** | |
| `airbyte_list_jobs` | 列出作业（按连接、类型、状态、日期过滤） |
| `airbyte_get_job` | 获取作业详情（状态、持续时间、同步的字节/行） |
| `airbyte_trigger_sync` | 触发同步或重置作业 |
| `airbyte_cancel_job` | 取消正在运行的作业 |
| `airbyte_trigger_refresh` | 触发按流刷新（内部 API，自我管理） |
| `airbyte_trigger_clear` | 清除流的目标数据（内部 API，自我管理） |
| `airbyte_wait_for_job` | 等待作业达到终端状态（内部 API） |
| **作业日志（内部 API — 仅限自我管理）** | |
| `airbyte_list_jobs_internal` | 列出所有作业类型，包括刷新和清除 |
| `airbyte_get_job_details` | 按尝试统计、失败原因和堆栈跟踪 |
| `airbyte_get_job_logs` | 所有尝试的结构化日志条目 |
| `airbyte_get_attempt_logs` | 特定尝试的结构化日志条目 |
| **作业日志（仅限云 —全文）** | |
| `airbyte_get_cloud_sync_logs` | 具有分页的全文本同步日志（云配置 API） |
| **流** | |
| `airbyte_get_stream_properties` | 获取源/目标对的流属性 |
| **标签** | |
| `airbyte_list_tags` | 列出标签 |
| `airbyte_create_tag` | 创建标签 |
| `airbyte_update_tag` | 更新标签 |
| `airbyte_delete_tag` | 删除标签 |
| **连接器定义** | |
| `airbyte_list_source_definitions` | 列出源连接器定义 |
| `airbyte_get_source_definition` | 获取源连接器定义 |
| `airbyte_list_destination_definitions` | 列出目标连接器定义 |
| `airbyte_get_destination_definition` | 获取目标连接器定义 |
参见 [docs/endpoints.md](docs/endpoints.md) 获取完整的 Airbyte API 端点检查清单。
## 先决条件
- 运行的 **Airbyte** 实例 — 既可以是：
- 通过 [abctl](https://docs.airbyte.com/platform/deploying-airbyte/abctl/) 自我管理（参见 [docs/local-setup.md](docs/local-setup.md)）
- [Airbyte Cloud](https://cloud.airbyte.com)
- 以下之一来运行服务器：
- **[uvx](https://docs.astral.sh/uv/)** （零安装；按需运行已发布的包 — 参见 [使用 uvx 运行](#使用-uvx-零安装)），或
- **[uv](https://docs.astral.sh/uv/)** + **Python 3.13+** （本地克隆/开发），或
- **[Docker](https://docs.docker.com/get-docker/)** （主机上不需要 Python/uv）
## 快速入门
### 1. 克隆和安装
```bash
git clone https://github.com/trustxai/airbyte-mcp.git
cd airbyte-mcp
uv sync
```
### 2. 配置凭证
```bash
cp .env.example .env
```
对于自我管理（abctl），获取凭证：
```bash
abctl local credentials
```
编辑 `.env` 文件，添加您的 `client-id` 和 `client-secret`。有关详细信息，请参见 [docs/authentication.md](docs/authentication.md)。
### 3. 运行服务器
```bash
uv run airbyte-mcp
```
## 使用 uvx（零安装）运行
[uvx](https://docs.astral.sh/uv/) 按需运行已发布的 PyPI 包 — 无需克隆，无需虚拟环境，无需持久安装。命令与包名匹配，因此不需要 `--from`：
```bash
uvx airbyte-mcp
```
> **与官方 Airbyte 复制 MCP 不同**。该 MCP 位于 `airbyte` 包中，并作为 `uvx --from=airbyte@latest airbyte-mcp` 调用。这里，PyPI 包 *是* `airbyte-mcp`，因此 `uvx airbyte-mcp` 无歧义地解析为此服务器。
凭证通过客户端的 `env` 块传递（如下所示），或在您的 shell 中导出以进行手动运行：
```bash
AIRBYTE_API_URL=http://localhost:8000/api/public/v1 \
AIRBYTE_CLIENT_ID=<your-client-id> \
AIRBYTE_CLIENT_SECRET=<your-client-secret> \
uvx airbyte-mcp
```
> **Python 版本：** 包针对 **Python 3.13+**。uv 自动提供匹配的解释器，因此这通常直接有效。如果您的环境固定了较旧的默认版本，请使用 `uvx --python=3.13 airbyte-mcp` 强制执行。
使用 uvx 的客户端配置（例如 Cursor `.cursor/mcp.json`）：
```json
{
"mcpServers": {
"airbyte": {
"command": "uvx",
"args": ["airbyte-mcp"],
"env": {
"AIRBYTE_API_URL": "http://localhost:8000/api/public/v1",
"AIRBYTE_CLIENT_ID": "<your-client-id>",
"AIRBYTE_CLIENT_SECRET": "<your-client-secret>"
}
}
}
}
```
## 客户端配置
每个 MCP 客户端（Cursor、Claude Desktop 等）都可以通过以下三种方式之一运行服务器：
- **uvx** — 零安装；按需运行已发布的包（参见 [使用 uvx 运行](#使用-uvx-零安装)）。
- **uv** — 从本地克隆；最适合开发。
- **Docker** — 主机上不需要 Python/uv；所有内容都在容器中运行。构建一次镜像，每个客户端配置都会重用它。
> **主机网络注意**: 如果 Airbyte 在您的主机上运行（例如，通过 `abctl`），在 Docker 容器中，`localhost` **不** 指向您的主机。在 macOS/Windows 上使用 `http://host.docker.internal:8000/api/public/v1`，或在 Linux 上为 `docker run` 参数添加 `--network=host`。
### 构建 Docker 镜像（一次性）
```bash
docker build -t airbyte-mcp:latest .
```
### Cursor
添加到 `.cursor/mcp.json`（项目级）或 `~/.cursor/mcp.json`（全局）：
**选项 A — uv**
```json
{
"mcpServers": {
"airbyte": {
"command": "uv",
"args": ["--directory", "/path/to/airbyte-mcp", "run", "airbyte-mcp"],
"env": {
"AIRBYTE_API_URL": "http://localhost:8000/api/public/v1",
"AIRBYTE_CLIENT_ID": "<your-client-id>",
"AIRBYTE_CLIENT_SECRET": "<your-client-secret>"
}
}
}
}
```
**选项 B — Docker**
```json
{
"mcpServers": {
"airbyte": {
"command": "docker",
"args": [
"run", "--rm", "-i",
"--name", "airbyte-mcp",
"-e", "AIRBYTE_API_URL",
"-e", "AIRBYTE_CLIENT_ID",
"-e", "AIRBYTE_CLIENT_SECRET",
"airbyte-mcp:latest"
],
"env": {
"AIRBYTE_API_URL": "http://host.docker.internal:8000/api/public/v1",
"AIRBYTE_CLIENT_ID": "<your-client-id>",
"AIRBYTE_CLIENT_SECRET": "<your-client-secret>"
}
}
}
}
```
### Claude Desktop
添加到 `claude_desktop_config.json`：
**选项 A — uv**
```json
{
"mcpServers": {
"airbyte": {
"command": "uv",
"args": ["--directory", "/path/to/airbyte-mcp", "run", "airbyte-mcp"],
"env": {
"AIRBYTE_API_URL": "http://localhost:8000/api/public/v1",
"AIRBYTE_CLIENT_ID": "<your-client-id>",
"AIRBYTE_CLIENT_SECRET": "<your-client-secret>"
}
}
}
}
```
**选项 B — Docker**
```json
{
"mcpServers": {
"airbyte": {
"command": "docker",
"args": [
"run", "--rm", "-i",
"--name", "airbyte-mcp",
"-e", "AIRBYTE_API_URL",
"-e", "AIRBYTE_CLIENT_ID",
"-e", "AIRBYTE_CLIENT_SECRET",
"airbyte-mcp:latest"
],
"env": {
"AIRBYTE_API_URL": "http://host.docker.internal:8000/api/public/v1",
"AIRBYTE_CLIENT_ID": "<your-client-id>",
"AIRBYTE_CLIENT_SECRET": "<your-client-secret>"
}
}
}
}
```
### Claude Code
```bash
claude mcp add \
--env AIRBYTE_API_URL=http://localhost:8000/api/public/v1 \
--env AIRBYTE_CLIENT_ID=<your-client-id> \
--env AIRBYTE_CLIENT_SECRET=<your-client-secret> \
--transport stdio \
airbyte \
-- uvx airbyte-mcp
```
### MCP Inspector
检查器可以直接启动_stdio_服务器：
```bash
npx @modelcontextprotocol/inspector uvx airbyte-mcp
```
在检查器的环境面板中设置 `AIRBYTE_API_URL`、`AIRBYTE_CLIENT_ID` 和 `AIRBYTE_CLIENT_SECRET`（或在 shell 中导出它们）。
## 手动运行（无需客户端）
如果您只想从 CLI 练习服务器：
```bash
# uvx — 无需克隆
uvx airbyte-mcp
# uv
uv run airbyte-mcp
# Docker
docker run --rm -i --env-file .env airbyte-mcp:latest
```
## 环境变量
| 变量 | 必需 | 默认值 | 描述 |
|---|---|---|---|
| `AIRBYTE_API_URL` | 否 | `http://localhost:8000/api/public/v1` | Airbyte API 基本 URL |
| `AIRBYTE_CLIENT_ID` | 是* | — | 应用程序客户端 ID |
| `AIRBYTE_CLIENT_SECRET` | 是* | — | 应用程序客户端密钥 |
| `AIRBYTE_ACCESS_TOKEN` | 否 | — | 预先获取的令牌（跳过交换） |
| `AIRBYTE_INTERNAL_API_URL` | 否 | *(派生)* | 重写配置 API 基本 URL（云：`https://cloud.airbyte.com/api/v1`） |
| `AIRBYTE_REQUEST_TIMEOUT_SECONDS` | 否 | `30` | 公共 API 调用默认 HTTP 超时 |
| `AIRBYTE_INTERNAL_LOG_TIMEOUT_SECONDS` | 否 | `120` | 内部/云配置 API 日志端点超时 |
*如果提供了 `AIRBYTE_ACCESS_TOKEN`，则不需要。
## 文档
- [身份验证](docs/authentication.md) — 令牌交换、凭证设置
- [架构](docs/architecture.md) — 系统设计、包布局、令牌生命周期
- [与官方 MCP 比较](docs/comparison.md) — 开源优先定位与官方 MCP
- [端点检查清单](docs/endpoints.md) — 完整的 Airbyte API 覆盖状态
- [本地设置](docs/local-setup.md) — abctl 安装演练
- [贡献](docs/CONTRIBUTING.md) — 开发工作流、PR 指南
- [安全](docs/SECURITY.md) — 漏洞报告
- [变更日志](CHANGELOG.md) — 发布历史
## 贡献
欢迎贡献！参见 [docs/CONTRIBUTING.md](docs/CONTRIBUTING.md) 开始。
## 许可证
Apache-2.0 — 参见 [LICENSE](LICENSE) 获取详细信息。
