airbyte.mcp.registry
Airbyte connector registry MCP operations.
registry module
MCP primitives registered by the registry module of the airbyte-mcp server: 4 tool(s), 0 prompt(s), 0 resource(s).
Tools (4)
get_api_docs_urls
Hints: read-only · idempotent
Get API documentation URLs for a connector.
This tool retrieves documentation URLs for a connector's upstream API from multiple sources:
- Registry metadata (documentationUrl, externalDocumentationUrls)
- Connector manifest.yaml file (data.externalDocumentationUrls)
Parameters
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
connector_name |
string |
yes | — | The canonical connector name (e.g., 'source-facebook-marketing', 'destination-snowflake') |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"connector_name": {
"description": "The canonical connector name (e.g., 'source-facebook-marketing', 'destination-snowflake')",
"type": "string"
}
},
"required": [
"connector_name"
],
"type": "object"
}
Show output JSON schema
{
"properties": {
"result": {
"anyOf": [
{
"items": {
"description": "API documentation URL information.",
"properties": {
"title": {
"type": "string"
},
"url": {
"type": "string"
},
"source": {
"type": "string"
},
"type": {
"default": "other",
"type": "string"
},
"requiresLogin": {
"default": false,
"type": "boolean"
}
},
"required": [
"title",
"url",
"source"
],
"type": "object"
},
"type": "array"
},
{
"const": "Connector not found.",
"type": "string"
}
]
}
},
"required": [
"result"
],
"type": "object",
"x-fastmcp-wrap-result": true
}
get_connector_info
Hints: read-only · idempotent
Get the documentation URL for a connector.
Parameters
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
connector_name |
string |
yes | — | The name of the connector to get information for. |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"connector_name": {
"description": "The name of the connector to get information for.",
"type": "string"
}
},
"required": [
"connector_name"
],
"type": "object"
}
Show output JSON schema
{
"properties": {
"result": {
"anyOf": [
{
"description": "@private Class to hold connector information.",
"properties": {
"connector_name": {
"type": "string"
},
"connector_metadata": {
"anyOf": [
{
"description": "Metadata for a connector.",
"properties": {
"name": {
"type": "string"
},
"latest_available_version": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
]
},
"pypi_package_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
]
},
"language": {
"anyOf": [
{
"description": "The language of a connector.",
"enum": [
"python",
"java",
"manifest-only"
],
"type": "string"
},
{
"type": "null"
}
]
},
"install_types": {
"items": {
"description": "The type of installation for a connector.",
"enum": [
"yaml",
"python",
"docker",
"java",
"installable",
"any"
],
"type": "string"
},
"type": "array",
"uniqueItems": true
},
"suggested_streams": {
"anyOf": [
{
"items": {
"type": "string"
},
"type": "array"
},
{
"type": "null"
}
],
"default": null
}
},
"required": [
"name",
"latest_available_version",
"pypi_package_name",
"language",
"install_types"
],
"type": "object"
},
{
"type": "null"
}
],
"default": null
},
"documentation_url": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null
},
"config_spec_jsonschema": {
"anyOf": [
{
"additionalProperties": true,
"type": "object"
},
{
"type": "null"
}
],
"default": null
},
"manifest_url": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null
}
},
"required": [
"connector_name"
],
"type": "object"
},
{
"const": "Connector not found.",
"type": "string"
}
]
}
},
"required": [
"result"
],
"type": "object",
"x-fastmcp-wrap-result": true
}
get_connector_version_history
Hints: read-only · idempotent
Get version history for a connector.
This tool retrieves the version history for a connector, including:
- Version number
- Release date (from changelog, with registry override for recent versions)
- DockerHub URL for the version
- Changelog URL
- PR URL and title (scraped from changelog)
For the most recent N versions (default 5), release dates are fetched from the registry for accuracy. For older versions, changelog dates are used.
Returns:
List of version information, sorted by most recent first.
Parameters
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
connector_name |
string |
yes | — | The name of the connector (e.g., 'source-faker', 'destination-postgres') |
num_versions_to_validate |
integer |
no | 5 |
Number of most recent versions to validate with registry data for accurate release dates. Defaults to 5. |
limit |
integer | null | null |
no | null |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"connector_name": {
"description": "The name of the connector (e.g., 'source-faker', 'destination-postgres')",
"type": "string"
},
"num_versions_to_validate": {
"default": 5,
"description": "Number of most recent versions to validate with registry data for accurate release dates. Defaults to 5.",
"type": "integer"
},
"limit": {
"anyOf": [
{
"anyOf": [
{
"type": "integer"
},
{
"type": "null"
}
],
"default": null,
"description": "DEPRECATED: Use num_versions_to_validate instead. Maximum number of versions to return (most recent first). If specified, only the first N versions will be returned."
},
{
"type": "null"
}
],
"default": null
}
},
"required": [
"connector_name"
],
"type": "object"
}
Show output JSON schema
{
"properties": {
"result": {
"anyOf": [
{
"items": {
"description": "Information about a specific connector version.",
"properties": {
"version": {
"type": "string"
},
"release_date": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null
},
"docker_image_url": {
"type": "string"
},
"changelog_url": {
"type": "string"
},
"pr_url": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null
},
"pr_title": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null
},
"parsing_errors": {
"items": {
"type": "string"
},
"type": "array"
}
},
"required": [
"version",
"docker_image_url",
"changelog_url"
],
"type": "object"
},
"type": "array"
},
{
"enum": [
"Connector not found.",
"Failed to fetch changelog."
],
"type": "string"
}
]
}
},
"required": [
"result"
],
"type": "object",
"x-fastmcp-wrap-result": true
}
list_connectors
Hints: read-only · idempotent
List available Airbyte connectors with optional filtering.
Returns:
List of connector names.
Parameters
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
keyword_filter |
string | null |
no | null |
Filter connectors by keyword. |
connector_type_filter |
enum("source", "destination") | null |
no | null |
Filter connectors by type ('source' or 'destination'). |
install_types |
enum("java", "python", "yaml", "docker") | array<enum("java", "python", "yaml", "docker")> | null |
no | null |
Filter connectors by install type. These are not mutually exclusive: - "python": Connectors that can be installed as Python packages. - "yaml": Connectors that can be installed simply via YAML download. These connectors are the fastest to install and run, as they do not require any additional dependencies. - "java": Connectors that can only be installed via Java. Since PyAirbyte does not currently ship with a JVM, these connectors will be run via Docker instead. In environments where Docker is not available, these connectors may not be runnable. - "docker": Connectors that can be installed via Docker. Note that all connectors can be run in Docker, so this filter should generally return the same results as not specifying a filter. If no install types are specified, all connectors will be returned. |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"keyword_filter": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Filter connectors by keyword."
},
"connector_type_filter": {
"anyOf": [
{
"enum": [
"source",
"destination"
],
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Filter connectors by type ('source' or 'destination')."
},
"install_types": {
"anyOf": [
{
"enum": [
"java",
"python",
"yaml",
"docker"
],
"type": "string"
},
{
"items": {
"enum": [
"java",
"python",
"yaml",
"docker"
],
"type": "string"
},
"type": "array"
},
{
"type": "null"
}
],
"default": null,
"description": "\n Filter connectors by install type.\n These are not mutually exclusive:\n - \"python\": Connectors that can be installed as Python packages.\n - \"yaml\": Connectors that can be installed simply via YAML download.\n These connectors are the fastest to install and run, as they do not require any\n additional dependencies.\n - \"java\": Connectors that can only be installed via Java. Since PyAirbyte does not\n currently ship with a JVM, these connectors will be run via Docker instead.\n In environments where Docker is not available, these connectors may not be\n runnable.\n - \"docker\": Connectors that can be installed via Docker. Note that all connectors\n can be run in Docker, so this filter should generally return the same results as\n not specifying a filter.\n If no install types are specified, all connectors will be returned.\n "
}
},
"type": "object"
}
Show output JSON schema
{
"properties": {
"result": {
"items": {
"type": "string"
},
"type": "array"
}
},
"required": [
"result"
],
"type": "object",
"x-fastmcp-wrap-result": true
}
1# Copyright (c) 2024 Airbyte, Inc., all rights reserved. 2"""Airbyte connector registry MCP operations. 3 4.. include:: ../../docs/mcp-generated/registry.md 5""" 6 7# No public Python API — MCP primitives are registered via decorators and 8# documented via the generated Markdown include above. Setting `__all__` to an 9# empty list tells pdoc (and other doc tools) not to surface the individual 10# tool / helper definitions as a redundant "API Documentation" list. 11__all__: list[str] = [] 12 13# Note: Deferred type evaluation must be avoided due to FastMCP/Pydantic needing 14# types to be available at import time for tool registration. 15import contextlib 16import logging 17from typing import Annotated, Any, Literal 18 19import requests 20from fastmcp import FastMCP 21from fastmcp_extensions import mcp_tool, register_mcp_tools 22from pydantic import BaseModel, Field 23 24from airbyte import exceptions as exc 25from airbyte._util.meta import is_docker_installed 26from airbyte.mcp._arg_resolvers import resolve_list_of_strings 27from airbyte.registry import ( 28 _DEFAULT_MANIFEST_URL, 29 ApiDocsUrl, 30 ConnectorMetadata, 31 ConnectorVersionInfo, 32 InstallType, 33 get_available_connectors, 34 get_connector_api_docs_urls, 35 get_connector_metadata, 36) 37from airbyte.registry import get_connector_version_history as _get_connector_version_history 38from airbyte.sources.util import get_source 39 40 41logger = logging.getLogger("airbyte.mcp") 42 43 44@mcp_tool( 45 read_only=True, 46 idempotent=True, 47) 48def list_connectors( 49 keyword_filter: Annotated[ 50 str | None, 51 Field( 52 description="Filter connectors by keyword.", 53 default=None, 54 ), 55 ], 56 connector_type_filter: Annotated[ 57 Literal["source", "destination"] | None, 58 Field( 59 description="Filter connectors by type ('source' or 'destination').", 60 default=None, 61 ), 62 ], 63 install_types: Annotated[ 64 Literal["java", "python", "yaml", "docker"] 65 | list[Literal["java", "python", "yaml", "docker"]] 66 | None, 67 Field( 68 description=( 69 """ 70 Filter connectors by install type. 71 These are not mutually exclusive: 72 - "python": Connectors that can be installed as Python packages. 73 - "yaml": Connectors that can be installed simply via YAML download. 74 These connectors are the fastest to install and run, as they do not require any 75 additional dependencies. 76 - "java": Connectors that can only be installed via Java. Since PyAirbyte does not 77 currently ship with a JVM, these connectors will be run via Docker instead. 78 In environments where Docker is not available, these connectors may not be 79 runnable. 80 - "docker": Connectors that can be installed via Docker. Note that all connectors 81 can be run in Docker, so this filter should generally return the same results as 82 not specifying a filter. 83 If no install types are specified, all connectors will be returned. 84 """ 85 ), 86 default=None, 87 ), 88 ], 89) -> list[str]: 90 """List available Airbyte connectors with optional filtering. 91 92 Returns: 93 List of connector names. 94 """ 95 # Start with the full list of known connectors (all support Docker): 96 connectors: list[str] = get_available_connectors(install_type=InstallType.ANY) 97 98 install_types_list: list[str] | None = resolve_list_of_strings( 99 install_types, # type: ignore[arg-type] # Type check doesn't understand literal is str 100 ) 101 102 if install_types_list: 103 # If install_types is provided, filter connectors based on the specified install types. 104 connectors = [ 105 connector 106 for connector in connectors 107 if any( 108 connector in get_available_connectors(install_type=install_type) 109 for install_type in install_types_list 110 ) 111 ] 112 113 if keyword_filter: 114 # Filter connectors by keyword, case-insensitive. 115 connectors = [ 116 connector for connector in connectors if keyword_filter.lower() in connector.lower() 117 ] 118 119 if connector_type_filter: 120 # Filter connectors by type ('source' or 'destination'). 121 # This assumes connector names are prefixed with 'source-' or 'destination-'. 122 connectors = [ 123 connector 124 for connector in connectors 125 if connector.startswith(f"{connector_type_filter}-") 126 ] 127 128 return sorted(connectors) 129 130 131class ConnectorInfo(BaseModel): 132 """@private Class to hold connector information.""" 133 134 connector_name: str 135 connector_metadata: ConnectorMetadata | None = None 136 documentation_url: str | None = None 137 config_spec_jsonschema: dict | None = None 138 manifest_url: str | None = None 139 140 141@mcp_tool( 142 read_only=True, 143 idempotent=True, 144) 145def get_connector_info( 146 connector_name: Annotated[ 147 str, 148 Field(description="The name of the connector to get information for."), 149 ], 150) -> ConnectorInfo | Literal["Connector not found."]: 151 """Get the documentation URL for a connector.""" 152 if connector_name not in get_available_connectors(): 153 return "Connector not found." 154 155 connector = get_source( 156 connector_name, 157 docker_image=is_docker_installed() or False, 158 install_if_missing=False, # Defer to avoid failing entirely if it can't be installed. 159 ) 160 161 connector_metadata: ConnectorMetadata | None = None 162 with contextlib.suppress(Exception): 163 connector_metadata = get_connector_metadata(connector_name) 164 165 config_spec_jsonschema: dict[str, Any] | None = None 166 with contextlib.suppress(Exception): 167 # This requires running the connector. Install it if it isn't already installed. 168 connector.install() 169 config_spec_jsonschema = connector.config_spec 170 171 manifest_url = _DEFAULT_MANIFEST_URL.format( 172 source_name=connector_name, 173 version="latest", 174 ) 175 176 return ConnectorInfo( 177 connector_name=connector.name, 178 connector_metadata=connector_metadata, 179 documentation_url=connector.docs_url, 180 config_spec_jsonschema=config_spec_jsonschema, 181 manifest_url=manifest_url, 182 ) 183 184 185@mcp_tool( 186 read_only=True, 187 idempotent=True, 188) 189def get_api_docs_urls( 190 connector_name: Annotated[ 191 str, 192 Field( 193 description=( 194 "The canonical connector name " 195 "(e.g., 'source-facebook-marketing', 'destination-snowflake')" 196 ) 197 ), 198 ], 199) -> list[ApiDocsUrl] | Literal["Connector not found."]: 200 """Get API documentation URLs for a connector. 201 202 This tool retrieves documentation URLs for a connector's upstream API from multiple sources: 203 - Registry metadata (documentationUrl, externalDocumentationUrls) 204 - Connector manifest.yaml file (data.externalDocumentationUrls) 205 """ 206 try: 207 return get_connector_api_docs_urls(connector_name) 208 except exc.AirbyteConnectorNotRegisteredError: 209 return "Connector not found." 210 211 212@mcp_tool( 213 read_only=True, 214 idempotent=True, 215) 216def get_connector_version_history( 217 connector_name: Annotated[ 218 str, 219 Field( 220 description="The name of the connector (e.g., 'source-faker', 'destination-postgres')" 221 ), 222 ], 223 num_versions_to_validate: Annotated[ 224 int, 225 Field( 226 description=( 227 "Number of most recent versions to validate with registry data for accurate " 228 "release dates. Defaults to 5." 229 ), 230 default=5, 231 ), 232 ] = 5, 233 limit: Annotated[ 234 int | None, 235 Field( 236 description=( 237 "DEPRECATED: Use num_versions_to_validate instead. " 238 "Maximum number of versions to return (most recent first). " 239 "If specified, only the first N versions will be returned." 240 ), 241 default=None, 242 ), 243 ] = None, 244) -> list[ConnectorVersionInfo] | Literal["Connector not found.", "Failed to fetch changelog."]: 245 """Get version history for a connector. 246 247 This tool retrieves the version history for a connector, including: 248 - Version number 249 - Release date (from changelog, with registry override for recent versions) 250 - DockerHub URL for the version 251 - Changelog URL 252 - PR URL and title (scraped from changelog) 253 254 For the most recent N versions (default 5), release dates are fetched from the 255 registry for accuracy. For older versions, changelog dates are used. 256 257 Returns: 258 List of version information, sorted by most recent first. 259 """ 260 try: 261 versions = _get_connector_version_history( 262 connector_name=connector_name, 263 num_versions_to_validate=num_versions_to_validate, 264 ) 265 except exc.AirbyteConnectorNotRegisteredError: 266 return "Connector not found." 267 except requests.exceptions.RequestException: 268 logger.exception(f"Failed to fetch changelog for {connector_name}") 269 return "Failed to fetch changelog." 270 else: 271 if limit is not None and limit > 0: 272 return versions[:limit] 273 return versions 274 275 276def register_registry_tools(app: FastMCP) -> None: 277 """Register registry tools with the FastMCP app. 278 279 Args: 280 app: FastMCP application instance 281 """ 282 register_mcp_tools(app, mcp_module=__name__)