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",
)
# OAuth protected resource metadata, published at
# /.well-known/oauth-protected-resource. Comma-separated where a list is
# expected, so every value stays a plain string that a runtime parameter or
# container env var can carry.
mcp_oauth_resource: str | None = Field(
default=None,
description="RFC 9728 `resource`: the canonical URI of this MCP server",
)
mcp_oauth_authorization_servers: str | None = Field(
default=None,
description="RFC 9728 `authorization_servers`, comma-separated",
)
# `scopes_supported` is deliberately not configurable: it is always derived
# from the scope requirements actually enforced, so the document can never
# advertise something the server does not check. Declare requirements with
# require_scopes(...) on the tool's auth= or MCP_OAUTH_TAG_SCOPES_<TAG> instead.
# The default is spelled out rather than left to None so it is visible
# wherever settings are listed. The type stays nullable: callers construct
# this config with an explicit None, and ScopeSource.parse reads None, "" and
# an unrecognised value all as `both` — refusing to start would turn a typo
# in one variable into an outage.
mcp_oauth_scope_source: str | None = Field(
default="both",
description=(
"Which scope declaration mechanism is live: `both` (the default — each "
"applies wherever it is declared), `code` (only the in-code require_scopes "
"declarations), or `tags` (only the MCP_OAUTH_TAG_SCOPES_<TAG> variables)"
),
)
# Per-tag scope requirements are one variable per tag —
# `MCP_OAUTH_TAG_SCOPES_<TAG>=scope,scope` — so they cannot be declared as
# fixed fields here. See drmcp.core.oauth_scopes.read_tag_scopes.
#
# There is no enforcement setting either: declared scopes are enforced per
# tools/call by the scope-validation middleware (see drmcp.core.middleware) and,
# against the same parsed token, by the checks on the components themselves,
# both gated by `mcp_enable_oauth_claim_validation` below (no token is parsed
# while it is off, and a tokenless request is admitted).
mcp_xaa_trusted_issuer: str | None = Field(
default=None,
description="Cross-Application Access token exchange trusted issuer",
)
mcp_xaa_exchange_audience: str | None = Field(
default=None,
description="Cross-Application Access token exchange audience",
)
mcp_xaa_token_url: str | None = Field(
default=None,
description="Cross-Application Access token request URL",
)
mcp_xaa_token_audience: str | None = Field(
default=None,
description="Cross-Application Access token request audience (optional)",
)
mcp_xaa_scopes: str | None = Field(
default=None,
description="Cross-Application Access token request scopes, comma-separated",
)
mcp_xaa_token_endpoint_auth_method: str | None = Field(
default=None,
description="Cross-Application Access token endpoint auth method",
)
mcp_enable_unauthenticated_well_known_route: bool = Field(
default=False,
description="Enable/disable unauthenticated well known route",
)
# 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,
)
mcp_enable_oauth_claim_validation: bool = Field(
default=False,
description="Enable OAuth claim validation (e.g., audience, scope)",
)