From 57cc1b73faaac3f8782396602b32cf36113f3d49 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 28 Oct 2026 17:48:16 +0000 Subject: [PATCH 1/3] fix(mcp): Expose connection-aware destination loading schemas Co-Authored-By: AJ Steers --- airbyte/_direct_connectors/connector_docs.py | 80 +++++++--- airbyte/mcp/cloud.py | 21 ++- tests/unit_tests/test_cloud_describe.py | 149 +++++++++++++++++++ 3 files changed, 229 insertions(+), 21 deletions(-) diff --git a/airbyte/_direct_connectors/connector_docs.py b/airbyte/_direct_connectors/connector_docs.py index 591a53845..affcdd01c 100644 --- a/airbyte/_direct_connectors/connector_docs.py +++ b/airbyte/_direct_connectors/connector_docs.py @@ -24,6 +24,7 @@ from typing import TYPE_CHECKING, Any, NamedTuple +from airbyte import exceptions as exc from airbyte._direct_connectors.models import ( _BIGQUERY_DESTINATION_DEFINITION_ID, _SNOWFLAKE_DESTINATION_DEFINITION_ID, @@ -182,6 +183,8 @@ def _destination_load_context( table_prefix = connection.table_prefix or None if connection.namespace_definition == "custom_format": schema_name = connection.namespace_format or None + if schema_name is not None and "${" in schema_name: + schema_name = None elif connection.namespace_definition == "source": schema_name = None @@ -255,7 +258,7 @@ def merge_destination_skill_docs( server_docs: DirectAccessGuidance, destination: _DestinationLike, ) -> DirectAccessGuidance: - """Prepend PyAirbyte's local table-layout intro to the server's overview content. + """Prepend table-layout and connection loading hints to the server's overview. Only the default (no-section) or `overview` read is augmented; the server outline and all other section reads pass through unchanged. @@ -263,15 +266,26 @@ def merge_destination_skill_docs( if server_docs.section_id not in {None, "overview"}: return server_docs dialect = _SQL_PASSTHROUGH_DESTINATION_DIALECTS[destination.definition_id] - content = ( - _destination_layout_blocks( - destination=destination, - dialect=dialect, - load_context=_destination_load_context(destination), - ) - + server_docs.content + content = _destination_layout_blocks( + destination=destination, + dialect=dialect, + load_context=_destination_load_context(destination), ) - return server_docs.model_copy(update={"content": content}) + metadata = server_docs.metadata + try: + content += _connections_section(destination, _destination_connections(destination)) + except exc.AirbyteCloudError: + metadata = metadata.model_copy( + update={ + "warnings": [ + *metadata.warnings, + "Connection-derived schema hints could not be loaded. " + "Use SQL discovery to verify schemas and table names.", + ] + } + ) + content += server_docs.content + return server_docs.model_copy(update={"content": content, "metadata": metadata}) def _location_sentence(dialect: str, load_context: _DestinationLoadContext) -> str: @@ -287,12 +301,12 @@ def _location_sentence(dialect: str, load_context: _DestinationLoadContext) -> s if value is not None ) return ( - f"Tables land in {location_text} unless a connection overrides the " - "namespace; unqualified table names resolve there" + f"The default query context is {location_text}; connection namespace " + "overrides change loading schemas, not where unqualified queries resolve" ) return ( - f"Tables land in the destination's configured {_NAMESPACE_NOUNS[dialect]}, " - "which is also where unqualified table names resolve" + f"Unqualified queries use the destination's configured {_NAMESPACE_NOUNS[dialect]}; " + "connections may load into other schemas" ) @@ -324,6 +338,16 @@ def _destination_layout_blocks( f"{_table_naming_sentence(dialect)}" ), }, + { + "type": "paragraph", + "text": ( + "Use database.schema.table names: BigQuery project.dataset.table; " + "Snowflake database.schema.table. Connection-derived schemas and table " + "prefixes below are expected loading locations, not exhaustive warehouse " + "inventory or an authorization boundary. Confirm actual tables through SQL " + "discovery; warehouse permissions and BigQuery query location still apply." + ), + }, { "type": "paragraph", "text": _DIALECT_NOTES[dialect] + _AIRBYTE_METADATA_COLUMNS, @@ -333,9 +357,12 @@ def _destination_layout_blocks( def _qualified_table_example(destination: _DestinationLike) -> str: """Return an example `SELECT` qualifying the table with the destination's namespace.""" - namespace = _destination_load_context(destination).schema_name + context = _destination_load_context(destination) + namespace = context.schema_name if namespace is None: return "SELECT * FROM LIMIT 10" + if context.database_name is not None: + namespace = f"{context.database_name}.{namespace}" if destination.definition_id == _BIGQUERY_DESTINATION_DEFINITION_ID: return f"SELECT * FROM `{namespace}.
` LIMIT 10" return f"SELECT * FROM {namespace}.
LIMIT 10" @@ -361,8 +388,9 @@ def _connections_section( f"{connection.name} (connection_id={connection.connection_id}) " f"syncs from source {source_name} (source_id={connection.source_id})" ) - if connection.table_prefix: - item += f", with table prefix {connection.table_prefix!r}" + item += ". " + _connection_namespace_note( + connection, _destination_load_context(destination, connection) + ) items.append(item) return [ {"type": "heading", "level": 2, "text": "Connections syncing into this destination"}, @@ -387,9 +415,21 @@ def _connection_namespace_note( ) namespace_definition = connection.namespace_definition if namespace_definition == "source": - note = "Streams land in a namespace mirroring the source's own namespace (e.g. its schema)" + note = ( + "Loading schemas mirror each source stream's namespace; a single schema " + "cannot be inferred from the destination default" + ) elif namespace_definition == "custom_format": - note = f"Streams land in namespace format `{connection.namespace_format}`" + if load_context.schema_name is not None: + note = ( + f"Expected loading schema `{load_context.schema_name}` " + "(connection namespace override)" + ) + else: + note = ( + f"Loading schema template `{connection.namespace_format}`; resolve it " + "per stream rather than treating it as a concrete schema" + ) else: note = "Streams land in the destination's default namespace" if load_context.schema_name is not None: @@ -415,8 +455,8 @@ def _streams_section( "type": "paragraph", "text": ( "Each enabled stream is written to a table named `` " - f"in the destination's configured {_NAMESPACE_NOUNS[dialect]}. Names below " - "assume the destination's default identifier casing and namespace." + "in its connection's loading schema, which may override the destination " + "default. Names below assume the destination's default identifier casing." ), }, ] diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py index 759852b0d..06e63de9d 100644 --- a/airbyte/mcp/cloud.py +++ b/airbyte/mcp/cloud.py @@ -1628,7 +1628,11 @@ def describe_connector( with_replication_details: Annotated[ bool, Field( - description="Include the connections that read from or write to this " "connector.", + description=( + "Include connections that read from or write to this connector, " + "including expected destination database, loading schema or namespace " + "format, and table prefix. These are advisory loading hints, not inventory." + ), default=False, ), ], @@ -1996,6 +2000,21 @@ def execute_external_sql_query( Only SQL-passthrough destinations (Snowflake/BigQuery) support this tool. + Use three-part table names: BigQuery `project.dataset.table` maps to + database.schema.table, as does Snowflake `database.schema.table`. `SHOW DATABASES` + identifies the configured project/database, not its datasets/schemas; `SHOW TABLES` + discovers tables across that context's schemas without changing the default query + dataset/schema. BigQuery metadata is scoped to the configured query location and + warehouse metadata permissions still apply. + + For expected loading schemas, call `describe_connector` with + `with_replication_details=True` or `with_direct_access_guidance=True`. Connection + namespace overrides and prefixes describe where each sync writes, not where + unqualified queries resolve. Source-derived namespaces and schema templates may + vary per stream. These hints are advisory, not exhaustive warehouse inventory or + an authorization boundary. Confirm table names with discovery and qualify queries + rather than assuming every connection loads into the destination default schema. + Useful, for example, for debugging pipelines and inspecting what a sync landed: discovering which tables a sync created, sampling rows to see how records are shaped and keyed, checking freshness or row counts, or exploring raw tables diff --git a/tests/unit_tests/test_cloud_describe.py b/tests/unit_tests/test_cloud_describe.py index e703c17aa..d820324cd 100644 --- a/tests/unit_tests/test_cloud_describe.py +++ b/tests/unit_tests/test_cloud_describe.py @@ -516,6 +516,155 @@ def test_get_direct_access_guidance_destination_merges_server_docs( ) +@pytest.mark.parametrize( + ("definition_id", "configuration", "expected_database"), + [ + ( + SNOWFLAKE_DEFINITION_ID, + {"database": "WAREHOUSE", "schema": "RAW_DEFAULT"}, + "WAREHOUSE", + ), + ( + BIGQUERY_DEFINITION_ID, + {"project_id": "my-project", "dataset_id": "RAW_DEFAULT"}, + "my-project", + ), + ], +) +@pytest.mark.parametrize( + ("mode", "namespace_format", "expected_schema", "expected_hint"), + [ + ("destination", None, "RAW_DEFAULT", "schema `RAW_DEFAULT`"), + ( + "custom_format", + "raw_salesforce", + "raw_salesforce", + "Expected loading schema `raw_salesforce`", + ), + ( + "custom_format", + "raw_${SOURCE_NAMESPACE}", + None, + "Loading schema template `raw_${SOURCE_NAMESPACE}`", + ), + ("source", None, None, "Loading schemas mirror each source stream's namespace"), + ], +) +def test_destination_guidance_exposes_connection_loading_schemas( + monkeypatch: pytest.MonkeyPatch, + definition_id: str, + configuration: dict[str, str], + expected_database: str, + mode: str, + namespace_format: str | None, + expected_schema: str | None, + expected_hint: str, +) -> None: + """Enrolled docs and structured details use connection namespaces, not query defaults.""" + workspace = _make_workspace(monkeypatch) + _patch_context_layer(monkeypatch) + destination = _seed_destination( + workspace, "destination-1", definition_id, configuration=configuration + ) + source = _seed_source(workspace, "source-1", "Salesforce") + connection = _fake_connection( + workspace, "connection-1", source.connector_id, destination.connector_id + ) + connection._connection_info.namespace_definition = mode # noqa: SLF001 + connection._connection_info.namespace_format = namespace_format # noqa: SLF001 + other_connection = _fake_connection( + workspace, "other-connection", source.connector_id, "other-destination" + ) + monkeypatch.setattr( + CloudWorkspace, + "list_connections", + lambda *_, **__: [connection, other_connection], + ) + monkeypatch.setattr(CloudWorkspace, "list_sources", lambda *_, **__: [source]) + monkeypatch.setattr( + CloudWorkspace, "list_destinations", lambda *_, **__: [destination] + ) + monkeypatch.setattr( + agents_api_util, + "read_cloud_skill_docs", + lambda **_: DESTINATION_SKILL_DOCS_RESPONSE, + ) + + guidance = destination.get_direct_access_guidance() + text = _content_text(guidance) + assert expected_hint in text + assert "default query context" in text + assert "not exhaustive warehouse inventory or an authorization boundary" in text + assert "table prefix 'raw_'" in text + assert "other-connection" not in text + assert guidance.content[-1]["text"] == "Server overview." + detail = connector_docs.build_connection_details(destination)[0] + assert detail.destination_database == expected_database + assert detail.destination_schema == expected_schema + assert detail.namespace_format == namespace_format + assert detail.table_prefix == "raw_" + + +def test_connection_hint_failure_preserves_server_docs( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A Cloud connection lookup failure is a warning, not missing server guidance.""" + workspace = _make_workspace(monkeypatch) + _patch_context_layer(monkeypatch) + destination = _seed_destination( + workspace, "dest", SNOWFLAKE_DEFINITION_ID, configuration={} + ) + monkeypatch.setattr( + agents_api_util, + "read_cloud_skill_docs", + lambda **_: DESTINATION_SKILL_DOCS_RESPONSE, + ) + + def fail(*_: Any, **__: Any) -> list[CloudConnection]: + raise AirbyteCloudError(context={"status_code": 403}) + + monkeypatch.setattr(CloudWorkspace, "list_connections", fail) + guidance = destination.get_direct_access_guidance() + assert guidance.content[-1]["text"] == "Server overview." + assert "could not be loaded" in guidance.metadata.warnings[-1] + assert "No connections sync" not in _content_text(guidance) + + +@pytest.mark.parametrize( + ("definition_id", "configuration", "expected"), + [ + ( + SNOWFLAKE_DEFINITION_ID, + {"database": "WAREHOUSE", "schema": "RAW"}, + "SELECT * FROM WAREHOUSE.RAW.
LIMIT 10", + ), + ( + BIGQUERY_DEFINITION_ID, + {"project_id": "my-project", "dataset_id": "raw"}, + "SELECT * FROM `my-project.raw.
` LIMIT 10", + ), + (BIGQUERY_DEFINITION_ID, {}, "SELECT * FROM
LIMIT 10"), + ], +) +def test_destination_examples_use_three_part_table_names( + monkeypatch: pytest.MonkeyPatch, + definition_id: str, + configuration: dict[str, str], + expected: str, +) -> None: + """Fallback examples qualify the configured project/database and dataset/schema.""" + workspace = _make_workspace(monkeypatch) + destination = _seed_destination( + workspace, "dest", definition_id, configuration=configuration + ) + monkeypatch.setattr(CloudWorkspace, "list_connections", lambda *_, **__: []) + guidance = connector_docs.build_direct_access_sql_guidance(destination) + assert ( + next(block["code"] for block in guidance.content if block["type"] == "code") + == expected + ) + + def test_get_direct_access_guidance_destination_server_error_propagates( monkeypatch: pytest.MonkeyPatch, ) -> None: From a14b74c5e8cb105cc4957abe159d1c3ed575ea4a Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 28 Oct 2026 17:54:43 +0000 Subject: [PATCH 2/3] fix(mcp): Preserve destination docs on hint transport failures Co-Authored-By: AJ Steers --- airbyte/_direct_connectors/connector_docs.py | 4 +++- tests/unit_tests/test_cloud_describe.py | 21 +++++++++++++++++--- 2 files changed, 21 insertions(+), 4 deletions(-) diff --git a/airbyte/_direct_connectors/connector_docs.py b/airbyte/_direct_connectors/connector_docs.py index affcdd01c..3b62c1bbd 100644 --- a/airbyte/_direct_connectors/connector_docs.py +++ b/airbyte/_direct_connectors/connector_docs.py @@ -24,6 +24,8 @@ from typing import TYPE_CHECKING, Any, NamedTuple +import requests + from airbyte import exceptions as exc from airbyte._direct_connectors.models import ( _BIGQUERY_DESTINATION_DEFINITION_ID, @@ -274,7 +276,7 @@ def merge_destination_skill_docs( metadata = server_docs.metadata try: content += _connections_section(destination, _destination_connections(destination)) - except exc.AirbyteCloudError: + except (exc.AirbyteCloudError, requests.RequestException): metadata = metadata.model_copy( update={ "warnings": [ diff --git a/tests/unit_tests/test_cloud_describe.py b/tests/unit_tests/test_cloud_describe.py index d820324cd..eb5762097 100644 --- a/tests/unit_tests/test_cloud_describe.py +++ b/tests/unit_tests/test_cloud_describe.py @@ -605,8 +605,17 @@ def test_destination_guidance_exposes_connection_loading_schemas( assert detail.table_prefix == "raw_" +@pytest.mark.parametrize("lookup", ["list_connections", "list_sources"]) +@pytest.mark.parametrize( + "error", + [ + AirbyteCloudError(context={"status_code": 403}), + requests.Timeout("timed out"), + requests.ConnectionError("connection failed"), + ], +) def test_connection_hint_failure_preserves_server_docs( - monkeypatch: pytest.MonkeyPatch, + monkeypatch: pytest.MonkeyPatch, lookup: str, error: Exception ) -> None: """A Cloud connection lookup failure is a warning, not missing server guidance.""" workspace = _make_workspace(monkeypatch) @@ -614,6 +623,12 @@ def test_connection_hint_failure_preserves_server_docs( destination = _seed_destination( workspace, "dest", SNOWFLAKE_DEFINITION_ID, configuration={} ) + connection = _fake_connection( + workspace, "connection-1", "source-1", destination.connector_id + ) + monkeypatch.setattr( + CloudWorkspace, "list_connections", lambda *_, **__: [connection] + ) monkeypatch.setattr( agents_api_util, "read_cloud_skill_docs", @@ -621,9 +636,9 @@ def test_connection_hint_failure_preserves_server_docs( ) def fail(*_: Any, **__: Any) -> list[CloudConnection]: - raise AirbyteCloudError(context={"status_code": 403}) + raise error - monkeypatch.setattr(CloudWorkspace, "list_connections", fail) + monkeypatch.setattr(CloudWorkspace, lookup, fail) guidance = destination.get_direct_access_guidance() assert guidance.content[-1]["text"] == "Server overview." assert "could not be loaded" in guidance.metadata.warnings[-1] From c6d665f17051937512a1c8d748bd4f9db96e814e Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 28 Oct 2026 18:02:39 +0000 Subject: [PATCH 3/3] fix(mcp): Quote Snowflake database names in SQL examples Co-Authored-By: AJ Steers --- airbyte/_direct_connectors/connector_docs.py | 6 ++++- tests/unit_tests/test_cloud_describe.py | 27 +++++++++++++++++++- 2 files changed, 31 insertions(+), 2 deletions(-) diff --git a/airbyte/_direct_connectors/connector_docs.py b/airbyte/_direct_connectors/connector_docs.py index 3b62c1bbd..396006f0e 100644 --- a/airbyte/_direct_connectors/connector_docs.py +++ b/airbyte/_direct_connectors/connector_docs.py @@ -25,6 +25,7 @@ from typing import TYPE_CHECKING, Any, NamedTuple import requests +from snowflake.sqlalchemy.snowdialect import SnowflakeDialect from airbyte import exceptions as exc from airbyte._direct_connectors.models import ( @@ -364,7 +365,10 @@ def _qualified_table_example(destination: _DestinationLike) -> str: if namespace is None: return "SELECT * FROM
LIMIT 10" if context.database_name is not None: - namespace = f"{context.database_name}.{namespace}" + database_name = context.database_name + if destination.definition_id == _SNOWFLAKE_DESTINATION_DEFINITION_ID: + database_name = SnowflakeDialect().identifier_preparer.quote(database_name) + namespace = f"{database_name}.{namespace}" if destination.definition_id == _BIGQUERY_DESTINATION_DEFINITION_ID: return f"SELECT * FROM `{namespace}.
` LIMIT 10" return f"SELECT * FROM {namespace}.
LIMIT 10" diff --git a/tests/unit_tests/test_cloud_describe.py b/tests/unit_tests/test_cloud_describe.py index eb5762097..5fe55d67a 100644 --- a/tests/unit_tests/test_cloud_describe.py +++ b/tests/unit_tests/test_cloud_describe.py @@ -651,7 +651,32 @@ def fail(*_: Any, **__: Any) -> list[CloudConnection]: ( SNOWFLAKE_DEFINITION_ID, {"database": "WAREHOUSE", "schema": "RAW"}, - "SELECT * FROM WAREHOUSE.RAW.
LIMIT 10", + 'SELECT * FROM "WAREHOUSE".RAW.
LIMIT 10', + ), + ( + SNOWFLAKE_DEFINITION_ID, + {"database": "warehouse", "schema": "RAW"}, + "SELECT * FROM warehouse.RAW.
LIMIT 10", + ), + ( + SNOWFLAKE_DEFINITION_ID, + {"database": "ANALYTICS-PROD", "schema": "RAW"}, + 'SELECT * FROM "ANALYTICS-PROD".RAW.
LIMIT 10', + ), + ( + SNOWFLAKE_DEFINITION_ID, + {"database": "ANALYTICS.PROD", "schema": "RAW"}, + 'SELECT * FROM "ANALYTICS.PROD".RAW.
LIMIT 10', + ), + ( + SNOWFLAKE_DEFINITION_ID, + {"database": 'ANALYTICS"PROD', "schema": "RAW"}, + 'SELECT * FROM "ANALYTICS""PROD".RAW.
LIMIT 10', + ), + ( + SNOWFLAKE_DEFINITION_ID, + {"database": "SELECT", "schema": "RAW"}, + 'SELECT * FROM "SELECT".RAW.
LIMIT 10', ), ( BIGQUERY_DEFINITION_ID,