airbyte_ops_mcp.mcp.server

  1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
  2"""Airbyte Admin MCP server implementation.
  3
  4This module provides the main MCP server for Airbyte admin operations.
  5
  6The server can run in two modes:
  7- **stdio mode** (default): For direct MCP client connections via stdin/stdout
  8- **HTTP mode**: HTTP transport is **always authenticated**, defaulting to
  9  Airbyte Cloud with zero auth config. This server maps its own `AIRBYTE_MCP_*`
 10  env vars into the typed configs that `fastmcp_extensions.build_mcp_auth`
 11  consumes, which 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` and
 14      `AIRBYTE_MCP_OIDC_CLIENT_SECRET` are supplied (the OIDC discovery URL
 15      defaults to Airbyte Cloud).
 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` against Airbyte Cloud's application-client realm by default
 20      (no browser, no stored/rotating refresh token).
 21  When both are active they are combined via `MultiAuth`.
 22
 23This module owns the Airbyte Cloud realm defaults (non-secret, publicly
 24discoverable) and maps its `AIRBYTE_MCP_OIDC_*` / `AIRBYTE_MCP_AUTH_*` env vars
 25into the typed `OIDCAuthConfig` / `JWTAuthConfig` objects that `build_mcp_auth`
 26consumes, so the extensions library stays provider-neutral and reads no env
 27itself. A self-hosted deployment pointing at its own Airbyte instance overrides
 28any default via the matching env var.
 29
 30An agent mints an Airbyte Cloud access token from its `AIRBYTE_CLOUD_CLIENT_ID` /
 31`AIRBYTE_CLOUD_CLIENT_SECRET` (the `<api_root>/applications/token` endpoint) and
 32sends it as `Authorization: Bearer`. That single token both authenticates
 33transport (verified here) and authorizes downstream Cloud API calls: the
 34downstream bearer is resolved from the transport-*verified* token
 35(`get_access_token`), not the raw `Authorization` header, so it works for both
 36headless (client-minted app token) and interactive (upstream Keycloak token,
 37where the raw header is only the proxy's reference JWT). An Airbyte-Cloud-issued
 38JWT is itself a valid Cloud API bearer.
 39
 40HTTP mode environment variables (the headless JWT-verifier vars default to
 41Airbyte Cloud and are optional overrides for self-hosted deployments; the
 42interactive OIDC client credentials have no default and must be supplied to
 43enable interactive login):
 44    MCP_SERVER_URL: Public base URL for the MCP server (also used for OIDC
 45        redirect callbacks). Defaults to `http://localhost:8080`.
 46    AIRBYTE_MCP_OIDC_CONFIG_URL: Keycloak OIDC discovery URL (defaults to
 47        Airbyte Cloud)
 48    AIRBYTE_MCP_OIDC_CLIENT_ID: OAuth client ID for interactive OIDC (no
 49        default; supply to activate interactive login)
 50    AIRBYTE_MCP_OIDC_CLIENT_SECRET: OAuth client secret for interactive OIDC
 51    AIRBYTE_MCP_AUTH_JWKS_URI: JWKS URL for verifying headless bearer tokens
 52    AIRBYTE_MCP_AUTH_JWT_PUBLIC_KEY: Static public key alternative to
 53        `AIRBYTE_MCP_AUTH_JWKS_URI`
 54    AIRBYTE_MCP_AUTH_ISSUER: Expected `iss` claim for headless tokens
 55    AIRBYTE_MCP_AUTH_AUDIENCE: Expected `aud` claim for headless tokens
 56    AIRBYTE_MCP_AUTH_ALGORITHM: JWT signing algorithm
 57    AIRBYTE_MCP_AUTH_ALLOW_CLIENT_CREDENTIALS: Set truthy to also accept
 58        `Authorization: Basic base64(client_id:client_secret)`. The server
 59        exchanges those long-lived credentials for a short-lived bearer token
 60        server-side and rewrites the request to `Authorization: Bearer <token>`
 61        so the headless verifier above validates it. Off by default. This is for
 62        headless agents that can only set a static `Authorization` header and
 63        cannot re-mint short-lived tokens themselves. See
 64        `airbyte_ops_mcp.mcp._client_credentials`.
 65    AIRBYTE_MCP_AUTH_CLIENT_CREDENTIALS_TOKEN_URL: Token endpoint used for the
 66        exchange above (defaults to Airbyte Cloud; override for self-hosted).
 67    AIRBYTE_MCP_AUTH_ACCEPT_USER_TOKENS: Set falsey to stop accepting Airbyte
 68        Cloud *user* realm tokens as headless bearers (on by default). These
 69        are tokens from the `airbyte` Keycloak realm — the same realm the
 70        interactive `OIDCProxy` already trusts upstream — so this server can
 71        accept a session token forwarded by a trusted sibling service (the
 72        AG-UI chat server relaying the Ops Webapp cookie) and still delegate
 73        it downstream, since it is a valid Cloud API bearer.
 74"""
 75
 76import asyncio
 77import logging
 78import os
 79import sys
 80from collections.abc import Mapping
 81from importlib.metadata import PackageNotFoundError, version
 82from pathlib import Path
 83from urllib.parse import urlparse
 84
 85from airbyte.cloud.auth import resolve_cloud_client_id, resolve_cloud_client_secret
 86from airbyte.constants import set_hosted_mcp_mode
 87from dotenv import load_dotenv
 88from fastmcp import FastMCP
 89from fastmcp.server.auth import AuthProvider, MultiAuth
 90from fastmcp.server.dependencies import get_access_token
 91from fastmcp_extensions import (
 92    # Re-exported so `server.ClientAllowlistJWTVerifier` stays importable for
 93    # tests and callers patching the shared verifier.
 94    ClientAllowlistJWTVerifier,  # noqa: F401
 95    JWTAuthConfig,
 96    MCPServerConfigArg,
 97    OIDCAuthConfig,
 98    ToolCallTelemetryMiddleware,
 99    build_mcp_auth,
100    mcp_server,
101    register_landing_page,
102    run_mcp_http_server,
103)
104from packaging.version import Version
105from pydantic import BaseModel
106from starlette.requests import Request
107from starlette.responses import JSONResponse
108
109from airbyte_ops_mcp._sentry import (
110    _SENTRY_DSN,
111    get_sentry_environment,
112    init_sentry_tracking,
113)
114from airbyte_ops_mcp.constants import (
115    HEADER_AIRBYTE_CLOUD_CLIENT_ID,
116    HEADER_AIRBYTE_CLOUD_CLIENT_SECRET,
117    MCP_SERVER_NAME,
118    ServerConfigKey,
119)
120from airbyte_ops_mcp.mcp._client_credentials import wrap_if_enabled
121from airbyte_ops_mcp.mcp._oidc_storage import resolve_oidc_client_storage
122from airbyte_ops_mcp.mcp.connection_medic import register_connection_medic_tools
123from airbyte_ops_mcp.mcp.connection_resources import register_connection_resource_tools
124from airbyte_ops_mcp.mcp.connector_qa import register_connector_qa_tools
125from airbyte_ops_mcp.mcp.connector_registry import register_connector_registry_tools
126from airbyte_ops_mcp.mcp.connector_versions import register_connector_version_tools
127from airbyte_ops_mcp.mcp.context_store_ops import register_context_store_ops_tools
128from airbyte_ops_mcp.mcp.data_ops import register_data_ops_tools
129from airbyte_ops_mcp.mcp.devin_ops import register_devin_ops_tools
130from airbyte_ops_mcp.mcp.feature_flags import register_launchdarkly_ops_tools
131from airbyte_ops_mcp.mcp.github_ops import register_github_ops_tools
132from airbyte_ops_mcp.mcp.human_in_the_loop import register_human_in_the_loop_tools
133from airbyte_ops_mcp.mcp.logging import register_logging_tools
134from airbyte_ops_mcp.mcp.organization_admin import register_organization_admin_tools
135from airbyte_ops_mcp.mcp.prod_db_ops import register_prod_db_ops_tools
136from airbyte_ops_mcp.mcp.prompts import register_prompts
137from airbyte_ops_mcp.mcp.zendesk_ops import register_zendesk_ops_tools
138from airbyte_ops_mcp.telemetry import _DEFAULT_SEGMENT_WRITE_KEY
139
140MCP_SERVER_INSTRUCTIONS = """
141Airbyte internal operations server for connector management, cloud administration,
142and production database queries.
143
144Use this server for:
145- Publishing connector prereleases and managing version overrides/pins
146- Running connector regression tests (single-version and comparison modes)
147- Querying the Airbyte Cloud production database for workspace, connector, sync,
148  and connection diagnostics
149- Triggering and monitoring GitHub Actions CI workflows
150- Looking up Cloud Logging errors for debugging connector issues
151- Performing repository operations on the Airbyte monorepo (for example, listing
152  connectors in the repo or inspecting connector definitions)
153
154Requirements:
155- GCP credentials for database queries and Cloud Logging access
156- Airbyte Cloud credentials for cloud administration operations
157- GitHub token for workflow dispatch and repository operations
158- Local checkout of the Airbyte repository for repo tools (typically at `../airbyte`)
159
160Note: This server is for Airbyte internal use only.
161""".strip()
162
163logger = logging.getLogger(__name__)
164
165# Default HTTP server configuration
166DEFAULT_HTTP_HOST = "0.0.0.0"
167DEFAULT_HTTP_PORT = 8080
168
169# Public base URL of this deployment, used to derive the mounted MCP path and the
170# OIDC redirect base.
171MCP_SERVER_URL_ENV = "MCP_SERVER_URL"
172
173# Default public base URL, mirroring the HTTP entrypoint default so the OIDC
174# redirect base is well-formed even when `MCP_SERVER_URL` is unset (local dev).
175DEFAULT_MCP_SERVER_URL = f"http://localhost:{DEFAULT_HTTP_PORT}"
176
177# Airbyte Cloud's public Keycloak realms. These are non-secret, publicly
178# discoverable endpoints used as the zero-config auth defaults so the hosted
179# Airbyte Cloud MCP server needs no auth env beyond its OIDC client credentials.
180# Interactive human login uses the `airbyte` realm; headless application-client
181# tokens are issued by (and verified against) the `_airbyte-application-clients`
182# realm. Because the same headless token is a valid Airbyte Cloud API bearer, one
183# token both authenticates transport and authorizes downstream Cloud API calls.
184AIRBYTE_CLOUD_OIDC_CONFIG_URL = (
185    "https://cloud.airbyte.com/auth/realms/airbyte/.well-known/openid-configuration"
186)
187AIRBYTE_CLOUD_ISSUER = (
188    "https://cloud.airbyte.com/auth/realms/_airbyte-application-clients"
189)
190AIRBYTE_CLOUD_JWKS_URI = f"{AIRBYTE_CLOUD_ISSUER}/protocol/openid-connect/certs"
191AIRBYTE_CLOUD_AUDIENCE = "account"
192AIRBYTE_CLOUD_ALGORITHM = "RS256"
193
194# Airbyte Cloud *user* realm (`airbyte`) — the realm the interactive
195# `OIDCProxy` already trusts upstream. User session tokens forwarded by a
196# trusted sibling service are accepted as headless bearers; issuer plus JWKS
197# signature is the trust boundary (`audience` varies by issuing client), the
198# same boundary `OIDCProxy` uses when delegating the upstream token. Which
199# clients' user tokens are trusted is pinned by the `azp` allowlist below.
200AIRBYTE_CLOUD_USER_ISSUER = "https://cloud.airbyte.com/auth/realms/airbyte"
201AIRBYTE_CLOUD_USER_JWKS_URI = (
202    f"{AIRBYTE_CLOUD_USER_ISSUER}/protocol/openid-connect/certs"
203)
204
205# Upstream authorize scopes requested for the interactive OIDC flow. `openid` is
206# required: without it Keycloak issues an identity-only token that Airbyte Cloud
207# APIs reject with `401`, even though the user is otherwise valid (the working
208# Ops Webapp OAuth client requests exactly `openid email profile`). These scopes
209# are advertised to MCP clients via DCR/`.well-known`, sent on the upstream
210# `/authorize`, and enforced on the verified upstream token.
211AIRBYTE_CLOUD_OIDC_SCOPES: str = "openid email profile"
212
213# Headless JWT verifier claim/algorithm family. This server's Airbyte-branded
214# env vars, each paired with the Airbyte Cloud default `_create_auth` applies.
215# Because these defaults are always present, HTTP transport always verifies
216# bearer tokens. Setting any matching env var overrides the Cloud default — the
217# escape hatch for self-hosted deployments pointing at their own Airbyte
218# instance. These carry the `AUTH` segment; `OIDC_*` vars keep `OIDC` alone (it
219# already denotes auth).
220#
221# The signing-key source (`AIRBYTE_MCP_AUTH_JWKS_URI` /
222# `AIRBYTE_MCP_AUTH_JWT_PUBLIC_KEY`) is resolved separately in `_create_auth`,
223# because the JWKS default must apply only when neither key source is set (see
224# `_resolve_signing_key`).
225JWT_ISSUER_ENV = "AIRBYTE_MCP_AUTH_ISSUER"
226JWT_AUDIENCE_ENV = "AIRBYTE_MCP_AUTH_AUDIENCE"
227JWT_ALGORITHM_ENV = "AIRBYTE_MCP_AUTH_ALGORITHM"
228
229# Signing-key sources for the headless JWT verifier. A deployment may point at a
230# JWKS endpoint (`AIRBYTE_MCP_AUTH_JWKS_URI`) or supply a static public key
231# (`AIRBYTE_MCP_AUTH_JWT_PUBLIC_KEY`, for self-hosted realms without a JWKS
232# endpoint). The Airbyte Cloud JWKS default applies only when neither is set.
233JWKS_URI_ENV = "AIRBYTE_MCP_AUTH_JWKS_URI"
234JWT_PUBLIC_KEY_ENV = "AIRBYTE_MCP_AUTH_JWT_PUBLIC_KEY"
235
236# Interactive OIDC env vars. The client credentials are secret, so they have no
237# default and must be supplied by the deployment to activate the interactive
238# path. The discovery URL defaults to Airbyte Cloud but is only injected when
239# the credentials are present (see `_create_auth`).
240OIDC_CLIENT_ID_ENV = "AIRBYTE_MCP_OIDC_CLIENT_ID"
241OIDC_CLIENT_SECRET_ENV = "AIRBYTE_MCP_OIDC_CLIENT_SECRET"
242OIDC_CONFIG_URL_ENV = "AIRBYTE_MCP_OIDC_CONFIG_URL"
243# CIMD (Client ID Metadata Document) is enabled by default so broad OAuth
244# clients that only implement CIMD can authenticate — notably Goose Desktop,
245# which hardcodes a metadata-document URL as its `client_id` and has no DCR
246# fallback. An operator can force it off with `...=false` to mitigate an auth
247# issue without a redeploy. `OIDCAuthConfig.enable_cimd` defaults to `False`
248# upstream, so this server opts in explicitly.
249OIDC_ENABLE_CIMD_ENV = "AIRBYTE_MCP_OIDC_ENABLE_CIMD"
250
251# Accept Airbyte Cloud user-realm tokens as headless bearers (default on).
252# An operator can force it off to shrink the accepted-token surface without a
253# redeploy.
254USER_TOKENS_ENABLED_ENV = "AIRBYTE_MCP_AUTH_ACCEPT_USER_TOKENS"
255
256# Comma-separated `azp` allowlist for user-realm bearer tokens (which Keycloak
257# client issued them). Only the Ops Webapp client is trusted by default.
258USER_TOKEN_CLIENT_IDS_ENV = "AIRBYTE_MCP_AUTH_USER_TOKEN_CLIENT_IDS"
259DEFAULT_USER_TOKEN_CLIENT_IDS = frozenset({"airbyte-ops-webapp-client"})
260
261# Issuer/JWKS overrides for the user-realm verifier (self-hosted escape hatch,
262# matching the `AIRBYTE_MCP_AUTH_*` claim overrides above). The JWKS default
263# derives from the *configured* issuer when unset.
264USER_ISSUER_ENV = "AIRBYTE_MCP_AUTH_USER_ISSUER"
265USER_JWKS_URI_ENV = "AIRBYTE_MCP_AUTH_USER_JWKS_URI"
266
267# Human-facing landing page shown when a browser GETs the MCP endpoint.
268MCP_LANDING_TITLE = "Airbyte Ops MCP Server"
269MCP_LANDING_DOCS_URL = "https://github.com/airbytehq/airbyte-ops-mcp#readme"
270RELEASE_TAG_URL_TEMPLATE = (
271    "https://github.com/airbytehq/airbyte-ops-mcp/releases/tag/v{}"
272)
273COMMIT_URL_TEMPLATE = "https://github.com/airbytehq/airbyte-ops-mcp/commit/{}"
274DISTRIBUTION_NAME = "airbyte-internal-ops"
275
276
277def _landing_version_str() -> str | None:
278    """Return the installed package version for the landing-page footer.
279
280    Returns `None` when the distribution metadata is unavailable (e.g. running
281    straight from a source tree), which omits the footer entirely.
282    """
283    try:
284        return f"v{version(DISTRIBUTION_NAME)}"
285    except PackageNotFoundError:
286        return None
287
288
289def _landing_version_url() -> str | None:
290    """Return the URL the landing-page version footer links to.
291
292    A tagged release links to its release page. A dev build carries the commit
293    it was cut from in the version's local segment
294    (`0.96.2.post5.dev0+1b1637b4`) and has no release of its own, so it links
295    to that commit instead.
296    """
297    try:
298        installed = Version(version(DISTRIBUTION_NAME))
299    except PackageNotFoundError:
300        return None
301
302    if installed.local:
303        commit_sha = installed.local.split(".")[0]
304        return COMMIT_URL_TEMPLATE.format(commit_sha)
305    return RELEASE_TAG_URL_TEMPLATE.format(installed.public)
306
307
308def _normalize_bearer_token(value: str) -> str | None:
309    """Extract bearer token from Authorization header value.
310
311    Parses "Bearer <token>" format (case-insensitive prefix).
312    Returns None if the value doesn't have the Bearer prefix.
313    """
314    if value.lower().startswith("bearer "):
315        token = value[7:].strip()
316        return token if token else None
317    return None
318
319
320def _resolve_transport_bearer_token() -> str:
321    """Resolve the verified transport bearer token if available.
322
323    FastMCP stores the access token of the current request after the transport
324    auth provider verifies it — behind `OIDCProxy` the token swap exposes the
325    upstream Keycloak token for interactive clients, and the client-minted JWT
326    for headless `JWTVerifier`. Both are Airbyte Cloud
327    tokens when the server verifies against Airbyte Cloud's realm, so reusing
328    the token as the downstream Cloud API bearer gives the caller's identity
329    delegated access without a second credential.
330
331    Returns empty string when no verified token is present (e.g. stdio mode).
332    """
333    access_token = get_access_token()
334    if access_token and access_token.token:
335        return access_token.token
336    return ""
337
338
339class ConnectedUser(BaseModel):
340    """Authenticated principal exposed by the server-info resource."""
341
342    sub: str | None = None
343    email: str | None = None
344    preferred_username: str | None = None
345    name: str | None = None
346
347
348def _server_info_identity() -> ConnectedUser | None:
349    """Return the authenticated principal for the current request."""
350    access_token = get_access_token()
351    if not access_token:
352        return None
353
354    raw_claims = getattr(access_token, "claims", {})
355    claims = raw_claims if isinstance(raw_claims, Mapping) else {}
356    sub = claims.get("sub")
357    email = claims.get("email")
358    preferred_username = claims.get("preferred_username")
359    name = claims.get("name")
360    return ConnectedUser(
361        sub=sub if isinstance(sub, str) else None,
362        email=email if isinstance(email, str) else None,
363        preferred_username=(
364            preferred_username if isinstance(preferred_username, str) else None
365        ),
366        name=name if isinstance(name, str) else None,
367    )
368
369
370def _server_info_provider() -> dict[str, object]:
371    """Serialize the authenticated principal for the server-info resource."""
372    identity = _server_info_identity()
373    return {
374        "connected_user": identity.model_dump(exclude_none=True) if identity else None
375    }
376
377
378def _env_or_default(name: str, default: str) -> str:
379    """Return the stripped value of env var `name`, or `default` when unset/blank.
380
381    An env var set to an empty or whitespace-only string is treated as unset, so
382    the baked default still applies and no blank value is propagated downstream
383    (a `"   "` JWKS URI or server URL would otherwise break auth resolution).
384    """
385    return os.getenv(name, "").strip() or default
386
387
388def _env_bool(name: str, *, default: bool) -> bool:
389    """Return the boolean value of env var `name`, or `default` when unset/blank.
390
391    Recognizes `true`/`false`, `1`/`0`, `yes`/`no`, `on`/`off` (case-insensitive).
392    A blank or whitespace-only value is treated as unset so the baked default
393    applies. An unrecognized value raises `ValueError` rather than silently
394    coercing a typo (e.g. `flase`) to `False`.
395    """
396    raw = os.getenv(name, "").strip().lower()
397    if not raw:
398        return default
399    if raw in ("true", "1", "yes", "on"):
400        return True
401    if raw in ("false", "0", "no", "off"):
402        return False
403    raise ValueError(
404        f"{name} must be a boolean (true/false/1/0/yes/no/on/off), got '{raw}'."
405    )
406
407
408def _resolve_signing_key() -> tuple[str, str]:
409    """Resolve the headless JWT verifier's signing-key source.
410
411    Returns the `(jwks_uri, public_key)` pair. A deployment may set either env
412    var to point at its own realm; the Airbyte Cloud JWKS default applies only
413    when *neither* is set, so a self-hosted static public key isn't shadowed by a
414    leftover Cloud JWKS URI. Blank or whitespace-only values are treated as
415    unset, and an unset member is returned as the empty string.
416    """
417    jwks_uri = os.getenv(JWKS_URI_ENV, "").strip()
418    public_key = os.getenv(JWT_PUBLIC_KEY_ENV, "").strip()
419    if not jwks_uri and not public_key:
420        jwks_uri = AIRBYTE_CLOUD_JWKS_URI
421    return jwks_uri, public_key
422
423
424def _create_auth() -> AuthProvider | None:
425    """Assemble the transport auth provider, defaulting to Airbyte Cloud.
426
427    Reads this server's `AIRBYTE_MCP_*` env vars (falling back to Airbyte Cloud's
428    public realm defaults), maps them into the typed `JWTAuthConfig` /
429    `OIDCAuthConfig` objects that `fastmcp_extensions.build_mcp_auth` consumes,
430    and lets it wire up a headless `JWTVerifier` and/or an interactive
431    `OIDCProxy`, combined via `MultiAuth`. Because a JWKS default is always
432    present, HTTP transport always verifies bearer tokens; the interactive path
433    additionally activates once the OIDC client credentials are supplied.
434    Unless `AIRBYTE_MCP_AUTH_ACCEPT_USER_TOKENS` is set falsey, a second
435    verifier (`ClientAllowlistJWTVerifier` from `fastmcp_extensions`) also
436    accepts user session tokens issued by the `airbyte` realm, pinned to the
437    `azp` allowlist — no `aud` check, since Keycloak user-token audiences
438    vary by client and issuer plus signature is the trust boundary.
439    """
440    base_url = _env_or_default(MCP_SERVER_URL_ENV, DEFAULT_MCP_SERVER_URL)
441
442    # Headless JWT verification is always configured (the Airbyte Cloud JWKS
443    # default is present whenever the deployment sets no key source of its own).
444    jwks_uri, public_key = _resolve_signing_key()
445    jwt = JWTAuthConfig(
446        jwks_uri=jwks_uri or None,
447        public_key=public_key or None,
448        issuer=_env_or_default(JWT_ISSUER_ENV, AIRBYTE_CLOUD_ISSUER),
449        audience=_env_or_default(JWT_AUDIENCE_ENV, AIRBYTE_CLOUD_AUDIENCE),
450        algorithm=_env_or_default(JWT_ALGORITHM_ENV, AIRBYTE_CLOUD_ALGORITHM),
451        base_url=base_url,
452    )
453
454    # Interactive OIDC activates only when both client credentials are present.
455    # Building it on the headless/bearer-only path would advertise an OIDC
456    # discovery URL with no credentials behind it.
457    oidc: OIDCAuthConfig | None = None
458    oidc_client_id = os.getenv(OIDC_CLIENT_ID_ENV, "").strip()
459    oidc_client_secret = os.getenv(OIDC_CLIENT_SECRET_ENV, "").strip()
460    if oidc_client_id and oidc_client_secret:
461        # Durable, encrypted backend for `OIDCProxy`'s OAuth state so interactive
462        # sessions survive restarts and span replicas. Returns `None` (keeping
463        # the in-memory default) unless `AIRBYTE_MCP_OIDC_STORAGE=firestore`. The
464        # encryption key is derived from the OIDC client secret this server
465        # already holds, so no separate encryption secret is provisioned.
466        oidc = OIDCAuthConfig(
467            config_url=_env_or_default(
468                OIDC_CONFIG_URL_ENV, AIRBYTE_CLOUD_OIDC_CONFIG_URL
469            ),
470            client_id=oidc_client_id,
471            client_secret=oidc_client_secret,
472            base_url=base_url,
473            # Advertise and accept the CIMD flow (URL `client_id`) so broad OAuth
474            # clients that only implement CIMD — notably Goose Desktop — can
475            # authenticate. The key-normalizing storage wrapper (see
476            # `_oidc_storage`) is what makes the URL `client_id` storable;
477            # without it the CIMD `/authorize` path crashes with a Firestore
478            # `InvalidArgument`.
479            enable_cimd=_env_bool(OIDC_ENABLE_CIMD_ENV, default=True),
480            # Request `openid` (plus email/profile) upstream so Keycloak issues
481            # an API-usable token, not an identity-only one that Airbyte Cloud
482            # rejects. Also advertised to clients so DCR/CIMD registrations may
483            # request them.
484            required_scopes=AIRBYTE_CLOUD_OIDC_SCOPES.split(),
485            client_storage=resolve_oidc_client_storage(
486                encryption_source_material=oidc_client_secret
487            ),
488        )
489
490    # Airbyte Cloud user-realm tokens (the Ops Webapp session token agui-server
491    # forwards) are accepted as bearers unless explicitly disabled. The
492    # `allowed_client_ids` config makes `build_mcp_auth` emit a
493    # `ClientAllowlistJWTVerifier` for this realm; `audience` stays unset —
494    # Keycloak user-token audiences vary by client.
495    jwt_configs = [jwt]
496    if _env_bool(USER_TOKENS_ENABLED_ENV, default=True):
497        user_issuer = _env_or_default(USER_ISSUER_ENV, AIRBYTE_CLOUD_USER_ISSUER)
498        user_jwks_uri = os.getenv(USER_JWKS_URI_ENV, "").strip() or (
499            f"{user_issuer}/protocol/openid-connect/certs"
500        )
501        jwt_configs.append(
502            JWTAuthConfig(
503                jwks_uri=user_jwks_uri,
504                issuer=user_issuer,
505                algorithm=AIRBYTE_CLOUD_ALGORITHM,
506                base_url=base_url,
507                allowed_client_ids=_user_token_client_ids(),
508            )
509        )
510
511    # One `build_mcp_auth` call composes the interactive `OIDCProxy` (when
512    # configured) with one verifier per JWT realm via `MultiAuth`; `oidc=None`
513    # yields the verifiers alone.
514    return build_mcp_auth(jwt=jwt_configs, oidc=oidc, base_url=base_url)
515
516
517def _user_token_client_ids() -> frozenset[str]:
518    """Parse `AIRBYTE_MCP_AUTH_USER_TOKEN_CLIENT_IDS` into the `azp` allowlist.
519
520    Comma-separated, whitespace-stripped, empty entries dropped; unset or
521    all-empty falls back to `DEFAULT_USER_TOKEN_CLIENT_IDS`.
522    """
523    parsed = frozenset(
524        entry.strip()
525        for entry in os.getenv(USER_TOKEN_CLIENT_IDS_ENV, "").split(",")
526        if entry.strip()
527    )
528    return parsed or DEFAULT_USER_TOKEN_CLIENT_IDS
529
530
531# Create the MCP server with built-in server info resource
532app = mcp_server(
533    name=MCP_SERVER_NAME,
534    instructions=MCP_SERVER_INSTRUCTIONS,
535    package_name="airbyte-internal-ops",
536    advertised_properties={
537        "docs_url": "https://github.com/airbytehq/airbyte-ops-mcp",
538        "release_history_url": "https://github.com/airbytehq/airbyte-ops-mcp/releases",
539    },
540    server_info_provider=_server_info_provider,
541    server_config_args=[
542        MCPServerConfigArg(
543            # The raw `Authorization` header is deliberately *not* a first-class
544            # source: behind `OAuthProxy`/`OIDCProxy` (interactive OIDC) it carries
545            # the proxy's self-minted reference JWT, which Airbyte Cloud rejects
546            # with `401`. Resolving via `_resolve_transport_bearer_token` uses the
547            # transport-*verified* upstream token (`get_access_token`) instead —
548            # the upstream Keycloak token for interactive, the client-minted app
549            # token for headless — both valid Airbyte Cloud API bearers. An
550            # explicit `AIRBYTE_CLOUD_BEARER_TOKEN` env still overrides.
551            name=ServerConfigKey.BEARER_TOKEN,
552            env_var="AIRBYTE_CLOUD_BEARER_TOKEN",
553            normalize_fn=_normalize_bearer_token,
554            default=_resolve_transport_bearer_token,
555            required=False,
556            sensitive=True,
557        ),
558        MCPServerConfigArg(
559            name=ServerConfigKey.CLIENT_ID,
560            http_header_key=HEADER_AIRBYTE_CLOUD_CLIENT_ID,
561            default=lambda: str(resolve_cloud_client_id()),
562            required=True,
563            sensitive=True,
564        ),
565        MCPServerConfigArg(
566            name=ServerConfigKey.CLIENT_SECRET,
567            http_header_key=HEADER_AIRBYTE_CLOUD_CLIENT_SECRET,
568            default=lambda: str(resolve_cloud_client_secret()),
569            required=True,
570            sensitive=True,
571        ),
572    ],
573    include_standard_tool_filters=True,
574    auth=_create_auth(),
575)
576
577
578def register_server_assets(app: FastMCP) -> None:
579    """Register all server assets (tools, prompts, resources) with the FastMCP app.
580
581    Tools are grouped into domain-oriented modules to keep the generated pdoc
582    reference navigable:
583
584    - `connector_versions`: cloud version overrides, rollouts, pre-release publish
585    - `connector_registry`: registry reads/yank plus monorepo list/bump
586    - `connector_qa`: regression tests and release blocking
587    - `connection_medic`: connection state/catalog reads plus emergency writes
588    - `prod_db_ops`: Prod Cloud DB-replica SQL queries
589    - `logging`: GCP Cloud Logging backend-error lookup
590    - `context_store_ops`: MotherDuck / context-store diagnostics
591    - `organization_admin`: is_agentic flag, payment config, customer tiers
592    - `github_ops`: CI workflow trigger/status, Docker image info, subscriptions
593    - `data_ops`: data warehouse (airbyte-warehouse) on-demand dbt dispatch
594    - `human_in_the_loop`: human escalation, team-roster lookup, Slack newsletter posting
595    - `devin_ops`: reminders, secret requests, session feedback and naming
596    - `zendesk_ops`: Zendesk Support ticket retrieval, updates, and spam alerts
597    - `prompts`: prompt templates for common workflows
598
599    Tools annotated with `requires_client_filesystem=True` are automatically
600    hidden when `MCP_NO_CLIENT_FILESYSTEM=1` via the standard tool filter.
601
602    Note: Server info resource is now built-in via `mcp_server()` helper.
603
604    Args:
605        app: FastMCP application instance
606    """
607    register_connector_version_tools(app)
608    register_connector_registry_tools(app)
609    register_connector_qa_tools(app)
610    register_connection_medic_tools(app)
611    register_connection_resource_tools(app)
612    register_prod_db_ops_tools(app)
613    register_logging_tools(app)
614    register_context_store_ops_tools(app)
615    register_organization_admin_tools(app)
616    register_launchdarkly_ops_tools(app)
617    register_github_ops_tools(app)
618    register_data_ops_tools(app)
619    register_human_in_the_loop_tools(app)
620    register_devin_ops_tools(app)
621    register_zendesk_ops_tools(app)
622    register_prompts(app)
623
624
625def _load_env() -> None:
626    """Load environment variables from .env file if present."""
627    env_file = Path.cwd() / ".env"
628    if env_file.exists():
629        load_dotenv(env_file)
630        print(f"Loaded environment from: {env_file}", flush=True, file=sys.stderr)
631
632
633register_server_assets(app)
634# Env must be loaded and Sentry initialized before the telemetry middleware
635# below, which otherwise calls `sentry_sdk.init` itself with default
636# integrations (Starlette included); a later `disabled_integrations` cannot
637# undo an already-installed one.
638_load_env()
639init_sentry_tracking(mode="mcp")
640app.add_middleware(
641    ToolCallTelemetryMiddleware(
642        package_name="airbyte-internal-ops",
643        sentry_dsn=_SENTRY_DSN if get_sentry_environment() is not None else None,
644        segment_write_key=_DEFAULT_SEGMENT_WRITE_KEY,
645    )
646)
647
648
649@app.custom_route("/health", methods=["GET"])
650async def health_check(request: Request) -> JSONResponse:
651    """Health check endpoint for Cloud Run liveness/readiness probes."""
652    return JSONResponse({"status": "ok"})
653
654
655def main() -> None:
656    """Main entry point for the Airbyte Admin MCP server (stdio mode).
657
658    This is the default entry point that runs the server in stdio mode,
659    suitable for direct MCP client connections.
660    """
661
662    print("=" * 60, flush=True, file=sys.stderr)
663    print("Starting Airbyte Admin MCP server (stdio mode).", file=sys.stderr)
664    try:
665        asyncio.run(app.run_stdio_async(show_banner=False))
666    except KeyboardInterrupt:
667        print("Airbyte Admin MCP server interrupted by user.", file=sys.stderr)
668
669    print("Airbyte Admin MCP server stopped.", file=sys.stderr)
670    print("=" * 60, flush=True, file=sys.stderr)
671
672
673def _advertise_root_mount_resource(auth: AuthProvider) -> None:
674    """Advertise the slash-less public URL as the RFC 8707 resource at a root mount.
675
676    Behind a path-stripping load balancer the MCP endpoint is mounted at root
677    (`mcp_path="/"`), and FastMCP derives the protected-resource identifier from
678    that mount path — appending a trailing slash (e.g. `.../ops-mcp/`). Strict
679    RFC 9728 clients canonicalize the connection URL to the slash-less form
680    (`.../ops-mcp`) and reject the mismatch, so they cannot attach. FastMCP
681    already returns the bare base URL for a *root* mount path (`None`/`""`), so
682    this maps the `"/"` mount path onto that root case, leaving non-root mounts
683    (e.g. the local `"/mcp"` default) untouched.
684
685    Applied to every provider in the tree because the protected-resource
686    metadata document and the `WWW-Authenticate` challenge are built from
687    different providers (the interactive server versus the top-level `MultiAuth`).
688    """
689    # FastMCP exposes no public seam for this, so we wrap the private accessor.
690    original = auth._get_resource_url
691
692    def resolve_resource_url(path: str | None = None):
693        normalized = path if path and path != "/" else None
694        return original(normalized)
695
696    auth._get_resource_url = resolve_resource_url  # ty: ignore[invalid-assignment]
697
698    if isinstance(auth, MultiAuth):
699        if auth.server is not None:
700            _advertise_root_mount_resource(auth.server)
701        for verifier in auth.verifiers:
702            _advertise_root_mount_resource(verifier)
703
704
705def main_http() -> None:
706    """HTTP entry point for the Airbyte Admin MCP server.
707
708    Runs the server in HTTP mode. When OIDC env vars are configured,
709    Keycloak authentication is enabled automatically.
710    """
711    set_hosted_mcp_mode()
712
713    host = DEFAULT_HTTP_HOST
714    port = DEFAULT_HTTP_PORT
715
716    # When deployed behind a path-stripping LB (MCP_SERVER_URL has a path
717    # component like /ops-mcp), serve the MCP endpoint at root so the
718    # public URL is just the base path. Otherwise keep the FastMCP default.
719    server_url = _env_or_default(MCP_SERVER_URL_ENV, DEFAULT_MCP_SERVER_URL)
720    mcp_path = "/" if urlparse(server_url).path.strip("/") else "/mcp"
721
722    if getattr(app, "auth", None) is None:
723        logger.warning(
724            "HTTP transport starting without authentication: no headless "
725            "bearer-token or interactive OIDC auth resolved, so every request "
726            "is unauthenticated. This is unexpected — headless verification "
727            "defaults to the Airbyte Cloud realm, so auth should normally always "
728            "be active. Reaching this state means the signing-key source could "
729            "not be resolved (e.g. `AIRBYTE_MCP_AUTH_JWKS_URI` set to an "
730            "unreachable URL). Verify your `AIRBYTE_MCP_AUTH_*` overrides."
731        )
732
733    # The advertised endpoint must match where the MCP route is actually mounted:
734    # the bare server URL when mounted at root, otherwise the server URL + mcp_path.
735    endpoint_url = server_url if mcp_path == "/" else server_url.rstrip("/") + mcp_path
736
737    # At a root mount FastMCP would advertise a trailing-slash resource that
738    # strict RFC 9728 clients reject; pin it to the slash-less public URL. Must
739    # run before `app.http_app()` below builds the protected-resource routes.
740    if mcp_path == "/" and app.auth is not None:
741        _advertise_root_mount_resource(app.auth)
742
743    # Serve a browser-friendly landing page on GET at the MCP path. In stateless
744    # mode FastMCP only binds POST/DELETE there, so this GET route does not
745    # interfere with MCP traffic.
746    register_landing_page(
747        app,
748        path=mcp_path,
749        title=MCP_LANDING_TITLE,
750        endpoint_url=endpoint_url,
751        docs_url=MCP_LANDING_DOCS_URL,
752        version_str=_landing_version_str(),
753        version_url=_landing_version_url(),
754    )
755
756    print("=" * 60, flush=True, file=sys.stderr)
757    print(
758        f"Starting Airbyte Admin MCP server (HTTP mode) on {host}:{port}"
759        f" (mcp_path={mcp_path!r})",
760        file=sys.stderr,
761    )
762    try:
763        run_mcp_http_server(
764            app,
765            path=mcp_path,
766            transport="streamable-http",
767            stateless_http=True,
768            wrapper=wrap_if_enabled,
769            host=host,
770            port=port,
771        )
772    except KeyboardInterrupt:
773        print("Airbyte Admin MCP server interrupted by user.", file=sys.stderr)
774
775    print("Airbyte Admin MCP server stopped.", file=sys.stderr)
776    print("=" * 60, flush=True, file=sys.stderr)
777
778
779if __name__ == "__main__":
780    main()