airbytehq / airbytehq/PyAirbyte

feat(mcp): Add tools to modify stream sync settings and refresh catalog on existing connections

オープン
#993 コメント 0 件 リアクション 0 件 担当者 0 名 GitHub で見る
主要言語
Python
スター
344
フォーク
77
平均マージ
1日 11時間
マージ済み PR(30日)
35

説明

## Summary

Two new MCP tool capabilities are needed to support automated incremental stream testing and configuration workflows:

1. **Modify stream sync settings on an existing connection** — A tool that can safely change the sync mode (e.g., `full_refresh` → `incremental`) for a specific stream or set of streams on a connection, including setting the cursor field and destination sync mode.

2. **Refresh/re-discover the catalog on an existing connection** — A tool that triggers a discover operation on a connection (equivalent to clicking "Refresh source schema" in the UI), so the catalog reflects the latest connector capabilities after a version pin or upgrade.

## Motivation

When adding incremental support to an existing connector stream, the typical workflow involves:
1. Publishing a pre-release connector version
2. Pinning the pre-release to a test connection
3. **Refreshing the catalog** so the connection picks up the newly-advertised incremental support
4. **Changing the stream sync mode** from `full_refresh` to `incremental`
5. Running syncs to verify

Steps 3 and 4 currently have no safe MCP tool support. The underlying mechanisms exist in PyAirbyte (`replace_connection_catalog` in `api_util.py` and the `withRefreshedCatalog: true` flag on `web_backend/connections/get`), but:

- They are not exposed as MCP tools
- Raw catalog manipulation by an LLM is risky — pulling the full catalog, transforming a specific stream, and pushing it back requires careful handling that should be encapsulated in a purpose-built tool with proper validation
- The ops MCP has `get_connection_catalog` (read-only) but no corresponding write tool

## Proposed Tools

### 1. `set_stream_sync_mode` (or similar)
- **Inputs**: `connection_id`, `stream_name`, `sync_mode` (`incremental` | `full_refresh`), optionally `destination_sync_mode` and `cursor_field`
- **Behavior**: Safely modifies only the specified stream(s) in the connection's syncCatalog, validates the configuration is legal (e.g., the stream actually supports incremental), and applies the change
- **Safety**: Should validate that the requested sync mode is in the stream's `supportedSyncModes` before applying

### 2. `refresh_connection_catalog` (or similar)
- **Inputs**: `connection_id`
- **Behavior**: Triggers a discover operation on the connection's source with the currently-pinned connector version, updating the connection's catalog with the latest stream definitions and supported sync modes
- **Implementation note**: This likely wraps a call to `web_backend/connections/get` with `withRefreshedCatalog: true`, or an equivalent mechanism

## Context

- Requested by @aaronsteers during development of the `add-incremental-stream-support` skill ([ai-skills PR #112](https://github.com/airbytehq/ai-skills/pull/112))
- The skill currently documents UI-based workarounds (Option A) and new-connection workarounds (Option B) because these tools don't exist yet
- Thread context: https://airbytehq-team.slack.com/archives/C07N1EGSM8E/p1773766113926719

---
[Devin session](https://app.devin.ai/sessions/1812bbf16ef24923b5f24b034dca96fc)

コントリビューションガイド

コントリビューションガイドを開く

評価

この issue はまだ評価されていません。

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。