fix(mcp): Expose connection-aware destination loading schemas

airbyte/_direct_connectors/connector_docs.py

@@ -24,6 +24,10 @@
 
 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 (
     _BIGQUERY_DESTINATION_DEFINITION_ID,
     _SNOWFLAKE_DESTINATION_DEFINITION_ID,
@@ -182,6 +186,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,23 +261,34 @@ 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.
     """
     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, requests.RequestException):
+        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 +304,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 +341,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 +360,15 @@ 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 <table> LIMIT 10"
+    if context.database_name is not None:
+        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}.<table>` LIMIT 10"
     return f"SELECT * FROM {namespace}.<table> LIMIT 10"
@@ -361,8 +394,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 +421,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 +461,8 @@ def _streams_section(
             "type": "paragraph",
             "text": (
                 "Each enabled stream is written to a table named `<table_prefix><stream_name>` "
-                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."
             ),
         },
     ]

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

tests/unit_tests/test_cloud_describe.py

@@ -516,6 +516,195 @@ 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_"
+
+
+@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, lookup: str, error: Exception
+) -> 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={}
+    )
+    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",
+        lambda **_: DESTINATION_SKILL_DOCS_RESPONSE,
+    )
+
+    def fail(*_: Any, **__: Any) -> list[CloudConnection]:
+        raise error
+
+    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]
+    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.<table> LIMIT 10',
+        ),
+        (
+            SNOWFLAKE_DEFINITION_ID,
+            {"database": "warehouse", "schema": "RAW"},
+            "SELECT * FROM warehouse.RAW.<table> LIMIT 10",
+        ),
+        (
+            SNOWFLAKE_DEFINITION_ID,
+            {"database": "ANALYTICS-PROD", "schema": "RAW"},
+            'SELECT * FROM "ANALYTICS-PROD".RAW.<table> LIMIT 10',
+        ),
+        (
+            SNOWFLAKE_DEFINITION_ID,
+            {"database": "ANALYTICS.PROD", "schema": "RAW"},
+            'SELECT * FROM "ANALYTICS.PROD".RAW.<table> LIMIT 10',
+        ),
+        (
+            SNOWFLAKE_DEFINITION_ID,
+            {"database": 'ANALYTICS"PROD', "schema": "RAW"},
+            'SELECT * FROM "ANALYTICS""PROD".RAW.<table> LIMIT 10',
+        ),
+        (
+            SNOWFLAKE_DEFINITION_ID,
+            {"database": "SELECT", "schema": "RAW"},
+            'SELECT * FROM "SELECT".RAW.<table> LIMIT 10',
+        ),
+        (
+            BIGQUERY_DEFINITION_ID,
+            {"project_id": "my-project", "dataset_id": "raw"},
+            "SELECT * FROM `my-project.raw.<table>` LIMIT 10",
+        ),
+        (BIGQUERY_DEFINITION_ID, {}, "SELECT * FROM <table> 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: