Skip to content

datarobot_genai.drmcp.core.dynamic_tools.deployment.controllers

controllers

register_tool_for_deployment_id async 

register_tool_for_deployment_id(deployment_id: str) -> Tool

Register a tool for a specific deployment ID.

Parameters:

Name Type Description Default
deployment_id str

The ID of the DataRobot deployment to register as a tool.

 required
Raises
DynamicToolRegistrationError: If registration fails at any step.
Returns
The registered Tool instance.
Source code in datarobot_genai/drmcp/core/dynamic_tools/deployment/controllers.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
async def register_tool_for_deployment_id(deployment_id: str) -> Tool:
    """Register a tool for a specific deployment ID.

    Args:
        deployment_id: The ID of the DataRobot deployment to register as a tool.

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

    Returns
    -------
        The registered Tool instance.
    """
    with request_user_dr_sdk(headers_auth_only=True):
        deployment = dr.Deployment.get(deployment_id)
    registered_tool = await register_tool_of_datarobot_deployment(deployment)
    try:
        linear_manager = LineageManager(mcp)
        await linear_manager.sync_mcp_tools()
    except LRSEnvVarIsNotSetError as error:
        error_message = f"MCP item metadata is not sync. {str(error)}"
        logger.warning(error_message)

    return registered_tool

get_registered_tool_deployments async 

get_registered_tool_deployments() -> dict[str, str]

Get the tool registered for the deployment in the MCP instance.

Source code in datarobot_genai/drmcp/core/dynamic_tools/deployment/controllers.py 
58
59
60
61
async def get_registered_tool_deployments() -> dict[str, str]:
    """Get the tool registered for the deployment in the MCP instance."""
    deployments = await mcp.get_deployment_mapping()
    return deployments

delete_registered_tool_deployment async 

delete_registered_tool_deployment(deployment_id: str) -> bool

Delete the tool registered for the deployment in the MCP instance.

Source code in datarobot_genai/drmcp/core/dynamic_tools/deployment/controllers.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
async def delete_registered_tool_deployment(deployment_id: str) -> bool:
    """Delete the tool registered for the deployment in the MCP instance."""
    deployments = await mcp.get_deployment_mapping()
    if deployment_id not in deployments:
        logger.debug(f"No tool registered for deployment {deployment_id}")
        return False

    tool_name = deployments[deployment_id]
    await mcp.remove_deployment_mapping(deployment_id)
    logger.info(f"Deleted tool {tool_name} for deployment {deployment_id}")
    try:
        linear_manager = LineageManager(mcp)
        await linear_manager.sync_mcp_tools()
    except LRSEnvVarIsNotSetError as error:
        error_message = f"MCP item metadata is not sync. {str(error)}"
        logger.warning(error_message)
    return True