fix(mcp): resolve get_connector_info spec from registry instead of installing

airbyte/mcp/registry.py

@@ -22,7 +22,7 @@
 from pydantic import BaseModel, Field
 
 from airbyte import exceptions as exc
-from airbyte._util.meta import is_docker_installed
+from airbyte._util.registry_spec import get_connector_spec_from_registry
 from airbyte.mcp._arg_resolvers import resolve_list_of_strings
 from airbyte.registry import (
     _DEFAULT_MANIFEST_URL,
@@ -148,25 +148,40 @@ def get_connector_info(
         Field(description="The name of the connector to get information for."),
     ],
 ) -> ConnectorInfo | Literal["Connector not found."]:
-    """Get the documentation URL for a connector."""
+    """Get metadata, documentation URL, config spec, and manifest URL for a connector.
+
+    `config_spec_jsonschema` is fetched from the public connector registry over
+    HTTP (no Docker or local install required), preferring the `cloud` spec and
+    falling back to `oss`. It is `None` when the registry has no spec available
+    for the connector.
+    """
     if connector_name not in get_available_connectors():
         return "Connector not found."
 
     connector = get_source(
         connector_name,
-        docker_image=is_docker_installed() or False,
         install_if_missing=False,  # Defer to avoid failing entirely if it can't be installed.
     )
 
     connector_metadata: ConnectorMetadata | None = None
     with contextlib.suppress(Exception):
         connector_metadata = get_connector_metadata(connector_name)
 
-    config_spec_jsonschema: dict[str, Any] | None = None
-    with contextlib.suppress(Exception):
-        # This requires running the connector. Install it if it isn't already installed.
-        connector.install()
-        config_spec_jsonschema = connector.config_spec
+    # Resolve the config spec from the public registry endpoint keyed by version,
+    # which works in any runtime (hosted or local) without installing the
+    # connector. Prefer the `cloud` spec and fall back to `oss`.
+    version = connector_metadata.latest_available_version if connector_metadata else None
+    config_spec_jsonschema: dict[str, Any] | None = get_connector_spec_from_registry(
+        connector_name,
+        version=version,
+        platform="cloud",
+    )
+    if config_spec_jsonschema is None:
+        config_spec_jsonschema = get_connector_spec_from_registry(
+            connector_name,
+            version=version,
+            platform="oss",
+        )
 
     manifest_url = _DEFAULT_MANIFEST_URL.format(
         source_name=connector_name,

tests/unit_tests/test_mcp_connector_registry.py

@@ -4,7 +4,7 @@
 from __future__ import annotations
 
 import asyncio
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock, call, patch
 
 import pytest
 from mcp.types import TextContent
@@ -17,7 +17,7 @@
     _list_public_registry_connectors,
 )
 from airbyte.mcp.interactive._shared_models import ConnectorType, SupportLevel
-from airbyte.mcp.registry import get_api_docs_urls
+from airbyte.mcp.registry import get_api_docs_urls, get_connector_info
 from airbyte.registry import (
     ApiDocsUrl,
     ConnectorMetadata,
@@ -569,3 +569,96 @@ def test_prefab_generative_tools_include_airbyte_annotations() -> None:
     assert tool.annotations is not None
     assert getattr(tool.annotations, "mcp_module") == "interactive"
     assert getattr(tool.annotations, INTERACTIVE_UI_ANNOTATION) is True
+
+
+def test_get_connector_info_resolves_spec_from_registry_without_docker() -> None:
+    """`get_connector_info` must resolve the config spec over HTTP, never via install.
+
+    The spec is fetched from the public connector registry keyed by version, so it
+    works in a hosted, no-Docker runtime without the unbounded `connector.install()`
+    fallback that previously hung requests for minutes.
+    """
+    connector = MagicMock()
+    connector.name = "source-faker"
+    connector.docs_url = "https://docs.airbyte.com/integrations/sources/faker"
+
+    cloud_spec = {"type": "object", "properties": {"count": {"type": "integer"}}}
+
+    with (
+        patch(
+            "airbyte.mcp.registry.get_available_connectors",
+            return_value=["source-faker"],
+        ),
+        patch("airbyte.mcp.registry.get_source", return_value=connector),
+        patch("airbyte.mcp.registry.get_connector_metadata", return_value=None),
+        patch(
+            "airbyte.mcp.registry.get_connector_spec_from_registry",
+            return_value=cloud_spec,
+        ) as mock_get_spec,
+    ):
+        result = get_connector_info("source-faker")
+
+    assert not isinstance(result, str)
+    assert result.config_spec_jsonschema == cloud_spec
+    assert result.manifest_url is not None
+    connector.install.assert_not_called()
+    mock_get_spec.assert_called_once_with(
+        "source-faker", version=None, platform="cloud"
+    )
+
+
+def test_get_connector_info_falls_back_to_oss_spec() -> None:
+    """When the `cloud` spec is unavailable, `get_connector_info` uses the `oss` spec."""
+    connector = MagicMock()
+    connector.name = "source-faker"
+    connector.docs_url = "https://docs.airbyte.com/integrations/sources/faker"
+
+    oss_spec = {"type": "object", "properties": {"seed": {"type": "integer"}}}
+
+    with (
+        patch(
+            "airbyte.mcp.registry.get_available_connectors",
+            return_value=["source-faker"],
+        ),
+        patch("airbyte.mcp.registry.get_source", return_value=connector),
+        patch("airbyte.mcp.registry.get_connector_metadata", return_value=None),
+        patch(
+            "airbyte.mcp.registry.get_connector_spec_from_registry",
+            side_effect=[None, oss_spec],
+        ) as mock_get_spec,
+    ):
+        result = get_connector_info("source-faker")
+
+    assert not isinstance(result, str)
+    assert result.config_spec_jsonschema == oss_spec
+    connector.install.assert_not_called()
+    assert mock_get_spec.call_args_list == [
+        call("source-faker", version=None, platform="cloud"),
+        call("source-faker", version=None, platform="oss"),
+    ]
+
+
+def test_get_connector_info_spec_none_when_registry_has_no_spec() -> None:
+    """`config_spec_jsonschema` is `None` when the registry has no spec for either platform."""
+    connector = MagicMock()
+    connector.name = "source-faker"
+    connector.docs_url = "https://docs.airbyte.com/integrations/sources/faker"
+
+    with (
+        patch(
+            "airbyte.mcp.registry.get_available_connectors",
+            return_value=["source-faker"],
+        ),
+        patch("airbyte.mcp.registry.get_source", return_value=connector),
+        patch("airbyte.mcp.registry.get_connector_metadata", return_value=None),
+        patch(
+            "airbyte.mcp.registry.get_connector_spec_from_registry",
+            return_value=None,
+        ),
+    ):
+        result = get_connector_info("source-faker")
+
+    assert not isinstance(result, str)
+    assert result.config_spec_jsonschema is None
+    assert result.manifest_url is not None
+    connector.install.assert_not_called()