Repository navigation
feat(mcp): add async source catalog discovery and comparison (discover_cloud_source_catalog) - #1392
Conversation
…loud connections Co-Authored-By: AJ Steers <[email protected]>
|
I'll fix CI failures and address comments from users with write access. I'll skip comments containing "(aside)".
|
👋 Greetings, Airbyte Team Member!Here are some helpful tips and reminders for your convenience. 💡 Show Tips and TricksTesting This PyAirbyte VersionYou can test this version of PyAirbyte using the following: # Run PyAirbyte CLI from this branch:
uvx --from 'git+https://github.com/airbytehq/PyAirbyte.git@devin/1792117913-rc-stream-sync-mode-refresh-catalog' pyairbyte --help
# Install PyAirbyte from this branch for development:
pip install 'git+https://github.com/airbytehq/PyAirbyte.git@devin/1792117913-rc-stream-sync-mode-refresh-catalog'PR Slash CommandsAirbyte Maintainers can execute the following slash commands on your PR:
📚 Show Repo GuidanceHelpful ResourcesCommunity SupportQuestions? Join the #pyairbyte channel in our Slack workspace. |
Code coverage overviewLanguages: Python Python / code-coverage/pytest-no-creds
View coverage of the 10 most impacted files.
Python / code-coverage/pytest
View coverage of the 10 most impacted files.
Python / code-coverage/pytest-unit
Line coverage data for the release-candidate/v1 branch is not yet available. View coverage of the 10 most covered files.
ℹ️ Calculations represent line coverage. Results may change as calculations are updated. Updated |
There was a problem hiding this comment.
🟡 Changes recommended
Default cursor persistence, null-namespace selection, and the trace-inventory fixture need correction.
3 open findings
What changed in this PR
Adds Cloud catalog refresh and per-stream sync configuration to the Python API and MCP, porting #994 to the RC trunk.
Changes:
- Adds catalog rediscovery and targeted stream configuration methods.
- Exposes guarded, destructive MCP tools.
- Adds mocked tests for validation and catalog preservation.
| File | Description |
|---|---|
| tests/unit_tests/test_cloud_connection_catalog_edits.py | Tests catalog refresh, validation, and stream preservation. |
| airbyte/mcp/cloud.py | Adds two guarded catalog-editing tools. |
| airbyte/cloud/connections.py | Implements refresh and per-stream sync configuration. |
| airbyte/_util/api_util.py | Adds the refreshed-catalog API request. |
🧠 Review effort: Balanced
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Co-Authored-By: AJ Steers <[email protected]>
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
There was a problem hiding this comment.
Devin Review found 3 potential issues.
2 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
Aaron ("AJ") Steers (aaronsteers)
left a comment
Aaron ("AJ") Steers (aaronsteers)
left a comment
There was a problem hiding this comment.
Devin, let's split this into two PRs. One for catalog refresh and one for set stream sync mode. This one can stay the refresh catalog feature, and the new one can tackle set stream sync mode. If helpful stack the new one on top of this one.
Co-Authored-By: AJ Steers <[email protected]>
…_catalog_refresh Co-Authored-By: AJ Steers <[email protected]>
…/1792117913-rc-stream-sync-mode-refresh-catalog
devin-ai-integration
Bot
changed the title
force_catalog_refresh)
|
Split done. This PR now covers only catalog refresh, exposed as |
Co-Authored-By: AJ Steers <[email protected]>
Co-Authored-By: AJ Steers <[email protected]>
devin-ai-integration
Bot
changed the title
force_catalog_refresh)discover_cloud_source_catalog)
Co-Authored-By: AJ Steers <[email protected]>
|
/evals full=true
|
|
Aaron ("AJ") Steers (@aaronsteers) sorry for the comment, just testing |
Overview
👉 TL;DR: Agents can start a source schema discovery in Cloud without waiting for it, check on it later, and compare the resulting catalog with another catalog or with a connection's current one. Nothing is applied to the connection yet.
Specifically, this adds the MCP tool
discover_cloud_source_catalog, which is backed byCloudSource.start_catalog_discover,CloudSource.get_catalog_discover_result,CloudSource.get_latest_saved_catalog, andCloudWorkspace.compare_source_catalogs.Depends on #1397:
update_cloud_connection_unsafe#1397Replaces (with #1397):
Changes
discover_cloud_source_catalog(source_id, discover_request_id, no_refresh, source_catalog_id, compare_to_source_catalog_id, compare_to_connection_id, workspace_id), markeddestructive=False, pipeline_change=False. It supports five modes:source_id: starts an async discover via/v1/commands/run/discoverand returns a generateddiscover_request_id.source_id+no_refresh=True: returns the latest saved catalog's timestamp and stream count, read from/v1/sources/most_recent_source_actor_catalog.discover_request_id: returns the status from/v1/commands/status. Once the discover completes, it also returns the outcome,source_catalog_id, stream count, and failure reason from/v1/commands/output/discover.source_catalog_id+compare_to_source_catalog_id: returns a bounded diff summary andschema_changefrom/v1/catalogs/diff(persist_changes=false).source_catalog_id+compare_to_connection_id: the same diff, against the connection's currentsourceCatalogId.source_idanddiscover_request_idare mutually exclusive.no_refreshrequiressource_id.source_catalog_idrequires exactly onecompare_to_*argument.api_utilwrappers:run_source_discover_command,get_command_status,get_discover_command_output,get_most_recent_source_catalog, andget_catalog_diff.CloudSourceCatalogDiscoverResultandCloudSourceCatalogDiffStreamSummary._otel.py: the new endpoints are added to the safe-URL list.CloudConnection.import_raw_catalog()now clears the cached_connection_info, sostream_namesand the other cached properties no longer go stale after a catalog import. This bug predates the PR.Review Spotlight
Reviewers with limited time, please review first:
mcp/cloud.py#L4136:discover_cloud_source_catalog, its mode routing and input validationworkspaces.py#L528:compare_source_catalogsconnectors.py#L1185: starting and polling a discoverRisks
/v1/catalogs/diffbehavior come from reading the platform code.source_catalog_id, becausemost_recent_source_actor_catalogdoesn't return one.Follow-ups
apply_source_catalog_idtoupdate_cloud_connection/update_cloud_connection_unsafe, in this PR, after feat(mcp): add batch stream config overrides andupdate_cloud_connection_unsafe#1397 merges.Test plan
AIRBYTE_MCP_TRUSTED_EXECUTION=1 uv run pytest tests/unit_tests/test_cloud_source_catalog_discovery.py tests/unit_tests/test_cloud_api_util.py tests/unit_tests/test_cloud_connection_catalog_edits.py tests/unit_tests/test_mcp_otel.py tests/unit_tests/test_mcp_arg_trace_parity.pyruff check,ruff format --check,pyrefly checkLink to Devin session: https://app.devin.ai/sessions/4d1ece6885c44c52a07034e025db11e6
Open in Devin Desktop: https://app.devin.ai/desktop/session/4d1ece6885c44c52a07034e025db11e6?variant=devin
Requested by: Aaron ("AJ") Steers (@aaronsteers)