airbytehq / airbytehq/PyAirbyte
feat(mcp): Add tools to modify stream sync settings and refresh catalog on existing connections
- Lenguaje dominante
- Python
- Estrellas
- 344
- Forks
- 77
- Merge medio
- 1 d 11 h
- PR fusionados (30 d)
- 35
Descripción
## 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)
Guía de contribución
Evaluación
Este issue todavía no se ha evaluado.