Skip to content

datarobot_genai.drmcp.core.dynamic_prompts.controllers

controllers

register_prompt_from_prompt_template_id_and_version async 

register_prompt_from_prompt_template_id_and_version(prompt_template_id: str, prompt_template_version_id: str | None) -> Prompt

Register a Prompt for a specific prompt template ID and version.

Parameters:

Name Type Description Default
prompt_template_id str

The ID of the DataRobot prompt template.

 required
prompt_template_version_id str | None

Optional ID of the DataRobot prompt template version. If not provided latest will be used

 required
Raises
DynamicPromptRegistrationError: If registration fails at any step.
Returns
The registered Prompt instance.
Source code in datarobot_genai/drmcp/core/dynamic_prompts/controllers.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
async def register_prompt_from_prompt_template_id_and_version(
    prompt_template_id: str, prompt_template_version_id: str | None
) -> Prompt:
    """Register a Prompt for a specific prompt template ID and version.

    Args:
        prompt_template_id: The ID of the DataRobot prompt template.
        prompt_template_version_id: Optional ID of the DataRobot prompt template version.
            If not provided latest will be used

    Raises
    ------
        DynamicPromptRegistrationError: If registration fails at any step.

    Returns
    -------
        The registered Prompt instance.
    """
    prompt_template = get_datarobot_prompt_template(prompt_template_id)

    if not prompt_template:
        raise DynamicPromptRegistrationError("Registration failed. Could not find prompt template.")

    if not prompt_template_version_id:
        registered_prompt = await register_prompt_from_datarobot_prompt_management(
            prompt_template=prompt_template
        )
        try:
            linear_manager = LineageManager(mcp)
            await linear_manager.sync_mcp_prompts()
        except LRSEnvVarIsNotSetError as error:
            error_message = f"MCP item metadata is not sync. {str(error)}"
            logger.warning(error_message)
        return registered_prompt

    prompt_template_version = get_datarobot_prompt_template_version(
        prompt_template_id, prompt_template_version_id
    )

    if not prompt_template_version:
        raise DynamicPromptRegistrationError(
            "Registration failed. Could not find prompt template version."
        )

    registered_prompt = await register_prompt_from_datarobot_prompt_management(
        prompt_template=prompt_template, prompt_template_version=prompt_template_version
    )
    try:
        linear_manager = LineageManager(mcp)
        await linear_manager.sync_mcp_prompts()
    except LRSEnvVarIsNotSetError as error:
        error_message = f"MCP item metadata is not sync. {str(error)}"
        logger.warning(error_message)
    return registered_prompt

delete_registered_prompt_template async 

delete_registered_prompt_template(prompt_template_id: str) -> bool

Delete the prompt registered for the prompt template id in the MCP instance.

Source code in datarobot_genai/drmcp/core/dynamic_prompts/controllers.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
async def delete_registered_prompt_template(prompt_template_id: str) -> bool:
    """Delete the prompt registered for the prompt template id in the MCP instance."""
    prompt_templates_mappings = await mcp.get_prompt_mapping()
    if prompt_template_id not in prompt_templates_mappings:
        logger.debug(f"No prompt registered for prompt template id {prompt_template_id}")
        return False

    prompt_template_version_id, prompt_name = prompt_templates_mappings[prompt_template_id]
    await mcp.remove_prompt_mapping(prompt_template_id, prompt_template_version_id)
    logger.info(
        f"Deleted prompt name {prompt_name} for prompt template id {prompt_template_id}, "
        f"version {prompt_template_version_id}"
    )
    try:
        linear_manager = LineageManager(mcp)
        await linear_manager.sync_mcp_prompts()
    except LRSEnvVarIsNotSetError as error:
        error_message = f"MCP item metadata is not sync. {str(error)}"
        logger.warning(error_message)
    return True

refresh_registered_prompt_template async 

refresh_registered_prompt_template(headers_auth_only: bool = False) -> None

Refresh all registered prompt templates in the MCP instance.

Source code in datarobot_genai/drmcp/core/dynamic_prompts/controllers.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
async def refresh_registered_prompt_template(headers_auth_only: bool = False) -> None:
    """Refresh all registered prompt templates in the MCP instance."""
    prompt_templates = get_datarobot_prompt_templates()
    prompt_templates_ids = {p.id for p in prompt_templates}
    prompt_templates_versions = get_datarobot_prompt_template_versions(
        list(prompt_templates_ids), headers_auth_only=headers_auth_only
    )

    mcp_prompt_templates_mappings = await mcp.get_prompt_mapping()

    for prompt_template in prompt_templates:
        prompt_template_versions = prompt_templates_versions.get(prompt_template.id)
        if not prompt_template_versions:
            continue

        latest_version = max(prompt_template_versions, key=lambda v: v.version)

        if prompt_template.id not in mcp_prompt_templates_mappings:
            # New prompt template -> add
            await register_prompt_from_datarobot_prompt_management(
                prompt_template=prompt_template, prompt_template_version=latest_version
            )
            continue

        mcp_prompt_template_version, mcp_prompt = mcp_prompt_templates_mappings[prompt_template.id]

        if mcp_prompt_template_version != latest_version:
            # Current version saved in MCP is not the latest one => update it
            await register_prompt_from_datarobot_prompt_management(
                prompt_template=prompt_template, prompt_template_version=latest_version
            )
            continue

        # Else => mcp_prompt_template_version == latest_version
        # For now it means nothing changed as there's no possibility to edit promp template version.

    for mcp_prompt_template_id, (
        mcp_prompt_template_version_id,
        _,
    ) in mcp_prompt_templates_mappings.items():
        if mcp_prompt_template_id not in prompt_templates_ids:
            # We need to also delete prompt templates that are
            await mcp.remove_prompt_mapping(mcp_prompt_template_id, mcp_prompt_template_version_id)

    try:
        linear_manager = LineageManager(mcp)
        await linear_manager.sync_mcp_prompts()
    except LRSEnvVarIsNotSetError as error:
        error_message = f"MCP item metadata is not sync. {str(error)}"
        logger.warning(error_message)