← Files Dataiku HeadlessARCHIVED FILE
dataiku_mcp/tools/plugins.py
53.2 KB · Nov 11, 2026 · 12:58 UTC
# Copyright 2026 Dataiku SAS
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Dataiku instance plugin inspection, installation, environment setup, and removal."""
from __future__ import annotations
import asyncio
import io
import re
import time
from pathlib import Path
from typing import Annotated, Literal
from zipfile import BadZipFile, ZipFile
from dataikuapi.dss.future import DSSFuture
from dataikuapi.utils import DataikuException
from fastmcp import Context
from pydantic import Field
from requests.exceptions import RequestException
from ..auth import get_dss_client
from ..config import request
from ..executors import run_blocking
from ..instance_policy import ADMIN_POLICY
from ..server import mcp
from .code_environments import PythonInterpreter
from .utils.errors import dataiku_message
from .utils.parsing import parse_json_object
from .utils.serialization import columnar, compact_json, omit_empty
from .utils.validation import (
require_allowed_value as _require_allowed_value,
require_git_repository_url as _require_git_repository_url,
require_int_in_range as _require_int_in_range,
require_non_empty_string as _require_non_empty_string,
require_non_negative_int as _require_non_negative_int,
require_positive_int as _require_positive_int,
require_relative_posix_path as _require_relative_posix_path,
)
PluginSource = Literal["store", "local_path", "git"]
_SEARCH_MODES = {"partial", "exact"}
_MAX_TIMEOUT_SECONDS = 3600
# Store updates unpack an archive; code-environment rebuilds also install dependencies.
_DEFAULT_TIMEOUT_SECONDS = 120
_BUILD_TIMEOUT_SECONDS = 600
# Automation futures omit their target. Retain only builds started by this process,
# scoped by caller and instance; after a restart, recover by environment name.
_BUILD_OWNERS: dict[tuple, str] = {}
_MIN_POLL_INTERVAL_SECONDS = 2
_MAX_POLL_INTERVAL_SECONDS = 30
# A deletion refusal provides a compact sample; list_plugin_usages paginates the
# complete impact before an agent asks the user to force deletion.
_MAX_REPORTED_USAGES = 50
# Excluded to keep the uploaded archive small, not as a security boundary.
_SKIPPED_DIRECTORY_NAMES = {".git", "__pycache__"}
_MANIFEST_NAME = "plugin.json"
# Noun phrases for error prose, keyed by the machine-readable operation name.
_OPERATION_NOUNS = {
"install": "installation",
"ensure_plugin_code_env": "code environment build",
"update": "update",
"delete": "deletion",
"rebuild_code_env": "code environment rebuild",
}
_FOLLOW_HINT = (
"Follow it with get_future_status(future_id, fetch_result=true). Do not start a "
"duplicate operation while it is running."
)
_USAGE_COLUMNS = [
"projectKey",
"objectType",
"objectId",
"elementKind",
"elementType",
]
_BLOCKER_FIXED_COLUMNS = ["kind", *_USAGE_COLUMNS, "pluginId", "missingType"]
_SUMMARY_COLUMNS = ["id", "version", "dev", "label", "deprecated"]
_DETAIL_COLUMNS = [
*_SUMMARY_COLUMNS,
"description",
"author",
"tags",
"category",
"support_level",
"license",
"code_env_name",
]
def _serialize_summary(raw: dict) -> dict:
"""Map one entry of Dataiku's installed-plugin listing to the MCP summary."""
meta = raw.get("meta") or {}
return {
"id": raw.get("id", ""),
"version": raw.get("version", ""),
"dev": bool(raw.get("isDev", False)),
"label": meta.get("label", ""),
"deprecated": bool(meta.get("deprecated", False)),
}
def _serialize_details(raw: dict, settings_raw: dict) -> dict:
"""Add catalog metadata and the bound code environment to a plugin summary.
Plugin-level ``config``, ``presets``, and ``parameterSets`` are deliberately never
read out: they routinely hold credentials, and parameter names are not a secrecy
boundary either.
"""
meta = raw.get("meta") or {}
return {
**_serialize_summary(raw),
"description": meta.get("description", ""),
"author": meta.get("author", ""),
"tags": meta.get("tags") or [],
"category": meta.get("category"),
"support_level": meta.get("supportLevel"),
"license": meta.get("licenseInfo", ""),
"code_env_name": settings_raw.get("codeEnvName") or None,
}
def _matches(raw: dict, query: str) -> bool:
"""Match a query against the identifiers a user is likely to know a plugin by."""
meta = raw.get("meta") or {}
values = [raw.get("id", ""), meta.get("label", ""), *(meta.get("tags") or [])]
return any(query in str(value).casefold() for value in values)
def _installed_plugin(client, plugin_id: str) -> dict:
"""Return one installed plugin's raw listing entry, or raise with the near misses."""
listing = client.list_plugins()
for raw in listing:
if raw.get("id") == plugin_id:
return raw
similar = sorted(
str(raw.get("id", ""))
for raw in listing
if plugin_id.casefold() in str(raw.get("id", "")).casefold()
)
message = f"Plugin '{plugin_id}' is not installed on this instance."
if similar:
message += f" Installed ids containing that text: {similar}."
raise ValueError(message)
def _require_absent(client, plugin_id: str) -> None:
if any(raw.get("id") == plugin_id for raw in client.list_plugins()):
raise ValueError(
f"Plugin '{plugin_id}' is already installed; use update_plugin."
)
def _archive_code_env(archive: io.BytesIO) -> dict:
"""Report which code-env descriptors the archive to upload declares."""
languages = []
with ZipFile(archive) as source:
# Dataiku's plugin layout names the R directory in upper case.
for language, directory in (("python", "python"), ("r", "R")):
descriptor = f"code-env/{directory}/desc.json"
if descriptor in source.namelist():
parse_json_object(source.read(descriptor).decode("utf-8"), descriptor)
languages.append(language)
archive.seek(0)
return {
"status": "required" if languages else "not_required",
"languages": languages,
}
def _find_manifest_member(archive: ZipFile) -> str:
"""Locate plugin.json at the ZIP root or inside a single wrapper directory.
Dataiku plugin exports put it at the root; a GitHub "Download ZIP" wraps
everything in one repository directory.
"""
names = [name for name in archive.namelist() if not name.endswith("/")]
if _MANIFEST_NAME in names:
return _MANIFEST_NAME
nested = [
name
for name in names
if name.count("/") == 1 and name.endswith(f"/{_MANIFEST_NAME}")
]
if len(nested) == 1:
return nested[0]
raise ValueError(
f"Plugin ZIP must contain {_MANIFEST_NAME} at its root or inside exactly one "
"top-level directory"
)
def _read_manifest_id(manifest_bytes: bytes) -> str:
manifest = parse_json_object(manifest_bytes.decode("utf-8"), _MANIFEST_NAME)
return _require_non_empty_string(
str(manifest.get("id", "")), f"{_MANIFEST_NAME} 'id'"
)
def _archive_from_directory(directory: Path) -> tuple[io.BytesIO, str]:
manifest = directory / _MANIFEST_NAME
if manifest.is_symlink():
raise ValueError(f"Plugin manifest '{manifest}' must not be a symlink")
if not manifest.is_file():
raise ValueError(
f"Plugin directory '{directory}' does not contain {_MANIFEST_NAME} at its root"
)
plugin_id = _read_manifest_id(manifest.read_bytes())
buffer = io.BytesIO()
with ZipFile(buffer, "w") as target:
for path in sorted(directory.rglob("*")):
relative = path.relative_to(directory)
if _SKIPPED_DIRECTORY_NAMES.intersection(relative.parts):
continue
if path.is_symlink():
raise ValueError(
f"Plugin directory '{directory}' contains symlink '{relative}'"
)
if path.is_file():
target.write(path, relative)
buffer.seek(0)
return buffer, plugin_id
def _archive_from_zip(zip_path: Path) -> tuple[io.BytesIO, str]:
try:
with ZipFile(zip_path) as source:
member = _find_manifest_member(source)
plugin_id = _read_manifest_id(source.read(member))
if member == _MANIFEST_NAME:
return io.BytesIO(zip_path.read_bytes()), plugin_id
# Dataiku expects plugin.json at the archive root, so drop the wrapper.
prefix = member[: -len(_MANIFEST_NAME)]
buffer = io.BytesIO()
with ZipFile(buffer, "w") as target:
for name in source.namelist():
if name.endswith("/") or not name.startswith(prefix):
continue
target.writestr(name[len(prefix) :], source.read(name))
except BadZipFile as exc:
raise ValueError(f"'{zip_path}' is not a readable ZIP archive") from exc
buffer.seek(0)
return buffer, plugin_id
def _local_target(plugin_id: str | None, local_path: str) -> tuple[io.BytesIO, str]:
"""Build an uploadable plugin ZIP and resolve which plugin it targets.
The archive's own manifest names the target, so a caller-supplied ``plugin_id`` is
only ever an assertion about it.
"""
path = Path(local_path).expanduser()
if path.is_dir():
buffer, archive_id = _archive_from_directory(path)
elif path.is_file():
buffer, archive_id = _archive_from_zip(path)
else:
raise ValueError(f"'local_path' does not exist: {local_path}")
if plugin_id is not None and plugin_id != archive_id:
raise ValueError(
f"'plugin_id' is '{plugin_id}' but {_MANIFEST_NAME} declares '{archive_id}'"
)
# requests names the multipart part from the stream's `name`; Dataiku's upload
# endpoints expect a filename to be present.
buffer.name = f"{archive_id}.zip"
return buffer, archive_id
def _resolve_source(
source: PluginSource, plugin_id: str | None, local_path: str | None
) -> tuple[str | None, str | None]:
"""Validate the argument ``source`` requires, and refuse local paths over HTTP."""
if source == "store":
if local_path is not None:
raise ValueError("'local_path' applies only to source='local_path'")
return _require_non_empty_string(plugin_id or "", "plugin_id"), None
if request.is_http_request():
raise ValueError(
"source='local_path' is unavailable in HTTP mode because it reads the MCP "
"host filesystem."
)
local_path = _require_non_empty_string(local_path or "", "local_path")
if plugin_id is not None:
plugin_id = _require_non_empty_string(plugin_id, "plugin_id")
return plugin_id, local_path
def _is_terminal(state: dict) -> bool:
"""Whether Dataiku has stopped working on a future.
``hasResult`` is the SDK's own completion predicate, and it also covers the case
where Dataiku answered inline with the result already in the start response. An
absent liveness flag is not evidence of completion, so polling continues.
"""
if state.get("hasResult", False) or state.get("aborted", False):
return True
return "alive" in state and not state["alive"]
async def _wait_for_future(future, timeout_seconds: int) -> tuple[bool, dict]:
"""Poll a Dataiku future until it stops running or the inline budget runs out.
The interval scales with the budget, so a ten-minute code-environment build is not
polled three hundred times while short waits stay responsive.
"""
interval = min(
max(_MIN_POLL_INTERVAL_SECONDS, timeout_seconds // 60),
_MAX_POLL_INTERVAL_SECONDS,
)
deadline = time.monotonic() + timeout_seconds
state = future.state or {}
while not _is_terminal(state):
remaining = deadline - time.monotonic()
if remaining <= 0:
return True, state
await asyncio.sleep(min(interval, remaining))
state = await run_blocking(future.get_state)
return False, state
def _terminal_result(operation: str, plugin_id: str | None, state: dict) -> dict:
"""Return a finished future's result, raising when it never produced one."""
noun = _OPERATION_NOUNS[operation]
target = f"plugin '{plugin_id}'" if plugin_id else "the Git plugin"
if state.get("aborted", False):
raise RuntimeError(f"Dataiku aborted the {noun} of {target}.")
if not state.get("hasResult", False):
raise RuntimeError(
f"Dataiku stopped the {noun} of {target} without reporting a "
"result. Inspect the plugin before retrying."
)
return state.get("result") or {}
def _successful_result(operation: str, plugin_id: str | None, state: dict) -> dict:
"""Return a completed action's result, raising when Dataiku reports it failed."""
result = _terminal_result(operation, plugin_id, state)
# Plugin actions can fail with HTTP 200 and success: false.
if result.get("success") is False:
error = result.get("installationError") or result.get("error") or {}
detail = error.get("message") or "Dataiku reported no error detail."
noun = _OPERATION_NOUNS[operation]
message = (
f"Dataiku reported the {noun} of plugin '{plugin_id}' failed: {detail}"
)
raise RuntimeError(message)
return result
def _build_outcome(result: dict, code_env_name: str) -> dict:
"""Report a code-environment build outcome instead of discarding the environment.
The environment exists by this point, so a failed dependency install is reported
alongside its name rather than raised.
"""
messages = result.get("messages")
if not isinstance(messages, dict) or not any(
key in messages for key in ("error", "fatal", "messages", "maxSeverity")
):
return {"status": "unknown", "code_env_name": code_env_name}
# messages.success can be false even on success; use error/fatal instead.
if not (messages.get("error") or messages.get("fatal")):
return {"status": "completed", "code_env_name": code_env_name}
reported = [
str(item.get("message") or item.get("details") or item.get("title") or "")
for item in messages.get("messages") or []
if item.get("severity") == "ERROR"
]
detail = "; ".join(filter(None, reported)) or "Dataiku reported no error detail."
return {
"status": "failed",
"code_env_name": code_env_name,
"error": detail,
# A rebuild keeps the interpreter and core-package set chosen at creation.
"hint": (
"Fix the plugin's requirements and rebuild. An interpreter or "
"core-package problem in desc.json needs the environment recreated in "
"the Dataiku UI."
),
}
def _in_flight(operation: str, plugin_id: str | None, future, started: bool) -> dict:
"""Describe an operation Dataiku is still working on.
The raw future state is deliberately left out: the caller is routed to
``get_future_status``, which exists to return it on demand.
"""
return omit_empty(
{
"status": "started" if started else "still_running",
"operation": operation,
"plugin_id": plugin_id,
"future_id": future.job_id,
"hint": _FOLLOW_HINT,
}
)
def _reload_flags(result: dict) -> dict:
return {
"needs_reload": bool(result.get("needsReload", False)),
"needs_restart": bool(result.get("needsRestart", False)),
}
def _deletion_blockers(raw: dict, plugin_id: str) -> list[dict]:
"""Return every plugin-specific record that blocks a normal deletion."""
usages = [{**item, "kind": "usage"} for item in raw.get("usages") or []]
unresolvable = [
{**item, "kind": "unresolvable_component"}
for item in raw.get("missingTypes") or []
if item.get("pluginId") == plugin_id
]
return [*usages, *unresolvable]
def _blocker_columns(blockers: list[dict]) -> list[str]:
"""Keep stable core columns while retaining every SDK-provided blocker field."""
fixed = set(_BLOCKER_FIXED_COLUMNS)
extras = sorted({key for blocker in blockers for key in blocker} - fixed)
return [*_BLOCKER_FIXED_COLUMNS, *extras]
async def _run_plugin_action(
operation: str,
plugin_id: str | None,
future,
wait_for_completion: bool,
timeout_seconds: int,
) -> tuple[dict | None, dict]:
"""Drive one plugin future to a reportable outcome.
Returns ``(result, payload)``. ``result`` is None when the action is still in
flight and ``payload`` describes how to follow it; otherwise ``payload`` carries
the reload flags of a verified success.
"""
if (
not wait_for_completion
and future.job_id
and not _is_terminal(future.state or {})
):
return None, _in_flight(operation, plugin_id, future, started=True)
timed_out, state = await _wait_for_future(future, timeout_seconds)
if timed_out:
return None, _in_flight(operation, plugin_id, future, started=False)
result = _successful_result(operation, plugin_id, state)
return result, omit_empty(
{
"status": "completed",
"operation": operation,
"plugin_id": plugin_id,
**_reload_flags(result),
}
)
@mcp.tool(
meta=ADMIN_POLICY,
title="List Installed Plugins",
description="Find installed plugins and inspect their metadata before an update or deletion.",
annotations={
"readOnlyHint": True,
"destructiveHint": False,
"openWorldHint": False,
},
)
async def list_plugins(
ctx: Context,
search: str = "",
search_mode: str = "partial",
dev: bool | None = None,
deprecated: bool | None = None,
include_details: bool = False,
offset: int = 0,
limit: int = 10,
) -> str:
"""List the plugins installed on the Dataiku instance.
Args:
search: Matches plugin id, label, and tags. Defaults to every plugin.
search_mode: ``partial`` for case-insensitive matching, or ``exact`` to
retrieve one plugin by its exact id. Ignored when ``search`` is empty.
dev: Keep only development plugins (true) or only regular ones (false).
deprecated: Keep only deprecated plugins (true) or only current ones (false).
include_details: Add catalog metadata and the bound code environment name.
Reads one plugin's settings per returned row, so pair it with ``search``
rather than listing every plugin in detail. Needs administrator or
plugin-developer permission.
offset: Zero-based offset within the matching plugins.
limit: Maximum plugins to return. Values above 100 are capped at 100.
"""
search = search.strip()
if search:
search_mode = _require_allowed_value(search_mode, "search_mode", _SEARCH_MODES)
offset = _require_non_negative_int(offset, "offset")
limit = min(_require_positive_int(limit, "limit"), 100)
await ctx.info("Listing installed Dataiku plugins...")
def _run():
client = get_dss_client()
listing = client.list_plugins()
matched = listing
if search and search_mode == "exact":
matched = [raw for raw in matched if raw.get("id") == search]
elif search:
query = search.casefold()
matched = [raw for raw in matched if _matches(raw, query)]
if dev is not None:
matched = [raw for raw in matched if bool(raw.get("isDev", False)) is dev]
if deprecated is not None:
matched = [
raw
for raw in matched
if bool((raw.get("meta") or {}).get("deprecated", False)) is deprecated
]
matched = sorted(matched, key=lambda raw: str(raw.get("id", "")).casefold())
page = matched[offset : offset + limit]
if include_details:
rows = [
_serialize_details(
raw, client.get_plugin(raw["id"]).get_settings().get_raw()
)
for raw in page
]
else:
rows = [_serialize_summary(raw) for raw in page]
return len(listing), len(matched), rows
total, matched, rows = await run_blocking(_run)
returned = len(rows)
return compact_json(
{
"total_plugins": total,
"matched_plugins": matched,
"returned_plugins": returned,
"next_offset": offset + returned if offset + returned < matched else None,
"plugins": columnar(
rows, _DETAIL_COLUMNS if include_details else _SUMMARY_COLUMNS
),
}
)
@mcp.tool(
meta=ADMIN_POLICY,
title="List Plugin Deletion Blockers",
description="Inspect every object or unresolved component that blocks deletion of an installed plugin.",
annotations={
"readOnlyHint": True,
"destructiveHint": False,
"openWorldHint": False,
},
)
async def list_plugin_usages(
plugin_id: str,
ctx: Context,
offset: int = 0,
limit: int = 50,
) -> str:
"""List every project object or unresolved component blocking plugin deletion.
Use this to inspect every page before requesting force deletion. Requires
administrator or plugin-developer permission.
Args:
plugin_id: Installed plugin id.
offset: Zero-based offset within deletion blockers.
limit: Maximum blockers to return. Values above 100 are capped at 100.
"""
plugin_id = _require_non_empty_string(plugin_id, "plugin_id")
offset = _require_non_negative_int(offset, "offset")
limit = min(_require_positive_int(limit, "limit"), 100)
await ctx.info(f"Listing deletion blockers for Dataiku plugin '{plugin_id}'...")
def _run():
client = get_dss_client()
_installed_plugin(client, plugin_id)
raw = client.get_plugin(plugin_id).list_usages().get_raw()
blockers = _deletion_blockers(raw, plugin_id)
return len(blockers), blockers[offset : offset + limit]
blocker_count, blockers = await run_blocking(_run)
returned = len(blockers)
return compact_json(
{
"plugin_id": plugin_id,
"blocker_count": blocker_count,
"returned_blockers": returned,
"next_offset": (
offset + returned if offset + returned < blocker_count else None
),
"blockers": columnar(blockers, _blocker_columns(blockers)),
}
)
def _git_error(operation: str, exc: Exception, future) -> RuntimeError:
# Git errors can include remote URLs, credentials, and subprocess output.
# Classify the problem, but never return the original text.
message = dataiku_message(exc).casefold()
auth_failure = any(
marker in message
for marker in (
"authentication failed",
"permission denied",
"publickey",
"could not read username",
"invalid username or password",
"unauthorized",
"not authorized",
)
)
problem = (
"Dataiku could not authenticate to or access the Git repository. "
if auth_failure
else f"The Git {_OPERATION_NOUNS[operation]} failed or its outcome could not be verified. "
)
recovery = (
f"Inspect future '{future.job_id}' with get_future_status before retrying. "
if future is not None and future.job_id
else "Check list_plugins before retrying. "
)
return RuntimeError(
problem
+ recovery
+ "Check the remote URL, checkout, and subpath. For Git access, an "
"administrator must allow the repository in Dataiku's Git settings "
"before access is possible. Configure SSH or HTTPS credentials in Dataiku. Credentials "
"on the Headless machine do not apply. See the plugin guide's Git access section."
)
async def _install_git(
repository_url: str,
checkout: str,
subpath: str | None,
wait_for_completion: bool,
timeout_seconds: int,
) -> str:
client = await run_blocking(get_dss_client)
future = None
try:
future = await run_blocking(
client.install_plugin_from_git, repository_url, checkout, subpath
)
result, payload = await _run_plugin_action(
"install", None, future, wait_for_completion, timeout_seconds
)
except Exception as exc:
raise _git_error("install", exc, future) from None
response = {**payload, "source": "git", "code_env": {"status": "unknown"}}
if result is None:
return compact_json(response)
# The remote manifest, not a caller-provided id, determines the installed plugin.
plugin_id = ((result.get("pluginDesc") or {}).get("desc") or {}).get("id")
if plugin_id:
try:
plugin = await run_blocking(
lambda: _serialize_summary(_installed_plugin(client, plugin_id))
)
except Exception:
pass
else:
return compact_json({**response, "plugin_id": plugin_id, "plugin": plugin})
return compact_json(
omit_empty(
{
**response,
"status": "verification_required",
"plugin_id": plugin_id,
"hint": "Dataiku reported installation completed. Verify the plugin with list_plugins before retrying or setting up its environment.",
}
)
)
@mcp.tool(
meta=ADMIN_POLICY,
title="Install Plugin",
description="Install a new instance plugin from the store, a local archive, or Git, then inspect its setup needs.",
annotations={
"readOnlyHint": False,
"destructiveHint": False,
"openWorldHint": False,
},
)
async def install_plugin(
source: PluginSource,
ctx: Context,
plugin_id: str | None = None,
local_path: str | None = None,
wait_for_completion: bool = True,
timeout_seconds: int = _DEFAULT_TIMEOUT_SECONDS,
repository_url: str | None = None,
checkout: str | None = None,
subpath: str | None = None,
) -> str:
"""Install a plugin that is not yet on the Dataiku instance.
Requires administrator or plugin-developer permission. ``code_env.status`` says
whether to follow up with ensure_plugin_code_env.
Args:
source: ``store``, ``local_path``, or ``git``. Dataiku fetches Git repositories.
plugin_id: Required for ``store``; optional manifest-id assertion for
``local_path``. Omit for ``git``; the remote manifest determines the id.
local_path: Plugin directory containing plugin.json at its root, or a plugin
ZIP. Required for ``local_path``.
wait_for_completion: Wait up to timeout_seconds for a verified outcome. When
false, return a future_id to follow instead.
timeout_seconds: Inline wait budget, checked between polls.
repository_url: Remote Git URL without inline credentials. Git source only.
checkout: Explicit branch, tag, or commit to fetch. Required for Git.
subpath: Optional "Path in repository": relative folder containing plugin.json.
Omit when plugin.json is at the repository root.
"""
timeout_seconds = _require_int_in_range(
timeout_seconds, "timeout_seconds", 1, _MAX_TIMEOUT_SECONDS
)
if source == "git":
if plugin_id is not None or local_path is not None:
raise ValueError(
"Omit 'plugin_id' and 'local_path' for source='git'; the remote plugin.json determines the id."
)
repository_url = _require_git_repository_url(repository_url or "")
checkout = _require_non_empty_string(checkout or "", "checkout")
if subpath is not None:
subpath = _require_relative_posix_path(subpath, "subpath")
await ctx.info("Installing Dataiku plugin from Git...")
return await _install_git(
repository_url, checkout, subpath, wait_for_completion, timeout_seconds
)
if any(value is not None for value in (repository_url, checkout, subpath)):
raise ValueError(
"'repository_url', 'checkout', and 'subpath' apply only to source='git'"
)
plugin_id, local_path = _resolve_source(source, plugin_id, local_path)
await ctx.info(f"Installing Dataiku plugin from {source}...")
def _start():
if source == "store":
client = get_dss_client()
_require_absent(client, plugin_id)
return (
client,
plugin_id,
client.install_plugin_from_store(plugin_id),
{"status": "unknown"},
)
archive, archive_id = _local_target(plugin_id, local_path)
requirements = _archive_code_env(archive)
client = get_dss_client()
_require_absent(client, archive_id)
return (
client,
archive_id,
client.start_install_plugin_from_archive(archive),
requirements,
)
client, resolved_id, future, requirements = await run_blocking(_start)
result, payload = await _run_plugin_action(
"install",
resolved_id,
future,
wait_for_completion,
timeout_seconds,
)
if result is None:
return compact_json({**payload, "source": source, "code_env": requirements})
plugin = await run_blocking(
lambda: _serialize_summary(_installed_plugin(client, resolved_id))
)
return compact_json(
{**payload, "source": source, "plugin": plugin, "code_env": requirements}
)
@mcp.tool(
meta=ADMIN_POLICY,
title="Update Plugin",
description="Update an installed plugin from the store, a local archive, or Git, optionally rebuilding its bound environment.",
annotations={
"readOnlyHint": False,
"destructiveHint": True,
"openWorldHint": False,
},
)
async def update_plugin(
source: PluginSource,
ctx: Context,
plugin_id: str | None = None,
local_path: str | None = None,
rebuild_code_env: bool = False,
wait_for_completion: bool = True,
timeout_seconds: int = _DEFAULT_TIMEOUT_SECONDS,
repository_url: str | None = None,
checkout: str | None = None,
subpath: str | None = None,
) -> str:
"""Update an installed plugin in place.
Requires administrator or plugin-developer permission. A store update needs the
plugin to have come from the store. Set ``rebuild_code_env`` after a change to the
plugin's dependencies; it rebuilds the bound environment from the plugin's own
specification and is reported separately under ``code_env_rebuild``, because the
plugin update has already landed by then.
Args:
source: ``store``, ``local_path``, or ``git``. Dataiku fetches Git repositories.
plugin_id: Installed plugin id. Required for ``store`` and ``git``. For ``local_path`` the
target is the id declared in plugin.json, and this argument, if given, must
match it.
local_path: Plugin directory containing plugin.json at its root, or a plugin
ZIP. Required for ``local_path``.
rebuild_code_env: Rebuild the plugin's bound code environment after the update.
wait_for_completion: Wait up to timeout_seconds for a verified outcome. When
false, return a future_id to follow instead, unless Dataiku already
answered with a completed result. Ignored while ``rebuild_code_env`` is
true, which needs the update to have completed.
timeout_seconds: Inline wait budget for each stage, checked between polls.
repository_url: Remote Git URL without inline credentials. Git source only.
checkout: Explicit branch, tag, or commit to fetch. Required for Git.
subpath: Optional "Path in repository": relative folder containing plugin.json.
Omit when plugin.json is at the repository root.
"""
if source == "git":
plugin_id = _require_non_empty_string(plugin_id or "", "plugin_id")
if local_path is not None:
raise ValueError("Omit 'local_path' for source='git'")
repository_url = _require_git_repository_url(repository_url or "")
checkout = _require_non_empty_string(checkout or "", "checkout")
if subpath is not None:
subpath = _require_relative_posix_path(subpath, "subpath")
else:
if any(value is not None for value in (repository_url, checkout, subpath)):
raise ValueError(
"'repository_url', 'checkout', and 'subpath' apply only to source='git'"
)
plugin_id, local_path = _resolve_source(source, plugin_id, local_path)
timeout_seconds = _require_int_in_range(
timeout_seconds, "timeout_seconds", 1, _MAX_TIMEOUT_SECONDS
)
if rebuild_code_env:
wait_for_completion = True
await ctx.info(f"Updating Dataiku plugin from {source}...")
def _start():
if source == "store":
client = get_dss_client()
_installed_plugin(client, plugin_id)
return client, plugin_id, client.get_plugin(plugin_id).update_from_store()
archive, archive_id = _local_target(plugin_id, local_path)
client = get_dss_client()
_installed_plugin(client, archive_id)
plugin = client.get_plugin(archive_id)
return client, archive_id, plugin.start_update_from_zip(archive)
if source == "git":
client = await run_blocking(get_dss_client)
await run_blocking(_installed_plugin, client, plugin_id)
resolved_id = plugin_id
future = None
try:
future = await run_blocking(
lambda: client.get_plugin(plugin_id).update_from_git(
repository_url, checkout, subpath
)
)
result, payload = await _run_plugin_action(
"update", resolved_id, future, wait_for_completion, timeout_seconds
)
except Exception as exc:
raise _git_error("update", exc, future) from None
if result is not None:
actual_id = ((result.get("pluginDesc") or {}).get("desc") or {}).get("id")
if actual_id != resolved_id:
return compact_json(
omit_empty(
{
**payload,
"status": "verification_required",
"source": source,
"reported_plugin_id": actual_id,
"hint": "Dataiku did not report the expected plugin id. Inspect list_plugins before retrying. No environment rebuild was started.",
}
)
)
else:
client, resolved_id, future = await run_blocking(_start)
result, payload = await _run_plugin_action(
"update", resolved_id, future, wait_for_completion, timeout_seconds
)
if result is None:
in_flight = {**payload, "source": source}
if rebuild_code_env:
# A rebuild needs the update to have landed, so it never started. Saying
# so keeps the caller from reading its absence as a completed rebuild.
in_flight["code_env_rebuild"] = {
"status": "not_started",
"reason": "The update has not completed, and a rebuild needs it to.",
"hint": (
"Follow the update to successful completion with "
"get_future_status(future_id, fetch_result=true), then ask the "
"user to rebuild the existing environment in the Dataiku UI."
),
}
return compact_json(in_flight)
plugin = await run_blocking(
lambda: _serialize_summary(_installed_plugin(client, resolved_id))
)
response = {**payload, "source": source, "plugin": plugin}
if rebuild_code_env:
response["code_env_rebuild"] = await _rebuild_code_env(
client, resolved_id, timeout_seconds
)
return compact_json(omit_empty(response))
async def _rebuild_code_env(client, plugin_id: str, timeout_seconds: int) -> dict:
"""Rebuild a plugin's bound code environment, reporting rather than raising.
The plugin update has already landed when this runs, so a rebuild problem is
reported alongside the successful update instead of discarding it.
"""
bound = None
future = None
def _inspect():
plugin = client.get_plugin(plugin_id)
return plugin, plugin.get_settings().get_raw().get("codeEnvName")
try:
plugin, bound = await run_blocking(_inspect)
if not bound:
return {
"status": "skipped",
"reason": "No code environment is bound to this plugin.",
"hint": (
"Call ensure_plugin_code_env to check whether the plugin needs an "
"environment and create or recover its managed environment."
),
}
future = await run_blocking(plugin.update_code_env)
timed_out, state = await _wait_for_future(future, timeout_seconds)
if timed_out:
return {
"status": "still_running",
"code_env_name": bound,
"future_id": future.job_id,
"hint": _FOLLOW_HINT,
}
result = _terminal_result("rebuild_code_env", plugin_id, state)
return _build_outcome(result, bound)
except Exception:
future_id = future.job_id if future is not None else None
return omit_empty(
{
"status": "unknown",
"code_env_name": bound,
"future_id": future_id,
"error": "The plugin update completed, but its code-environment rebuild could not be verified.",
"hint": (
"Do not repeat update_plugin. "
+ (
"Check this future_id with get_future_status before taking further action."
if future_id
else "Inspect the plugin's bound environment in Dataiku before rebuilding; a rebuild may already have started."
)
),
}
)
def _managed_env_name(plugin_id: str, env_name: str) -> bool:
return bool(
re.fullmatch(
re.escape(f"plugin_{plugin_id}_managed") + r"(?:_[0-9]+)?", env_name
)
)
def _managed_env_names(client, plugin_id: str) -> list[str]:
return sorted(
raw["envName"]
for raw in client.list_code_envs()
if raw.get("deploymentMode") == "PLUGIN_MANAGED"
and _managed_env_name(plugin_id, raw.get("envName", ""))
)
def _build_owner_key(client, future_id: str) -> tuple:
return (request.get_request_owner(), client.host, future_id)
def _resume_build(
client, plugin_id: str, future_id: str, retained_owner: str | None = None
) -> DSSFuture:
"""Verify this plugin's build before reading its result.
Dataiku forgets a finished future once its result is read, so peek first.
"""
future = DSSFuture(client, future_id)
try:
state = future.peek_state()
except DataikuException as exc:
if dataiku_message(exc) != "No future by that id":
raise
raise ValueError(
f"Dataiku has no future '{future.job_id}'; its result was already read. "
"Bind the environment with code_env_name."
) from None
payload = state.get("payload") or {}
target = str(payload.get("displayName", "")).rpartition(": ")[2]
identified = _managed_env_name(plugin_id, target) or (
payload.get("displayName") == "Import env" and retained_owner == plugin_id
)
if payload.get("action") != "import_env" or not identified:
raise ValueError(
f"Future '{future.job_id}' is not a code environment build of plugin "
f"'{plugin_id}'."
)
future.get_state()
return future
def _start_build(plugin, plugin_id: str, python_interpreter: str | None):
"""Start the managed-env build, or return None when the plugin declares none."""
try:
return plugin.create_code_env(python_interpreter=python_interpreter)
except DataikuException as exc:
# The SDK may omit the error code, leaving only the endpoint's message.
message = dataiku_message(exc)
if re.search(r"\bERR_PLUGIN_WITHOUT_CODEENV\b", message) or message == (
f"The plugin {plugin_id} doesn't specify a code env, cannot create it"
):
return None
raise RuntimeError(
f"Dataiku refused plugin code-environment creation: {message}"
) from None
async def _bind_code_env(plugin, env_name: str, build: dict, created: bool) -> dict:
"""Bind an environment unless another binding appeared meanwhile, and report."""
def _bind():
settings = plugin.get_settings()
current = settings.get_raw().get("codeEnvName")
if current and current != env_name:
return current
settings.set_code_env(env_name)
settings.save()
return None
try:
changed = await run_blocking(_bind)
except DataikuException as exc:
return {
"status": "binding_failed",
"code_env_name": env_name,
"build": build,
"error": dataiku_message(exc),
"hint": "Fix the cause, then retry with this code_env_name.",
}
except RequestException:
return {
"status": "binding_unknown",
"code_env_name": env_name,
"build": build,
"hint": "The binding request lost its connection. Inspect the plugin binding, then retry with this code_env_name if needed.",
}
if changed:
return {
"status": "binding_changed",
"code_env_name": env_name,
"bound_code_env_name": changed,
"build": build,
}
return {
"status": "completed",
"created": created,
"code_env_name": env_name,
"build": build,
}
@mcp.tool(
meta=ADMIN_POLICY,
title="Ensure Plugin Code Environment",
description="Create and bind a plugin-managed code environment, resume its build, or bind an existing unbound one.",
annotations={
"readOnlyHint": False,
"destructiveHint": False,
"openWorldHint": False,
},
)
async def ensure_plugin_code_env(
plugin_id: str,
ctx: Context,
python_interpreter: Annotated[
PythonInterpreter,
Field(
description="Omit to let Dataiku select an interpreter based on the plugin's "
"accepted versions and server availability. Only applies to a new environment."
),
] = None,
future_id: str | None = None,
code_env_name: str | None = None,
timeout_seconds: int = _BUILD_TIMEOUT_SECONDS,
) -> str:
"""Create and bind a plugin's managed code environment, or finish an interrupted setup.
Args:
plugin_id: Installed plugin id.
python_interpreter: Omit to let Dataiku select an interpreter based on the
plugin's accepted versions and server availability. Only applies to a
new environment.
future_id: Resume a build this tool returned as still_running. Dataiku
forgets a build once its result is read, so this works only while it runs.
code_env_name: Bind this existing, unbound PLUGIN_MANAGED environment of the
plugin instead of creating one, as listed by recovery_required.
timeout_seconds: Inline wait budget for a build, checked between polls.
"""
plugin_id = _require_non_empty_string(plugin_id, "plugin_id")
timeout_seconds = _require_int_in_range(
timeout_seconds, "timeout_seconds", 1, _MAX_TIMEOUT_SECONDS
)
if future_id is not None:
future_id = _require_non_empty_string(future_id, "future_id")
if code_env_name is not None:
code_env_name = _require_non_empty_string(code_env_name, "code_env_name")
if (
sum(
value is not None
for value in (python_interpreter, future_id, code_env_name)
)
> 1
):
raise ValueError(
"Pass at most one of python_interpreter, future_id, and code_env_name."
)
await ctx.info(f"Preparing the code environment of plugin '{plugin_id}'...")
def _inspect():
client = get_dss_client()
_installed_plugin(client, plugin_id)
plugin = client.get_plugin(plugin_id)
return client, plugin, plugin.get_settings().get_raw().get("codeEnvName")
client, plugin, bound = await run_blocking(_inspect)
common = {"operation": "ensure_plugin_code_env", "plugin_id": plugin_id}
if future_id:
future = await run_blocking(
_resume_build,
client,
plugin_id,
future_id,
_BUILD_OWNERS.get(_build_owner_key(client, future_id)),
)
elif bound:
if python_interpreter or code_env_name not in (None, bound):
raise ValueError(
f"Plugin '{plugin_id}' is already bound to '{bound}'; nothing was "
"changed."
)
return compact_json(
{
**common,
"status": "completed",
"created": False,
"code_env_name": bound,
"build": {"status": "unknown"},
}
)
else:
candidates = await run_blocking(_managed_env_names, client, plugin_id)
if code_env_name:
if code_env_name not in candidates:
raise ValueError(
f"'{code_env_name}' is not an existing managed environment of "
f"plugin '{plugin_id}'. Candidates: {candidates}."
)
bound_env = await _bind_code_env(
plugin, code_env_name, {"status": "unknown"}, created=False
)
return compact_json({**common, **bound_env})
# An unbound environment may still be building, so neither guess nor
# create a duplicate.
if candidates:
return compact_json(
{
**common,
"status": "recovery_required",
"code_env_names": candidates,
"hint": "Resume its future_id if still building; otherwise bind one with code_env_name.",
}
)
future = await run_blocking(_start_build, plugin, plugin_id, python_interpreter)
if future is None:
return compact_json(
{
**common,
"status": "not_required",
}
)
if future.job_id:
if len(_BUILD_OWNERS) >= 256:
_BUILD_OWNERS.pop(next(iter(_BUILD_OWNERS)))
_BUILD_OWNERS[_build_owner_key(client, future.job_id)] = plugin_id
timed_out, state = await _wait_for_future(future, timeout_seconds)
if timed_out:
return compact_json(
{
**_in_flight(
"ensure_plugin_code_env", plugin_id, future, started=False
),
"hint": "Call ensure_plugin_code_env again with this future_id; do not start another build.",
}
)
result = _terminal_result("ensure_plugin_code_env", plugin_id, state)
_BUILD_OWNERS.pop(_build_owner_key(client, future.job_id), None)
env_name = _require_non_empty_string(
str(result.get("envName", "")), "created code environment name"
)
if not _managed_env_name(plugin_id, env_name):
raise ValueError(
"The build returned another plugin's environment; no binding was changed."
)
build = _build_outcome(result, env_name)
return compact_json(
{**common, **await _bind_code_env(plugin, env_name, build, created=True)}
)
def _unresolved_components_refusal(plugin_id: str, reason: str) -> dict:
"""Refuse a deletion Dataiku will not perform until it can resolve the plugin.
Dataiku raises this from the usage analysis and from the delete endpoint itself,
and it is the usual state for a plugin installed or updated since the last backend
reload. Its own wording says the deletion may be forced, which is exactly what the
``force`` flag is for.
"""
return {
"status": "refused",
"operation": "delete",
"plugin_id": plugin_id,
"deleted": False,
"error": "Dataiku could not resolve this plugin's components.",
"reason": reason,
"hint": (
"Reloading the Dataiku backend clears the common case, a plugin installed "
"or updated since the last reload. Otherwise pass force=true to delete "
"without the usage check."
),
}
@mcp.tool(
meta=ADMIN_POLICY,
title="Delete Plugin",
description="Remove an installed plugin after checking whether its components are in use.",
annotations={
"readOnlyHint": False,
"destructiveHint": True,
"openWorldHint": False,
},
)
async def delete_plugin(
plugin_id: str,
ctx: Context,
force: bool = False,
timeout_seconds: int = _DEFAULT_TIMEOUT_SECONDS,
) -> str:
"""Delete an installed plugin, refusing while its components are still in use.
Requires administrator or plugin-developer permission. Usages are checked first and
deletion is refused with ``deleted: false`` when any are found. ``force=true``
deletes anyway, which breaks every listed object. Do not infer that deletion
happened until ``deleted`` is true.
Args:
plugin_id: Installed plugin id.
force: Delete despite usages or a failed usage analysis.
timeout_seconds: Inline wait budget, checked between polls.
"""
plugin_id = _require_non_empty_string(plugin_id, "plugin_id")
timeout_seconds = _require_int_in_range(
timeout_seconds, "timeout_seconds", 1, _MAX_TIMEOUT_SECONDS
)
await ctx.info(f"Deleting Dataiku plugin '{plugin_id}'...")
def _inspect():
client = get_dss_client()
_installed_plugin(client, plugin_id)
plugin = client.get_plugin(plugin_id)
try:
raw = plugin.list_usages().get_raw()
except Exception as exc:
# Dataiku raises here when it cannot resolve the plugin's component types,
# the same condition its delete endpoint enforces.
return plugin, None, None, dataiku_message(exc)
usages = raw.get("usages") or []
# Dataiku reports unresolvable component types for the whole instance here,
# most of them unrelated to this plugin.
unresolvable = [
item
for item in raw.get("missingTypes") or []
if item.get("pluginId") == plugin_id
]
return plugin, usages, unresolvable, None
plugin, usages, unresolvable, analysis_error = await run_blocking(_inspect)
if analysis_error is not None and not force:
if "may be forced" not in analysis_error:
raise RuntimeError(
f"Dataiku could not analyze usages of plugin '{plugin_id}': "
f"{analysis_error}"
)
return compact_json(_unresolved_components_refusal(plugin_id, analysis_error))
if (usages or unresolvable) and not force:
return compact_json(
omit_empty(
{
"status": "refused",
"operation": "delete",
"plugin_id": plugin_id,
"deleted": False,
"error": "Plugin cannot be deleted because it is still in use.",
"usage_count": len(usages),
"blocker_count": len(usages) + len(unresolvable),
"usages": columnar(usages[:_MAX_REPORTED_USAGES], _USAGE_COLUMNS),
"unresolvable_components": unresolvable[:_MAX_REPORTED_USAGES],
"hint": (
"Inspect every blocker with list_plugin_usages before removing "
"or replacing them. Pass force=true only after the user has "
"reviewed the complete impact and explicitly asked to delete "
"any remaining blockers."
),
}
)
)
try:
future = await run_blocking(lambda: plugin.delete(force=force))
except Exception as exc:
# Dataiku enforces the same resolvability check on the delete endpoint, so
# this is reachable even when the usage analysis itself came back clean.
message = dataiku_message(exc)
if force or "may be forced" not in message:
raise
return compact_json(_unresolved_components_refusal(plugin_id, message))
timed_out, state = await _wait_for_future(future, timeout_seconds)
if timed_out:
return compact_json(
{**_in_flight("delete", plugin_id, future, started=False), "deleted": False}
)
result = _successful_result("delete", plugin_id, state)
return compact_json(
{
"status": "completed",
"operation": "delete",
"plugin_id": plugin_id,
"deleted": True,
"forced": force,
**_reload_flags(result),
}
)
SHA-256: d83b503026dcc3648f96eb76a440f26eb51c0d70f3203c7fa194d71a182ed446