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