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