feat(scenarios): create Python-script scenarios and edit their script

CODING_STANDARDS_AND_STRUCTURE.md

@@ -103,9 +103,9 @@ export DKU_API_KEY="your-api-key"
 - Give a fixed-value parameter a `Literal` so its allowed values ride in the schema, and drop the `require_allowed_value` check it replaces. Keep the check and describe the values in prose instead when it normalizes input (for example `.upper()`), since a `Literal` would reject values that work today, or when the value set is large enough that listing it in every catalog send costs more than a description.
 
 ## Cobuild Write-Routing Convention
-- This server intentionally does not expose direct create/update/delete tools for in-project flow and analytic assets (recipes, ML analyses, dashboards, insights, agents, agent tools, scenarios, webapps, wiki articles, data quality rules, knowledge banks, semantic models, evaluation stores). Project-level building goes through `dataiku_mcp/tools/cobuild.py`'s Cobuild conversation tools instead. `set_container_exec_config` is the narrow, placement-only exception; all other changes stay with Cobuild.
+- This server intentionally does not expose direct create/update/delete tools for in-project flow and analytic assets (recipes, ML analyses, dashboards, insights, agents, agent tools, scenarios, webapps, wiki articles, data quality rules, knowledge banks, semantic models, evaluation stores). Project-level building goes through `dataiku_mcp/tools/cobuild.py`'s Cobuild conversation tools instead. `set_container_exec_config` is the narrow, placement-only exception, and `create_python_scenario` and `set_python_scenario_script` are the narrow Python-script scenario exceptions; all other changes stay with Cobuild.
 - Do not add a new direct write tool for an in-project asset type beyond the fixed exceptions below. If a gap in Cobuild's coverage is found, note it in the relevant SKILL.md rather than adding an MCP write tool around it.
-- A new direct write tool is only justified when the operation is cross-project, instance-level, or must happen before a project/Cobuild conversation exists. The fixed bootstrap exceptions are `create_project`, `create_upload_dataset`, `create_managed_folder`, `upload_file_to_managed_folder`, `write_project_library_file`, and `set_project_variables`; all but `create_project` write into a project, and each creates a container or carries local content rather than building logic. The other fixed exceptions are deterministic execution of existing assets (`build_datasets`, `run_recipe`, `run_scenario`, `abort_job`, which stops an existing job, and `get_dataset_metrics`, which recomputes a dataset's existing metric probes so a read does not return stale stored values) and deterministic container execution placement for an existing object (`set_container_exec_config`).
+- A new direct write tool is only justified when the operation is cross-project, instance-level, or must happen before a project/Cobuild conversation exists. The fixed bootstrap exceptions are `create_project`, `create_upload_dataset`, `create_managed_folder`, `upload_file_to_managed_folder`, `write_project_library_file`, and `set_project_variables`; all but `create_project` write into a project, and each creates a container or carries local content rather than building logic. The other fixed exceptions are deterministic execution of existing assets (`build_datasets`, `run_recipe`, `run_scenario`, `abort_job`, which stops an existing job, and `get_dataset_metrics`, which recomputes a dataset's existing metric probes so a read does not return stale stored values), deterministic container execution placement for an existing object (`set_container_exec_config`), and Python-script scenario authoring (`create_python_scenario`, which creates an inactive scenario without triggers, and `set_python_scenario_script`), because Cobuild creates only step-based scenarios and cannot edit a scenario's script.
 
 ## Fixed Tool Surface
 

README.md

@@ -134,7 +134,7 @@ npx skills add dataiku/dataiku-headless
 
 Dataiku Headless is an MCP server and agent skill library for operating Dataiku from an AI agent. Connect it to a Dataiku instance to inspect projects, gather context, and use Cobuild—Dataiku's agent for building data pipelines, analytics, machine learning models, multi-agent workflows, applications, and automation pipelines.
 
-Cobuild runs as a retained conversation through MCP tools. This repository intentionally keeps its own tool surface small: inspection tools, three deterministic executions of existing assets (`build_datasets`, `run_recipe`, and `run_scenario`), narrow configuration actions such as selecting which container an existing object runs in, and a few bootstrap actions that Cobuild cannot perform, such as creating a project or uploading a local file.
+Cobuild runs as a retained conversation through MCP tools. This repository intentionally keeps its own tool surface small: inspection tools, three deterministic executions of existing assets (`build_datasets`, `run_recipe`, and `run_scenario`), narrow configuration actions such as selecting which container an existing object runs in, Python-script scenario authoring that Cobuild does not support, and a few bootstrap actions that Cobuild cannot perform, such as creating a project or uploading a local file.
 
 ## Capability reference
 
@@ -296,7 +296,7 @@ The Dataiku Headless marketplace plugin uses stdio transport. For customer-manag
 │   │   ├── plugins.py         # Instance plugin lifecycle and managed code environments
 │   │   ├── project_folders.py # Project folder hierarchy inspection and organization tools
 │   │   ├── projects.py        # Project inspection, creation, variables, and settings
-│   │   ├── scenarios.py       # Scenario/run-history/messaging-channel inspection tools
+│   │   ├── scenarios.py       # Scenario inspection, runs, and Python-script scenarios
 │   │   ├── semantic_models.py # Semantic model inspection tools
 │   │   ├── groups.py          # Instance group administration tools
 │   │   ├── licensing.py       # Instance licensing status inspection tool

dataiku_mcp/tools/scenarios.py

@@ -12,17 +12,19 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Scenario inspection and execution tools for Dataiku."""
+"""Scenario inspection, execution, and Python-script scenario authoring tools."""
 
 import asyncio
 import time
 from typing import Annotated
 
+from dataikuapi.utils import DataikuException
 from fastmcp import Context
 from pydantic import Field
 
 from ..instance_policy import (
     ADMIN_POLICY,
+    PROJECT_EDIT_POLICY,
     PROJECT_EXECUTION_POLICY,
     PROJECT_READ_POLICY,
 )
@@ -42,6 +44,42 @@
 SCENARIO_RUN_ID_RESOLVE_BUDGET_SECONDS = 4
 MAX_SCENARIO_WAIT_SECONDS = 3600
 
+PythonScript = Annotated[
+    str, Field(description="Complete Python source; replaces the whole script.")
+]
+
+
+def _python_scenario_settings(scenario, scenario_id: str):
+    """Read one scenario's settings, refusing a scenario without a Python script."""
+    settings = scenario.get_settings()
+    scenario_type = settings.get_raw().get("type")
+    if scenario_type != "custom_python":
+        raise ValueError(
+            f"Scenario '{scenario_id}' is a '{scenario_type}' scenario; only "
+            "Python-script (custom_python) scenarios have a script"
+        )
+    return settings
+
+
+def _save_script(settings, script: str) -> None:
+    # The SDK saves the definition before the script. That definition save runs
+    # Dataiku's code-permission and license checks, which the script endpoint
+    # alone does not, so a refused caller never reaches the script write.
+    settings.code = script
+    settings.save()
+
+
+def _python_scenario_summary(project_key: str, settings) -> dict:
+    raw = settings.get_raw()
+    return {
+        "project_key": project_key,
+        "scenario_id": raw.get("id"),
+        "name": raw.get("name"),
+        "type": raw.get("type"),
+        "active": raw.get("active", False),
+        "trigger_count": len(raw.get("triggers") or []),
+    }
+
 
 async def _resolve_scenario_run_id(trigger_fire, budget_seconds: int):
     """Resolve the scenario run id after a trigger fires."""
@@ -148,21 +186,149 @@ async def get_scenario_settings(
     scenario_id: str,
     ctx: Context,
 ) -> str:
-    """Read what a scenario does, on what trigger, and who it reports to."""
+    """Read a scenario's steps or Python script, its triggers, and who it reports to."""
     project_key = _require_non_empty_string(project_key, "project_key")
     scenario_id = _require_non_empty_string(scenario_id, "scenario_id")
     await ctx.info(f"Fetching settings for scenario {scenario_id} in {project_key}...")
 
-    raw = await run_blocking(
-        lambda: (
-            get_dss_client()
-            .get_project(project_key)
-            .get_scenario(scenario_id)
-            .get_settings()
-            .get_raw()
+    def _run():
+        scenario = get_dss_client().get_project(project_key).get_scenario(scenario_id)
+        settings = scenario.get_settings()
+        raw = settings.get_raw()
+        if raw.get("type") == "custom_python":
+            return {**raw, "script": settings.code}
+        return raw
+
+    return compact_json(await run_blocking(_run))
+
+
+@mcp.tool(
+    meta=PROJECT_EDIT_POLICY,
+    title="Create Python Scenario",
+    annotations={
+        "readOnlyHint": False,
+        "destructiveHint": False,
+        "idempotentHint": False,
+        "openWorldHint": False,
+    },
+)
+async def create_python_scenario(
+    project_key: str,
+    name: str,
+    script: PythonScript,
+    ctx: Context,
+) -> str:
+    """Create a scenario defined by a Python script, inactive and without triggers."""
+    project_key = _require_non_empty_string(project_key, "project_key")
+    name = _require_non_empty_string(name, "name")
+    # Validate only: the script is saved exactly as given.
+    _require_non_empty_string(script, "script")
+    await ctx.info(f"Creating Python scenario '{name}' in {project_key}...")
+
+    def _run():
+        project = get_dss_client().get_project(project_key)
+        # Dataiku accepts duplicate names, so a retried call would otherwise
+        # leave a second scenario holding Dataiku's sample script.
+        existing = next(
+            (item for item in project.list_scenarios() if item.get("name") == name),
+            None,
         )
-    )
-    return compact_json(raw)
+        if existing is not None:
+            raise ValueError(
+                f"Project '{project_key}' already has a scenario named '{name}' "
+                f"(id '{existing.get('id')}'). Choose another name, or use "
+                "set_python_scenario_script to change that scenario's script"
+            )
+        try:
+            scenario = project.create_scenario(
+                name,
+                "custom_python",
+                definition={
+                    "params": {},
+                    "active": False,
+                    "triggers": [],
+                    "reporters": [],
+                },
+            )
+        except DataikuException:
+            # Dataiku answered and refused, so nothing was created.
+            raise
+        except Exception as exc:
+            raise RuntimeError(
+                f"Dataiku did not confirm creating scenario '{name}'. Check "
+                "list_scenarios before retrying because the request may have "
+                "reached Dataiku."
+            ) from exc
+
+        # Dataiku creates the scenario with a sample script, so until the real
+        # script is confirmed the result must say the scenario is not ready.
+        error = None
+        try:
+            _save_script(_python_scenario_settings(scenario, scenario.id), script)
+            saved = _python_scenario_settings(scenario, scenario.id)
+            if saved.code == script:
+                return {
+                    "status": "python_scenario_created",
+                    **_python_scenario_summary(project_key, saved),
+                }
+        except Exception as exc:
+            error = exc
+        return {
+            "status": "python_scenario_script_unconfirmed",
+            "project_key": project_key,
+            "scenario_id": scenario.id,
+            "name": name,
+            **(
+                {"error_type": type(error).__name__, "error": _safe_error_text(error)}
+                if error is not None
+                else {}
+            ),
+            "hint": (
+                "The scenario exists, inactive and without triggers, but its script "
+                "was not confirmed and may still be Dataiku's sample script. Do not "
+                "run it. Call set_python_scenario_script with this scenario_id and "
+                "the same script; do not create the scenario again."
+            ),
+        }
+
+    return compact_json(await run_blocking(_run))
+
+
+@mcp.tool(
+    meta=PROJECT_EDIT_POLICY,
+    title="Set Python Scenario Script",
+    annotations={
+        "readOnlyHint": False,
+        "destructiveHint": True,
+        "idempotentHint": True,
+        "openWorldHint": False,
+    },
+)
+async def set_python_scenario_script(
+    project_key: str,
+    scenario_id: str,
+    script: PythonScript,
+    ctx: Context,
+) -> str:
+    """Replace the script of an existing Python-script scenario, keeping its settings."""
+    project_key = _require_non_empty_string(project_key, "project_key")
+    scenario_id = _require_non_empty_string(scenario_id, "scenario_id")
+    # Validate only: the script is saved exactly as given.
+    _require_non_empty_string(script, "script")
+    await ctx.info(f"Saving the script of scenario {scenario_id} in {project_key}...")
+
+    def _run():
+        scenario = get_dss_client().get_project(project_key).get_scenario(scenario_id)
+        _save_script(_python_scenario_settings(scenario, scenario_id), script)
+        saved = _python_scenario_settings(scenario, scenario_id)
+        if saved.code != script:
+            raise RuntimeError(
+                f"Dataiku did not confirm the new script of scenario '{scenario_id}'. "
+                "Inspect it with get_scenario_settings; repeating this call is safe."
+            )
+        return _python_scenario_summary(project_key, saved)
+
+    return compact_json(await run_blocking(_run))
 
 
 @mcp.tool(