Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 30 additions & 4 deletions airbyte/mcp/http_main.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,8 @@
import os
from urllib.parse import urlparse

from fastmcp_extensions import register_landing_page

from airbyte.mcp.server import (
DEFAULT_HTTP_HOST,
DEFAULT_HTTP_PORT,
Expand All @@ -47,6 +49,18 @@

logger = logging.getLogger(__name__)

# Human-facing landing page shown when a browser GETs the MCP endpoint.
MCP_LANDING_TITLE = "Airbyte MCP Server"
MCP_LANDING_DOCS_URL = "https://docs.airbyte.com/ai-agents/"


def _get_server_url() -> str:
Comment thread
aaronsteers marked this conversation as resolved.
"""Return the public base URL from `MCP_SERVER_URL`, defaulting to localhost."""
return os.getenv(
MCP_SERVER_URL_ENV,
f"http://localhost:{DEFAULT_HTTP_PORT}",
)


def main() -> None:
"""Start the Airbyte MCP server with HTTP transport."""
Expand All @@ -55,12 +69,24 @@ def main() -> None:
# When deployed behind a path-stripping LB (MCP_SERVER_URL has a path
# component like /cloud-mcp), serve the MCP endpoint at root so the
# public URL is just the base path. Otherwise keep the FastMCP default.
server_url = os.getenv(
MCP_SERVER_URL_ENV,
f"http://localhost:{DEFAULT_HTTP_PORT}",
)
server_url = _get_server_url()
mcp_path = "/" if urlparse(server_url).path.strip("/") else "/mcp"

# The advertised endpoint must match where the MCP route is actually mounted:
# the bare server URL when mounted at root, otherwise the server URL + mcp_path.
endpoint_url = server_url if mcp_path == "/" else server_url.rstrip("/") + mcp_path
Comment thread
aaronsteers marked this conversation as resolved.

# Serve a browser-friendly landing page on GET at the MCP path. In stateless
# mode FastMCP only binds POST/DELETE there, so this GET route does not
# interfere with MCP traffic.
register_landing_page(
app,
path=mcp_path,
title=MCP_LANDING_TITLE,
endpoint_url=endpoint_url,
docs_url=MCP_LANDING_DOCS_URL,
)

if getattr(app, "auth", None) is None:
logger.warning(
"HTTP transport starting without authentication: no interactive "
Expand Down
Loading