class MCPServerConfig(DataRobotAppFrameworkBaseSettings):
"""MCP Server configuration.
Extends ``DataRobotAppFrameworkBaseSettings`` so each field resolves from env
vars (including ``MLOPS_RUNTIME_PARAM_`` runtime parameters), ``.env``, file
secrets, and ``pulumi_config.json``. Fields map by name: ``mcp_server_name``
reads ``MCP_SERVER_NAME``.
"""
mcp_server_name: str = Field(
default="datarobot-mcp-server",
description="Name of the MCP server",
)
mcp_server_port: int = Field(
default=8080,
description="Port number for the MCP server",
)
mcp_server_log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = Field(
default="WARNING",
description="Log level for the MCP server",
)
mcp_server_host: str = Field(
default="0.0.0.0",
description="Host address for the MCP server",
)
app_log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = Field(
default="INFO",
description="App log level",
)
# When the server is run in a custom model, it is important to mount all routes under the
# prefix provided in the URL_PREFIX
mount_path: str = Field(default="/", alias="URL_PREFIX")
@staticmethod
def _get_default_otel_endpoint() -> str:
"""Get the default OpenTelemetry endpoint e.g. https://app.datarobot.com/otel."""
parsed_url = urlparse(os.environ.get("DATAROBOT_ENDPOINT", DEFAULT_DATAROBOT_ENDPOINT))
stripped_url = (parsed_url.scheme, parsed_url.netloc, "otel", "", "", "")
return urlunparse(stripped_url)
otel_collector_base_url: str = Field(
default=_get_default_otel_endpoint(),
description="Base URL for the OpenTelemetry collector",
)
otel_entity_id: str = Field(
default="",
description="Entity ID for tracing",
)
otel_attributes: dict[str, Any] = Field(
default={},
description="Attributes for tracing (as JSON string)",
)
otel_enabled: bool = Field(
default=True,
description="Enable/disable OpenTelemetry",
)
otel_enabled_http_instrumentors: bool = Field(
default=False,
description="Enable/disable HTTP instrumentors",
)
otel_exporter_otlp_endpoint: str = Field(
default="",
description="Standard OTel OTLP endpoint. Takes priority over otel_collector_base_url.",
)
otel_exporter_otlp_headers: str = Field(
default="",
description="Standard OTel OTLP headers. Takes priority over entity_id construction.",
)
@field_validator("otel_exporter_otlp_headers", mode="before")
@classmethod
def _assemble_otel_headers(cls, v: object, info: ValidationInfo) -> object:
if v:
return v
entity_id = (info.data or {}).get("otel_entity_id", "")
api_token = os.environ.get("DATAROBOT_API_TOKEN", "")
if entity_id and api_token:
return f"x-datarobot-entity-id={entity_id},x-datarobot-api-key={api_token}"
return v
mcp_server_register_dynamic_tools_on_startup: bool = Field(
default=False,
description="Register dynamic tools on startup. When enabled, the MCP server will "
"automatically register all DataRobot tool deployments as MCP tools during startup.",
)
mcp_server_tool_registration_allow_empty_schema: bool = Field(
default=False,
description="Allow registration of tools with no input parameters. When enabled, "
"tools can be registered with empty schemas for static endpoints that don't require any "
"inputs. "
"Disabled by default, as this is not typical use case and can hide potential issues with "
"schema.",
)
mcp_server_tool_registration_duplicate_behavior: DuplicateBehavior = Field(
default="warn",
description="Behavior when a tool with the same name already exists in the MCP server. "
" - 'warn': will log a warning and replace the existing tool. "
" - 'replace': will replace the existing tool without a warning. "
" - 'error': will raise an error and prevent registration. "
" - 'ignore': will skip registration of the new tool.",
)
mcp_server_register_dynamic_prompts_on_startup: bool = Field(
default=False,
description="Register dynamic prompts on startup. When enabled, the MCP server will "
"automatically register all prompts from DataRobot Prompt Management "
"as MCP prompts during startup.",
)
mcp_server_prompt_registration_duplicate_behavior: DuplicateBehavior = Field(
default="warn",
description="Behavior when a prompt with the same name already exists in the MCP server. "
" - 'warn': will log a warning and replace the existing tool. "
" - 'replace': will replace the existing tool without a warning. "
" - 'error': will raise an error and prevent registration. "
" - 'ignore': will skip registration of the new tool.",
)
mcp_cli_configs: str | None = Field(
default=None,
description="Comma-separated list of features to enable: dynamic_tools, dynamic_prompts, "
"predictive, gdrive, microsoft_graph, jira, confluence, perplexity, tavily. "
"When unset (None), defaults apply. When set to empty string, all listed features are "
"disabled. Individual env vars (e.g. ENABLE_PREDICTIVE_TOOLS) take precedence when set.",
)
tool_config: MCPToolConfig = Field(
default_factory=MCPToolConfig,
description="Tool configuration",
)
# Treat empty env values as unset so a runtime parameter (resolved by the base
# settings sources) is not shadowed by an empty plain env var.
model_config = SettingsConfigDict(
env_file=".env",
extra="ignore",
env_ignore_empty=True,
)