Skip to content

feat(mcp): Add optional workspace_id parameter to cloud tools with conditional schema hiding #876

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Aaron ("AJ") Steers (aaronsteers) merged 2 commits into from
Nov 25, 2025
Merged

feat(mcp): Add optional workspace_id parameter to cloud tools with conditional schema hiding #876

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ad201e6
feat(mcp): Add optional workspace_id parameter to cloud tools with co…
devin-ai-integration[bot] Nov 25, 2025
4203f57
remove python-level default arg values
aaronsteers Nov 25, 2025
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
20 changes: 17 additions & 3 deletions airbyte/mcp/_tool_utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

from __future__ import annotations

import inspect
import os
from collections.abc import Callable
from typing import Any, Literal, TypeVar
Expand All @@ -25,6 +26,7 @@
os.environ.get("AIRBYTE_CLOUD_MCP_READONLY_MODE", "").strip() == "1"
)
AIRBYTE_CLOUD_MCP_SAFE_MODE = os.environ.get("AIRBYTE_CLOUD_MCP_SAFE_MODE", "1").strip() != "0"
AIRBYTE_CLOUD_WORKSPACE_ID_IS_SET = bool(os.environ.get("AIRBYTE_CLOUD_WORKSPACE_ID", "").strip())

_REGISTERED_TOOLS: list[tuple[Callable[..., Any], dict[str, Any]]] = []
_GUIDS_CREATED_IN_SESSION: set[str] = set()
Expand Down Expand Up @@ -158,8 +160,20 @@ def register_tools(app: Any, domain: Literal["cloud", "local", "registry"]) -> N
for func, tool_annotations in get_registered_tools(domain):
if should_register_tool(tool_annotations):
extra_help_text = getattr(func, "_mcp_extra_help_text", None)
description: str | None = None
if extra_help_text:
description = (func.__doc__ or "").rstrip() + "\n" + extra_help_text
app.tool(func, annotations=tool_annotations, description=description)
else:
app.tool(func, annotations=tool_annotations)

# For cloud tools, conditionally hide workspace_id parameter when env var is set
exclude_args: list[str] | None = None
if domain == "cloud" and AIRBYTE_CLOUD_WORKSPACE_ID_IS_SET:
params = set(inspect.signature(func).parameters.keys())
excluded = [name for name in ["workspace_id"] if name in params]
exclude_args = excluded or None

app.tool(
func,
annotations=tool_annotations,
description=description,
exclude_args=exclude_args,
)
Loading
Loading