airbyte.mcp.connector_registry

  1# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
  2"""Airbyte Cloud MCP operations."""
  3
  4# Note: Deferred type evaluation must be avoided due to FastMCP/Pydantic needing
  5# types to be available at import time for tool registration.
  6import contextlib
  7from typing import Annotated, Any, Literal
  8
  9from fastmcp import FastMCP
 10from pydantic import BaseModel, Field
 11
 12from airbyte._executors.util import DEFAULT_MANIFEST_URL
 13from airbyte._util.meta import is_docker_installed
 14from airbyte.mcp._tool_utils import mcp_tool, register_tools
 15from airbyte.mcp._util import resolve_list_of_strings
 16from airbyte.sources import get_available_connectors
 17from airbyte.sources.registry import ConnectorMetadata, get_connector_metadata
 18from airbyte.sources.util import get_source
 19
 20
 21@mcp_tool(
 22    domain="registry",
 23    read_only=True,
 24    idempotent=True,
 25)
 26def list_connectors(
 27    keyword_filter: Annotated[
 28        str | None,
 29        Field(
 30            description="Filter connectors by keyword.",
 31            default=None,
 32        ),
 33    ],
 34    connector_type_filter: Annotated[
 35        Literal["source", "destination"] | None,
 36        Field(
 37            description="Filter connectors by type ('source' or 'destination').",
 38            default=None,
 39        ),
 40    ],
 41    install_types: Annotated[
 42        Literal["java", "python", "yaml", "docker"]
 43        | list[Literal["java", "python", "yaml", "docker"]]
 44        | None,
 45        Field(
 46            description=(
 47                """
 48                Filter connectors by install type.
 49                These are not mutually exclusive:
 50                - "python": Connectors that can be installed as Python packages.
 51                - "yaml": Connectors that can be installed simply via YAML download.
 52                    These connectors are the fastest to install and run, as they do not require any
 53                    additional dependencies.
 54                - "java": Connectors that can only be installed via Java. Since PyAirbyte does not
 55                    currently ship with a JVM, these connectors will be run via Docker instead.
 56                    In environments where Docker is not available, these connectors may not be
 57                    runnable.
 58                - "docker": Connectors that can be installed via Docker. Note that all connectors
 59                    can be run in Docker, so this filter should generally return the same results as
 60                    not specifying a filter.
 61                If no install types are specified, all connectors will be returned.
 62                """
 63            ),
 64            default=None,
 65        ),
 66    ],
 67) -> list[str]:
 68    """List available Airbyte connectors with optional filtering.
 69
 70    Returns:
 71        List of connector names.
 72    """
 73    connectors: list[str] = get_available_connectors()
 74
 75    install_types_list: list[str] | None = resolve_list_of_strings(
 76        install_types,  # type: ignore[arg-type]  # Type check doesn't understand literal is str
 77    )
 78
 79    if install_types_list:
 80        # If install_types is provided, filter connectors based on the specified install types.
 81        connectors = [
 82            connector
 83            for connector in connectors
 84            if any(
 85                connector in get_available_connectors(install_type=install_type)
 86                for install_type in install_types_list
 87            )
 88        ]
 89
 90    if keyword_filter:
 91        # Filter connectors by keyword, case-insensitive.
 92        connectors = [
 93            connector for connector in connectors if keyword_filter.lower() in connector.lower()
 94        ]
 95
 96    if connector_type_filter:
 97        # Filter connectors by type ('source' or 'destination').
 98        # This assumes connector names are prefixed with 'source-' or 'destination-'.
 99        connectors = [
100            connector
101            for connector in connectors
102            if connector.startswith(f"{connector_type_filter}-")
103        ]
104
105    return sorted(connectors)
106
107
108class ConnectorInfo(BaseModel):
109    """@private Class to hold connector information."""
110
111    connector_name: str
112    connector_metadata: ConnectorMetadata | None = None
113    documentation_url: str | None = None
114    config_spec_jsonschema: dict | None = None
115    manifest_url: str | None = None
116
117
118@mcp_tool(
119    domain="registry",
120    read_only=True,
121    idempotent=True,
122)
123def get_connector_info(
124    connector_name: Annotated[
125        str,
126        Field(description="The name of the connector to get information for."),
127    ],
128) -> ConnectorInfo | Literal["Connector not found."]:
129    """Get the documentation URL for a connector."""
130    if connector_name not in get_available_connectors():
131        return "Connector not found."
132
133    connector = get_source(
134        connector_name,
135        docker_image=is_docker_installed() or False,
136        install_if_missing=False,  # Defer to avoid failing entirely if it can't be installed.
137    )
138
139    connector_metadata: ConnectorMetadata | None = None
140    with contextlib.suppress(Exception):
141        connector_metadata = get_connector_metadata(connector_name)
142
143    config_spec_jsonschema: dict[str, Any] | None = None
144    with contextlib.suppress(Exception):
145        # This requires running the connector. Install it if it isn't already installed.
146        connector.install()
147        config_spec_jsonschema = connector.config_spec
148
149    manifest_url = DEFAULT_MANIFEST_URL.format(
150        source_name=connector_name,
151        version="latest",
152    )
153
154    return ConnectorInfo(
155        connector_name=connector.name,
156        connector_metadata=connector_metadata,
157        documentation_url=connector.docs_url,
158        config_spec_jsonschema=config_spec_jsonschema,
159        manifest_url=manifest_url,
160    )
161
162
163def register_connector_registry_tools(app: FastMCP) -> None:
164    """@private Register tools with the FastMCP app.
165
166    This is an internal function and should not be called directly.
167    """
168    register_tools(app, domain="registry")