Skip to content
Draft
Show file tree
Hide file tree
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
73 changes: 72 additions & 1 deletion airbyte/mcp/guidance.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,9 @@
__all__: list[str] = []

import contextlib
from typing import TYPE_CHECKING, Annotated, Any, Literal
from enum import StrEnum
from typing import TYPE_CHECKING, Annotated, Any, Literal, cast
from urllib.parse import quote, urlencode

import requests
from fastmcp import Context, FastMCP
Expand Down Expand Up @@ -56,6 +58,23 @@
_KAPA_TIMEOUT_SECONDS = 30.0

KNOWLEDGE_SEARCH_CAPABILITY = "io.airbyte/knowledge-search"
GITHUB_ISSUES_REPO = "airbytehq/PyAirbyte"
MAX_GITHUB_ISSUE_URL_LENGTH = 8000


class GitHubIssueCategory(StrEnum):
"""Issue categories, each matching an issue form in `.github/ISSUE_TEMPLATE/`."""

CLOUD_MCP_BUG_REPORT = "cloud-mcp-bug-report"
CLOUD_MCP_FEATURE_REQUEST = "cloud-mcp-feature-request"


class GitHubIssueCreationLink(BaseModel):
"""A pre-filled GitHub issue creation link."""

url: str
warning: str | None = None


KAPA_API_KEY_CONFIG_ARG = MCPServerConfigArg(
name="kapa_api_key",
Expand Down Expand Up @@ -333,6 +352,58 @@ def get_api_docs_urls(
return "Connector not found."


@mcp_tool(read_only=True, idempotent=True)
def get_github_issue_creation_link(
category: Annotated[
GitHubIssueCategory,
Field(description="The GitHub issue form to use for the Airbyte Cloud MCP server."),
TraceArg.VALUE,
],
title: Annotated[
str,
Field(description="A short, specific issue title.", min_length=1, max_length=200),
TraceArg.FINGERPRINT,
],
description: Annotated[
str,
Field(
description=(
"A Markdown description of what happened, what was expected, "
"and the tool calls involved."
),
min_length=1,
),
TraceArg.FINGERPRINT,
],
) -> GitHubIssueCreationLink:
"""Create a link to a pre-filled public GitHub issue for the Airbyte Cloud MCP server.

Use this when a user encounters a bug in this server's tools or asks for a capability it
lacks. Always give the returned URL to the user as a clickable link. If `warning` is set, the
link may be too long to open; follow its guidance. This tool does not file anything: the user
must be signed in to GitHub to review and submit the issue under their own account. Issues are
public, so never include secrets, credentials, connector config values, or customer data. Draft
the description in Markdown with what happened, what was expected, and the tool calls involved.
"""
url = f"https://github.com/{GITHUB_ISSUES_REPO}/issues/new?" + urlencode(
{
"template": f"{category.value}.yml",
"title": title,
"description": description,
},
quote_via=cast(Any, quote),
)
warning = None
if len(url) > MAX_GITHUB_ISSUE_URL_LENGTH:
warning = (
f"This link is {len(url)} characters after URL encoding, above the "
f"{MAX_GITHUB_ISSUE_URL_LENGTH}-character limit some browsers and servers enforce. "
"Give the user the link anyway. If it fails to open, shorten the description and "
"create a new link."
)
return GitHubIssueCreationLink(url=url, warning=warning)


def register_guidance_tools(app: FastMCP) -> None:
"""Register guidance tools and prompts with the FastMCP app."""
register_mcp_tools(app, mcp_module=__name__)
Expand Down
5 changes: 5 additions & 0 deletions tests/unit_tests/fixtures/arg_trace_plan.json
Original file line number Diff line number Diff line change
Expand Up @@ -138,6 +138,11 @@
"workspace_id": "omit"
},
"get_default_cloud_context": {},
"get_github_issue_creation_link": {
"category": "value allowed=[\"cloud-mcp-bug-report\",\"cloud-mcp-feature-request\"]",
"description": "fingerprint",
"title": "fingerprint"
},
"get_source_stream_json_schema": {
"config": "presence",
"config_file": "fingerprint",
Expand Down
82 changes: 82 additions & 0 deletions tests/unit_tests/test_mcp_guidance.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import asyncio
import json
from typing import cast
from urllib.parse import parse_qs, urlsplit

import pytest
import requests
Expand Down Expand Up @@ -47,6 +48,87 @@ def _airbyte_mcp_app() -> FastMCP:
return app


@pytest.mark.parametrize(
("category", "slug"),
[
pytest.param(
guidance.GitHubIssueCategory.CLOUD_MCP_BUG_REPORT,
"cloud-mcp-bug-report",
id="bug-report",
),
pytest.param(
guidance.GitHubIssueCategory.CLOUD_MCP_FEATURE_REQUEST,
"cloud-mcp-feature-request",
id="feature-request",
),
],
)
def test_get_github_issue_creation_link(
category: guidance.GitHubIssueCategory, slug: str
) -> None:
title = "Bug & regression #42 + edge case? caf茅"
description = (
"What happened: the `tool+name` call failed & returned #1? Unexpectedly.\n"
"Expected: success.\nTool calls: `get_cloud_sync_status`."
)

result = guidance.get_github_issue_creation_link(category, title, description)
parsed_url = urlsplit(result.url)
query = parse_qs(parsed_url.query)

assert result.warning is None
assert parsed_url.scheme == "https"
assert parsed_url.netloc == "github.com"
assert parsed_url.path == "/airbytehq/PyAirbyte/issues/new"
assert set(query) == {"template", "title", "description"}
assert query["template"] == [f"{slug}.yml"]
assert query["title"] == [title]
assert query["description"] == [description]


def test_get_github_issue_creation_link_is_listed_by_app() -> None:
assert "get_github_issue_creation_link" in _list_tool_names(_airbyte_mcp_app())


def test_get_github_issue_creation_link_warns_for_oversized_encoded_description() -> (
None
):
result = guidance.get_github_issue_creation_link(
guidance.GitHubIssueCategory.CLOUD_MCP_BUG_REPORT,
"Issue",
"馃榾" * 4000,
)

assert result.warning is not None
assert "shorten the description" in result.warning
assert result.url.startswith("https://github.com/airbytehq/PyAirbyte/issues/new?")


def test_get_github_issue_creation_link_accepts_ascii_description_at_field_limit() -> (
None
):
result = guidance.get_github_issue_creation_link(
guidance.GitHubIssueCategory.CLOUD_MCP_BUG_REPORT,
"Issue",
"a" * 4000,
)

assert len(result.url) <= guidance.MAX_GITHUB_ISSUE_URL_LENGTH
assert result.warning is None


def test_get_github_issue_creation_link_warns_for_long_ascii_description() -> None:
result = guidance.get_github_issue_creation_link(
guidance.GitHubIssueCategory.CLOUD_MCP_BUG_REPORT,
"Issue",
"a" * 10000,
)

assert result.warning is not None
assert "shorten the description" in result.warning
assert result.url.startswith("https://github.com/airbytehq/PyAirbyte/issues/new?")


@pytest.mark.parametrize(
("hosted", "api_key", "retrieval_api_url", "expected_visible"),
[
Expand Down
13 changes: 12 additions & 1 deletion tests/unit_tests/test_mcp_tool_annotations.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
from __future__ import annotations

import airbyte.mcp.server # noqa: F401 # Importing registers every MCP tool module.
from airbyte.mcp import cloud as cloud_mcp
from airbyte.mcp import cloud as cloud_mcp, guidance as guidance_mcp
from fastmcp_extensions.decorators import _REGISTERED_TOOLS # noqa: PLC2701

READ_ONLY_NAME_PREFIXES = ("list_", "describe_", "get_", "check_")
Expand All @@ -27,6 +27,17 @@ def test_check_cloud_connector_is_read_only_and_idempotent() -> None:
assert annotations["idempotentHint"] is True


def test_get_github_issue_creation_link_is_read_only_and_idempotent() -> None:
"""`get_github_issue_creation_link` only constructs a GitHub URL."""
(annotations,) = [
a
for f, a in _REGISTERED_TOOLS
if f is guidance_mcp.get_github_issue_creation_link
]
assert annotations["readOnlyHint"] is True
assert annotations["idempotentHint"] is True


def test_read_prefixed_cloud_tools_are_all_read_only() -> None:
"""Every cloud tool named list_*/describe_*/get_*/check_* must be read-only."""
violations = [
Expand Down
Loading