|
7 | 7 |
|
8 | 8 | from __future__ import annotations |
9 | 9 |
|
| 10 | +import inspect |
10 | 11 | import os |
11 | 12 | from collections.abc import Callable |
12 | 13 | from typing import Any, Literal, TypeVar |
|
25 | 26 | os.environ.get("AIRBYTE_CLOUD_MCP_READONLY_MODE", "").strip() == "1" |
26 | 27 | ) |
27 | 28 | AIRBYTE_CLOUD_MCP_SAFE_MODE = os.environ.get("AIRBYTE_CLOUD_MCP_SAFE_MODE", "1").strip() != "0" |
| 29 | +AIRBYTE_CLOUD_WORKSPACE_ID_IS_SET = bool(os.environ.get("AIRBYTE_CLOUD_WORKSPACE_ID", "").strip()) |
28 | 30 |
|
29 | 31 | _REGISTERED_TOOLS: list[tuple[Callable[..., Any], dict[str, Any]]] = [] |
30 | 32 | _GUIDS_CREATED_IN_SESSION: set[str] = set() |
@@ -158,8 +160,20 @@ def register_tools(app: Any, domain: Literal["cloud", "local", "registry"]) -> N |
158 | 160 | for func, tool_annotations in get_registered_tools(domain): |
159 | 161 | if should_register_tool(tool_annotations): |
160 | 162 | extra_help_text = getattr(func, "_mcp_extra_help_text", None) |
| 163 | + description: str | None = None |
161 | 164 | if extra_help_text: |
162 | 165 | description = (func.__doc__ or "").rstrip() + "\n" + extra_help_text |
163 | | - app.tool(func, annotations=tool_annotations, description=description) |
164 | | - else: |
165 | | - app.tool(func, annotations=tool_annotations) |
| 166 | + |
| 167 | + # For cloud tools, conditionally hide workspace_id parameter when env var is set |
| 168 | + exclude_args: list[str] | None = None |
| 169 | + if domain == "cloud" and AIRBYTE_CLOUD_WORKSPACE_ID_IS_SET: |
| 170 | + params = set(inspect.signature(func).parameters.keys()) |
| 171 | + excluded = [name for name in ["workspace_id"] if name in params] |
| 172 | + exclude_args = excluded or None |
| 173 | + |
| 174 | + app.tool( |
| 175 | + func, |
| 176 | + annotations=tool_annotations, |
| 177 | + description=description, |
| 178 | + exclude_args=exclude_args, |
| 179 | + ) |
0 commit comments