airbytehq / airbytehq/PyAirbyte

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

Abierto
#993 0 comentarios 0 reacciones 0 asignados Ver en GitHub
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

Abrir la guía de contribución

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.