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