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.). Auth is not
  7  enforced; the provider assembled below is ignored by the stdio transport.
  8- **HTTP**: For hosted deployment. Start via `airbyte-mcp-http` entry point or
  9  `poe mcp-serve-http`. This server maps its own branded `AIRBYTE_MCP_*` env vars
 10  into the typed configs that `fastmcp_extensions.build_mcp_auth` consumes, which
 11  supports two client shapes on the same deployment:
 12    - **Interactive** (humans in a browser): Keycloak Authorization Code + PKCE
 13      via `OIDCProxy`, active once `AIRBYTE_MCP_OIDC_CLIENT_ID`,
 14      `AIRBYTE_MCP_OIDC_CLIENT_SECRET`, and `AIRBYTE_MCP_OIDC_CONFIG_URL` (the
 15      OIDC discovery URL) are supplied.
 16    - **Headless** (agents, CI): the client mints its own short-lived bearer
 17      token via the OAuth 2.0 client credentials grant and sends it as
 18      `Authorization: Bearer <token>`. The server verifies it with a
 19      `JWTVerifier`, active once a signing-key source (`AIRBYTE_MCP_AUTH_JWKS_URI`
 20      or `AIRBYTE_MCP_AUTH_JWT_PUBLIC_KEY`) is configured (no browser, no
 21      stored/rotating refresh token).
 22  When both are active they are combined via `MultiAuth`; when neither is
 23  configured `_create_auth` returns `None` and HTTP transport runs
 24  unauthenticated (a startup warning is logged in `http_main`).
 25
 26This module declares only the env var *names* and maps their values into the
 27typed `OIDCAuthConfig` / `JWTAuthConfig` objects that `build_mcp_auth` consumes,
 28so the extensions library stays provider-neutral and reads no env itself. It
 29embeds no provider-specific configuration *values* (a realm's discovery URL,
 30issuer, JWKS URI, audience, algorithm, etc.); those are supplied at deploy time
 31by the deployment's own repo — e.g. the hosted Cloud MCP image in
 32`airbyte-ops-mcp` sets the `AIRBYTE_MCP_*` env for the Airbyte Cloud realm.
 33
 34For the headless path, an agent mints an access token from its client id/secret
 35(via the deployment's `<api_root>/applications/token` endpoint) and sends it as
 36`Authorization: Bearer`. When the deployment's realm is Airbyte Cloud, that
 37single token both authenticates transport (verified here) and authorizes
 38downstream Cloud API calls, because an Airbyte-Cloud-issued JWT is itself a valid
 39Cloud API bearer.
 40"""
 41
 42from __future__ import annotations
 43
 44import asyncio
 45import logging
 46import os
 47import pkgutil
 48import sys
 49from typing import TYPE_CHECKING, Protocol
 50
 51from fastmcp_extensions import (
 52    JWTAuthConfig,
 53    OIDCAuthConfig,
 54    build_mcp_auth,
 55    mcp_server,
 56)
 57from starlette.responses import JSONResponse
 58
 59
 60if TYPE_CHECKING:
 61    from fastmcp.server.auth import AuthProvider
 62    from key_value.aio.protocols.key_value import AsyncKeyValue
 63    from starlette.requests import Request
 64
 65from airbyte._util.meta import set_mcp_mode
 66from airbyte.mcp._config import load_secrets_to_env_vars
 67from airbyte.mcp._tool_utils import (
 68    AIRBYTE_EXCLUDE_MODULES_CONFIG_ARG,
 69    AIRBYTE_INCLUDE_MODULES_CONFIG_ARG,
 70    AIRBYTE_READONLY_MODE_CONFIG_ARG,
 71    API_URL_CONFIG_ARG,
 72    BEARER_TOKEN_CONFIG_ARG,
 73    CLIENT_ID_CONFIG_ARG,
 74    CLIENT_SECRET_CONFIG_ARG,
 75    CONFIG_API_URL_CONFIG_ARG,
 76    TRUSTED_EXECUTION_CONFIG_ARG,
 77    WORKSPACE_ID_CONFIG_ARG,
 78    airbyte_module_filter,
 79    airbyte_readonly_mode_filter,
 80    airbyte_ui_support_filter,
 81    validate_airbyte_domains,
 82)
 83from airbyte.mcp.cloud import register_cloud_tools
 84from airbyte.mcp.interactive import register_interactive_tools
 85from airbyte.mcp.local import register_local_tools
 86from airbyte.mcp.prompts import register_prompts
 87from airbyte.mcp.registry import register_registry_tools
 88
 89
 90# =============================================================================
 91# Server Instructions
 92# =============================================================================
 93# This text is provided to AI agents via the MCP protocol's "instructions" field.
 94# It helps agents understand when to use this server's tools, especially when
 95# tool search is enabled. For more context, see:
 96# - FastMCP docs: https://gofastmcp.com/servers/overview
 97# - Claude tool search: https://www.anthropic.com/news/tool-use-improvements
 98# =============================================================================
 99
100MCP_SERVER_INSTRUCTIONS = """
101PyAirbyte connector management and data integration server for discovering,
102deploying, and running Airbyte connectors.
103
104Use this server for:
105- Discovering connectors from the Airbyte registry (sources and destinations)
106- Deploying sources, destinations, and connections to Airbyte Cloud
107- Running cloud syncs and monitoring sync status
108- Managing custom connector definitions in Airbyte Cloud
109- Local connector execution for data extraction without cloud deployment
110- Listing and describing environment variables for connector configuration
111
112Operational modes:
113- Cloud operations: Deploy and manage connectors on Airbyte Cloud (use request
114  headers when connecting to a hosted MCP server, or AIRBYTE_CLOUD_CLIENT_ID +
115  AIRBYTE_CLOUD_CLIENT_SECRET (or AIRBYTE_CLOUD_BEARER_TOKEN) plus
116  AIRBYTE_CLOUD_WORKSPACE_ID for local or stdio connections)
117- Local operations: Run connectors locally for data extraction (requires
118  AIRBYTE_PROJECT_DIR for artifact storage)
119
120Safety features:
121- Safe mode (default): Restricts destructive operations to objects created in
122  the current session
123- Read-only mode: Disables all write operations for cloud resources
124""".strip()
125
126logger = logging.getLogger(__name__)
127
128# This server's own transport-auth env vars. It owns these *names* and maps the
129# *values* into the typed `OIDCAuthConfig` / `JWTAuthConfig` objects that
130# `build_mcp_auth` consumes; the extensions library reads no env itself. The
131# auth vars use the branded `AIRBYTE_MCP_*` namespace as an added layer over
132# generic OAuth names; `MCP_SERVER_URL` (a deployment URL, not an auth var)
133# stays unbranded. Only names live here — the concrete values (e.g. a specific
134# realm's endpoints) are supplied at deploy time by the deployment's own repo,
135# keeping infrastructure configuration out of this generic library.
136
137# Public base URL of this deployment (also used for OIDC redirect callbacks);
138# `http_main` reuses it to derive the mounted MCP path.
139MCP_SERVER_URL_ENV = "MCP_SERVER_URL"
140
141# Interactive OIDC (`OIDCProxy`). Client id + secret gate the interactive path;
142# the discovery URL comes from the deployment.
143OIDC_CLIENT_ID_ENV = "AIRBYTE_MCP_OIDC_CLIENT_ID"
144OIDC_CLIENT_SECRET_ENV = "AIRBYTE_MCP_OIDC_CLIENT_SECRET"
145OIDC_CONFIG_URL_ENV = "AIRBYTE_MCP_OIDC_CONFIG_URL"
146
147# Upstream authorize scopes requested for the interactive OIDC flow, also
148# advertised to clients via DCR/`.well-known` and enforced on the verified
149# upstream token. `openid` is required for OIDC: without it the IdP may issue an
150# identity-only token that downstream APIs reject.
151AIRBYTE_CLOUD_REQUIRED_OIDC_SCOPES: str = "openid email profile"
152
153# Headless JWT verifier. A signing-key source (`JWKS_URI_ENV` or
154# `JWT_PUBLIC_KEY_ENV`) activates it; issuer/audience/algorithm refine it.
155JWKS_URI_ENV = "AIRBYTE_MCP_AUTH_JWKS_URI"
156JWT_PUBLIC_KEY_ENV = "AIRBYTE_MCP_AUTH_JWT_PUBLIC_KEY"
157JWT_ISSUER_ENV = "AIRBYTE_MCP_AUTH_ISSUER"
158JWT_AUDIENCE_ENV = "AIRBYTE_MCP_AUTH_AUDIENCE"
159JWT_ALGORITHM_ENV = "AIRBYTE_MCP_AUTH_ALGORITHM"
160
161# Names a durable-storage factory (`"package.module:callable"`) for the
162# interactive `OIDCProxy`'s OAuth state. The concrete backend (and its infra
163# config) lives in the deployment's own package, keeping PyAirbyte generic.
164OIDC_CLIENT_STORAGE_FACTORY_ENV = "AIRBYTE_MCP_OIDC_CLIENT_STORAGE_FACTORY"
165
166DEFAULT_HTTP_HOST = "0.0.0.0"
167DEFAULT_HTTP_PORT = 8080
168DEFAULT_MCP_SERVER_URL = f"http://localhost:{DEFAULT_HTTP_PORT}"
169
170
171class _ClientStorageFactory(Protocol):
172    """Callable that builds a durable `OIDCProxy` OAuth-state backend.
173
174    A deployment names its factory via
175    `AIRBYTE_MCP_OIDC_CLIENT_STORAGE_FACTORY` (`"package.module:callable"`). The
176    callable receives the OIDC client secret as `encryption_source_material` so
177    it can derive an at-rest encryption key, and returns an `AsyncKeyValue`
178    store. Keeping the concrete backend (Firestore, Redis, ...) behind this hook
179    lets PyAirbyte stay generic — the infrastructure-specific factory ships in
180    the deployment's own package (e.g. the hosted Cloud MCP image), not here.
181    """
182
183    def __call__(self, *, encryption_source_material: str) -> AsyncKeyValue: ...
184
185
186def _env_or_default(name: str, default: str) -> str:
187    """Return the stripped value of env var `name`, or `default` when blank/unset.
188
189    Blank and whitespace-only values are treated as unset so an empty deployment
190    override falls back to `default` rather than an empty string.
191    """
192    value = os.getenv(name, "").strip()
193    return value or default
194
195
196def _resolve_client_storage(*, encryption_source_material: str) -> AsyncKeyValue | None:
197    """Resolve the durable `OIDCProxy` OAuth-state store, if one is configured.
198
199    Reads `AIRBYTE_MCP_OIDC_CLIENT_STORAGE_FACTORY` (`"package.module:callable"`),
200    imports the named factory, and calls it to build the store. Returns `None`
201    when the var is unset/blank, keeping `OIDCProxy`'s in-memory default (fine
202    for single-instance local dev). PyAirbyte stays backend-agnostic: it never
203    imports a concrete store, so the infrastructure-specific factory (e.g. the
204    Fernet-wrapped Firestore store for the hosted Cloud MCP image) ships in the
205    deployment's own package.
206
207    Raises `ValueError` (naming the env var and expected format) when the
208    factory reference is malformed or points at a missing symbol, so a
209    misconfigured deployment fails with a clear message instead of a bare
210    import traceback.
211    """
212    factory_spec = os.getenv(OIDC_CLIENT_STORAGE_FACTORY_ENV, "").strip()
213    if not factory_spec:
214        return None
215    try:
216        factory: _ClientStorageFactory = pkgutil.resolve_name(factory_spec)
217    except (ImportError, AttributeError, ValueError) as exc:
218        msg = (
219            f"{OIDC_CLIENT_STORAGE_FACTORY_ENV}={factory_spec!r} could not be "
220            "resolved; expected a 'package.module:callable' reference to an "
221            "importable OAuth-state store factory."
222        )
223        raise ValueError(msg) from exc
224    return factory(encryption_source_material=encryption_source_material)
225
226
227def _create_auth() -> AuthProvider | None:
228    """Assemble the transport auth provider from this server's env configuration.
229
230    Reads this server's branded `AIRBYTE_MCP_*` env vars and maps them into the
231    typed `JWTAuthConfig` / `OIDCAuthConfig` objects that
232    `fastmcp_extensions.build_mcp_auth` consumes, which wires up a headless
233    `JWTVerifier` and/or an interactive `OIDCProxy`, combined via `MultiAuth`.
234    The headless verifier activates once a signing-key source
235    (`AIRBYTE_MCP_AUTH_JWKS_URI` or `AIRBYTE_MCP_AUTH_JWT_PUBLIC_KEY`) is
236    configured; the interactive path activates once the OIDC client credentials
237    are supplied. Returns `None` when neither is configured, so the server falls
238    back to unauthenticated local behavior. The `stdio` transport ignores the
239    provider entirely.
240
241    This server declares only the env var *names*; the concrete values (e.g. a
242    deployment's realm endpoints, issuer, audience, and discovery URL) are
243    supplied at deploy time by the deployment's own repo, keeping
244    infrastructure configuration out of this generic library.
245    """
246    base_url = _env_or_default(MCP_SERVER_URL_ENV, DEFAULT_MCP_SERVER_URL)
247
248    jwt: JWTAuthConfig | None = None
249    jwks_uri = os.getenv(JWKS_URI_ENV, "").strip()
250    public_key = os.getenv(JWT_PUBLIC_KEY_ENV, "").strip()
251    if jwks_uri or public_key:
252        jwt = JWTAuthConfig(
253            jwks_uri=jwks_uri or None,
254            public_key=public_key or None,
255            issuer=os.getenv(JWT_ISSUER_ENV, "").strip() or None,
256            audience=os.getenv(JWT_AUDIENCE_ENV, "").strip() or None,
257            algorithm=os.getenv(JWT_ALGORITHM_ENV, "").strip() or None,
258            base_url=base_url,
259        )
260
261    oidc: OIDCAuthConfig | None = None
262    oidc_client_id = os.getenv(OIDC_CLIENT_ID_ENV, "").strip()
263    oidc_client_secret = os.getenv(OIDC_CLIENT_SECRET_ENV, "").strip()
264    if bool(oidc_client_id) != bool(oidc_client_secret):
265        present, missing = (
266            (OIDC_CLIENT_ID_ENV, OIDC_CLIENT_SECRET_ENV)
267            if oidc_client_id
268            else (OIDC_CLIENT_SECRET_ENV, OIDC_CLIENT_ID_ENV)
269        )
270        msg = (
271            f"{present} is set but {missing} is not; the interactive OIDC path "
272            "needs both client credentials. Set both, or neither."
273        )
274        raise ValueError(msg)
275    if oidc_client_id and oidc_client_secret:
276        config_url = os.getenv(OIDC_CONFIG_URL_ENV, "").strip()
277        if not config_url:
278            msg = (
279                f"{OIDC_CLIENT_ID_ENV} and {OIDC_CLIENT_SECRET_ENV} are set but "
280                f"{OIDC_CONFIG_URL_ENV} is not; the interactive OIDC path needs "
281                "an OpenID Connect discovery URL."
282            )
283            raise ValueError(msg)
284        oidc = OIDCAuthConfig(
285            config_url=config_url,
286            client_id=oidc_client_id,
287            client_secret=oidc_client_secret,
288            base_url=base_url,
289            required_scopes=AIRBYTE_CLOUD_REQUIRED_OIDC_SCOPES.split(),
290            client_storage=_resolve_client_storage(encryption_source_material=oidc_client_secret),
291        )
292
293    return build_mcp_auth(oidc=oidc, jwt=jwt, base_url=base_url)
294
295
296set_mcp_mode()
297load_secrets_to_env_vars()
298
299app = mcp_server(
300    name="airbyte-mcp",
301    package_name="airbyte",
302    instructions=MCP_SERVER_INSTRUCTIONS,
303    include_standard_tool_filters=True,
304    server_config_args=[
305        AIRBYTE_READONLY_MODE_CONFIG_ARG,
306        AIRBYTE_EXCLUDE_MODULES_CONFIG_ARG,
307        AIRBYTE_INCLUDE_MODULES_CONFIG_ARG,
308        WORKSPACE_ID_CONFIG_ARG,
309        BEARER_TOKEN_CONFIG_ARG,
310        CLIENT_ID_CONFIG_ARG,
311        CLIENT_SECRET_CONFIG_ARG,
312        API_URL_CONFIG_ARG,
313        CONFIG_API_URL_CONFIG_ARG,
314        TRUSTED_EXECUTION_CONFIG_ARG,
315    ],
316    tool_filters=[
317        airbyte_readonly_mode_filter,
318        airbyte_module_filter,
319        airbyte_ui_support_filter,
320    ],
321    auth=_create_auth(),
322)
323"""The Airbyte MCP Server application instance."""
324
325# Register tools from each module
326register_cloud_tools(app)
327register_local_tools(app)
328register_registry_tools(app)
329register_interactive_tools(app)
330register_prompts(app)
331
332validate_airbyte_domains(app)
333
334
335@app.custom_route("/health", methods=["GET"])
336async def health_check(request: Request) -> JSONResponse:  # noqa: ARG001, RUF029
337    """Health check endpoint for load balancer probes."""
338    return JSONResponse({"status": "ok"})
339
340
341def main() -> None:
342    """@private Main entry point for the MCP server.
343
344    This function starts the FastMCP server to handle MCP requests.
345
346    It should not be called directly; instead, consult the MCP client documentation
347    for instructions on how to connect to the server.
348    """
349    print("Starting Airbyte MCP server.", file=sys.stderr)
350    try:
351        asyncio.run(app.run_stdio_async())
352    except KeyboardInterrupt:
353        print("Airbyte MCP server interrupted by user.", file=sys.stderr)
354    except Exception as ex:
355        print(f"Error running Airbyte MCP server: {ex}", file=sys.stderr)
356        sys.exit(1)
357
358    print("Airbyte MCP server stopped.", file=sys.stderr)
359
360
361if __name__ == "__main__":
362    main()