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