Skip to content

datarobot_genai.drmcp.core.dynamic_tools.deployment.register

register

register_tools_of_datarobot_deployments async 

register_tools_of_datarobot_deployments() -> None

Register tools for all deployments tagged as 'tool' in DataRobot.

Source code in datarobot_genai/drmcp/core/dynamic_tools/deployment/register.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
async def register_tools_of_datarobot_deployments() -> None:
    """Register tools for all deployments tagged as 'tool' in DataRobot."""
    deployment_ids = get_datarobot_tool_deployments()
    logger.info(f"Found {len(deployment_ids)} tool deployments.")

    # Try to register each tool deployment, continue on failure
    for deployment_id in deployment_ids:
        try:
            deployment = dr.Deployment.get(deployment_id)
            await register_tool_of_datarobot_deployment(deployment)
        except DynamicToolRegistrationError:
            pass
        except Exception as exc:
            logger.error(f"Unexpected error for deployment {deployment_id}: {exc}")
            pass

register_tool_of_datarobot_deployment async 

register_tool_of_datarobot_deployment(deployment: Deployment) -> Tool

Register a single tool for a given deployment.

Parameters:

Name Type Description Default
deployment Deployment

The tool deployment within DataRobot.

 required
Raises
DynamicToolRegistrationError: If registration fails at any step.
Returns
The registered Tool instance.
Source code in datarobot_genai/drmcp/core/dynamic_tools/deployment/register.py 
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
async def register_tool_of_datarobot_deployment(
    deployment: dr.Deployment,
) -> Tool:
    """Register a single tool for a given deployment.

    Args:
        deployment: The tool deployment within DataRobot.

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

    Returns
    -------
        The registered Tool instance.
    """
    logger.info(f"Found tool: id: {deployment.id}, label: {deployment.label}")

    try:
        # Create the configuration object with all necessary information
        # This includes fetching metadata and assembling all deployment-specific data
        config = create_deployment_tool_config(deployment)

        # Register using generic external tool registration with the config
        tool = await register_external_tool(config, deployment_id=deployment.id)

        return tool

    except Exception as exc:
        logger.error(f"Skipping deployment {deployment.id}. Registration failed: {exc}")
        raise DynamicToolRegistrationError("Registration failed. Could not create tool.") from exc

get_datarobot_tool_deployments 

get_datarobot_tool_deployments() -> list[str]

Fetch deployments from DataRobot that are tagged as 'tool'.

Source code in datarobot_genai/drmcp/core/dynamic_tools/deployment/register.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def get_datarobot_tool_deployments() -> list[str]:
    """Fetch deployments from DataRobot that are tagged as 'tool'."""
    # Replace this with dr.Deployment.list when the 3.9.0 version
    # of datarobot python SDK is released.
    with request_user_dr_client(headers_auth_only=False) as client:
        deployments_data = dr.utils.pagination.unpaginate(
            initial_url="deployments/",
            initial_params={"tag_values": "tool", "tag_keys": "tool"},
            client=client,
        )

    # The API filters with OR logic for tags, so we need to filter for AND logic
    valid_deployment_ids = []
    for deployment in deployments_data:
        for tag in deployment.get("tags", []):
            if tag.get("name") == "tool" and tag.get("value") == "tool":
                valid_deployment_ids.append(deployment["id"])
                break

    return valid_deployment_ids