airbyte.mcp.server

  1# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
  2"""MCP (Model Context Protocol) server for PyAirbyte connector management.
  3
  4Supports two transport modes:
  5
  6- **stdio** (default): For local MCP clients (Claude Desktop, etc.)
  7- **HTTP**: For hosted deployment. Start via `airbyte-mcp-http` entry point or
  8  `poe mcp-serve-http`. Transport auth is assembled by
  9  `fastmcp_extensions.resolve_mcp_auth`, which supports two client shapes on the
 10  same deployment:
 11    - **Interactive** (humans in a browser): Keycloak Authorization Code + PKCE
 12      via `OIDCProxy`, enabled when `OIDC_CONFIG_URL`, `OIDC_CLIENT_ID`, and
 13      `OIDC_CLIENT_SECRET` are all set.
 14    - **Headless** (agents, CI): the client mints its own short-lived bearer
 15      token via the OAuth 2.0 client credentials grant and sends it as
 16      `Authorization: Bearer <token>`. The server verifies it with a
 17      `JWTVerifier` (no browser, no stored/rotating refresh token), enabled
 18      when `MCP_AUTH_JWKS_URI` (or `MCP_AUTH_JWT_PUBLIC_KEY`) is set.
 19  When both are configured they are combined via `MultiAuth`.
 20
 21For Airbyte Cloud, set `MCP_AUTH_AIRBYTE_CLOUD=true` to verify against Airbyte
 22Cloud's application-client realm without hand-configuring URLs. An agent then
 23mints an Airbyte Cloud access token from its `AIRBYTE_CLOUD_CLIENT_ID` /
 24`AIRBYTE_CLOUD_CLIENT_SECRET` (the `<api_root>/applications/token` endpoint) and
 25sends it as `Authorization: Bearer`. That single token both authenticates
 26transport (verified here) and authorizes downstream Cloud API calls (the same
 27header feeds the Cloud bearer token), because an Airbyte-Cloud-issued JWT is
 28itself a valid Cloud API bearer.
 29"""
 30
 31from __future__ import annotations
 32
 33import asyncio
 34import logging
 35import os
 36import sys
 37from typing import TYPE_CHECKING
 38
 39from fastmcp_extensions import (
 40    JWTAuthConfig,
 41    mcp_server,
 42    resolve_mcp_auth,
 43)
 44from starlette.responses import JSONResponse
 45
 46
 47if TYPE_CHECKING:
 48    from fastmcp.server.auth import AuthProvider
 49    from starlette.requests import Request
 50
 51from airbyte._util.meta import set_mcp_mode
 52from airbyte.mcp._config import load_secrets_to_env_vars
 53from airbyte.mcp._tool_utils import (
 54    AIRBYTE_EXCLUDE_MODULES_CONFIG_ARG,
 55    AIRBYTE_INCLUDE_MODULES_CONFIG_ARG,
 56    AIRBYTE_READONLY_MODE_CONFIG_ARG,
 57    API_URL_CONFIG_ARG,
 58    BEARER_TOKEN_CONFIG_ARG,
 59    CLIENT_ID_CONFIG_ARG,
 60    CLIENT_SECRET_CONFIG_ARG,
 61    CONFIG_API_URL_CONFIG_ARG,
 62    TRUSTED_EXECUTION_CONFIG_ARG,
 63    WORKSPACE_ID_CONFIG_ARG,
 64    airbyte_module_filter,
 65    airbyte_readonly_mode_filter,
 66    airbyte_ui_support_filter,
 67    validate_airbyte_domains,
 68)
 69from airbyte.mcp.cloud import register_cloud_tools
 70from airbyte.mcp.interactive import register_interactive_tools
 71from airbyte.mcp.local import register_local_tools
 72from airbyte.mcp.prompts import register_prompts
 73from airbyte.mcp.registry import register_registry_tools
 74
 75
 76# =============================================================================
 77# Server Instructions
 78# =============================================================================
 79# This text is provided to AI agents via the MCP protocol's "instructions" field.
 80# It helps agents understand when to use this server's tools, especially when
 81# tool search is enabled. For more context, see:
 82# - FastMCP docs: https://gofastmcp.com/servers/overview
 83# - Claude tool search: https://www.anthropic.com/news/tool-use-improvements
 84# =============================================================================
 85
 86MCP_SERVER_INSTRUCTIONS = """
 87PyAirbyte connector management and data integration server for discovering,
 88deploying, and running Airbyte connectors.
 89
 90Use this server for:
 91- Discovering connectors from the Airbyte registry (sources and destinations)
 92- Deploying sources, destinations, and connections to Airbyte Cloud
 93- Running cloud syncs and monitoring sync status
 94- Managing custom connector definitions in Airbyte Cloud
 95- Local connector execution for data extraction without cloud deployment
 96- Listing and describing environment variables for connector configuration
 97
 98Operational modes:
 99- Cloud operations: Deploy and manage connectors on Airbyte Cloud (requires
100  AIRBYTE_CLOUD_CLIENT_ID, AIRBYTE_CLOUD_CLIENT_SECRET, AIRBYTE_CLOUD_WORKSPACE_ID)
101- Local operations: Run connectors locally for data extraction (requires
102  AIRBYTE_PROJECT_DIR for artifact storage)
103
104Safety features:
105- Safe mode (default): Restricts destructive operations to objects created in
106  the current session
107- Read-only mode: Disables all write operations for cloud resources
108""".strip()
109
110logger = logging.getLogger(__name__)
111
112# Public base URL of this deployment; consumed by `http_main` to derive the
113# mounted MCP path. All other transport auth env vars (`OIDC_*` interactive,
114# `MCP_AUTH_*` headless) are read by `fastmcp_extensions.resolve_mcp_auth`.
115MCP_SERVER_URL_ENV = "MCP_SERVER_URL"
116
117# Flag that opts headless verification into Airbyte Cloud's application-client
118# realm; this server owns only this provider literal.
119MCP_AUTH_AIRBYTE_CLOUD_ENV = "MCP_AUTH_AIRBYTE_CLOUD"
120
121# Airbyte Cloud's application-client realm. Tokens minted from an Airbyte Cloud
122# `client_id`/`client_secret` via `<api_root>/applications/token` are RS256 JWTs
123# issued by this realm, and the same token is a valid Airbyte Cloud API bearer.
124# Verifying against this realm lets one token both authenticate transport and
125# authorize downstream Cloud calls. Enable with `MCP_AUTH_AIRBYTE_CLOUD=true`.
126AIRBYTE_CLOUD_REALM_ISSUER = "https://cloud.airbyte.com/auth/realms/_airbyte-application-clients"
127AIRBYTE_CLOUD_JWKS_URI = f"{AIRBYTE_CLOUD_REALM_ISSUER}/protocol/openid-connect/certs"
128AIRBYTE_CLOUD_JWT_AUDIENCE = "account"
129AIRBYTE_CLOUD_JWT_ALGORITHM = "RS256"
130
131DEFAULT_HTTP_HOST = "0.0.0.0"
132DEFAULT_HTTP_PORT = 8080
133
134
135def _create_auth() -> AuthProvider | None:
136    """Assemble the transport auth provider from environment configuration.
137
138    Delegates env parsing to `fastmcp_extensions.resolve_mcp_auth`, which wires
139    up interactive `OIDCProxy` (from `OIDC_*`) and/or headless `JWTVerifier`
140    (from `MCP_AUTH_*`), combining them via `MultiAuth` when both are set and
141    returning `None` when neither is — so the server falls back to standard
142    local (no-auth) behavior.
143
144    When `MCP_AUTH_AIRBYTE_CLOUD` is truthy, the headless verifier defaults to
145    Airbyte Cloud's application-client realm (JWKS / issuer / audience /
146    algorithm); individual `MCP_AUTH_*` vars still override those fields. This
147    is the only provider literal the server owns.
148    """
149    jwt_defaults: JWTAuthConfig | None = None
150    if os.getenv(MCP_AUTH_AIRBYTE_CLOUD_ENV, "").strip().lower() in {"1", "true", "yes"}:
151        jwt_defaults = JWTAuthConfig(
152            jwks_uri=AIRBYTE_CLOUD_JWKS_URI,
153            issuer=AIRBYTE_CLOUD_REALM_ISSUER,
154            audience=AIRBYTE_CLOUD_JWT_AUDIENCE,
155            algorithm=AIRBYTE_CLOUD_JWT_ALGORITHM,
156        )
157    return resolve_mcp_auth(jwt_defaults=jwt_defaults)
158
159
160set_mcp_mode()
161load_secrets_to_env_vars()
162
163app = mcp_server(
164    name="airbyte-mcp",
165    package_name="airbyte",
166    instructions=MCP_SERVER_INSTRUCTIONS,
167    include_standard_tool_filters=True,
168    server_config_args=[
169        AIRBYTE_READONLY_MODE_CONFIG_ARG,
170        AIRBYTE_EXCLUDE_MODULES_CONFIG_ARG,
171        AIRBYTE_INCLUDE_MODULES_CONFIG_ARG,
172        WORKSPACE_ID_CONFIG_ARG,
173        BEARER_TOKEN_CONFIG_ARG,
174        CLIENT_ID_CONFIG_ARG,
175        CLIENT_SECRET_CONFIG_ARG,
176        API_URL_CONFIG_ARG,
177        CONFIG_API_URL_CONFIG_ARG,
178        TRUSTED_EXECUTION_CONFIG_ARG,
179    ],
180    tool_filters=[
181        airbyte_readonly_mode_filter,
182        airbyte_module_filter,
183        airbyte_ui_support_filter,
184    ],
185    auth=_create_auth(),
186)
187"""The Airbyte MCP Server application instance."""
188
189# Register tools from each module
190register_cloud_tools(app)
191register_local_tools(app)
192register_registry_tools(app)
193register_interactive_tools(app)
194register_prompts(app)
195
196validate_airbyte_domains(app)
197
198
199@app.custom_route("/health", methods=["GET"])
200async def health_check(request: Request) -> JSONResponse:  # noqa: ARG001, RUF029
201    """Health check endpoint for load balancer probes."""
202    return JSONResponse({"status": "ok"})
203
204
205def main() -> None:
206    """@private Main entry point for the MCP server.
207
208    This function starts the FastMCP server to handle MCP requests.
209
210    It should not be called directly; instead, consult the MCP client documentation
211    for instructions on how to connect to the server.
212    """
213    print("Starting Airbyte MCP server.", file=sys.stderr)
214    try:
215        asyncio.run(app.run_stdio_async())
216    except KeyboardInterrupt:
217        print("Airbyte MCP server interrupted by user.", file=sys.stderr)
218    except Exception as ex:
219        print(f"Error running Airbyte MCP server: {ex}", file=sys.stderr)
220        sys.exit(1)
221
222    print("Airbyte MCP server stopped.", file=sys.stderr)
223
224
225if __name__ == "__main__":
226    main()