diff --git a/airbyte/_direct_connectors/connector_docs.py b/airbyte/_direct_connectors/connector_docs.py
index 591a53845..396006f0e 100644
--- a/airbyte/_direct_connectors/connector_docs.py
+++ b/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,7 +261,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 +269,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, 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
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}.` LIMIT 10"
return f"SELECT * FROM {namespace}. 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 `` "
- 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..5fe55d67a 100644
--- a/tests/unit_tests/test_cloud_describe.py
+++ b/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. 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,
+ {"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: