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 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.
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"
},
"display_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null
},
"connector_type": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null
},
"definition_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null
},
"docker_repository": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null
},
"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
},
"support_level": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null
},
"release_stage": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null
},
"source_type": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null
},
"documentation_url": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null
},
"release_date": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null
},
"github_issue_label": {
"anyOf": [
{
"type": "string"
},
{
"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.registry_spec import get_connector_spec_from_registry 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 metadata, documentation URL, config spec, and manifest URL for a connector. 152 153 `config_spec_jsonschema` is fetched from the public connector registry over 154 HTTP (no Docker or local install required), preferring the `cloud` spec and 155 falling back to `oss`. It is `None` when the registry has no spec available 156 for the connector. 157 """ 158 if connector_name not in get_available_connectors(): 159 return "Connector not found." 160 161 connector = get_source( 162 connector_name, 163 install_if_missing=False, # Defer to avoid failing entirely if it can't be installed. 164 ) 165 166 connector_metadata: ConnectorMetadata | None = None 167 with contextlib.suppress(Exception): 168 connector_metadata = get_connector_metadata(connector_name) 169 170 # Resolve the config spec from the public registry endpoint keyed by version, 171 # which works in any runtime (hosted or local) without installing the 172 # connector. Prefer the `cloud` spec and fall back to `oss`. 173 version = connector_metadata.latest_available_version if connector_metadata else None 174 config_spec_jsonschema: dict[str, Any] | None = get_connector_spec_from_registry( 175 connector_name, 176 version=version, 177 platform="cloud", 178 ) 179 if config_spec_jsonschema is None: 180 config_spec_jsonschema = get_connector_spec_from_registry( 181 connector_name, 182 version=version, 183 platform="oss", 184 ) 185 186 manifest_url = _DEFAULT_MANIFEST_URL.format( 187 source_name=connector_name, 188 version="latest", 189 ) 190 191 return ConnectorInfo( 192 connector_name=connector.name, 193 connector_metadata=connector_metadata, 194 documentation_url=connector.docs_url, 195 config_spec_jsonschema=config_spec_jsonschema, 196 manifest_url=manifest_url, 197 ) 198 199 200@mcp_tool( 201 read_only=True, 202 idempotent=True, 203) 204def get_api_docs_urls( 205 connector_name: Annotated[ 206 str, 207 Field( 208 description=( 209 "The canonical connector name " 210 "(e.g., 'source-facebook-marketing', 'destination-snowflake')" 211 ) 212 ), 213 ], 214) -> list[ApiDocsUrl] | Literal["Connector not found."]: 215 """Get API documentation URLs for a connector. 216 217 This tool retrieves documentation URLs for a connector's upstream API from multiple sources: 218 - Registry metadata (documentationUrl, externalDocumentationUrls) 219 - Connector manifest.yaml file (data.externalDocumentationUrls) 220 """ 221 try: 222 return get_connector_api_docs_urls(connector_name) 223 except exc.AirbyteConnectorNotRegisteredError: 224 return "Connector not found." 225 226 227@mcp_tool( 228 read_only=True, 229 idempotent=True, 230) 231def get_connector_version_history( 232 connector_name: Annotated[ 233 str, 234 Field( 235 description="The name of the connector (e.g., 'source-faker', 'destination-postgres')" 236 ), 237 ], 238 num_versions_to_validate: Annotated[ 239 int, 240 Field( 241 description=( 242 "Number of most recent versions to validate with registry data for accurate " 243 "release dates. Defaults to 5." 244 ), 245 default=5, 246 ), 247 ] = 5, 248 limit: Annotated[ 249 int | None, 250 Field( 251 description=( 252 "DEPRECATED: Use num_versions_to_validate instead. " 253 "Maximum number of versions to return (most recent first). " 254 "If specified, only the first N versions will be returned." 255 ), 256 default=None, 257 ), 258 ] = None, 259) -> list[ConnectorVersionInfo] | Literal["Connector not found.", "Failed to fetch changelog."]: 260 """Get version history for a connector. 261 262 This tool retrieves the version history for a connector, including: 263 - Version number 264 - Release date (from changelog, with registry override for recent versions) 265 - DockerHub URL for the version 266 - Changelog URL 267 - PR URL and title (scraped from changelog) 268 269 For the most recent N versions (default 5), release dates are fetched from the 270 registry for accuracy. For older versions, changelog dates are used. 271 272 Returns: 273 List of version information, sorted by most recent first. 274 """ 275 try: 276 versions = _get_connector_version_history( 277 connector_name=connector_name, 278 num_versions_to_validate=num_versions_to_validate, 279 ) 280 except exc.AirbyteConnectorNotRegisteredError: 281 return "Connector not found." 282 except requests.exceptions.RequestException: 283 logger.exception(f"Failed to fetch changelog for {connector_name}") 284 return "Failed to fetch changelog." 285 else: 286 if limit is not None and limit > 0: 287 return versions[:limit] 288 return versions 289 290 291def register_registry_tools(app: FastMCP) -> None: 292 """Register registry tools with the FastMCP app. 293 294 Args: 295 app: FastMCP application instance 296 """ 297 register_mcp_tools(app, mcp_module=__name__)