fix(mcp): return direct-access enablement link from skill docs and rename tool to get_external_access_skill_docs

airbyte/_direct_connectors/connector_docs.py

@@ -4,7 +4,7 @@
 Three paths serve destination docs:
 
 - `CloudWorkspace.get_agent_skill_docs` / `CloudConnector.read_agent_skill_docs`
-  (MCP tool `get_agent_skill_docs`) read destination skill docs straight from
+  (MCP tool `get_external_access_skill_docs`) read destination skill docs straight from
   the Agents API, with no local enrichment.
 - `CloudConnector.get_direct_access_guidance` (MCP tool `describe_cloud_connector`
   with `with_direct_access_guidance=True`) prepends a short local intro --

airbyte/cloud/connectors.py

@@ -921,19 +921,16 @@ def _direct_access_guidance_id(self) -> str | None:
     def read_agent_skill_docs(self, *, section: str | None = None) -> DirectAccessGuidance:
         """Read this connector's skill docs from the Agents API without enrichment.
 
-        Sources are addressed by the `docs_skill_id` reported by Context Layer
-        `inspect`; SQL passthrough destinations by their conventional
-        `connector-destination:` skill ID. `section` is passed through unchanged
-        and Agents API errors (including not-enabled 403/404s) propagate. Other
-        destinations do not support direct access docs.
+        Sources are addressed by their `connector-source:` skill ID; SQL passthrough
+        destinations by their conventional `connector-destination:` skill ID.
+        `section` is passed through unchanged. Not-enabled 403/404s raise
+        `AirbyteExternalAccessNotEnabledError` with an `enablement_url`; when the
+        Context Layer API is unavailable, that error has no URL. Other errors
+        propagate. Other destinations do not support direct access docs.
         """
         if self.connector_type == ConnectorType.SOURCE:
-            skill_id: str | None = self._direct_access_guidance_id()
-            if skill_id is None:
-                raise exc.AirbyteExternalAccessNotEnabledError(
-                    connector_name=self.name,
-                    connector_id=self.connector_id,
-                )
+            self._require_context_layer_api()
+            skill_id = connector_docs.source_skill_id(self.connector_id)
         elif self.definition_id in _SQL_PASSTHROUGH_DESTINATION_DIALECTS:
             skill_id = connector_docs.destination_skill_id(self.connector_id)
         else:
@@ -948,14 +945,22 @@ def read_agent_skill_docs(self, *, section: str | None = None) -> DirectAccessGu
                     "definition_id": self.definition_id,
                 },
             )
-        return DirectAccessGuidance.model_validate(
-            agents_api_util.read_cloud_skill_docs(
-                workspace_id=self.workspace.workspace_id,
-                skill_id=skill_id,
-                credentials=self.workspace._credentials,  # noqa: SLF001
-                section=section,
+        try:
+            return DirectAccessGuidance.model_validate(
+                agents_api_util.read_cloud_skill_docs(
+                    workspace_id=self.workspace.workspace_id,
+                    skill_id=skill_id,
+                    credentials=self.workspace._credentials,  # noqa: SLF001
+                    section=section,
+                )
             )
-        )
+        except exc.AirbyteCloudError as error:
+            self._raise_if_feature_not_enabled(
+                error,
+                ConnectorFeature.DIRECT_ACCESS,
+                enablement_url=self._direct_access_enablement_url,
+            )
+            raise
 
     def get_direct_access_guidance(self, *, section: str | None = None) -> DirectAccessGuidance:
         """Read this connector's direct-access docs, optionally scoped to a section.

airbyte/cloud/workspaces.py

@@ -635,6 +635,10 @@ def get_agent_skill_docs(
             raise exc.AirbyteLibInputError(
                 message="Provide exactly one of `docs_skill_id` or `connector_id`.",
             )
+        if docs_skill_id.startswith(connector_docs.SOURCE_SKILL_PREFIX):
+            return self.get_source(
+                connector_docs.connector_id_from_skill_id(docs_skill_id)
+            ).read_agent_skill_docs(section=section)
         if docs_skill_id.startswith(connector_docs.DESTINATION_SKILL_PREFIX):
             destination = self.get_destination(
                 connector_docs.connector_id_from_skill_id(docs_skill_id)

airbyte/mcp/_docs_results.py

@@ -1,7 +1,7 @@
 # Copyright (c) 2025 Airbyte, Inc., all rights reserved.
 """Docs-result models and renderers for the Cloud MCP tools.
 
-This module provides the `CloudConnectorDocsResult` and `AgentSkillDocsResult`
+This module provides the `CloudConnectorDocsResult` and `ExternalAccessSkillDocsResult`
 result models plus the helpers that render `DirectAccessGuidance` into them.
 """
 
@@ -48,8 +48,8 @@ class CloudConnectorDocsResult(BaseModel):
     """Non-fatal issues reported while reading or rendering the docs."""
 
 
-class AgentSkillDocsResult(BaseModel):
-    """Docs for a single agent skill (skills for agents), returned by `get_agent_skill_docs`."""
+class ExternalAccessSkillDocsResult(BaseModel):
+    """Docs for a single external-access skill, returned by `get_external_access_skill_docs`."""
 
     model_config = ConfigDict(extra="allow")
 
@@ -71,6 +71,9 @@ class AgentSkillDocsResult(BaseModel):
     warnings: list[str] = Field(default_factory=list)
     """Non-fatal issues reported while reading or rendering the docs."""
 
+    errors: list[str] = Field(default_factory=list)
+    """Errors to relay to the user, for example when direct access is disabled for the connector."""
+
 
 def render_connector_docs_result(docs: DirectAccessGuidance) -> CloudConnectorDocsResult:
     """Render `DirectAccessGuidance` into a `CloudConnectorDocsResult`."""
@@ -84,9 +87,11 @@ def render_connector_docs_result(docs: DirectAccessGuidance) -> CloudConnectorDo
     )
 
 
-def render_agent_skill_docs_result(docs: DirectAccessGuidance) -> AgentSkillDocsResult:
-    """Render `DirectAccessGuidance` into an `AgentSkillDocsResult`."""
-    return AgentSkillDocsResult(
+def render_external_access_skill_docs_result(
+    docs: DirectAccessGuidance,
+) -> ExternalAccessSkillDocsResult:
+    """Render `DirectAccessGuidance` into an `ExternalAccessSkillDocsResult`."""
+    return ExternalAccessSkillDocsResult(
         skill_id=docs.metadata.id,
         title=docs.metadata.title,
         content=render_docs_content_markdown(docs.content),

airbyte/mcp/cloud.py

@@ -149,11 +149,11 @@ def _manifest_and_testing_values_hint() -> str | None:
 SKILL_DOCS_SECTION_HINT = """\
 Skill doc sections for direct API queries and actions are named with the pattern
 `actions.<entity_type>.<action>`, matching the `entity_type` and `action` arguments of the
-`execute_external_api_query` tool. Call `get_agent_skill_docs` with the deployed `connector_id`
-(or a `docs_skill_id`) and no `section` to get the overview and list of available sections, then
-retrieve the relevant section(s) before calling an action. E.g. for GitHub `entity_type='issues'`,
-`action='list'`, the `'actions.issues.list'` section will teach that its expected
-`api_args` keys are `owner`, `repo`, `states`, etc.\
+`execute_external_api_query` tool. Call `get_external_access_skill_docs` with the deployed
+`connector_id` (or a `docs_skill_id`) and no `section` to get the overview and list of
+available sections, then retrieve the relevant section(s) before calling an action. E.g. for
+GitHub `entity_type='issues'`, `action='list'`, the `'actions.issues.list'` section will teach
+that its expected `api_args` keys are `owner`, `repo`, `states`, etc.\
 """
 CONNECTOR_CHECK_FAILURE_FALLBACK = "Connector check failed without a failure message."
 DEFER_CREDENTIALS_TIP_TEXT = (
@@ -1657,10 +1657,19 @@ def _direct_access_disabled_result(
     label = error.connector_name or f"{connector.connector_type} {connector.connector_id}"
     return ExternalApiExecuteResult(
         status="error",
-        errors=[
-            f"Show the user this message as-is: Direct access is currently disabled for "
-            f"{label}. [Click here]({error.enablement_url}) to enable it."
-        ],
+        errors=[_direct_access_disabled_message(error, label=label)],
+    )
+
+
+def _direct_access_disabled_message(
+    error: AirbyteExternalAccessNotEnabledError,
+    *,
+    label: str,
+) -> str:
+    """Return the user-facing message for disabled direct access."""
+    return (
+        f"Show the user this message as-is: Direct access is currently disabled for "
+        f"{redact_agent_text(label)}. [Click here]({error.enablement_url}) to enable it."
     )
 
 
@@ -1684,7 +1693,7 @@ def execute_external_api_query(  # noqa: PLR0913  # Explicit args mirror the con
         Field(
             description=(
                 "The type of entity to query, for example 'issues'. Call "
-                "`get_agent_skill_docs` for supported entity types."
+                "`get_external_access_skill_docs` for supported entity types."
             ),
         ),
         TraceArg.VALUE,
@@ -1809,7 +1818,7 @@ def _execute_external_api_action(  # noqa: PLR0913  # Explicit args mirror the c
         Field(
             description=(
                 "The type of entity to act on, for example 'issues'. Call "
-                "`get_agent_skill_docs` for supported entity types."
+                "`get_external_access_skill_docs` for supported entity types."
             ),
         ),
     ],
@@ -1874,7 +1883,7 @@ def _execute_external_api_action(  # noqa: PLR0913  # Explicit args mirror the c
 
     Creates, updates, or deletes data in the external system.
 
-    Before calling, read the connector's action docs with `get_agent_skill_docs`
+    Before calling, read the connector's action docs with `get_external_access_skill_docs`
     (or `describe_cloud_connector` with `with_direct_access_guidance=True`) to
     learn the entity types, actions, and required `api_args`; argument names
     differ per connector and are not guessable.

airbyte/mcp/guidance.py

@@ -28,11 +28,16 @@
 from airbyte import exceptions as exc
 from airbyte._util.registry_spec import get_connector_spec_from_registry
 from airbyte.constants import is_hosted_mcp_mode
-from airbyte.mcp._docs_results import AgentSkillDocsResult, render_agent_skill_docs_result
+from airbyte.exceptions import AirbyteExternalAccessNotEnabledError
+from airbyte.mcp._docs_results import (
+    ExternalAccessSkillDocsResult,
+    render_external_access_skill_docs_result,
+)
 from airbyte.mcp._tool_utils import mcp_tool
 from airbyte.mcp.cloud import (
     SKILL_DOCS_SECTION_HINT,
     WORKSPACE_ID_TIP_TEXT,
+    _direct_access_disabled_message,
     _get_cloud_workspace,
 )
 from airbyte.registry import (
@@ -176,7 +181,7 @@ def test_my_tools_prompt(
     extra_help_text=SKILL_DOCS_SECTION_HINT,
     external_access=True,
 )
-def get_agent_skill_docs(
+def get_external_access_skill_docs(
     ctx: Context,
     *,
     docs_skill_id: Annotated[
@@ -219,7 +224,7 @@ def get_agent_skill_docs(
         ),
         TraceArg.OMIT,
     ],
-) -> AgentSkillDocsResult:
+) -> ExternalAccessSkillDocsResult:
     """Read a deployed connector's direct-access docs (agent skill docs).
 
     These docs describe the entity types, actions, and arguments that
@@ -231,11 +236,25 @@ def get_agent_skill_docs(
     Exactly one of `connector_id` or `docs_skill_id` is required. `connector_id` must be
     the ID of a source or destination deployed in the workspace (for example from
     `list_cloud_connectors`), not a connector definition ID from the registry.
+
+    When direct access is disabled for the connector, `errors` holds a message with an
+    enablement link to show the user.
     """
     workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id)
-    return render_agent_skill_docs_result(
-        workspace.get_agent_skill_docs(docs_skill_id, connector_id=connector_id, section=section)
-    )
+    try:
+        docs = workspace.get_agent_skill_docs(
+            docs_skill_id, connector_id=connector_id, section=section
+        )
+    except AirbyteExternalAccessNotEnabledError as error:
+        if error.enablement_url is None:
+            raise
+        label = error.connector_name or f"connector {error.connector_id}"
+        return ExternalAccessSkillDocsResult(
+            skill_id=docs_skill_id,
+            content="",
+            errors=[_direct_access_disabled_message(error, label=label)],
+        )
+    return render_external_access_skill_docs_result(docs)
 
 
 class ConnectorInfo(BaseModel):

tests/unit_tests/fixtures/arg_trace_plan.json

@@ -110,12 +110,6 @@
     "sql_dialect": "value str",
     "workspace_id": "omit"
   },
-  "get_agent_skill_docs": {
-    "connector_id": "hash",
-    "docs_skill_id": "hash",
-    "section": "fingerprint",
-    "workspace_id": "omit"
-  },
   "get_api_docs_urls": {
     "connector_name": "fingerprint"
   },
@@ -181,6 +175,12 @@
     "workspace_id": "omit"
   },
   "get_default_cloud_context": {},
+  "get_external_access_skill_docs": {
+    "connector_id": "hash",
+    "docs_skill_id": "hash",
+    "section": "fingerprint",
+    "workspace_id": "omit"
+  },
   "get_source_stream_json_schema": {
     "config": "presence",
     "config_file": "fingerprint",

tests/unit_tests/test_cloud_describe.py

@@ -15,7 +15,7 @@
 from airbyte._direct_connectors import connector_docs
 from airbyte.cloud import workspaces as cloud_workspaces
 from airbyte.mcp._docs_results import (
-    render_agent_skill_docs_result,
+    render_external_access_skill_docs_result,
     render_connector_docs_result,
 )
 from airbyte.cloud.connections import CloudConnection
@@ -298,9 +298,10 @@ def test_direct_access_guidance_id_source_probe_not_enabled(
     monkeypatch: pytest.MonkeyPatch,
     probe_error: AirbyteCloudError,
 ) -> None:
-    """A 403/404 docs probe leaves no guidance ID; reads raise not-enabled."""
+    """A 403/404 probe leaves no guidance ID and reads return an enablement URL."""
     workspace = _make_workspace(monkeypatch)
     _patch_context_layer(monkeypatch)
+    monkeypatch.setattr(CloudConnector, "is_feature_enabled", lambda *_: False)
     monkeypatch.setattr(
         agents_api_util,
         "read_cloud_skill_docs",
@@ -309,8 +310,12 @@ def test_direct_access_guidance_id_source_probe_not_enabled(
     source = _seed_source(workspace, "source-1", "GitHub")
 
     assert source._direct_access_guidance_id() is None  # noqa: SLF001
-    with pytest.raises(AirbyteExternalAccessNotEnabledError):
+    with pytest.raises(AirbyteExternalAccessNotEnabledError) as exc_info:
         source.get_direct_access_guidance()
+    assert (
+        exc_info.value.enablement_url
+        == f"{source.connector_url}/enable-direct-access?skip-consent-screen=false"
+    )
 
 
 @pytest.mark.parametrize(
@@ -417,7 +422,7 @@ def test_get_direct_access_guidance_destination_not_enabled_adds_notice(
     assert guidance.content[0] == {"type": "paragraph", "text": notice}
     assert guidance.metadata.warnings == [notice]
     assert render_connector_docs_result(guidance).warnings == [notice]
-    assert render_agent_skill_docs_result(guidance).warnings == [notice]
+    assert render_external_access_skill_docs_result(guidance).warnings == [notice]
     assert guidance.outline == []
     assert any(block.get("type") == "table" for block in guidance.content)
     assert "execute_external_sql_query" not in _content_text(guidance)
@@ -558,7 +563,7 @@ def test_get_direct_access_guidance_destination_no_context_layer_notices(
     assert guidance.content[0] == {"type": "paragraph", "text": notice}
     assert guidance.metadata.warnings == [notice]
     assert render_connector_docs_result(guidance).warnings == [notice]
-    assert render_agent_skill_docs_result(guidance).warnings == [notice]
+    assert render_external_access_skill_docs_result(guidance).warnings == [notice]
     assert guidance.outline == []
     assert "execute_external_sql_query" not in _content_text(guidance)
     assert "SHOW TABLES" not in _content_text(guidance)