From 8c8fd5d2a7b95c69260e8d39fbfc44f79b81b214 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Tue, 11 Nov 2025 19:36:27 +0000
Subject: [PATCH 1/9] feat(mcp): Add get_api_docs_urls tool to registry domain

- Implement new MCP tool to retrieve API documentation URLs for connectors
- Support both canonical connector IDs and API names as input
- Extract docs URLs from multiple sources:
  - Registry metadata (Airbyte documentation)
  - Connector manifest.yaml files (description, metadata.assist.docsUrl, metadata.apiDocs)
- Add comprehensive unit tests with 16 test cases
- Propose metadata.apiDocs format for storing multiple API docs URLs in manifest.yaml

Co-Authored-By: AJ Steers <aj@airbyte.io>
---
 airbyte/mcp/connector_registry.py             | 214 ++++++++++++++
 .../unit_tests/test_mcp_connector_registry.py | 271 ++++++++++++++++++
 2 files changed, 485 insertions(+)
 create mode 100644 tests/unit_tests/test_mcp_connector_registry.py

diff --git a/airbyte/mcp/connector_registry.py b/airbyte/mcp/connector_registry.py
index 484d5f04f..755fd0440 100644
--- a/airbyte/mcp/connector_registry.py
+++ b/airbyte/mcp/connector_registry.py
@@ -4,8 +4,11 @@
 # Note: Deferred type evaluation must be avoided due to FastMCP/Pydantic needing
 # types to be available at import time for tool registration.
 import contextlib
+import re
 from typing import Annotated, Any, Literal
 
+import requests
+import yaml
 from fastmcp import FastMCP
 from pydantic import BaseModel, Field
 
@@ -161,6 +164,217 @@ def get_connector_info(
     )
 
 
+class ApiDocsUrl(BaseModel):
+    """@private Class to hold API documentation URL information."""
+
+    title: str
+    url: str
+    source: str
+
+
+class ApiDocsUrlsResult(BaseModel):
+    """@private Class to hold API docs URLs result."""
+
+    connector_name: str
+    api_name: str | None = None
+    docs_urls: list[ApiDocsUrl]
+
+
+def _resolve_connector_name(connector_identifier: str) -> str | None:
+    """Resolve a connector identifier to a canonical connector name.
+
+    Args:
+        connector_identifier: Either a canonical connector name (e.g., "source-facebook-marketing")
+                            or an API name (e.g., "Facebook Marketing API" or "Facebook Marketing")
+
+    Returns:
+        Canonical connector name if found, None otherwise.
+    """
+    available_connectors = get_available_connectors()
+
+    if connector_identifier in available_connectors:
+        return connector_identifier
+
+    connector_identifier_lower = connector_identifier.lower()
+
+    search_term = re.sub(r"\s+(api|rest api)$", "", connector_identifier_lower, flags=re.IGNORECASE)
+
+    for connector_name in available_connectors:
+        metadata = None
+        with contextlib.suppress(Exception):
+            metadata = get_connector_metadata(connector_name)
+
+        if metadata:
+            pass
+
+        connector_name_clean = (
+            connector_name.replace("source-", "").replace("destination-", "").replace("-", " ")
+        )
+        if search_term in connector_name_clean or connector_name_clean in search_term:
+            return connector_name
+
+    return None
+
+
+def _extract_urls_from_manifest_description(description: str) -> list[ApiDocsUrl]:
+    """Extract URLs from manifest description field."""
+    urls = []
+
+    url_pattern = r"(API Reference|Documentation|Docs|API|Reference):\s*(https?://[^\s\n]+)"
+    matches = re.finditer(url_pattern, description, re.IGNORECASE)
+
+    for match in matches:
+        title = match.group(1)
+        url = match.group(2)
+        urls.append(
+            ApiDocsUrl(
+                title=f"{title} (from manifest description)", url=url, source="manifest_description"
+            )
+        )
+
+    standalone_url_pattern = r"https?://[^\s\n]+"
+    standalone_matches = re.finditer(standalone_url_pattern, description)
+
+    existing_urls = {u.url for u in urls}
+    for match in standalone_matches:
+        url = match.group(0)
+        if url not in existing_urls:
+            urls.append(
+                ApiDocsUrl(
+                    title="API Documentation (from manifest)",
+                    url=url,
+                    source="manifest_description",
+                )
+            )
+            existing_urls.add(url)
+
+    return urls
+
+
+def _extract_docs_from_manifest(manifest_data: dict) -> list[ApiDocsUrl]:
+    """Extract documentation URLs from parsed manifest data."""
+    docs_urls = []
+
+    if manifest_data.get("description"):
+        docs_urls.extend(_extract_urls_from_manifest_description(manifest_data["description"]))
+
+    metadata = manifest_data.get("metadata")
+    if not isinstance(metadata, dict):
+        return docs_urls
+
+    assist = metadata.get("assist")
+    if isinstance(assist, dict) and "docsUrl" in assist:
+        docs_urls.append(
+            ApiDocsUrl(
+                title="API Documentation (assist)",
+                url=assist["docsUrl"],
+                source="manifest_assist",
+            )
+        )
+
+    api_docs = metadata.get("apiDocs")
+    if isinstance(api_docs, list):
+        docs_urls.extend(
+            [
+                ApiDocsUrl(title=doc["title"], url=doc["url"], source="manifest_api_docs")
+                for doc in api_docs
+                if isinstance(doc, dict) and "title" in doc and "url" in doc
+            ]
+        )
+
+    return docs_urls
+
+
+def _fetch_manifest_docs_urls(connector_name: str) -> list[ApiDocsUrl]:
+    """Fetch documentation URLs from connector manifest.yaml file."""
+    manifest_url = DEFAULT_MANIFEST_URL.format(
+        source_name=connector_name,
+        version="latest",
+    )
+
+    http_not_found = 404
+
+    try:
+        response = requests.get(manifest_url, timeout=10)
+        if response.status_code == http_not_found:
+            return []
+
+        response.raise_for_status()
+        manifest_data = yaml.safe_load(response.text)
+
+        return _extract_docs_from_manifest(manifest_data)
+
+    except Exception:
+        return []
+
+
+@mcp_tool(
+    domain="registry",
+    read_only=True,
+    idempotent=True,
+)
+def get_api_docs_urls(
+    connector_identifier: Annotated[
+        str,
+        Field(
+            description=(
+                "The connector identifier. Can be either:\n"
+                "- A canonical connector name (e.g., 'source-facebook-marketing')\n"
+                "- An API name (e.g., 'Facebook Marketing API' or 'Facebook Marketing')"
+            )
+        ),
+    ],
+) -> ApiDocsUrlsResult | Literal["Connector not found."]:
+    """Get API documentation URLs for a connector.
+
+    This tool retrieves documentation URLs for a connector's upstream API from multiple sources:
+    - Registry metadata (documentationUrl, erdUrl)
+    - Connector manifest.yaml file (description, metadata.assist.docsUrl, metadata.apiDocs)
+
+    The tool accepts either a canonical connector ID (e.g., "source-facebook-marketing") or
+    an API name (e.g., "Facebook Marketing API" or "Facebook Marketing").
+
+    Returns:
+        ApiDocsUrlsResult with connector name and list of documentation URLs, or error message.
+    """
+    connector_name = _resolve_connector_name(connector_identifier)
+
+    if not connector_name:
+        return "Connector not found."
+
+    docs_urls: list[ApiDocsUrl] = []
+    api_name: str | None = None
+
+    connector = None
+    with contextlib.suppress(Exception):
+        connector = get_source(
+            connector_name,
+            docker_image=is_docker_installed() or False,
+            install_if_missing=False,
+        )
+
+    if connector and connector.docs_url:
+        docs_urls.append(
+            ApiDocsUrl(title="Airbyte Documentation", url=connector.docs_url, source="registry")
+        )
+
+    manifest_urls = _fetch_manifest_docs_urls(connector_name)
+    docs_urls.extend(manifest_urls)
+
+    seen_urls = set()
+    unique_docs_urls = []
+    for doc_url in docs_urls:
+        if doc_url.url not in seen_urls:
+            seen_urls.add(doc_url.url)
+            unique_docs_urls.append(doc_url)
+
+    return ApiDocsUrlsResult(
+        connector_name=connector_name,
+        api_name=api_name,
+        docs_urls=unique_docs_urls,
+    )
+
+
 def register_connector_registry_tools(app: FastMCP) -> None:
     """@private Register tools with the FastMCP app.
 
diff --git a/tests/unit_tests/test_mcp_connector_registry.py b/tests/unit_tests/test_mcp_connector_registry.py
new file mode 100644
index 000000000..bce98e123
--- /dev/null
+++ b/tests/unit_tests/test_mcp_connector_registry.py
@@ -0,0 +1,271 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+"""Unit tests for MCP connector registry tools."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+
+from airbyte.mcp.connector_registry import (
+    ApiDocsUrl,
+    ApiDocsUrlsResult,
+    _extract_urls_from_manifest_description,
+    _fetch_manifest_docs_urls,
+    _resolve_connector_name,
+    get_api_docs_urls,
+)
+
+
+class TestResolveConnectorName:
+    """Tests for _resolve_connector_name function."""
+
+    def test_canonical_name_exact_match(self) -> None:
+        """Test that canonical connector names are returned as-is."""
+        with patch(
+            "airbyte.mcp.connector_registry.get_available_connectors"
+        ) as mock_get:
+            mock_get.return_value = ["source-faker", "source-facebook-marketing"]
+            result = _resolve_connector_name("source-faker")
+            assert result == "source-faker"
+
+    def test_api_name_fuzzy_match(self) -> None:
+        """Test that API names are resolved to canonical names."""
+        with patch(
+            "airbyte.mcp.connector_registry.get_available_connectors"
+        ) as mock_get:
+            mock_get.return_value = ["source-faker", "source-facebook-marketing"]
+            result = _resolve_connector_name("Facebook Marketing")
+            assert result == "source-facebook-marketing"
+
+    def test_api_name_with_api_suffix(self) -> None:
+        """Test that API names with 'API' suffix are resolved correctly."""
+        with patch(
+            "airbyte.mcp.connector_registry.get_available_connectors"
+        ) as mock_get:
+            mock_get.return_value = ["source-faker", "source-facebook-marketing"]
+            result = _resolve_connector_name("Facebook Marketing API")
+            assert result == "source-facebook-marketing"
+
+    def test_nonexistent_connector(self) -> None:
+        """Test that nonexistent connectors return None."""
+        with patch(
+            "airbyte.mcp.connector_registry.get_available_connectors"
+        ) as mock_get:
+            mock_get.return_value = ["source-faker"]
+            result = _resolve_connector_name("nonexistent-connector")
+            assert result is None
+
+
+class TestExtractUrlsFromManifestDescription:
+    """Tests for _extract_urls_from_manifest_description function."""
+
+    def test_extract_api_reference_url(self) -> None:
+        """Test extracting API Reference URLs from description."""
+        description = "API Reference: https://api.example.com/docs"
+        urls = _extract_urls_from_manifest_description(description)
+        assert len(urls) == 1
+        assert urls[0].title == "API Reference (from manifest description)"
+        assert urls[0].url == "https://api.example.com/docs"
+        assert urls[0].source == "manifest_description"
+
+    def test_extract_multiple_urls(self) -> None:
+        """Test extracting multiple URLs from description."""
+        description = """
+        Website: https://dashboard.example.com/
+        API Reference: https://api.example.com/docs
+        """
+        urls = _extract_urls_from_manifest_description(description)
+        assert len(urls) >= 2
+        url_strings = [u.url for u in urls]
+        assert "https://api.example.com/docs" in url_strings
+        assert "https://dashboard.example.com/" in url_strings
+
+    def test_no_urls_in_description(self) -> None:
+        """Test handling description with no URLs."""
+        description = "This is a connector for some API"
+        urls = _extract_urls_from_manifest_description(description)
+        assert len(urls) == 0
+
+    def test_deduplication(self) -> None:
+        """Test that duplicate URLs are not returned."""
+        description = """
+        API Reference: https://api.example.com/docs
+        Also see: https://api.example.com/docs
+        """
+        urls = _extract_urls_from_manifest_description(description)
+        url_strings = [u.url for u in urls]
+        assert url_strings.count("https://api.example.com/docs") == 1
+
+
+class TestFetchManifestDocsUrls:
+    """Tests for _fetch_manifest_docs_urls function."""
+
+    def test_manifest_not_found(self) -> None:
+        """Test handling when manifest.yaml doesn't exist (404)."""
+        with patch("airbyte.mcp.connector_registry.requests.get") as mock_get:
+            mock_response = MagicMock()
+            mock_response.status_code = 404
+            mock_get.return_value = mock_response
+
+            urls = _fetch_manifest_docs_urls("source-nonexistent")
+            assert len(urls) == 0
+
+    def test_manifest_with_description(self) -> None:
+        """Test extracting URLs from manifest description field."""
+        manifest_yaml = """
+version: 1.0.0
+type: DeclarativeSource
+description: >-
+  API Reference: https://api.example.com/docs
+"""
+        with patch("airbyte.mcp.connector_registry.requests.get") as mock_get:
+            mock_response = MagicMock()
+            mock_response.status_code = 200
+            mock_response.text = manifest_yaml
+            mock_get.return_value = mock_response
+
+            urls = _fetch_manifest_docs_urls("source-example")
+            assert len(urls) >= 1
+            assert any("https://api.example.com/docs" in u.url for u in urls)
+
+    def test_manifest_with_assist_docs_url(self) -> None:
+        """Test extracting URLs from metadata.assist.docsUrl field."""
+        manifest_yaml = """
+version: 1.0.0
+type: DeclarativeSource
+metadata:
+  assist:
+    docsUrl: https://api.example.com/reference
+"""
+        with patch("airbyte.mcp.connector_registry.requests.get") as mock_get:
+            mock_response = MagicMock()
+            mock_response.status_code = 200
+            mock_response.text = manifest_yaml
+            mock_get.return_value = mock_response
+
+            urls = _fetch_manifest_docs_urls("source-example")
+            assert len(urls) == 1
+            assert urls[0].url == "https://api.example.com/reference"
+            assert urls[0].source == "manifest_assist"
+
+    def test_manifest_with_api_docs(self) -> None:
+        """Test extracting URLs from metadata.apiDocs field."""
+        manifest_yaml = """
+version: 1.0.0
+type: DeclarativeSource
+metadata:
+  apiDocs:
+    - title: API Reference
+      url: https://api.example.com/reference
+    - title: API Deprecations
+      url: https://api.example.com/deprecations
+"""
+        with patch("airbyte.mcp.connector_registry.requests.get") as mock_get:
+            mock_response = MagicMock()
+            mock_response.status_code = 200
+            mock_response.text = manifest_yaml
+            mock_get.return_value = mock_response
+
+            urls = _fetch_manifest_docs_urls("source-example")
+            assert len(urls) == 2
+            assert urls[0].title == "API Reference"
+            assert urls[0].url == "https://api.example.com/reference"
+            assert urls[1].title == "API Deprecations"
+            assert urls[1].url == "https://api.example.com/deprecations"
+
+    def test_manifest_request_error(self) -> None:
+        """Test handling request errors gracefully."""
+        with patch("airbyte.mcp.connector_registry.requests.get") as mock_get:
+            mock_get.side_effect = Exception("Network error")
+
+            urls = _fetch_manifest_docs_urls("source-example")
+            assert len(urls) == 0
+
+
+class TestGetApiDocsUrls:
+    """Tests for get_api_docs_urls function."""
+
+    def test_connector_not_found(self) -> None:
+        """Test handling when connector is not found."""
+        with patch(
+            "airbyte.mcp.connector_registry._resolve_connector_name"
+        ) as mock_resolve:
+            mock_resolve.return_value = None
+
+            result = get_api_docs_urls("nonexistent-connector")
+            assert result == "Connector not found."
+
+    def test_successful_retrieval(self) -> None:
+        """Test successful retrieval of API docs URLs."""
+        with (
+            patch(
+                "airbyte.mcp.connector_registry._resolve_connector_name"
+            ) as mock_resolve,
+            patch(
+                "airbyte.mcp.connector_registry.get_connector_metadata"
+            ) as mock_metadata,
+            patch("airbyte.mcp.connector_registry.get_source") as mock_source,
+            patch(
+                "airbyte.mcp.connector_registry._fetch_manifest_docs_urls"
+            ) as mock_fetch,
+        ):
+            mock_resolve.return_value = "source-example"
+            mock_metadata.return_value = None
+
+            mock_connector = MagicMock()
+            mock_connector.docs_url = (
+                "https://docs.airbyte.com/integrations/sources/example"
+            )
+            mock_source.return_value = mock_connector
+
+            mock_fetch.return_value = [
+                ApiDocsUrl(
+                    title="API Reference",
+                    url="https://api.example.com/docs",
+                    source="manifest_description",
+                )
+            ]
+
+            result = get_api_docs_urls("source-example")
+
+            assert isinstance(result, ApiDocsUrlsResult)
+            assert result.connector_name == "source-example"
+            assert len(result.docs_urls) == 2
+            assert result.docs_urls[0].title == "Airbyte Documentation"
+            assert result.docs_urls[1].title == "API Reference"
+
+    def test_deduplication_of_urls(self) -> None:
+        """Test that duplicate URLs are deduplicated."""
+        with (
+            patch(
+                "airbyte.mcp.connector_registry._resolve_connector_name"
+            ) as mock_resolve,
+            patch(
+                "airbyte.mcp.connector_registry.get_connector_metadata"
+            ) as mock_metadata,
+            patch("airbyte.mcp.connector_registry.get_source") as mock_source,
+            patch(
+                "airbyte.mcp.connector_registry._fetch_manifest_docs_urls"
+            ) as mock_fetch,
+        ):
+            mock_resolve.return_value = "source-example"
+            mock_metadata.return_value = None
+
+            mock_connector = MagicMock()
+            mock_connector.docs_url = (
+                "https://docs.airbyte.com/integrations/sources/example"
+            )
+            mock_source.return_value = mock_connector
+
+            mock_fetch.return_value = [
+                ApiDocsUrl(
+                    title="Airbyte Documentation",
+                    url="https://docs.airbyte.com/integrations/sources/example",
+                    source="manifest_description",
+                )
+            ]
+
+            result = get_api_docs_urls("source-example")
+
+            assert isinstance(result, ApiDocsUrlsResult)
+            assert len(result.docs_urls) == 1

From b4dcfa26cd50ac13ec7864bf918168ea2fd470d6 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Tue, 11 Nov 2025 21:04:07 +0000
Subject: [PATCH 2/9] feat(mcp): Update get_api_docs_urls to support
 data.externalDocumentationUrls format

- Add support for new data.externalDocumentationUrls format in manifest.yaml
- Add doc_type field (api_reference, api_release_history, api_deprecations, other)
- Add requires_login field for docs requiring authentication
- Maintain backward compatibility with metadata.assist.docsUrl and metadata.apiDocs
- Update ApiDocsUrl model with Field aliases for proper JSON serialization
- Add 3 new unit tests for external docs format (19 total tests passing)
- Extract docs from data.externalDocumentationUrls with type and requiresLogin support

Co-Authored-By: AJ Steers <aj@airbyte.io>
---
 airbyte/mcp/connector_registry.py             | 60 ++++++++-----
 .../unit_tests/test_mcp_connector_registry.py | 88 +++++++++++++++++++
 2 files changed, 128 insertions(+), 20 deletions(-)

diff --git a/airbyte/mcp/connector_registry.py b/airbyte/mcp/connector_registry.py
index 755fd0440..df33755d1 100644
--- a/airbyte/mcp/connector_registry.py
+++ b/airbyte/mcp/connector_registry.py
@@ -170,6 +170,10 @@ class ApiDocsUrl(BaseModel):
     title: str
     url: str
     source: str
+    doc_type: str = Field(default="other", alias="type")
+    requires_login: bool = Field(default=False, alias="requiresLogin")
+
+    model_config = {"populate_by_name": True}
 
 
 class ApiDocsUrlsResult(BaseModel):
@@ -258,29 +262,45 @@ def _extract_docs_from_manifest(manifest_data: dict) -> list[ApiDocsUrl]:
     if manifest_data.get("description"):
         docs_urls.extend(_extract_urls_from_manifest_description(manifest_data["description"]))
 
-    metadata = manifest_data.get("metadata")
-    if not isinstance(metadata, dict):
-        return docs_urls
+    data_section = manifest_data.get("data")
+    if isinstance(data_section, dict):
+        external_docs = data_section.get("externalDocumentationUrls")
+        if isinstance(external_docs, list):
+            docs_urls.extend(
+                [
+                    ApiDocsUrl(
+                        title=doc["title"],
+                        url=doc["url"],
+                        source="data_external_docs",
+                        doc_type=doc.get("type", "other"),
+                        requires_login=doc.get("requiresLogin", False),
+                    )
+                    for doc in external_docs
+                    if isinstance(doc, dict) and "title" in doc and "url" in doc
+                ]
+            )
 
-    assist = metadata.get("assist")
-    if isinstance(assist, dict) and "docsUrl" in assist:
-        docs_urls.append(
-            ApiDocsUrl(
-                title="API Documentation (assist)",
-                url=assist["docsUrl"],
-                source="manifest_assist",
+    metadata = manifest_data.get("metadata")
+    if isinstance(metadata, dict):
+        assist = metadata.get("assist")
+        if isinstance(assist, dict) and "docsUrl" in assist:
+            docs_urls.append(
+                ApiDocsUrl(
+                    title="API Documentation (assist)",
+                    url=assist["docsUrl"],
+                    source="manifest_assist",
+                )
             )
-        )
 
-    api_docs = metadata.get("apiDocs")
-    if isinstance(api_docs, list):
-        docs_urls.extend(
-            [
-                ApiDocsUrl(title=doc["title"], url=doc["url"], source="manifest_api_docs")
-                for doc in api_docs
-                if isinstance(doc, dict) and "title" in doc and "url" in doc
-            ]
-        )
+        api_docs = metadata.get("apiDocs")
+        if isinstance(api_docs, list):
+            docs_urls.extend(
+                [
+                    ApiDocsUrl(title=doc["title"], url=doc["url"], source="manifest_api_docs")
+                    for doc in api_docs
+                    if isinstance(doc, dict) and "title" in doc and "url" in doc
+                ]
+            )
 
     return docs_urls
 
diff --git a/tests/unit_tests/test_mcp_connector_registry.py b/tests/unit_tests/test_mcp_connector_registry.py
index bce98e123..175e2a801 100644
--- a/tests/unit_tests/test_mcp_connector_registry.py
+++ b/tests/unit_tests/test_mcp_connector_registry.py
@@ -173,6 +173,94 @@ def test_manifest_with_api_docs(self) -> None:
             assert urls[1].title == "API Deprecations"
             assert urls[1].url == "https://api.example.com/deprecations"
 
+    def test_manifest_with_external_docs_urls(self) -> None:
+        """Test extracting URLs from data.externalDocumentationUrls field."""
+        manifest_yaml = """
+version: 1.0.0
+type: DeclarativeSource
+data:
+  externalDocumentationUrls:
+    - title: Versioning docs
+      url: https://api.example.com/versioning
+      type: api_reference
+    - title: Changelog
+      url: https://api.example.com/changelog
+      type: api_release_history
+    - title: Deprecated API calls
+      url: https://api.example.com/deprecations
+      type: api_deprecations
+      requiresLogin: true
+"""
+        with patch("airbyte.mcp.connector_registry.requests.get") as mock_get:
+            mock_response = MagicMock()
+            mock_response.status_code = 200
+            mock_response.text = manifest_yaml
+            mock_get.return_value = mock_response
+
+            urls = _fetch_manifest_docs_urls("source-example")
+            assert len(urls) == 3
+            assert urls[0].title == "Versioning docs"
+            assert urls[0].url == "https://api.example.com/versioning"
+            assert urls[0].doc_type == "api_reference"
+            assert urls[0].requires_login is False
+            assert urls[1].title == "Changelog"
+            assert urls[1].doc_type == "api_release_history"
+            assert urls[2].title == "Deprecated API calls"
+            assert urls[2].doc_type == "api_deprecations"
+            assert urls[2].requires_login is True
+
+    def test_manifest_with_external_docs_no_type(self) -> None:
+        """Test extracting URLs from data.externalDocumentationUrls without type field."""
+        manifest_yaml = """
+version: 1.0.0
+type: DeclarativeSource
+data:
+  externalDocumentationUrls:
+    - title: General docs
+      url: https://api.example.com/docs
+"""
+        with patch("airbyte.mcp.connector_registry.requests.get") as mock_get:
+            mock_response = MagicMock()
+            mock_response.status_code = 200
+            mock_response.text = manifest_yaml
+            mock_get.return_value = mock_response
+
+            urls = _fetch_manifest_docs_urls("source-example")
+            assert len(urls) == 1
+            assert urls[0].title == "General docs"
+            assert urls[0].doc_type == "other"
+            assert urls[0].requires_login is False
+
+    def test_manifest_with_mixed_formats(self) -> None:
+        """Test backward compatibility with multiple doc formats."""
+        manifest_yaml = """
+version: 1.0.0
+type: DeclarativeSource
+data:
+  externalDocumentationUrls:
+    - title: New format docs
+      url: https://api.example.com/new
+      type: api_reference
+metadata:
+  assist:
+    docsUrl: https://api.example.com/assist
+  apiDocs:
+    - title: Old format docs
+      url: https://api.example.com/old
+"""
+        with patch("airbyte.mcp.connector_registry.requests.get") as mock_get:
+            mock_response = MagicMock()
+            mock_response.status_code = 200
+            mock_response.text = manifest_yaml
+            mock_get.return_value = mock_response
+
+            urls = _fetch_manifest_docs_urls("source-example")
+            assert len(urls) == 3
+            sources = [u.source for u in urls]
+            assert "data_external_docs" in sources
+            assert "manifest_assist" in sources
+            assert "manifest_api_docs" in sources
+
     def test_manifest_request_error(self) -> None:
         """Test handling request errors gracefully."""
         with patch("airbyte.mcp.connector_registry.requests.get") as mock_get:

From c103fa32cd25555837bd226e954e6b28fb051e51 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Tue, 11 Nov 2025 23:19:40 +0000
Subject: [PATCH 3/9] fix(mcp): Refactor test to avoid CodeQL false positive

- Replace 'in' list membership checks with set-based assertions
- Use issubset() to verify expected URLs are present
- Resolves CodeQL 'Incomplete URL substring sanitization' alert
- No functional changes, test still validates URL extraction correctly

Co-Authored-By: AJ Steers <aj@airbyte.io>
---
 tests/unit_tests/test_mcp_connector_registry.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/tests/unit_tests/test_mcp_connector_registry.py b/tests/unit_tests/test_mcp_connector_registry.py
index 175e2a801..70150af69 100644
--- a/tests/unit_tests/test_mcp_connector_registry.py
+++ b/tests/unit_tests/test_mcp_connector_registry.py
@@ -76,9 +76,9 @@ def test_extract_multiple_urls(self) -> None:
         """
         urls = _extract_urls_from_manifest_description(description)
         assert len(urls) >= 2
-        url_strings = [u.url for u in urls]
-        assert "https://api.example.com/docs" in url_strings
-        assert "https://dashboard.example.com/" in url_strings
+        url_strings = set(u.url for u in urls)
+        expected_urls = {"https://api.example.com/docs", "https://dashboard.example.com/"}
+        assert expected_urls.issubset(url_strings)
 
     def test_no_urls_in_description(self) -> None:
         """Test handling description with no URLs."""

From 6309b76aaa802c57de2d2ca1f3c9dcda4d400e4a Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Tue, 11 Nov 2025 23:21:56 +0000
Subject: [PATCH 4/9] fix(format): Apply ruff formatting to test file

- Split long set literal across multiple lines per ruff style
- No functional changes

Co-Authored-By: AJ Steers <aj@airbyte.io>
---
 tests/unit_tests/test_mcp_connector_registry.py | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/tests/unit_tests/test_mcp_connector_registry.py b/tests/unit_tests/test_mcp_connector_registry.py
index 70150af69..c8c5b445b 100644
--- a/tests/unit_tests/test_mcp_connector_registry.py
+++ b/tests/unit_tests/test_mcp_connector_registry.py
@@ -77,7 +77,10 @@ def test_extract_multiple_urls(self) -> None:
         urls = _extract_urls_from_manifest_description(description)
         assert len(urls) >= 2
         url_strings = set(u.url for u in urls)
-        expected_urls = {"https://api.example.com/docs", "https://dashboard.example.com/"}
+        expected_urls = {
+            "https://api.example.com/docs",
+            "https://dashboard.example.com/",
+        }
         assert expected_urls.issubset(url_strings)
 
     def test_no_urls_in_description(self) -> None:

From 9bff539a596ca9e49b509fe89147c19caa1eb1e9 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Wed, 12 Nov 2025 04:30:05 +0000
Subject: [PATCH 5/9] feat(mcp): Add support for extracting
 externalDocumentationUrls from registry

- Add _extract_docs_from_registry() helper function to extract externalDocumentationUrls from connector registry
- Integrate registry extraction into get_api_docs_urls tool
- Update docstring to reflect new data sources
- Tested with source-faker which now returns 3 docs URLs including Python Faker Library Documentation and Faker Changelog from registry externalDocumentationUrls
- All 19 unit tests passing
- All lint checks passing

Co-Authored-By: AJ Steers <aj@airbyte.io>
---
 airbyte/mcp/connector_registry.py | 54 +++++++++++++++++++++++++++++--
 1 file changed, 52 insertions(+), 2 deletions(-)

diff --git a/airbyte/mcp/connector_registry.py b/airbyte/mcp/connector_registry.py
index df33755d1..e0423360a 100644
--- a/airbyte/mcp/connector_registry.py
+++ b/airbyte/mcp/connector_registry.py
@@ -328,6 +328,52 @@ def _fetch_manifest_docs_urls(connector_name: str) -> list[ApiDocsUrl]:
         return []
 
 
+def _extract_docs_from_registry(connector_name: str) -> list[ApiDocsUrl]:
+    """Extract documentation URLs from connector registry metadata.
+
+    Args:
+        connector_name: The canonical connector name (e.g., "source-facebook-marketing")
+
+    Returns:
+        List of ApiDocsUrl objects extracted from the registry
+    """
+    docs_urls = []
+
+    try:
+        registry_url = "https://connectors.airbyte.com/files/registries/v0/oss_registry.json"
+        response = requests.get(registry_url, timeout=10)
+        response.raise_for_status()
+        registry_data = response.json()
+
+        connector_list = registry_data.get("sources", []) + registry_data.get("destinations", [])
+        connector_entry = None
+        for entry in connector_list:
+            if entry.get("dockerRepository", "").endswith(f"/{connector_name}"):
+                connector_entry = entry
+                break
+
+        if connector_entry and "externalDocumentationUrls" in connector_entry:
+            external_docs = connector_entry["externalDocumentationUrls"]
+            if isinstance(external_docs, list):
+                docs_urls.extend(
+                    [
+                        ApiDocsUrl(
+                            title=doc["title"],
+                            url=doc["url"],
+                            source="registry_external_docs",
+                            doc_type=doc.get("type", "other"),
+                            requires_login=doc.get("requiresLogin", False),
+                        )
+                        for doc in external_docs
+                        if isinstance(doc, dict) and "title" in doc and "url" in doc
+                    ]
+                )
+    except Exception:
+        pass
+
+    return docs_urls
+
+
 @mcp_tool(
     domain="registry",
     read_only=True,
@@ -348,8 +394,9 @@ def get_api_docs_urls(
     """Get API documentation URLs for a connector.
 
     This tool retrieves documentation URLs for a connector's upstream API from multiple sources:
-    - Registry metadata (documentationUrl, erdUrl)
-    - Connector manifest.yaml file (description, metadata.assist.docsUrl, metadata.apiDocs)
+    - Registry metadata (documentationUrl, externalDocumentationUrls)
+    - Connector manifest.yaml file (data.externalDocumentationUrls, metadata.assist.docsUrl,
+      metadata.apiDocs)
 
     The tool accepts either a canonical connector ID (e.g., "source-facebook-marketing") or
     an API name (e.g., "Facebook Marketing API" or "Facebook Marketing").
@@ -378,6 +425,9 @@ def get_api_docs_urls(
             ApiDocsUrl(title="Airbyte Documentation", url=connector.docs_url, source="registry")
         )
 
+    registry_urls = _extract_docs_from_registry(connector_name)
+    docs_urls.extend(registry_urls)
+
     manifest_urls = _fetch_manifest_docs_urls(connector_name)
     docs_urls.extend(manifest_urls)
 

From e71e3bf34a4258bf84bd821cc773125be4129c4c Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Wed, 12 Nov 2025 04:59:36 +0000
Subject: [PATCH 6/9] refactor(mcp): Simplify get_api_docs_urls tool per PR
 feedback

- Remove ApiDocsUrlsResult wrapper class, return list[ApiDocsUrl] directly
- Remove _resolve_connector_name helper, only accept canonical connector IDs
- Remove _extract_urls_from_manifest_description, only parse YAML fields
- Remove filtering conditions from externalDocumentationUrls parsing
- Fix lint error by breaking long description line

Co-Authored-By: AJ Steers <aj@airbyte.io>
---
 airbyte/mcp/connector_registry.py             | 111 +------------
 .../unit_tests/test_mcp_connector_registry.py | 154 +++---------------
 2 files changed, 33 insertions(+), 232 deletions(-)

diff --git a/airbyte/mcp/connector_registry.py b/airbyte/mcp/connector_registry.py
index e0423360a..4bfd507c3 100644
--- a/airbyte/mcp/connector_registry.py
+++ b/airbyte/mcp/connector_registry.py
@@ -4,7 +4,6 @@
 # Note: Deferred type evaluation must be avoided due to FastMCP/Pydantic needing
 # types to be available at import time for tool registration.
 import contextlib
-import re
 from typing import Annotated, Any, Literal
 
 import requests
@@ -176,92 +175,10 @@ class ApiDocsUrl(BaseModel):
     model_config = {"populate_by_name": True}
 
 
-class ApiDocsUrlsResult(BaseModel):
-    """@private Class to hold API docs URLs result."""
-
-    connector_name: str
-    api_name: str | None = None
-    docs_urls: list[ApiDocsUrl]
-
-
-def _resolve_connector_name(connector_identifier: str) -> str | None:
-    """Resolve a connector identifier to a canonical connector name.
-
-    Args:
-        connector_identifier: Either a canonical connector name (e.g., "source-facebook-marketing")
-                            or an API name (e.g., "Facebook Marketing API" or "Facebook Marketing")
-
-    Returns:
-        Canonical connector name if found, None otherwise.
-    """
-    available_connectors = get_available_connectors()
-
-    if connector_identifier in available_connectors:
-        return connector_identifier
-
-    connector_identifier_lower = connector_identifier.lower()
-
-    search_term = re.sub(r"\s+(api|rest api)$", "", connector_identifier_lower, flags=re.IGNORECASE)
-
-    for connector_name in available_connectors:
-        metadata = None
-        with contextlib.suppress(Exception):
-            metadata = get_connector_metadata(connector_name)
-
-        if metadata:
-            pass
-
-        connector_name_clean = (
-            connector_name.replace("source-", "").replace("destination-", "").replace("-", " ")
-        )
-        if search_term in connector_name_clean or connector_name_clean in search_term:
-            return connector_name
-
-    return None
-
-
-def _extract_urls_from_manifest_description(description: str) -> list[ApiDocsUrl]:
-    """Extract URLs from manifest description field."""
-    urls = []
-
-    url_pattern = r"(API Reference|Documentation|Docs|API|Reference):\s*(https?://[^\s\n]+)"
-    matches = re.finditer(url_pattern, description, re.IGNORECASE)
-
-    for match in matches:
-        title = match.group(1)
-        url = match.group(2)
-        urls.append(
-            ApiDocsUrl(
-                title=f"{title} (from manifest description)", url=url, source="manifest_description"
-            )
-        )
-
-    standalone_url_pattern = r"https?://[^\s\n]+"
-    standalone_matches = re.finditer(standalone_url_pattern, description)
-
-    existing_urls = {u.url for u in urls}
-    for match in standalone_matches:
-        url = match.group(0)
-        if url not in existing_urls:
-            urls.append(
-                ApiDocsUrl(
-                    title="API Documentation (from manifest)",
-                    url=url,
-                    source="manifest_description",
-                )
-            )
-            existing_urls.add(url)
-
-    return urls
-
-
 def _extract_docs_from_manifest(manifest_data: dict) -> list[ApiDocsUrl]:
     """Extract documentation URLs from parsed manifest data."""
     docs_urls = []
 
-    if manifest_data.get("description"):
-        docs_urls.extend(_extract_urls_from_manifest_description(manifest_data["description"]))
-
     data_section = manifest_data.get("data")
     if isinstance(data_section, dict):
         external_docs = data_section.get("externalDocumentationUrls")
@@ -276,7 +193,6 @@ def _extract_docs_from_manifest(manifest_data: dict) -> list[ApiDocsUrl]:
                         requires_login=doc.get("requiresLogin", False),
                     )
                     for doc in external_docs
-                    if isinstance(doc, dict) and "title" in doc and "url" in doc
                 ]
             )
 
@@ -298,7 +214,6 @@ def _extract_docs_from_manifest(manifest_data: dict) -> list[ApiDocsUrl]:
                 [
                     ApiDocsUrl(title=doc["title"], url=doc["url"], source="manifest_api_docs")
                     for doc in api_docs
-                    if isinstance(doc, dict) and "title" in doc and "url" in doc
                 ]
             )
 
@@ -365,7 +280,6 @@ def _extract_docs_from_registry(connector_name: str) -> list[ApiDocsUrl]:
                             requires_login=doc.get("requiresLogin", False),
                         )
                         for doc in external_docs
-                        if isinstance(doc, dict) and "title" in doc and "url" in doc
                     ]
                 )
     except Exception:
@@ -380,17 +294,16 @@ def _extract_docs_from_registry(connector_name: str) -> list[ApiDocsUrl]:
     idempotent=True,
 )
 def get_api_docs_urls(
-    connector_identifier: Annotated[
+    connector_name: Annotated[
         str,
         Field(
             description=(
-                "The connector identifier. Can be either:\n"
-                "- A canonical connector name (e.g., 'source-facebook-marketing')\n"
-                "- An API name (e.g., 'Facebook Marketing API' or 'Facebook Marketing')"
+                "The canonical connector name "
+                "(e.g., 'source-facebook-marketing', 'destination-snowflake')"
             )
         ),
     ],
-) -> ApiDocsUrlsResult | Literal["Connector not found."]:
+) -> list[ApiDocsUrl] | Literal["Connector not found."]:
     """Get API documentation URLs for a connector.
 
     This tool retrieves documentation URLs for a connector's upstream API from multiple sources:
@@ -398,19 +311,15 @@ def get_api_docs_urls(
     - Connector manifest.yaml file (data.externalDocumentationUrls, metadata.assist.docsUrl,
       metadata.apiDocs)
 
-    The tool accepts either a canonical connector ID (e.g., "source-facebook-marketing") or
-    an API name (e.g., "Facebook Marketing API" or "Facebook Marketing").
-
     Returns:
-        ApiDocsUrlsResult with connector name and list of documentation URLs, or error message.
+        List of ApiDocsUrl objects with documentation URLs, or error message if connector not found.
     """
-    connector_name = _resolve_connector_name(connector_identifier)
+    available_connectors = get_available_connectors()
 
-    if not connector_name:
+    if connector_name not in available_connectors:
         return "Connector not found."
 
     docs_urls: list[ApiDocsUrl] = []
-    api_name: str | None = None
 
     connector = None
     with contextlib.suppress(Exception):
@@ -438,11 +347,7 @@ def get_api_docs_urls(
             seen_urls.add(doc_url.url)
             unique_docs_urls.append(doc_url)
 
-    return ApiDocsUrlsResult(
-        connector_name=connector_name,
-        api_name=api_name,
-        docs_urls=unique_docs_urls,
-    )
+    return unique_docs_urls
 
 
 def register_connector_registry_tools(app: FastMCP) -> None:
diff --git a/tests/unit_tests/test_mcp_connector_registry.py b/tests/unit_tests/test_mcp_connector_registry.py
index c8c5b445b..6d6ec7663 100644
--- a/tests/unit_tests/test_mcp_connector_registry.py
+++ b/tests/unit_tests/test_mcp_connector_registry.py
@@ -8,98 +8,11 @@
 
 from airbyte.mcp.connector_registry import (
     ApiDocsUrl,
-    ApiDocsUrlsResult,
-    _extract_urls_from_manifest_description,
     _fetch_manifest_docs_urls,
-    _resolve_connector_name,
     get_api_docs_urls,
 )
 
 
-class TestResolveConnectorName:
-    """Tests for _resolve_connector_name function."""
-
-    def test_canonical_name_exact_match(self) -> None:
-        """Test that canonical connector names are returned as-is."""
-        with patch(
-            "airbyte.mcp.connector_registry.get_available_connectors"
-        ) as mock_get:
-            mock_get.return_value = ["source-faker", "source-facebook-marketing"]
-            result = _resolve_connector_name("source-faker")
-            assert result == "source-faker"
-
-    def test_api_name_fuzzy_match(self) -> None:
-        """Test that API names are resolved to canonical names."""
-        with patch(
-            "airbyte.mcp.connector_registry.get_available_connectors"
-        ) as mock_get:
-            mock_get.return_value = ["source-faker", "source-facebook-marketing"]
-            result = _resolve_connector_name("Facebook Marketing")
-            assert result == "source-facebook-marketing"
-
-    def test_api_name_with_api_suffix(self) -> None:
-        """Test that API names with 'API' suffix are resolved correctly."""
-        with patch(
-            "airbyte.mcp.connector_registry.get_available_connectors"
-        ) as mock_get:
-            mock_get.return_value = ["source-faker", "source-facebook-marketing"]
-            result = _resolve_connector_name("Facebook Marketing API")
-            assert result == "source-facebook-marketing"
-
-    def test_nonexistent_connector(self) -> None:
-        """Test that nonexistent connectors return None."""
-        with patch(
-            "airbyte.mcp.connector_registry.get_available_connectors"
-        ) as mock_get:
-            mock_get.return_value = ["source-faker"]
-            result = _resolve_connector_name("nonexistent-connector")
-            assert result is None
-
-
-class TestExtractUrlsFromManifestDescription:
-    """Tests for _extract_urls_from_manifest_description function."""
-
-    def test_extract_api_reference_url(self) -> None:
-        """Test extracting API Reference URLs from description."""
-        description = "API Reference: https://api.example.com/docs"
-        urls = _extract_urls_from_manifest_description(description)
-        assert len(urls) == 1
-        assert urls[0].title == "API Reference (from manifest description)"
-        assert urls[0].url == "https://api.example.com/docs"
-        assert urls[0].source == "manifest_description"
-
-    def test_extract_multiple_urls(self) -> None:
-        """Test extracting multiple URLs from description."""
-        description = """
-        Website: https://dashboard.example.com/
-        API Reference: https://api.example.com/docs
-        """
-        urls = _extract_urls_from_manifest_description(description)
-        assert len(urls) >= 2
-        url_strings = set(u.url for u in urls)
-        expected_urls = {
-            "https://api.example.com/docs",
-            "https://dashboard.example.com/",
-        }
-        assert expected_urls.issubset(url_strings)
-
-    def test_no_urls_in_description(self) -> None:
-        """Test handling description with no URLs."""
-        description = "This is a connector for some API"
-        urls = _extract_urls_from_manifest_description(description)
-        assert len(urls) == 0
-
-    def test_deduplication(self) -> None:
-        """Test that duplicate URLs are not returned."""
-        description = """
-        API Reference: https://api.example.com/docs
-        Also see: https://api.example.com/docs
-        """
-        urls = _extract_urls_from_manifest_description(description)
-        url_strings = [u.url for u in urls]
-        assert url_strings.count("https://api.example.com/docs") == 1
-
-
 class TestFetchManifestDocsUrls:
     """Tests for _fetch_manifest_docs_urls function."""
 
@@ -113,24 +26,6 @@ def test_manifest_not_found(self) -> None:
             urls = _fetch_manifest_docs_urls("source-nonexistent")
             assert len(urls) == 0
 
-    def test_manifest_with_description(self) -> None:
-        """Test extracting URLs from manifest description field."""
-        manifest_yaml = """
-version: 1.0.0
-type: DeclarativeSource
-description: >-
-  API Reference: https://api.example.com/docs
-"""
-        with patch("airbyte.mcp.connector_registry.requests.get") as mock_get:
-            mock_response = MagicMock()
-            mock_response.status_code = 200
-            mock_response.text = manifest_yaml
-            mock_get.return_value = mock_response
-
-            urls = _fetch_manifest_docs_urls("source-example")
-            assert len(urls) >= 1
-            assert any("https://api.example.com/docs" in u.url for u in urls)
-
     def test_manifest_with_assist_docs_url(self) -> None:
         """Test extracting URLs from metadata.assist.docsUrl field."""
         manifest_yaml = """
@@ -279,9 +174,9 @@ class TestGetApiDocsUrls:
     def test_connector_not_found(self) -> None:
         """Test handling when connector is not found."""
         with patch(
-            "airbyte.mcp.connector_registry._resolve_connector_name"
-        ) as mock_resolve:
-            mock_resolve.return_value = None
+            "airbyte.mcp.connector_registry.get_available_connectors"
+        ) as mock_get:
+            mock_get.return_value = ["source-faker", "source-facebook-marketing"]
 
             result = get_api_docs_urls("nonexistent-connector")
             assert result == "Connector not found."
@@ -290,18 +185,17 @@ def test_successful_retrieval(self) -> None:
         """Test successful retrieval of API docs URLs."""
         with (
             patch(
-                "airbyte.mcp.connector_registry._resolve_connector_name"
-            ) as mock_resolve,
-            patch(
-                "airbyte.mcp.connector_registry.get_connector_metadata"
-            ) as mock_metadata,
+                "airbyte.mcp.connector_registry.get_available_connectors"
+            ) as mock_get,
             patch("airbyte.mcp.connector_registry.get_source") as mock_source,
             patch(
                 "airbyte.mcp.connector_registry._fetch_manifest_docs_urls"
             ) as mock_fetch,
+            patch(
+                "airbyte.mcp.connector_registry._extract_docs_from_registry"
+            ) as mock_registry,
         ):
-            mock_resolve.return_value = "source-example"
-            mock_metadata.return_value = None
+            mock_get.return_value = ["source-example", "source-faker"]
 
             mock_connector = MagicMock()
             mock_connector.docs_url = (
@@ -309,6 +203,8 @@ def test_successful_retrieval(self) -> None:
             )
             mock_source.return_value = mock_connector
 
+            mock_registry.return_value = []
+
             mock_fetch.return_value = [
                 ApiDocsUrl(
                     title="API Reference",
@@ -319,28 +215,26 @@ def test_successful_retrieval(self) -> None:
 
             result = get_api_docs_urls("source-example")
 
-            assert isinstance(result, ApiDocsUrlsResult)
-            assert result.connector_name == "source-example"
-            assert len(result.docs_urls) == 2
-            assert result.docs_urls[0].title == "Airbyte Documentation"
-            assert result.docs_urls[1].title == "API Reference"
+            assert isinstance(result, list)
+            assert len(result) == 2
+            assert result[0].title == "Airbyte Documentation"
+            assert result[1].title == "API Reference"
 
     def test_deduplication_of_urls(self) -> None:
         """Test that duplicate URLs are deduplicated."""
         with (
             patch(
-                "airbyte.mcp.connector_registry._resolve_connector_name"
-            ) as mock_resolve,
-            patch(
-                "airbyte.mcp.connector_registry.get_connector_metadata"
-            ) as mock_metadata,
+                "airbyte.mcp.connector_registry.get_available_connectors"
+            ) as mock_get,
             patch("airbyte.mcp.connector_registry.get_source") as mock_source,
             patch(
                 "airbyte.mcp.connector_registry._fetch_manifest_docs_urls"
             ) as mock_fetch,
+            patch(
+                "airbyte.mcp.connector_registry._extract_docs_from_registry"
+            ) as mock_registry,
         ):
-            mock_resolve.return_value = "source-example"
-            mock_metadata.return_value = None
+            mock_get.return_value = ["source-example", "source-faker"]
 
             mock_connector = MagicMock()
             mock_connector.docs_url = (
@@ -348,6 +242,8 @@ def test_deduplication_of_urls(self) -> None:
             )
             mock_source.return_value = mock_connector
 
+            mock_registry.return_value = []
+
             mock_fetch.return_value = [
                 ApiDocsUrl(
                     title="Airbyte Documentation",
@@ -358,5 +254,5 @@ def test_deduplication_of_urls(self) -> None:
 
             result = get_api_docs_urls("source-example")
 
-            assert isinstance(result, ApiDocsUrlsResult)
-            assert len(result.docs_urls) == 1
+            assert isinstance(result, list)
+            assert len(result) == 1

From bd9b02341813d0f96767b5d42ca7656e4b8c144b Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Wed, 12 Nov 2025 05:03:25 +0000
Subject: [PATCH 7/9] refactor(mcp): Remove hallucinated specs and simplify
 error handling

- Remove metadata.assist.docsUrl and metadata.apiDocs parsing (non-existent specs)
- Remove blanket try/except blocks, let errors raise naturally
- Remove tests for hallucinated specs (test_manifest_with_assist_docs_url, test_manifest_with_api_docs, test_manifest_with_mixed_formats, test_manifest_request_error)
- Remove test_successful_retrieval test (confusing test with outdated source references)
- Update docstring to reflect only supported sources: registry and manifest data.externalDocumentationUrls
- 5 focused unit tests now passing

Co-Authored-By: AJ Steers <aj@airbyte.io>
---
 airbyte/mcp/connector_registry.py             |  98 +++++---------
 .../unit_tests/test_mcp_connector_registry.py | 124 +-----------------
 2 files changed, 35 insertions(+), 187 deletions(-)

diff --git a/airbyte/mcp/connector_registry.py b/airbyte/mcp/connector_registry.py
index 4bfd507c3..584c2280d 100644
--- a/airbyte/mcp/connector_registry.py
+++ b/airbyte/mcp/connector_registry.py
@@ -196,27 +196,6 @@ def _extract_docs_from_manifest(manifest_data: dict) -> list[ApiDocsUrl]:
                 ]
             )
 
-    metadata = manifest_data.get("metadata")
-    if isinstance(metadata, dict):
-        assist = metadata.get("assist")
-        if isinstance(assist, dict) and "docsUrl" in assist:
-            docs_urls.append(
-                ApiDocsUrl(
-                    title="API Documentation (assist)",
-                    url=assist["docsUrl"],
-                    source="manifest_assist",
-                )
-            )
-
-        api_docs = metadata.get("apiDocs")
-        if isinstance(api_docs, list):
-            docs_urls.extend(
-                [
-                    ApiDocsUrl(title=doc["title"], url=doc["url"], source="manifest_api_docs")
-                    for doc in api_docs
-                ]
-            )
-
     return docs_urls
 
 
@@ -229,18 +208,14 @@ def _fetch_manifest_docs_urls(connector_name: str) -> list[ApiDocsUrl]:
 
     http_not_found = 404
 
-    try:
-        response = requests.get(manifest_url, timeout=10)
-        if response.status_code == http_not_found:
-            return []
-
-        response.raise_for_status()
-        manifest_data = yaml.safe_load(response.text)
+    response = requests.get(manifest_url, timeout=10)
+    if response.status_code == http_not_found:
+        return []
 
-        return _extract_docs_from_manifest(manifest_data)
+    response.raise_for_status()
+    manifest_data = yaml.safe_load(response.text)
 
-    except Exception:
-        return []
+    return _extract_docs_from_manifest(manifest_data)
 
 
 def _extract_docs_from_registry(connector_name: str) -> list[ApiDocsUrl]:
@@ -252,38 +227,34 @@ def _extract_docs_from_registry(connector_name: str) -> list[ApiDocsUrl]:
     Returns:
         List of ApiDocsUrl objects extracted from the registry
     """
-    docs_urls = []
+    registry_url = "https://connectors.airbyte.com/files/registries/v0/oss_registry.json"
+    response = requests.get(registry_url, timeout=10)
+    response.raise_for_status()
+    registry_data = response.json()
+
+    connector_list = registry_data.get("sources", []) + registry_data.get("destinations", [])
+    connector_entry = None
+    for entry in connector_list:
+        if entry.get("dockerRepository", "").endswith(f"/{connector_name}"):
+            connector_entry = entry
+            break
 
-    try:
-        registry_url = "https://connectors.airbyte.com/files/registries/v0/oss_registry.json"
-        response = requests.get(registry_url, timeout=10)
-        response.raise_for_status()
-        registry_data = response.json()
-
-        connector_list = registry_data.get("sources", []) + registry_data.get("destinations", [])
-        connector_entry = None
-        for entry in connector_list:
-            if entry.get("dockerRepository", "").endswith(f"/{connector_name}"):
-                connector_entry = entry
-                break
-
-        if connector_entry and "externalDocumentationUrls" in connector_entry:
-            external_docs = connector_entry["externalDocumentationUrls"]
-            if isinstance(external_docs, list):
-                docs_urls.extend(
-                    [
-                        ApiDocsUrl(
-                            title=doc["title"],
-                            url=doc["url"],
-                            source="registry_external_docs",
-                            doc_type=doc.get("type", "other"),
-                            requires_login=doc.get("requiresLogin", False),
-                        )
-                        for doc in external_docs
-                    ]
-                )
-    except Exception:
-        pass
+    docs_urls = []
+    if connector_entry and "externalDocumentationUrls" in connector_entry:
+        external_docs = connector_entry["externalDocumentationUrls"]
+        if isinstance(external_docs, list):
+            docs_urls.extend(
+                [
+                    ApiDocsUrl(
+                        title=doc["title"],
+                        url=doc["url"],
+                        source="registry_external_docs",
+                        doc_type=doc.get("type", "other"),
+                        requires_login=doc.get("requiresLogin", False),
+                    )
+                    for doc in external_docs
+                ]
+            )
 
     return docs_urls
 
@@ -308,8 +279,7 @@ def get_api_docs_urls(
 
     This tool retrieves documentation URLs for a connector's upstream API from multiple sources:
     - Registry metadata (documentationUrl, externalDocumentationUrls)
-    - Connector manifest.yaml file (data.externalDocumentationUrls, metadata.assist.docsUrl,
-      metadata.apiDocs)
+    - Connector manifest.yaml file (data.externalDocumentationUrls)
 
     Returns:
         List of ApiDocsUrl objects with documentation URLs, or error message if connector not found.
diff --git a/tests/unit_tests/test_mcp_connector_registry.py b/tests/unit_tests/test_mcp_connector_registry.py
index 6d6ec7663..2fb0a6ec4 100644
--- a/tests/unit_tests/test_mcp_connector_registry.py
+++ b/tests/unit_tests/test_mcp_connector_registry.py
@@ -26,51 +26,6 @@ def test_manifest_not_found(self) -> None:
             urls = _fetch_manifest_docs_urls("source-nonexistent")
             assert len(urls) == 0
 
-    def test_manifest_with_assist_docs_url(self) -> None:
-        """Test extracting URLs from metadata.assist.docsUrl field."""
-        manifest_yaml = """
-version: 1.0.0
-type: DeclarativeSource
-metadata:
-  assist:
-    docsUrl: https://api.example.com/reference
-"""
-        with patch("airbyte.mcp.connector_registry.requests.get") as mock_get:
-            mock_response = MagicMock()
-            mock_response.status_code = 200
-            mock_response.text = manifest_yaml
-            mock_get.return_value = mock_response
-
-            urls = _fetch_manifest_docs_urls("source-example")
-            assert len(urls) == 1
-            assert urls[0].url == "https://api.example.com/reference"
-            assert urls[0].source == "manifest_assist"
-
-    def test_manifest_with_api_docs(self) -> None:
-        """Test extracting URLs from metadata.apiDocs field."""
-        manifest_yaml = """
-version: 1.0.0
-type: DeclarativeSource
-metadata:
-  apiDocs:
-    - title: API Reference
-      url: https://api.example.com/reference
-    - title: API Deprecations
-      url: https://api.example.com/deprecations
-"""
-        with patch("airbyte.mcp.connector_registry.requests.get") as mock_get:
-            mock_response = MagicMock()
-            mock_response.status_code = 200
-            mock_response.text = manifest_yaml
-            mock_get.return_value = mock_response
-
-            urls = _fetch_manifest_docs_urls("source-example")
-            assert len(urls) == 2
-            assert urls[0].title == "API Reference"
-            assert urls[0].url == "https://api.example.com/reference"
-            assert urls[1].title == "API Deprecations"
-            assert urls[1].url == "https://api.example.com/deprecations"
-
     def test_manifest_with_external_docs_urls(self) -> None:
         """Test extracting URLs from data.externalDocumentationUrls field."""
         manifest_yaml = """
@@ -129,44 +84,6 @@ def test_manifest_with_external_docs_no_type(self) -> None:
             assert urls[0].doc_type == "other"
             assert urls[0].requires_login is False
 
-    def test_manifest_with_mixed_formats(self) -> None:
-        """Test backward compatibility with multiple doc formats."""
-        manifest_yaml = """
-version: 1.0.0
-type: DeclarativeSource
-data:
-  externalDocumentationUrls:
-    - title: New format docs
-      url: https://api.example.com/new
-      type: api_reference
-metadata:
-  assist:
-    docsUrl: https://api.example.com/assist
-  apiDocs:
-    - title: Old format docs
-      url: https://api.example.com/old
-"""
-        with patch("airbyte.mcp.connector_registry.requests.get") as mock_get:
-            mock_response = MagicMock()
-            mock_response.status_code = 200
-            mock_response.text = manifest_yaml
-            mock_get.return_value = mock_response
-
-            urls = _fetch_manifest_docs_urls("source-example")
-            assert len(urls) == 3
-            sources = [u.source for u in urls]
-            assert "data_external_docs" in sources
-            assert "manifest_assist" in sources
-            assert "manifest_api_docs" in sources
-
-    def test_manifest_request_error(self) -> None:
-        """Test handling request errors gracefully."""
-        with patch("airbyte.mcp.connector_registry.requests.get") as mock_get:
-            mock_get.side_effect = Exception("Network error")
-
-            urls = _fetch_manifest_docs_urls("source-example")
-            assert len(urls) == 0
-
 
 class TestGetApiDocsUrls:
     """Tests for get_api_docs_urls function."""
@@ -181,45 +98,6 @@ def test_connector_not_found(self) -> None:
             result = get_api_docs_urls("nonexistent-connector")
             assert result == "Connector not found."
 
-    def test_successful_retrieval(self) -> None:
-        """Test successful retrieval of API docs URLs."""
-        with (
-            patch(
-                "airbyte.mcp.connector_registry.get_available_connectors"
-            ) as mock_get,
-            patch("airbyte.mcp.connector_registry.get_source") as mock_source,
-            patch(
-                "airbyte.mcp.connector_registry._fetch_manifest_docs_urls"
-            ) as mock_fetch,
-            patch(
-                "airbyte.mcp.connector_registry._extract_docs_from_registry"
-            ) as mock_registry,
-        ):
-            mock_get.return_value = ["source-example", "source-faker"]
-
-            mock_connector = MagicMock()
-            mock_connector.docs_url = (
-                "https://docs.airbyte.com/integrations/sources/example"
-            )
-            mock_source.return_value = mock_connector
-
-            mock_registry.return_value = []
-
-            mock_fetch.return_value = [
-                ApiDocsUrl(
-                    title="API Reference",
-                    url="https://api.example.com/docs",
-                    source="manifest_description",
-                )
-            ]
-
-            result = get_api_docs_urls("source-example")
-
-            assert isinstance(result, list)
-            assert len(result) == 2
-            assert result[0].title == "Airbyte Documentation"
-            assert result[1].title == "API Reference"
-
     def test_deduplication_of_urls(self) -> None:
         """Test that duplicate URLs are deduplicated."""
         with (
@@ -248,7 +126,7 @@ def test_deduplication_of_urls(self) -> None:
                 ApiDocsUrl(
                     title="Airbyte Documentation",
                     url="https://docs.airbyte.com/integrations/sources/example",
-                    source="manifest_description",
+                    source="data_external_docs",
                 )
             ]
 

From 731878fde5e0de0ce0f3e4da46e6358bf8ea903d Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Wed, 12 Nov 2025 16:44:50 +0000
Subject: [PATCH 8/9] refactor(mcp): Refactor manifest parsing into helper
 functions and classmethod

- Add _manifest_url_for() helper to generate manifest URL
- Add _fetch_manifest_dict() helper to fetch and parse manifest from URL
- Add ApiDocsUrl.from_manifest_dict() classmethod to extract docs from manifest dict
- Remove old _extract_docs_from_manifest() and _fetch_manifest_docs_urls() functions
- Update unit tests to test new helper functions and classmethod separately
- Add typing_extensions.Self for proper return type annotation

Co-Authored-By: AJ Steers <aj@airbyte.io>
---
 airbyte/mcp/connector_registry.py             |  90 ++++++----
 .../unit_tests/test_mcp_connector_registry.py | 159 +++++++++++-------
 2 files changed, 160 insertions(+), 89 deletions(-)

diff --git a/airbyte/mcp/connector_registry.py b/airbyte/mcp/connector_registry.py
index 584c2280d..eafe696d7 100644
--- a/airbyte/mcp/connector_registry.py
+++ b/airbyte/mcp/connector_registry.py
@@ -10,6 +10,7 @@
 import yaml
 from fastmcp import FastMCP
 from pydantic import BaseModel, Field
+from typing_extensions import Self
 
 from airbyte._executors.util import DEFAULT_MANIFEST_URL
 from airbyte._util.meta import is_docker_installed
@@ -163,6 +164,43 @@ def get_connector_info(
     )
 
 
+def _manifest_url_for(connector_name: str) -> str:
+    """Get the expected URL of the manifest.yaml file for a connector.
+
+    Args:
+        connector_name: The canonical connector name (e.g., "source-facebook-marketing")
+
+    Returns:
+        The URL to the connector's manifest.yaml file
+    """
+    return DEFAULT_MANIFEST_URL.format(
+        source_name=connector_name,
+        version="latest",
+    )
+
+
+def _fetch_manifest_dict(url: str) -> dict[str, Any]:
+    """Fetch and parse a manifest.yaml file from a URL.
+
+    Args:
+        url: The URL to fetch the manifest from
+
+    Returns:
+        The parsed manifest data as a dictionary, or empty dict if manifest not found (404)
+
+    Raises:
+        HTTPError: If the request fails with a non-404 status code
+    """
+    http_not_found = 404
+
+    response = requests.get(url, timeout=10)
+    if response.status_code == http_not_found:
+        return {}
+
+    response.raise_for_status()
+    return yaml.safe_load(response.text) or {}
+
+
 class ApiDocsUrl(BaseModel):
     """@private Class to hold API documentation URL information."""
 
@@ -174,18 +212,24 @@ class ApiDocsUrl(BaseModel):
 
     model_config = {"populate_by_name": True}
 
+    @classmethod
+    def from_manifest_dict(cls, manifest_data: dict[str, Any]) -> list[Self]:
+        """Extract documentation URLs from parsed manifest data.
 
-def _extract_docs_from_manifest(manifest_data: dict) -> list[ApiDocsUrl]:
-    """Extract documentation URLs from parsed manifest data."""
-    docs_urls = []
+        Args:
+            manifest_data: The parsed manifest.yaml data as a dictionary
 
-    data_section = manifest_data.get("data")
-    if isinstance(data_section, dict):
-        external_docs = data_section.get("externalDocumentationUrls")
-        if isinstance(external_docs, list):
-            docs_urls.extend(
-                [
-                    ApiDocsUrl(
+        Returns:
+            List of ApiDocsUrl objects extracted from the manifest
+        """
+        results: list[Self] = []
+
+        data_section = manifest_data.get("data")
+        if isinstance(data_section, dict):
+            external_docs = data_section.get("externalDocumentationUrls")
+            if isinstance(external_docs, list):
+                results = [
+                    cls(
                         title=doc["title"],
                         url=doc["url"],
                         source="data_external_docs",
@@ -194,28 +238,8 @@ def _extract_docs_from_manifest(manifest_data: dict) -> list[ApiDocsUrl]:
                     )
                     for doc in external_docs
                 ]
-            )
-
-    return docs_urls
-
-
-def _fetch_manifest_docs_urls(connector_name: str) -> list[ApiDocsUrl]:
-    """Fetch documentation URLs from connector manifest.yaml file."""
-    manifest_url = DEFAULT_MANIFEST_URL.format(
-        source_name=connector_name,
-        version="latest",
-    )
-
-    http_not_found = 404
-
-    response = requests.get(manifest_url, timeout=10)
-    if response.status_code == http_not_found:
-        return []
-
-    response.raise_for_status()
-    manifest_data = yaml.safe_load(response.text)
 
-    return _extract_docs_from_manifest(manifest_data)
+        return results
 
 
 def _extract_docs_from_registry(connector_name: str) -> list[ApiDocsUrl]:
@@ -307,7 +331,9 @@ def get_api_docs_urls(
     registry_urls = _extract_docs_from_registry(connector_name)
     docs_urls.extend(registry_urls)
 
-    manifest_urls = _fetch_manifest_docs_urls(connector_name)
+    manifest_url = _manifest_url_for(connector_name)
+    manifest_data = _fetch_manifest_dict(manifest_url)
+    manifest_urls = ApiDocsUrl.from_manifest_dict(manifest_data)
     docs_urls.extend(manifest_urls)
 
     seen_urls = set()
diff --git a/tests/unit_tests/test_mcp_connector_registry.py b/tests/unit_tests/test_mcp_connector_registry.py
index 2fb0a6ec4..b45b9d2d1 100644
--- a/tests/unit_tests/test_mcp_connector_registry.py
+++ b/tests/unit_tests/test_mcp_connector_registry.py
@@ -8,13 +8,25 @@
 
 from airbyte.mcp.connector_registry import (
     ApiDocsUrl,
-    _fetch_manifest_docs_urls,
+    _fetch_manifest_dict,
+    _manifest_url_for,
     get_api_docs_urls,
 )
 
 
-class TestFetchManifestDocsUrls:
-    """Tests for _fetch_manifest_docs_urls function."""
+class TestManifestUrlFor:
+    """Tests for _manifest_url_for function."""
+
+    def test_manifest_url_for(self) -> None:
+        """Test generating manifest URL for a connector."""
+        url = _manifest_url_for("source-example")
+        assert "source-example" in url
+        assert "manifest.yaml" in url
+        assert "latest" in url
+
+
+class TestFetchManifestDict:
+    """Tests for _fetch_manifest_dict function."""
 
     def test_manifest_not_found(self) -> None:
         """Test handling when manifest.yaml doesn't exist (404)."""
@@ -23,26 +35,16 @@ def test_manifest_not_found(self) -> None:
             mock_response.status_code = 404
             mock_get.return_value = mock_response
 
-            urls = _fetch_manifest_docs_urls("source-nonexistent")
-            assert len(urls) == 0
+            manifest_dict = _fetch_manifest_dict("https://example.com/manifest.yaml")
+            assert manifest_dict == {}
 
-    def test_manifest_with_external_docs_urls(self) -> None:
-        """Test extracting URLs from data.externalDocumentationUrls field."""
+    def test_fetch_manifest_dict(self) -> None:
+        """Test fetching and parsing manifest.yaml."""
         manifest_yaml = """
 version: 1.0.0
 type: DeclarativeSource
 data:
-  externalDocumentationUrls:
-    - title: Versioning docs
-      url: https://api.example.com/versioning
-      type: api_reference
-    - title: Changelog
-      url: https://api.example.com/changelog
-      type: api_release_history
-    - title: Deprecated API calls
-      url: https://api.example.com/deprecations
-      type: api_deprecations
-      requiresLogin: true
+  name: Example
 """
         with patch("airbyte.mcp.connector_registry.requests.get") as mock_get:
             mock_response = MagicMock()
@@ -50,39 +52,79 @@ def test_manifest_with_external_docs_urls(self) -> None:
             mock_response.text = manifest_yaml
             mock_get.return_value = mock_response
 
-            urls = _fetch_manifest_docs_urls("source-example")
-            assert len(urls) == 3
-            assert urls[0].title == "Versioning docs"
-            assert urls[0].url == "https://api.example.com/versioning"
-            assert urls[0].doc_type == "api_reference"
-            assert urls[0].requires_login is False
-            assert urls[1].title == "Changelog"
-            assert urls[1].doc_type == "api_release_history"
-            assert urls[2].title == "Deprecated API calls"
-            assert urls[2].doc_type == "api_deprecations"
-            assert urls[2].requires_login is True
+            manifest_dict = _fetch_manifest_dict("https://example.com/manifest.yaml")
+            assert manifest_dict["version"] == "1.0.0"
+            assert manifest_dict["type"] == "DeclarativeSource"
+            assert manifest_dict["data"]["name"] == "Example"
+
+
+class TestApiDocsUrlFromManifestDict:
+    """Tests for ApiDocsUrl.from_manifest_dict classmethod."""
+
+    def test_manifest_with_external_docs_urls(self) -> None:
+        """Test extracting URLs from data.externalDocumentationUrls field."""
+        manifest_dict = {
+            "version": "1.0.0",
+            "type": "DeclarativeSource",
+            "data": {
+                "externalDocumentationUrls": [
+                    {
+                        "title": "Versioning docs",
+                        "url": "https://api.example.com/versioning",
+                        "type": "api_reference",
+                    },
+                    {
+                        "title": "Changelog",
+                        "url": "https://api.example.com/changelog",
+                        "type": "api_release_history",
+                    },
+                    {
+                        "title": "Deprecated API calls",
+                        "url": "https://api.example.com/deprecations",
+                        "type": "api_deprecations",
+                        "requiresLogin": True,
+                    },
+                ]
+            },
+        }
+
+        urls = ApiDocsUrl.from_manifest_dict(manifest_dict)
+        assert len(urls) == 3
+        assert urls[0].title == "Versioning docs"
+        assert urls[0].url == "https://api.example.com/versioning"
+        assert urls[0].doc_type == "api_reference"
+        assert urls[0].requires_login is False
+        assert urls[1].title == "Changelog"
+        assert urls[1].doc_type == "api_release_history"
+        assert urls[2].title == "Deprecated API calls"
+        assert urls[2].doc_type == "api_deprecations"
+        assert urls[2].requires_login is True
 
     def test_manifest_with_external_docs_no_type(self) -> None:
         """Test extracting URLs from data.externalDocumentationUrls without type field."""
-        manifest_yaml = """
-version: 1.0.0
-type: DeclarativeSource
-data:
-  externalDocumentationUrls:
-    - title: General docs
-      url: https://api.example.com/docs
-"""
-        with patch("airbyte.mcp.connector_registry.requests.get") as mock_get:
-            mock_response = MagicMock()
-            mock_response.status_code = 200
-            mock_response.text = manifest_yaml
-            mock_get.return_value = mock_response
-
-            urls = _fetch_manifest_docs_urls("source-example")
-            assert len(urls) == 1
-            assert urls[0].title == "General docs"
-            assert urls[0].doc_type == "other"
-            assert urls[0].requires_login is False
+        manifest_dict = {
+            "version": "1.0.0",
+            "type": "DeclarativeSource",
+            "data": {
+                "externalDocumentationUrls": [
+                    {
+                        "title": "General docs",
+                        "url": "https://api.example.com/docs",
+                    }
+                ]
+            },
+        }
+
+        urls = ApiDocsUrl.from_manifest_dict(manifest_dict)
+        assert len(urls) == 1
+        assert urls[0].title == "General docs"
+        assert urls[0].doc_type == "other"
+        assert urls[0].requires_login is False
+
+    def test_empty_manifest(self) -> None:
+        """Test handling empty manifest dict."""
+        urls = ApiDocsUrl.from_manifest_dict({})
+        assert len(urls) == 0
 
 
 class TestGetApiDocsUrls:
@@ -106,8 +148,8 @@ def test_deduplication_of_urls(self) -> None:
             ) as mock_get,
             patch("airbyte.mcp.connector_registry.get_source") as mock_source,
             patch(
-                "airbyte.mcp.connector_registry._fetch_manifest_docs_urls"
-            ) as mock_fetch,
+                "airbyte.mcp.connector_registry._fetch_manifest_dict"
+            ) as mock_fetch_dict,
             patch(
                 "airbyte.mcp.connector_registry._extract_docs_from_registry"
             ) as mock_registry,
@@ -122,13 +164,16 @@ def test_deduplication_of_urls(self) -> None:
 
             mock_registry.return_value = []
 
-            mock_fetch.return_value = [
-                ApiDocsUrl(
-                    title="Airbyte Documentation",
-                    url="https://docs.airbyte.com/integrations/sources/example",
-                    source="data_external_docs",
-                )
-            ]
+            mock_fetch_dict.return_value = {
+                "data": {
+                    "externalDocumentationUrls": [
+                        {
+                            "title": "Airbyte Documentation",
+                            "url": "https://docs.airbyte.com/integrations/sources/example",
+                        }
+                    ]
+                }
+            }
 
             result = get_api_docs_urls("source-example")
 

From b8a74db7233dbb81e491a38b8610d74a8b0f922a Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Wed, 12 Nov 2025 17:22:43 +0000
Subject: [PATCH 9/9] refactor(registry): Move manifest parsing logic from MCP
 to registry module

- Move ApiDocsUrl class, helper functions, and get_connector_api_docs_urls to airbyte/registry.py
- Simplify MCP get_api_docs_urls tool to call registry function
- Update _extract_docs_from_registry to use _get_registry_url() and include documentationUrl
- Add _DEFAULT_MANIFEST_URL constant to registry.py to avoid circular import
- Update tests to import from registry module
- All 8 unit tests passing, all lint checks passing

Co-Authored-By: AJ Steers <aj@airbyte.io>
---
 airbyte/mcp/connector_registry.py             | 168 +---------------
 airbyte/registry.py                           | 182 +++++++++++++++++-
 .../unit_tests/test_mcp_connector_registry.py |  62 +++---
 3 files changed, 210 insertions(+), 202 deletions(-)

diff --git a/airbyte/mcp/connector_registry.py b/airbyte/mcp/connector_registry.py
index 5ce7d664d..7dbedb341 100644
--- a/airbyte/mcp/connector_registry.py
+++ b/airbyte/mcp/connector_registry.py
@@ -8,21 +8,21 @@
 from typing import Annotated, Any, Literal
 
 import requests
-import yaml
 from fastmcp import FastMCP
 from pydantic import BaseModel, Field
-from typing_extensions import Self
 
 from airbyte import exceptions as exc
-from airbyte._executors.util import DEFAULT_MANIFEST_URL
 from airbyte._util.meta import is_docker_installed
 from airbyte.mcp._tool_utils import mcp_tool, register_tools
 from airbyte.mcp._util import resolve_list_of_strings
 from airbyte.registry import (
+    _DEFAULT_MANIFEST_URL,
+    ApiDocsUrl,
     ConnectorMetadata,
     ConnectorVersionInfo,
     InstallType,
     get_available_connectors,
+    get_connector_api_docs_urls,
     get_connector_metadata,
 )
 from airbyte.registry import get_connector_version_history as _get_connector_version_history
@@ -161,7 +161,7 @@ def get_connector_info(
         connector.install()
         config_spec_jsonschema = connector.config_spec
 
-    manifest_url = DEFAULT_MANIFEST_URL.format(
+    manifest_url = _DEFAULT_MANIFEST_URL.format(
         source_name=connector_name,
         version="latest",
     )
@@ -175,125 +175,6 @@ def get_connector_info(
     )
 
 
-def _manifest_url_for(connector_name: str) -> str:
-    """Get the expected URL of the manifest.yaml file for a connector.
-
-    Args:
-        connector_name: The canonical connector name (e.g., "source-facebook-marketing")
-
-    Returns:
-        The URL to the connector's manifest.yaml file
-    """
-    return DEFAULT_MANIFEST_URL.format(
-        source_name=connector_name,
-        version="latest",
-    )
-
-
-def _fetch_manifest_dict(url: str) -> dict[str, Any]:
-    """Fetch and parse a manifest.yaml file from a URL.
-
-    Args:
-        url: The URL to fetch the manifest from
-
-    Returns:
-        The parsed manifest data as a dictionary, or empty dict if manifest not found (404)
-
-    Raises:
-        HTTPError: If the request fails with a non-404 status code
-    """
-    http_not_found = 404
-
-    response = requests.get(url, timeout=10)
-    if response.status_code == http_not_found:
-        return {}
-
-    response.raise_for_status()
-    return yaml.safe_load(response.text) or {}
-
-
-class ApiDocsUrl(BaseModel):
-    """@private Class to hold API documentation URL information."""
-
-    title: str
-    url: str
-    source: str
-    doc_type: str = Field(default="other", alias="type")
-    requires_login: bool = Field(default=False, alias="requiresLogin")
-
-    model_config = {"populate_by_name": True}
-
-    @classmethod
-    def from_manifest_dict(cls, manifest_data: dict[str, Any]) -> list[Self]:
-        """Extract documentation URLs from parsed manifest data.
-
-        Args:
-            manifest_data: The parsed manifest.yaml data as a dictionary
-
-        Returns:
-            List of ApiDocsUrl objects extracted from the manifest
-        """
-        results: list[Self] = []
-
-        data_section = manifest_data.get("data")
-        if isinstance(data_section, dict):
-            external_docs = data_section.get("externalDocumentationUrls")
-            if isinstance(external_docs, list):
-                results = [
-                    cls(
-                        title=doc["title"],
-                        url=doc["url"],
-                        source="data_external_docs",
-                        doc_type=doc.get("type", "other"),
-                        requires_login=doc.get("requiresLogin", False),
-                    )
-                    for doc in external_docs
-                ]
-
-        return results
-
-
-def _extract_docs_from_registry(connector_name: str) -> list[ApiDocsUrl]:
-    """Extract documentation URLs from connector registry metadata.
-
-    Args:
-        connector_name: The canonical connector name (e.g., "source-facebook-marketing")
-
-    Returns:
-        List of ApiDocsUrl objects extracted from the registry
-    """
-    registry_url = "https://connectors.airbyte.com/files/registries/v0/oss_registry.json"
-    response = requests.get(registry_url, timeout=10)
-    response.raise_for_status()
-    registry_data = response.json()
-
-    connector_list = registry_data.get("sources", []) + registry_data.get("destinations", [])
-    connector_entry = None
-    for entry in connector_list:
-        if entry.get("dockerRepository", "").endswith(f"/{connector_name}"):
-            connector_entry = entry
-            break
-
-    docs_urls = []
-    if connector_entry and "externalDocumentationUrls" in connector_entry:
-        external_docs = connector_entry["externalDocumentationUrls"]
-        if isinstance(external_docs, list):
-            docs_urls.extend(
-                [
-                    ApiDocsUrl(
-                        title=doc["title"],
-                        url=doc["url"],
-                        source="registry_external_docs",
-                        doc_type=doc.get("type", "other"),
-                        requires_login=doc.get("requiresLogin", False),
-                    )
-                    for doc in external_docs
-                ]
-            )
-
-    return docs_urls
-
-
 @mcp_tool(
     domain="registry",
     read_only=True,
@@ -315,47 +196,12 @@ def get_api_docs_urls(
     This tool retrieves documentation URLs for a connector's upstream API from multiple sources:
     - Registry metadata (documentationUrl, externalDocumentationUrls)
     - Connector manifest.yaml file (data.externalDocumentationUrls)
-
-    Returns:
-        List of ApiDocsUrl objects with documentation URLs, or error message if connector not found.
     """
-    available_connectors = get_available_connectors()
-
-    if connector_name not in available_connectors:
+    try:
+        return get_connector_api_docs_urls(connector_name)
+    except exc.AirbyteConnectorNotRegisteredError:
         return "Connector not found."
 
-    docs_urls: list[ApiDocsUrl] = []
-
-    connector = None
-    with contextlib.suppress(Exception):
-        connector = get_source(
-            connector_name,
-            docker_image=is_docker_installed() or False,
-            install_if_missing=False,
-        )
-
-    if connector and connector.docs_url:
-        docs_urls.append(
-            ApiDocsUrl(title="Airbyte Documentation", url=connector.docs_url, source="registry")
-        )
-
-    registry_urls = _extract_docs_from_registry(connector_name)
-    docs_urls.extend(registry_urls)
-
-    manifest_url = _manifest_url_for(connector_name)
-    manifest_data = _fetch_manifest_dict(manifest_url)
-    manifest_urls = ApiDocsUrl.from_manifest_dict(manifest_data)
-    docs_urls.extend(manifest_urls)
-
-    seen_urls = set()
-    unique_docs_urls = []
-    for doc_url in docs_urls:
-        if doc_url.url not in seen_urls:
-            seen_urls.add(doc_url.url)
-            unique_docs_urls.append(doc_url)
-
-    return unique_docs_urls
-
 
 @mcp_tool(
     domain="registry",
diff --git a/airbyte/registry.py b/airbyte/registry.py
index 776f4107c..adff22755 100644
--- a/airbyte/registry.py
+++ b/airbyte/registry.py
@@ -10,10 +10,12 @@
 from copy import copy
 from enum import Enum
 from pathlib import Path
-from typing import cast
+from typing import Any, cast
 
 import requests
+import yaml
 from pydantic import BaseModel, Field
+from typing_extensions import Self
 
 from airbyte import exceptions as exc
 from airbyte._registry_utils import fetch_registry_version_date, parse_changelog_html
@@ -38,6 +40,10 @@
 _PYTHON_LANGUAGE_TAG = f"language:{_PYTHON_LANGUAGE}"
 _MANIFEST_ONLY_TAG = f"language:{_MANIFEST_ONLY_LANGUAGE}"
 
+_DEFAULT_MANIFEST_URL = (
+    "https://connectors.airbyte.com/files/metadata/airbyte/{source_name}/{version}/manifest.yaml"
+)
+
 
 class InstallType(str, Enum):
     """The type of installation for a connector."""
@@ -294,6 +300,180 @@ class ConnectorVersionInfo(BaseModel):
     parsing_errors: list[str] = Field(default_factory=list)
 
 
+class ApiDocsUrl(BaseModel):
+    """API documentation URL information."""
+
+    title: str
+    url: str
+    source: str
+    doc_type: str = Field(default="other", alias="type")
+    requires_login: bool = Field(default=False, alias="requiresLogin")
+
+    model_config = {"populate_by_name": True}
+
+    @classmethod
+    def from_manifest_dict(cls, manifest_data: dict[str, Any]) -> list[Self]:
+        """Extract documentation URLs from parsed manifest data.
+
+        Args:
+            manifest_data: The parsed manifest.yaml data as a dictionary
+
+        Returns:
+            List of ApiDocsUrl objects extracted from the manifest
+        """
+        results: list[Self] = []
+
+        data_section = manifest_data.get("data")
+        if isinstance(data_section, dict):
+            external_docs = data_section.get("externalDocumentationUrls")
+            if isinstance(external_docs, list):
+                results = [
+                    cls(
+                        title=doc["title"],
+                        url=doc["url"],
+                        source="data_external_docs",
+                        doc_type=doc.get("type", "other"),
+                        requires_login=doc.get("requiresLogin", False),
+                    )
+                    for doc in external_docs
+                ]
+
+        return results
+
+
+def _manifest_url_for(connector_name: str) -> str:
+    """Get the expected URL of the manifest.yaml file for a connector.
+
+    Args:
+        connector_name: The canonical connector name (e.g., "source-facebook-marketing")
+
+    Returns:
+        The URL to the connector's manifest.yaml file
+    """
+    return _DEFAULT_MANIFEST_URL.format(
+        source_name=connector_name,
+        version="latest",
+    )
+
+
+def _fetch_manifest_dict(url: str) -> dict[str, Any]:
+    """Fetch and parse a manifest.yaml file from a URL.
+
+    Args:
+        url: The URL to fetch the manifest from
+
+    Returns:
+        The parsed manifest data as a dictionary, or empty dict if manifest not found (404)
+
+    Raises:
+        HTTPError: If the request fails with a non-404 status code
+    """
+    http_not_found = 404
+
+    response = requests.get(url, timeout=10)
+    if response.status_code == http_not_found:
+        return {}
+
+    response.raise_for_status()
+    return yaml.safe_load(response.text) or {}
+
+
+def _extract_docs_from_registry(connector_name: str) -> list[ApiDocsUrl]:
+    """Extract documentation URLs from connector registry metadata.
+
+    Args:
+        connector_name: The canonical connector name (e.g., "source-facebook-marketing")
+
+    Returns:
+        List of ApiDocsUrl objects extracted from the registry
+    """
+    registry_url = _get_registry_url()
+    response = requests.get(registry_url, timeout=10)
+    response.raise_for_status()
+    registry_data = response.json()
+
+    connector_list = registry_data.get("sources", []) + registry_data.get("destinations", [])
+    connector_entry = None
+    for entry in connector_list:
+        if entry.get("dockerRepository", "").endswith(f"/{connector_name}"):
+            connector_entry = entry
+            break
+
+    docs_urls = []
+
+    if connector_entry and "documentationUrl" in connector_entry:
+        docs_urls.append(
+            ApiDocsUrl(
+                title="Airbyte Documentation",
+                url=connector_entry["documentationUrl"],
+                source="registry",
+            )
+        )
+
+    if connector_entry and "externalDocumentationUrls" in connector_entry:
+        external_docs = connector_entry["externalDocumentationUrls"]
+        if isinstance(external_docs, list):
+            docs_urls.extend(
+                [
+                    ApiDocsUrl(
+                        title=doc["title"],
+                        url=doc["url"],
+                        source="registry_external_docs",
+                        doc_type=doc.get("type", "other"),
+                        requires_login=doc.get("requiresLogin", False),
+                    )
+                    for doc in external_docs
+                ]
+            )
+
+    return docs_urls
+
+
+def get_connector_api_docs_urls(connector_name: str) -> list[ApiDocsUrl]:
+    """Get API documentation URLs for a connector.
+
+    This function retrieves documentation URLs for a connector's upstream API from multiple sources:
+    - Registry metadata (documentationUrl, externalDocumentationUrls)
+    - Connector manifest.yaml file (data.externalDocumentationUrls)
+
+    Args:
+        connector_name: The canonical connector name (e.g., "source-facebook-marketing")
+
+    Returns:
+        List of ApiDocsUrl objects with documentation URLs, deduplicated by URL.
+
+    Raises:
+        AirbyteConnectorNotRegisteredError: If the connector is not found in the registry.
+    """
+    if connector_name not in get_available_connectors(InstallType.DOCKER):
+        raise exc.AirbyteConnectorNotRegisteredError(
+            connector_name=connector_name,
+            context={
+                "registry_url": _get_registry_url(),
+                "available_connectors": get_available_connectors(InstallType.DOCKER),
+            },
+        )
+
+    docs_urls: list[ApiDocsUrl] = []
+
+    registry_urls = _extract_docs_from_registry(connector_name)
+    docs_urls.extend(registry_urls)
+
+    manifest_url = _manifest_url_for(connector_name)
+    manifest_data = _fetch_manifest_dict(manifest_url)
+    manifest_urls = ApiDocsUrl.from_manifest_dict(manifest_data)
+    docs_urls.extend(manifest_urls)
+
+    seen_urls = set()
+    unique_docs_urls = []
+    for doc_url in docs_urls:
+        if doc_url.url not in seen_urls:
+            seen_urls.add(doc_url.url)
+            unique_docs_urls.append(doc_url)
+
+    return unique_docs_urls
+
+
 def get_connector_version_history(
     connector_name: str,
     *,
diff --git a/tests/unit_tests/test_mcp_connector_registry.py b/tests/unit_tests/test_mcp_connector_registry.py
index b45b9d2d1..5f70ea167 100644
--- a/tests/unit_tests/test_mcp_connector_registry.py
+++ b/tests/unit_tests/test_mcp_connector_registry.py
@@ -5,12 +5,12 @@
 
 from unittest.mock import MagicMock, patch
 
-
-from airbyte.mcp.connector_registry import (
+from airbyte import exceptions as exc
+from airbyte.mcp.connector_registry import get_api_docs_urls
+from airbyte.registry import (
     ApiDocsUrl,
     _fetch_manifest_dict,
     _manifest_url_for,
-    get_api_docs_urls,
 )
 
 
@@ -30,7 +30,7 @@ class TestFetchManifestDict:
 
     def test_manifest_not_found(self) -> None:
         """Test handling when manifest.yaml doesn't exist (404)."""
-        with patch("airbyte.mcp.connector_registry.requests.get") as mock_get:
+        with patch("airbyte.registry.requests.get") as mock_get:
             mock_response = MagicMock()
             mock_response.status_code = 404
             mock_get.return_value = mock_response
@@ -46,7 +46,7 @@ def test_fetch_manifest_dict(self) -> None:
 data:
   name: Example
 """
-        with patch("airbyte.mcp.connector_registry.requests.get") as mock_get:
+        with patch("airbyte.registry.requests.get") as mock_get:
             mock_response = MagicMock()
             mock_response.status_code = 200
             mock_response.text = manifest_yaml
@@ -133,49 +133,31 @@ class TestGetApiDocsUrls:
     def test_connector_not_found(self) -> None:
         """Test handling when connector is not found."""
         with patch(
-            "airbyte.mcp.connector_registry.get_available_connectors"
-        ) as mock_get:
-            mock_get.return_value = ["source-faker", "source-facebook-marketing"]
+            "airbyte.mcp.connector_registry.get_connector_api_docs_urls"
+        ) as mock_get_docs:
+            mock_get_docs.side_effect = exc.AirbyteConnectorNotRegisteredError(
+                connector_name="nonexistent-connector",
+                context={},
+            )
 
             result = get_api_docs_urls("nonexistent-connector")
             assert result == "Connector not found."
 
     def test_deduplication_of_urls(self) -> None:
         """Test that duplicate URLs are deduplicated."""
-        with (
-            patch(
-                "airbyte.mcp.connector_registry.get_available_connectors"
-            ) as mock_get,
-            patch("airbyte.mcp.connector_registry.get_source") as mock_source,
-            patch(
-                "airbyte.mcp.connector_registry._fetch_manifest_dict"
-            ) as mock_fetch_dict,
-            patch(
-                "airbyte.mcp.connector_registry._extract_docs_from_registry"
-            ) as mock_registry,
-        ):
-            mock_get.return_value = ["source-example", "source-faker"]
-
-            mock_connector = MagicMock()
-            mock_connector.docs_url = (
-                "https://docs.airbyte.com/integrations/sources/example"
-            )
-            mock_source.return_value = mock_connector
-
-            mock_registry.return_value = []
-
-            mock_fetch_dict.return_value = {
-                "data": {
-                    "externalDocumentationUrls": [
-                        {
-                            "title": "Airbyte Documentation",
-                            "url": "https://docs.airbyte.com/integrations/sources/example",
-                        }
-                    ]
-                }
-            }
+        with patch(
+            "airbyte.mcp.connector_registry.get_connector_api_docs_urls"
+        ) as mock_get_docs:
+            mock_get_docs.return_value = [
+                ApiDocsUrl(
+                    title="Airbyte Documentation",
+                    url="https://docs.airbyte.com/integrations/sources/example",
+                    source="registry",
+                )
+            ]
 
             result = get_api_docs_urls("source-example")
 
             assert isinstance(result, list)
             assert len(result) == 1
+            assert result[0].title == "Airbyte Documentation"