feat(mcp): Add Kapa knowledge search tool

airbyte/mcp/kapa.py

@@ -0,0 +1,112 @@
+# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+"""Kapa knowledge source MCP operations."""
+
+from __future__ import annotations
+
+import os
+from typing import TYPE_CHECKING, Annotated
+
+import requests
+from fastmcp_extensions import mcp_tool, register_mcp_tools
+from pydantic import Field
+
+
+if TYPE_CHECKING:
+    from fastmcp import FastMCP
+
+
+_KAPA_RETRIEVAL_URL_TEMPLATE = "https://api.kapa.ai/query/v1/projects/{project_id}/retrieval/"
+_KAPA_TIMEOUT_SECONDS = 30.0
+
+
+def _kapa_credentials_configured() -> bool:
+    return bool(
+        os.getenv("KAPA_API_KEY")
+        or os.getenv("KAPA_DOCS_MCP_BEARER_TOKEN")
+        or os.getenv("KAPA_BEARER_TOKEN")
+    )
+
+
+def _kapa_auth_headers() -> dict[str, str]:
+    api_key = (os.getenv("KAPA_API_KEY") or "").strip()
+    if api_key:
+        return {"X-API-KEY": api_key}
+
+    bearer_token = (
+        (os.getenv("KAPA_DOCS_MCP_BEARER_TOKEN") or os.getenv("KAPA_BEARER_TOKEN")) or ""
+    ).strip()
+    if bearer_token:
+        return {"Authorization": f"Bearer {bearer_token}"}
+
+    raise ValueError(
+        "Kapa docs search is not configured. Set KAPA_API_KEY, "
+        "KAPA_DOCS_MCP_BEARER_TOKEN, or KAPA_BEARER_TOKEN."
+    )
+
+
+def _kapa_retrieval_url() -> str:
+    project_id = (os.getenv("KAPA_PROJECT_ID") or "").strip()
+    if not project_id:
+        raise ValueError("KAPA_PROJECT_ID must be set to use Kapa docs search.")
+
+    return _KAPA_RETRIEVAL_URL_TEMPLATE.format(project_id=project_id)
+
+
+def _kapa_request_payload(query: str) -> dict[str, str]:
+    payload = {"query": query}
+    integration_id = (os.getenv("KAPA_INTEGRATION_ID") or "").strip()
+    if integration_id:
+        payload["integration_id"] = integration_id
+    return payload
+
+
+def _normalize_kapa_results(data: object) -> list[dict[str, str]]:
+    if not isinstance(data, list):
+        raise TypeError("Kapa retrieval API returned an unexpected response shape.")
+
+    results: list[dict[str, str]] = []
+    for item in data:
+        if not isinstance(item, dict):
+            raise TypeError("Kapa retrieval API returned an unexpected result item.")
+
+        source_url = item.get("source_url")
+        content = item.get("content")
+        if not isinstance(source_url, str) or not isinstance(content, str):
+            raise TypeError(
+                "Kapa retrieval API returned a result without source_url and content strings."
+            )
+
+        results.append({"source_url": source_url, "content": content})
+
+    return results
+
+
+@mcp_tool(
+    read_only=True,
+    idempotent=True,
+)
+def search_airbyte_knowledge_sources(
+    query: Annotated[
+        str,
+        Field(
+            description=(
+                "A single, well-formed natural-language query. Must be a complete sentence."
+            ),
+        ),
+    ],
+) -> list[dict[str, str]]:
+    """Search Airbyte knowledge sources."""
+    response = requests.post(
+        _kapa_retrieval_url(),
+        headers=_kapa_auth_headers(),
+        json=_kapa_request_payload(query),
+        timeout=_KAPA_TIMEOUT_SECONDS,
+    )
+    response.raise_for_status()
+    return _normalize_kapa_results(response.json())
+
+
+def register_kapa_tools(app: FastMCP) -> None:
+    """Register Kapa tools with the FastMCP app."""
+    if _kapa_credentials_configured():
+        register_mcp_tools(app, mcp_module=__name__)

airbyte/mcp/server.py

@@ -24,6 +24,7 @@
     airbyte_readonly_mode_filter,
 )
 from airbyte.mcp.cloud import register_cloud_tools
+from airbyte.mcp.kapa import register_kapa_tools
 from airbyte.mcp.local import register_local_tools
 from airbyte.mcp.prompts import register_prompts
 from airbyte.mcp.registry import register_registry_tools
@@ -93,6 +94,7 @@
 register_cloud_tools(app)
 register_local_tools(app)
 register_registry_tools(app)
+register_kapa_tools(app)
 register_prompts(app)
 
 

tests/unit_tests/test_mcp_kapa.py

@@ -0,0 +1,98 @@
+# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+"""Unit tests for Kapa MCP tools."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+import responses
+
+from airbyte.mcp import kapa
+
+
+@pytest.fixture(autouse=True)
+def clear_kapa_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Clear Kapa environment variables before each test."""
+    monkeypatch.delenv("KAPA_API_KEY", raising=False)
+    monkeypatch.delenv("KAPA_DOCS_MCP_BEARER_TOKEN", raising=False)
+    monkeypatch.delenv("KAPA_BEARER_TOKEN", raising=False)
+    monkeypatch.delenv("KAPA_PROJECT_ID", raising=False)
+    monkeypatch.delenv("KAPA_INTEGRATION_ID", raising=False)
+
+
+@pytest.mark.parametrize(
+    ("env_name", "expected_headers"),
+    [
+        pytest.param("KAPA_API_KEY", {"X-API-KEY": "secret"}, id="api-key"),
+        pytest.param(
+            "KAPA_DOCS_MCP_BEARER_TOKEN",
+            {"Authorization": "Bearer secret"},
+            id="docs-mcp-bearer-token",
+        ),
+        pytest.param(
+            "KAPA_BEARER_TOKEN", {"Authorization": "Bearer secret"}, id="bearer-token"
+        ),
+    ],
+)
+def test_kapa_auth_headers(
+    env_name: str,
+    expected_headers: dict[str, str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Test Kapa auth header selection from supported env vars."""
+    monkeypatch.setenv(env_name, "secret")
+
+    assert kapa._kapa_auth_headers() == expected_headers  # noqa: SLF001
+
+
+@responses.activate
+def test_search_airbyte_knowledge_sources_calls_kapa_retrieval_api(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Test Kapa search request construction and response normalization."""
+    monkeypatch.setenv("KAPA_API_KEY", "secret")
+    monkeypatch.setenv("KAPA_PROJECT_ID", "project-id")
+    monkeypatch.setenv("KAPA_INTEGRATION_ID", "integration-id")
+    responses.add(
+        responses.POST,
+        "https://api.kapa.ai/query/v1/projects/project-id/retrieval/",
+        json=[{"source_url": "https://docs.airbyte.com", "content": "Airbyte docs"}],
+        status=200,
+    )
+
+    result = kapa.search_airbyte_knowledge_sources("How do I set up a source?")
+
+    assert result == [
+        {"source_url": "https://docs.airbyte.com", "content": "Airbyte docs"}
+    ]
+    assert len(responses.calls) == 1
+    request = responses.calls[0].request
+    assert request.headers["X-API-KEY"] == "secret"
+    assert (
+        request.body
+        == b'{"query": "How do I set up a source?", "integration_id": "integration-id"}'
+    )
+
+
+def test_register_kapa_tools_skips_registration_without_credentials() -> None:
+    """Test that Kapa tools are hidden when credentials are absent."""
+    app = MagicMock()
+
+    with patch("airbyte.mcp.kapa.register_mcp_tools") as register:
+        kapa.register_kapa_tools(app)
+
+    register.assert_not_called()
+
+
+def test_register_kapa_tools_registers_when_credentials_are_configured(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Test that Kapa tools are visible when credentials are configured."""
+    app = MagicMock()
+    monkeypatch.setenv("KAPA_API_KEY", "secret")
+
+    with patch("airbyte.mcp.kapa.register_mcp_tools") as register:
+        kapa.register_kapa_tools(app)
+
+    register.assert_called_once_with(app, mcp_module="airbyte.mcp.kapa")