This is an automated email from the ASF dual-hosted git repository.
FreeOnePlus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-mcp-server.git
The following commit(s) were added to refs/heads/master by this push:
new c0c56cc refactor: centralize tool definitions (#145)
c0c56cc is described below
commit c0c56cc695db578d1ed1252be6607c708f5a6a82
Author: Yijia Su <[email protected]>
AuthorDate: Thu Jul 30 12:07:13 2026 +0800
refactor: centralize tool definitions (#145)
---
CHANGELOG.md | 3 +
README.md | 85 +-
docs/tool-registry.md | 33 +
doris_mcp_server/auth/doris_oauth_scope_policy.py | 45 +-
doris_mcp_server/auth/operation_policy.py | 90 +--
doris_mcp_server/tools/tool_registry.py | 356 +++++++++
doris_mcp_server/tools/tools_manager.py | 897 ++--------------------
doris_mcp_server/utils/config.py | 21 +-
test/protocol/test_mcp_v2_protocol.py | 96 +++
test/protocol/tool_registry_server.py | 95 +++
test/tools/test_tool_registry.py | 200 +++++
11 files changed, 956 insertions(+), 965 deletions(-)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index d955da7..dc4da6c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -54,6 +54,9 @@ under **Unreleased** until a new version is selected and
published.
capability-honest.
- Added bounded JSON Schema 2020-12 validation for tool definitions, arguments,
and declared structured outputs without external reference fetching.
+- Consolidated tool schemas, execution handlers, Doris OAuth policy, safe
+ audit metadata, compatibility aliases, and generated documentation in one
+ validated Tool Definition Registry.
- Positioned Dynamic Client Registration as a compatibility fallback behind
preconfigured clients and Client ID Metadata Documents.
- Persisted static bearer tokens as self-describing SHA-256/SHA-512 digests,
diff --git a/README.md b/README.md
index a1e409f..df0b48b 100644
--- a/README.md
+++ b/README.md
@@ -342,39 +342,12 @@ cp .env.example .env
### Available MCP Tools
-The following table lists the main tools currently available for invocation
via an MCP client:
-
-| Tool Name | Description
| Parameters |
-|-----------------------------|--------------------------------------------------------------|--------------------------------------------------------------|
-| `exec_query` | Execute SQL query and return results.
| `sql` (string, Required), `db_name` (string, Optional),
`catalog_name` (string, Optional), `max_rows` (integer, Optional), `timeout`
(integer, Optional) |
-| `get_table_schema` | Get detailed table structure information.
| `table_name` (string, Required), `db_name` (string, Optional),
`catalog_name` (string, Optional) |
-| `get_db_table_list` | Get list of all table names in specified
database. | `db_name` (string, Optional), `catalog_name` (string,
Optional) |
-| `get_db_list` | Get list of all database names.
| `catalog_name` (string, Optional) |
-| `get_table_comment` | Get table comment information.
| `table_name` (string, Required), `db_name` (string, Optional),
`catalog_name` (string, Optional) |
-| `get_table_column_comments` | Get comment information for all columns in
table. | `table_name` (string, Required), `db_name` (string,
Optional), `catalog_name` (string, Optional) |
-| `get_table_indexes` | Get index information for specified table.
| `table_name` (string, Required), `db_name` (string, Optional),
`catalog_name` (string, Optional) |
-| `get_recent_audit_logs` | Get audit log records for recent period.
| `days` (integer, Optional), `limit` (integer, Optional) |
-| `get_catalog_list` | Get list of all catalog names.
| `random_string` (string, Required) |
-| `get_sql_explain` | Get SQL execution plan with configurable
content truncation and file export for LLM analysis. | `sql`
(string, Required), `verbose` (boolean, Optional), `db_name` (string,
Optional), `catalog_name` (string, Optional) |
-| `get_sql_profile` | Get SQL execution profile with content
management and file export for LLM optimization workflows. |
`sql` (string, Required), `db_name` (string, Optional), `catalog_name` (string,
Optional), `timeout` (integer, Optional) |
-| `get_table_data_size` | Get table data size information via FE HTTP
API. | `db_name` (string, Optional), `table_name` (string, Optional),
`single_replica` (boolean, Optional) |
-| `get_monitoring_metrics_info` | Get Doris monitoring metrics definitions and
descriptions. | `role` (string, Optional), `monitor_type` (string, Optional),
`priority` (string, Optional) |
-| `get_monitoring_metrics_data` | Get actual Doris monitoring metrics data
from nodes with flexible BE discovery. | `role` (string, Optional),
`monitor_type` (string, Optional), `priority` (string, Optional) |
-| `get_realtime_memory_stats` | Get real-time memory statistics via BE Memory
Tracker with auto/manual BE discovery. | `tracker_type` (string, Optional),
`include_details` (boolean, Optional) |
-| `get_historical_memory_stats` | Get historical memory statistics via BE Bvar
interface with flexible BE configuration. | `tracker_names` (array,
Optional), `time_range` (string, Optional) |
-| `analyze_data_quality` | Comprehensive data quality analysis combining
completeness and distribution analysis. | `table_name` (string, Required),
`analysis_scope` (string, Optional), `sample_size` (integer, Optional),
`business_rules` (array, Optional) |
-| `trace_column_lineage` | End-to-end column lineage tracking through SQL
analysis and dependency mapping. | `target_columns` (array, Required),
`analysis_depth` (integer, Optional), `include_transformations` (boolean,
Optional) |
-| `monitor_data_freshness` | Real-time data staleness monitoring with
configurable freshness thresholds. | `table_names` (array, Optional),
`freshness_threshold_hours` (integer, Optional), `include_update_patterns`
(boolean, Optional) |
-| `analyze_data_access_patterns` | User behavior analysis and security anomaly
detection with access pattern monitoring. | `days` (integer, Optional),
`include_system_users` (boolean, Optional), `min_query_threshold` (integer,
Optional) |
-| `analyze_data_flow_dependencies` | Data flow impact analysis and dependency
mapping between tables and views. | `target_table` (string, Optional),
`analysis_depth` (integer, Optional), `include_views` (boolean, Optional) |
-| `analyze_slow_queries_topn` | Performance bottleneck identification with
top-N slow query analysis and patterns. | `days` (integer, Optional), `top_n`
(integer, Optional), `min_execution_time_ms` (integer, Optional),
`include_patterns` (boolean, Optional) |
-| `analyze_resource_growth_curves` | Capacity planning with resource growth
analysis and trend forecasting. | `days` (integer, Optional), `resource_types`
(array, Optional), `include_predictions` (boolean, Optional) |
-| `exec_adbc_query` | High-performance SQL execution using ADBC (Arrow Flight
SQL) protocol. | `sql` (string, Required), `max_rows` (integer, Optional),
`timeout` (integer, Optional), `return_format` (string, Optional) |
-| `get_adbc_connection_info` | ADBC connection diagnostics and status
monitoring for Arrow Flight SQL. | No parameters required |
-
-**Note:** All metadata tools support catalog federation for multi-catalog
environments. Enhanced monitoring tools provide comprehensive memory tracking
and metrics collection capabilities. **New in v0.5.0**: 7 advanced analytics
tools for enterprise data governance and 2 ADBC tools for high-performance data
transfer with 3-10x performance improvements for large datasets.
-
-**Doris-backed OAuth note:** The table above describes global server
capabilities. Doris-backed OAuth uses configuration gates for its operation
surface. MCP resources are available with resource metadata caching disabled.
Reviewed metadata tools are callable when `DORIS_OAUTH_DB_TOOLS_ENABLED=true`;
`exec_query` and `get_sql_explain` are callable when their Doris OAuth
query/explain gates are enabled. These MySQL-channel operations run through the
logged-in Doris user pool, so Doris RBA [...]
+The canonical catalog is generated from the same `ToolDefinitionRegistry` used
+for MCP JSON Schemas, execution dispatch, authorization policy, and audit
+metadata. See [docs/tool-registry.md](docs/tool-registry.md). The checked-in
+catalog is verified against the runtime registry by the test suite.
+
+**Doris-backed OAuth note:** The generated catalog describes global server
capabilities. Doris-backed OAuth uses configuration gates for its operation
surface. MCP resources are available with resource metadata caching disabled.
Reviewed metadata tools are callable when `DORIS_OAUTH_DB_TOOLS_ENABLED=true`;
`exec_query` and `get_sql_explain` are callable when their Doris OAuth
query/explain gates are enabled. These MySQL-channel operations run through the
logged-in Doris user pool, so Dor [...]
### 4. Run the Service
@@ -1440,14 +1413,20 @@ The server provides comprehensive utility modules for
common database operations
### 2. Implement Tool Logic
-Add your new tool to the `DorisToolsManager` class in
`doris_mcp_server/tools/tools_manager.py`. The tools manager provides a
centralized approach to tool registration and execution with unified interfaces.
+Add a private handler to `DorisToolsManager` in
+`doris_mcp_server/tools/tools_manager.py`. Handler names follow
+`_<tool_name>_tool`; the registry resolves this name and validates that the
+handler exists during manager construction.
**Example:** Adding a new analysis tool:
```python
# In doris_mcp_server/tools/tools_manager.py
-async def your_new_analysis_tool(self, arguments: Dict[str, Any]) ->
List[Dict[str, Any]]:
+async def _your_new_analysis_tool(
+ self,
+ arguments: dict[str, Any],
+) -> dict[str, Any]:
"""
Your new analysis tool implementation
@@ -1455,7 +1434,7 @@ async def your_new_analysis_tool(self, arguments:
Dict[str, Any]) -> List[Dict[s
arguments: Tool arguments from MCP client
Returns:
- List of MCP response messages
+ JSON-serializable tool result
"""
try:
# Use existing utilities
@@ -1464,30 +1443,24 @@ async def your_new_analysis_tool(self, arguments:
Dict[str, Any]) -> List[Dict[s
max_rows=arguments.get("max_rows", 100)
)
- return [{
- "type": "text",
- "text": json.dumps(result, ensure_ascii=False, indent=2)
- }]
+ return result
except Exception as e:
logger.error(f"Tool execution failed: {str(e)}", exc_info=True)
- return [{
- "type": "text",
- "text": f"Error: {str(e)}"
- }]
+ return {"success": False, "error": "Analysis failed"}
```
-### 3. Register the Tool
+### 3. Add the Registry Definition
-Add your tool to the `_register_tools` method in the same class:
+Add one `Tool` schema to `DorisToolsManager._build_tool_registry`, then
classify
+its policy once in `doris_mcp_server/tools/tool_registry.py`. Do not add a
+decorator wrapper or an `if/elif` dispatch branch:
```python
-# In the _register_tools method of DorisToolsManager
-
[email protected](
+Tool(
name="your_new_analysis_tool",
description="Description of your new analysis tool",
- inputSchema={
+ input_schema={
"type": "object",
"properties": {
"parameter1": {
@@ -1500,13 +1473,17 @@ Add your tool to the `_register_tools` method in the
same class:
"default": 100
}
},
- "required": ["parameter1"]
- }
+ "required": ["parameter1"],
+ },
)
-async def your_new_analysis_tool_wrapper(arguments: Dict[str, Any]) ->
List[Dict[str, Any]]:
- return await self.your_new_analysis_tool(arguments)
```
+The registry derives the execution handler, safe audit fields, and generated
+documentation from that definition. Add the name to exactly one policy class:
+metadata, query, explain, or restricted. Run the registry tests and refresh
+`docs/tool-registry.md` from `ToolDefinitionRegistry.render_markdown()`; the
+test suite rejects documentation drift.
+
### 4. Advanced Features
For more complex tools, you can leverage the comprehensive framework:
diff --git a/docs/tool-registry.md b/docs/tool-registry.md
new file mode 100644
index 0000000..2001621
--- /dev/null
+++ b/docs/tool-registry.md
@@ -0,0 +1,33 @@
+# Doris MCP Tool Registry
+
+<!-- Generated from ToolDefinitionRegistry; do not edit by hand. -->
+
+| Tool | Policy | Risk | Handler | Audit event | Parameters |
+|---|---|---|---|---|---|
+| `exec_query` | `query` | `query` | `_exec_query_tool` |
`mcp.tool.call.exec_query` | `catalog_name`, `db_name`, `max_rows`, `sql`*,
`timeout` |
+| `get_table_schema` | `metadata` | `metadata` | `_get_table_schema_tool` |
`mcp.tool.call.get_table_schema` | `catalog_name`, `db_name`, `table_name`* |
+| `get_db_table_list` | `metadata` | `metadata` | `_get_db_table_list_tool` |
`mcp.tool.call.get_db_table_list` | `catalog_name`, `db_name` |
+| `get_db_list` | `metadata` | `metadata` | `_get_db_list_tool` |
`mcp.tool.call.get_db_list` | `catalog_name` |
+| `get_table_comment` | `metadata` | `metadata` | `_get_table_comment_tool` |
`mcp.tool.call.get_table_comment` | `catalog_name`, `db_name`, `table_name`* |
+| `get_table_column_comments` | `metadata` | `metadata` |
`_get_table_column_comments_tool` | `mcp.tool.call.get_table_column_comments` |
`catalog_name`, `db_name`, `table_name`* |
+| `get_table_indexes` | `metadata` | `metadata` | `_get_table_indexes_tool` |
`mcp.tool.call.get_table_indexes` | `catalog_name`, `db_name`, `table_name`* |
+| `get_recent_audit_logs` | `restricted` | `high` |
`_get_recent_audit_logs_tool` | `mcp.tool.call.get_recent_audit_logs` | `days`,
`limit` |
+| `get_catalog_list` | `metadata` | `metadata` | `_get_catalog_list_tool` |
`mcp.tool.call.get_catalog_list` | `random_string`* |
+| `get_sql_explain` | `explain` | `explain` | `_get_sql_explain_tool` |
`mcp.tool.call.get_sql_explain` | `catalog_name`, `db_name`, `sql`*, `verbose` |
+| `get_sql_profile` | `restricted` | `high` | `_get_sql_profile_tool` |
`mcp.tool.call.get_sql_profile` | `catalog_name`, `db_name`, `sql`*, `timeout` |
+| `get_table_data_size` | `restricted` | `high` | `_get_table_data_size_tool`
| `mcp.tool.call.get_table_data_size` | `db_name`, `single_replica`,
`table_name` |
+| `get_monitoring_metrics` | `restricted` | `high` |
`_get_monitoring_metrics_tool` | `mcp.tool.call.get_monitoring_metrics` |
`content_type`, `include_raw_metrics`, `monitor_type`, `priority`, `role` |
+| `get_memory_stats` | `restricted` | `high` | `_get_memory_stats_tool` |
`mcp.tool.call.get_memory_stats` | `data_type`, `include_details`,
`time_range`, `tracker_names`, `tracker_type` |
+| `get_table_basic_info` | `restricted` | `high` |
`_get_table_basic_info_tool` | `mcp.tool.call.get_table_basic_info` |
`catalog_name`, `db_name`, `table_name`* |
+| `analyze_columns` | `restricted` | `high` | `_analyze_columns_tool` |
`mcp.tool.call.analyze_columns` | `analysis_types`, `catalog_name`, `columns`*,
`db_name`, `detailed_response`, `sample_size`, `table_name`* |
+| `analyze_table_storage` | `restricted` | `high` |
`_analyze_table_storage_tool` | `mcp.tool.call.analyze_table_storage` |
`catalog_name`, `db_name`, `detailed_response`, `table_name`* |
+| `trace_column_lineage` | `restricted` | `high` |
`_trace_column_lineage_tool` | `mcp.tool.call.trace_column_lineage` |
`analysis_depth`, `catalog_name`, `include_transformations`, `target_columns`* |
+| `monitor_data_freshness` | `restricted` | `high` |
`_monitor_data_freshness_tool` | `mcp.tool.call.monitor_data_freshness` |
`catalog_name`, `db_name`, `freshness_threshold_hours`,
`include_update_patterns`, `table_names` |
+| `analyze_data_access_patterns` | `restricted` | `high` |
`_analyze_data_access_patterns_tool` |
`mcp.tool.call.analyze_data_access_patterns` | `days`, `include_system_users`,
`min_query_threshold` |
+| `analyze_data_flow_dependencies` | `restricted` | `high` |
`_analyze_data_flow_dependencies_tool` |
`mcp.tool.call.analyze_data_flow_dependencies` | `analysis_depth`,
`catalog_name`, `db_name`, `include_views`, `target_table` |
+| `analyze_slow_queries_topn` | `restricted` | `high` |
`_analyze_slow_queries_topn_tool` | `mcp.tool.call.analyze_slow_queries_topn` |
`days`, `include_patterns`, `min_execution_time_ms`, `top_n` |
+| `analyze_resource_growth_curves` | `restricted` | `high` |
`_analyze_resource_growth_curves_tool` |
`mcp.tool.call.analyze_resource_growth_curves` | `days`, `detailed_response`,
`include_predictions`, `resource_types` |
+| `exec_adbc_query` | `restricted` | `high` | `_exec_adbc_query_tool` |
`mcp.tool.call.exec_adbc_query` | `max_rows`, `return_format`, `sql`*,
`timeout` |
+| `get_adbc_connection_info` | `restricted` | `high` |
`_get_adbc_connection_info_tool` | `mcp.tool.call.get_adbc_connection_info` |
None |
+
+Required parameters are marked with `*`. Tool descriptions and JSON Schemas
are exposed directly by MCP `tools/list` from the same registry entries.
diff --git a/doris_mcp_server/auth/doris_oauth_scope_policy.py
b/doris_mcp_server/auth/doris_oauth_scope_policy.py
index 29c5128..517a5f5 100644
--- a/doris_mcp_server/auth/doris_oauth_scope_policy.py
+++ b/doris_mcp_server/auth/doris_oauth_scope_policy.py
@@ -4,6 +4,11 @@
from collections.abc import Iterable
from typing import TYPE_CHECKING
+from ..tools.tool_registry import (
+ DORIS_OAUTH_EXPLAIN_TOOL_SET,
+ DORIS_OAUTH_METADATA_TOOL_NAMES,
+ DORIS_OAUTH_QUERY_TOOL_SET,
+)
from .doris_oauth_types import TokenEndpointError
if TYPE_CHECKING:
@@ -12,15 +17,7 @@ if TYPE_CHECKING:
BASE_DORIS_OAUTH_SCOPES = frozenset({"tool:list"})
RESOURCE_SCOPES = frozenset({"resource:list", "resource:read"})
-DORIS_OAUTH_METADATA_TOOLS = (
- "get_db_list",
- "get_db_table_list",
- "get_table_schema",
- "get_table_comment",
- "get_table_column_comments",
- "get_table_indexes",
- "get_catalog_list",
-)
+DORIS_OAUTH_METADATA_TOOLS = DORIS_OAUTH_METADATA_TOOL_NAMES
METADATA_DB_SCOPES = frozenset(
{
@@ -28,12 +25,20 @@ METADATA_DB_SCOPES = frozenset(
}
)
-QUERY_DB_SCOPES = frozenset({"tool:call:exec_query"})
-EXPLAIN_DB_SCOPES = frozenset({"tool:call:get_sql_explain"})
+QUERY_DB_SCOPES = frozenset(
+ f"tool:call:{tool_name}" for tool_name in DORIS_OAUTH_QUERY_TOOL_SET
+)
+EXPLAIN_DB_SCOPES = frozenset(
+ f"tool:call:{tool_name}" for tool_name in DORIS_OAUTH_EXPLAIN_TOOL_SET
+)
PENDING_DB_SCOPES = METADATA_DB_SCOPES | QUERY_DB_SCOPES | EXPLAIN_DB_SCOPES
TOOL_SCOPE_ALIASES = {
tool_name: f"tool:call:{tool_name}"
- for tool_name in (*DORIS_OAUTH_METADATA_TOOLS, "exec_query",
"get_sql_explain")
+ for tool_name in (
+ *DORIS_OAUTH_METADATA_TOOLS,
+ *DORIS_OAUTH_QUERY_TOOL_SET,
+ *DORIS_OAUTH_EXPLAIN_TOOL_SET,
+ )
}
FORBIDDEN_DORIS_OAUTH_SCOPES = frozenset(
@@ -99,16 +104,24 @@ class DorisOAuthScopePolicy:
if getattr(security_config, "doris_oauth_query_tools_enabled", False):
tool_names = self._configured_tool_names(
- getattr(security_config, "doris_oauth_query_tool_allowlist",
("exec_query",))
+ getattr(
+ security_config,
+ "doris_oauth_query_tool_allowlist",
+ tuple(DORIS_OAUTH_QUERY_TOOL_SET),
+ )
)
- if "exec_query" in tool_names:
+ if DORIS_OAUTH_QUERY_TOOL_SET.intersection(tool_names):
allowed.update(QUERY_DB_SCOPES)
if getattr(security_config, "doris_oauth_explain_tools_enabled",
False):
tool_names = self._configured_tool_names(
- getattr(security_config, "doris_oauth_explain_tool_allowlist",
("get_sql_explain",))
+ getattr(
+ security_config,
+ "doris_oauth_explain_tool_allowlist",
+ tuple(DORIS_OAUTH_EXPLAIN_TOOL_SET),
+ )
)
- if "get_sql_explain" in tool_names:
+ if DORIS_OAUTH_EXPLAIN_TOOL_SET.intersection(tool_names):
allowed.update(EXPLAIN_DB_SCOPES)
return allowed
diff --git a/doris_mcp_server/auth/operation_policy.py
b/doris_mcp_server/auth/operation_policy.py
index dea8d20..28ebf06 100644
--- a/doris_mcp_server/auth/operation_policy.py
+++ b/doris_mcp_server/auth/operation_policy.py
@@ -20,6 +20,15 @@
from dataclasses import dataclass
from typing import Any
+from ..tools.tool_registry import (
+ DORIS_OAUTH_EXPLAIN_TOOL_SET,
+ DORIS_OAUTH_METADATA_TOOL_NAMES,
+ DORIS_OAUTH_METADATA_TOOL_SET,
+ DORIS_OAUTH_QUERY_TOOL_SET,
+ RESTRICTED_TOOL_NAMES,
+ policy_definition_for_tool,
+)
+
@dataclass(frozen=True)
class OperationPolicy:
@@ -62,29 +71,10 @@ class OperationAuthorizationError(Exception):
return payload
-P4_DORIS_OAUTH_METADATA_TOOLS = frozenset(
- {
- "get_db_list",
- "get_db_table_list",
- "get_table_schema",
- "get_table_comment",
- "get_table_column_comments",
- "get_table_indexes",
- "get_catalog_list",
- }
-)
-
-P4_DORIS_OAUTH_QUERY_TOOLS = frozenset({"exec_query"})
-P4_DORIS_OAUTH_EXPLAIN_TOOLS = frozenset({"get_sql_explain"})
-P4_DORIS_OAUTH_DEFAULT_METADATA_TOOL_ALLOWLIST = (
- "get_db_list",
- "get_db_table_list",
- "get_table_schema",
- "get_table_comment",
- "get_table_column_comments",
- "get_table_indexes",
- "get_catalog_list",
- )
+P4_DORIS_OAUTH_METADATA_TOOLS = DORIS_OAUTH_METADATA_TOOL_SET
+P4_DORIS_OAUTH_QUERY_TOOLS = DORIS_OAUTH_QUERY_TOOL_SET
+P4_DORIS_OAUTH_EXPLAIN_TOOLS = DORIS_OAUTH_EXPLAIN_TOOL_SET
+P4_DORIS_OAUTH_DEFAULT_METADATA_TOOL_ALLOWLIST =
DORIS_OAUTH_METADATA_TOOL_NAMES
# Phase 4 keeps DB-backed Doris OAuth tools fail-closed unless configuration
and
# exact tool scope both allow a reviewed metadata tool.
@@ -96,28 +86,7 @@ DORIS_OAUTH_QUERY_TOOL_ALLOWLIST = ("exec_query",)
DORIS_OAUTH_EXPLAIN_TOOLS_ENABLED = False
DORIS_OAUTH_EXPLAIN_TOOL_ALLOWLIST = ("get_sql_explain",)
-HIGH_RISK_TOOLS = {
- "get_recent_audit_logs",
- "get_sql_profile",
- "get_table_data_size",
- "get_monitoring_metrics",
- "get_memory_stats",
- "get_monitoring_metrics_info",
- "get_monitoring_metrics_data",
- "get_realtime_memory_stats",
- "get_historical_memory_stats",
- "get_table_basic_info",
- "analyze_columns",
- "analyze_table_storage",
- "trace_column_lineage",
- "monitor_data_freshness",
- "analyze_data_access_patterns",
- "analyze_data_flow_dependencies",
- "analyze_slow_queries_topn",
- "analyze_resource_growth_curves",
- "exec_adbc_query",
- "get_adbc_connection_info",
-}
+HIGH_RISK_TOOLS = RESTRICTED_TOOL_NAMES
def normalize_doris_oauth_metadata_tool_allowlist(tools: Any = None) ->
tuple[str, ...]:
@@ -240,7 +209,16 @@ def _allowed_metadata_tool_policy(tool_name: str) ->
OperationPolicy:
def _tool_policy(tool_name: str, auth_context: Any = None) -> OperationPolicy:
- if tool_name in P4_DORIS_OAUTH_METADATA_TOOLS:
+ registry_policy = policy_definition_for_tool(tool_name)
+ if registry_policy is None:
+ return _denied_tool_policy(
+ tool_name,
+ channel="unknown",
+ risk="unknown",
+ error_code="UNKNOWN_OPERATION",
+ )
+
+ if registry_policy.policy_class == "metadata":
if not _metadata_tools_enabled(auth_context):
return _denied_tool_policy(
tool_name,
@@ -266,7 +244,7 @@ def _tool_policy(tool_name: str, auth_context: Any = None)
-> OperationPolicy:
)
return _allowed_metadata_tool_policy(tool_name)
- if tool_name in P4_DORIS_OAUTH_QUERY_TOOLS:
+ if registry_policy.policy_class == "query":
if not _query_tools_enabled(auth_context):
return _denied_tool_policy(
tool_name,
@@ -289,7 +267,7 @@ def _tool_policy(tool_name: str, auth_context: Any = None)
-> OperationPolicy:
risk="query",
)
- if tool_name in P4_DORIS_OAUTH_EXPLAIN_TOOLS:
+ if registry_policy.policy_class == "explain":
if not _explain_tools_enabled(auth_context):
return _denied_tool_policy(
tool_name,
@@ -312,20 +290,16 @@ def _tool_policy(tool_name: str, auth_context: Any =
None) -> OperationPolicy:
risk="explain",
)
- if tool_name in HIGH_RISK_TOOLS:
+ if registry_policy.policy_class == "restricted":
return _denied_tool_policy(
tool_name,
- channel="unsupported",
- risk="high",
- error_code="UNSUPPORTED_FOR_DORIS_OAUTH",
+ channel=registry_policy.channel,
+ risk=registry_policy.risk,
+ error_code=registry_policy.error_code
+ or "UNSUPPORTED_FOR_DORIS_OAUTH",
)
- return _denied_tool_policy(
- tool_name,
- channel="unknown",
- risk="unknown",
- error_code="UNKNOWN_OPERATION",
- )
+ raise AssertionError(f"Unhandled registry policy:
{registry_policy.policy_class}")
OPERATION_POLICIES: dict[str, OperationPolicy] = {
diff --git a/doris_mcp_server/tools/tool_registry.py
b/doris_mcp_server/tools/tool_registry.py
new file mode 100644
index 0000000..2f742ef
--- /dev/null
+++ b/doris_mcp_server/tools/tool_registry.py
@@ -0,0 +1,356 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Single source of truth for Doris MCP tool definitions."""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable, Iterable
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import Any, Literal, Protocol, cast
+
+from mcp.types import Tool
+
+ToolPolicyClass = Literal["metadata", "query", "explain", "restricted"]
+ToolHandler = Callable[[dict[str, Any]], Awaitable[dict[str, Any]]]
+
+DORIS_OAUTH_METADATA_TOOL_NAMES = (
+ "get_db_list",
+ "get_db_table_list",
+ "get_table_schema",
+ "get_table_comment",
+ "get_table_column_comments",
+ "get_table_indexes",
+ "get_catalog_list",
+)
+DORIS_OAUTH_METADATA_TOOL_SET = frozenset(DORIS_OAUTH_METADATA_TOOL_NAMES)
+DORIS_OAUTH_QUERY_TOOL_SET = frozenset({"exec_query"})
+DORIS_OAUTH_EXPLAIN_TOOL_SET = frozenset({"get_sql_explain"})
+
+_LEGACY_ALIASES: tuple[tuple[str, str, tuple[tuple[str, Any], ...]], ...] = (
+ (
+ "get_monitoring_metrics_info",
+ "get_monitoring_metrics",
+ (("content_type", "definitions"),),
+ ),
+ (
+ "get_monitoring_metrics_data",
+ "get_monitoring_metrics",
+ (("content_type", "data"),),
+ ),
+ (
+ "get_realtime_memory_stats",
+ "get_memory_stats",
+ (("data_type", "realtime"),),
+ ),
+ (
+ "get_historical_memory_stats",
+ "get_memory_stats",
+ (("data_type", "historical"),),
+ ),
+)
+LEGACY_TOOL_ALIAS_NAMES = frozenset(alias for alias, _, _ in _LEGACY_ALIASES)
+RESTRICTED_TOOL_NAMES = frozenset(
+ {
+ "get_recent_audit_logs",
+ "get_sql_profile",
+ "get_table_data_size",
+ "get_monitoring_metrics",
+ "get_memory_stats",
+ "get_table_basic_info",
+ "analyze_columns",
+ "analyze_table_storage",
+ "trace_column_lineage",
+ "monitor_data_freshness",
+ "analyze_data_access_patterns",
+ "analyze_data_flow_dependencies",
+ "analyze_slow_queries_topn",
+ "analyze_resource_growth_curves",
+ "exec_adbc_query",
+ "get_adbc_connection_info",
+ }
+) | LEGACY_TOOL_ALIAS_NAMES
+
+
+class ToolRegistryError(ValueError):
+ """Raised when the tool registry is invalid or a tool is unknown."""
+
+
+class ToolHandlerOwner(Protocol):
+ """Structural type for objects that own registry handlers."""
+
+
+@dataclass(frozen=True)
+class ToolPolicyDefinition:
+ """Authorization metadata consumed by the operation policy."""
+
+ policy_class: ToolPolicyClass
+ channel: str
+ risk: str
+ error_code: str | None = None
+
+
+@dataclass(frozen=True)
+class ToolAuditDefinition:
+ """Safe audit metadata derived from the advertised tool schema."""
+
+ event_name: str
+ category: str
+ risk: str
+ argument_names: tuple[str, ...]
+ sensitive_arguments: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class ToolDefinition:
+ """Schema, policy, handler, audit, and documentation for one tool."""
+
+ name: str
+ canonical_name: str
+ tool: Tool | None
+ handler_name: str
+ policy: ToolPolicyDefinition
+ audit: ToolAuditDefinition
+ advertised: bool = True
+ argument_overrides: tuple[tuple[str, Any], ...] = ()
+
+ def bind_handler(self, owner: ToolHandlerOwner) -> ToolHandler:
+ handler = getattr(owner, self.handler_name, None)
+ if not callable(handler):
+ raise ToolRegistryError(
+ f"Tool {self.name!r} references missing handler
{self.handler_name!r}"
+ )
+ return cast(ToolHandler, handler)
+
+ def prepare_arguments(self, arguments: dict[str, Any]) -> dict[str, Any]:
+ prepared = dict(arguments)
+ prepared.update(self.argument_overrides)
+ return prepared
+
+
+def _policy_for_name(name: str) -> ToolPolicyDefinition:
+ if name in DORIS_OAUTH_METADATA_TOOL_SET:
+ return ToolPolicyDefinition("metadata", "mysql_metadata", "metadata")
+ if name in DORIS_OAUTH_QUERY_TOOL_SET:
+ return ToolPolicyDefinition("query", "mysql_query", "query")
+ if name in DORIS_OAUTH_EXPLAIN_TOOL_SET:
+ return ToolPolicyDefinition("explain", "mysql_explain", "explain")
+ if name in RESTRICTED_TOOL_NAMES:
+ return ToolPolicyDefinition(
+ "restricted",
+ "unsupported",
+ "high",
+ "UNSUPPORTED_FOR_DORIS_OAUTH",
+ )
+ raise ToolRegistryError(f"Tool {name!r} has no policy definition")
+
+
+def _audit_category(name: str, policy: ToolPolicyDefinition) -> str:
+ if name == "get_recent_audit_logs":
+ return "audit"
+ if name.startswith(("get_monitoring_", "get_memory_", "get_realtime_")):
+ return "monitoring"
+ if name.startswith("get_historical_") or name == "get_table_data_size":
+ return "monitoring"
+ if "adbc" in name:
+ return "adbc"
+ if name.startswith(("analyze_", "trace_", "monitor_")):
+ return "analytics"
+ if name in {"get_sql_profile", "get_sql_explain", "exec_query"}:
+ return "query"
+ return policy.policy_class
+
+
+def _audit_for_tool(
+ name: str,
+ tool: Tool | None,
+ policy: ToolPolicyDefinition,
+) -> ToolAuditDefinition:
+ properties: dict[str, Any] = {}
+ if tool is not None:
+ raw_properties = tool.input_schema.get("properties", {})
+ if isinstance(raw_properties, dict):
+ properties = raw_properties
+ argument_names = tuple(sorted(str(key) for key in properties))
+ sensitive_arguments = tuple(
+ argument for argument in argument_names if argument in {"sql"}
+ )
+ return ToolAuditDefinition(
+ event_name=f"mcp.tool.call.{name}",
+ category=_audit_category(name, policy),
+ risk=policy.risk,
+ argument_names=argument_names,
+ sensitive_arguments=sensitive_arguments,
+ )
+
+
+class ToolDefinitionRegistry:
+ """Immutable registry used by listing, dispatch, policy, audit, and
docs."""
+
+ def __init__(
+ self,
+ definitions: Iterable[ToolDefinition],
+ owner: ToolHandlerOwner,
+ ) -> None:
+ definitions_by_name: dict[str, ToolDefinition] = {}
+ ordered_names: list[str] = []
+ for definition in definitions:
+ if definition.name in definitions_by_name:
+ raise ToolRegistryError(
+ f"Duplicate tool definition: {definition.name}"
+ )
+ definition.bind_handler(owner)
+ if definition.advertised and definition.tool is None:
+ raise ToolRegistryError(
+ f"Advertised tool {definition.name!r} has no MCP schema"
+ )
+ definitions_by_name[definition.name] = definition
+ ordered_names.append(definition.name)
+
+ self._definitions = MappingProxyType(definitions_by_name)
+ self._ordered_names = tuple(ordered_names)
+
+ @classmethod
+ def from_tools(
+ cls,
+ tools: Iterable[Tool],
+ owner: ToolHandlerOwner,
+ ) -> ToolDefinitionRegistry:
+ tools_by_name: dict[str, Tool] = {}
+ definitions: list[ToolDefinition] = []
+ for tool in tools:
+ if tool.name in tools_by_name:
+ raise ToolRegistryError(f"Duplicate tool schema: {tool.name}")
+ tools_by_name[tool.name] = tool
+ policy = _policy_for_name(tool.name)
+ definitions.append(
+ ToolDefinition(
+ name=tool.name,
+ canonical_name=tool.name,
+ tool=tool,
+ handler_name=f"_{tool.name}_tool",
+ policy=policy,
+ audit=_audit_for_tool(tool.name, tool, policy),
+ )
+ )
+
+ for alias, canonical_name, overrides in _LEGACY_ALIASES:
+ if canonical_name not in tools_by_name:
+ raise ToolRegistryError(
+ f"Legacy alias {alias!r} references unknown tool "
+ f"{canonical_name!r}"
+ )
+ policy = _policy_for_name(alias)
+ definitions.append(
+ ToolDefinition(
+ name=alias,
+ canonical_name=canonical_name,
+ tool=tools_by_name[canonical_name],
+ handler_name=f"_{canonical_name}_tool",
+ policy=policy,
+ audit=_audit_for_tool(
+ alias,
+ tools_by_name[canonical_name],
+ policy,
+ ),
+ advertised=False,
+ argument_overrides=overrides,
+ )
+ )
+
+ return cls(definitions, owner)
+
+ @property
+ def definitions(self) -> tuple[ToolDefinition, ...]:
+ return tuple(
+ self._definitions[name] for name in self._ordered_names
+ )
+
+ @property
+ def advertised_definitions(self) -> tuple[ToolDefinition, ...]:
+ return tuple(
+ definition
+ for definition in self.definitions
+ if definition.advertised
+ )
+
+ @property
+ def advertised_names(self) -> tuple[str, ...]:
+ return tuple(
+ definition.name for definition in self.advertised_definitions
+ )
+
+ def resolve(self, name: str) -> ToolDefinition:
+ try:
+ return self._definitions[name]
+ except KeyError as exc:
+ raise ToolRegistryError(f"Unknown tool: {name}") from exc
+
+ def listed_tools(self) -> list[Tool]:
+ return [
+ cast(Tool, definition.tool)
+ for definition in self.advertised_definitions
+ ]
+
+ def render_markdown(self) -> str:
+ """Render the checked-in public catalog from registry definitions."""
+ lines = [
+ "# Doris MCP Tool Registry",
+ "",
+ "<!-- Generated from ToolDefinitionRegistry; do not edit by hand.
-->",
+ "",
+ "| Tool | Policy | Risk | Handler | Audit event | Parameters |",
+ "|---|---|---|---|---|---|",
+ ]
+ for definition in self.advertised_definitions:
+ tool = cast(Tool, definition.tool)
+ required = set(tool.input_schema.get("required", []))
+ parameters = ", ".join(
+ f"`{name}`{'*' if name in required else ''}"
+ for name in definition.audit.argument_names
+ )
+ lines.append(
+ "| "
+ f"`{definition.name}` | "
+ f"`{definition.policy.policy_class}` | "
+ f"`{definition.policy.risk}` | "
+ f"`{definition.handler_name}` | "
+ f"`{definition.audit.event_name}` | "
+ f"{parameters or 'None'} |"
+ )
+ lines.extend(
+ [
+ "",
+ "Required parameters are marked with `*`. Tool descriptions
and "
+ "JSON Schemas are exposed directly by MCP `tools/list` from
the "
+ "same registry entries.",
+ "",
+ ]
+ )
+ return "\n".join(lines)
+
+
+def policy_definition_for_tool(name: str) -> ToolPolicyDefinition | None:
+ """Return policy metadata for a registered canonical tool or legacy
alias."""
+ if (
+ name in DORIS_OAUTH_METADATA_TOOL_SET
+ or name in DORIS_OAUTH_QUERY_TOOL_SET
+ or name in DORIS_OAUTH_EXPLAIN_TOOL_SET
+ or name in RESTRICTED_TOOL_NAMES
+ ):
+ return _policy_for_name(name)
+ return None
diff --git a/doris_mcp_server/tools/tools_manager.py
b/doris_mcp_server/tools/tools_manager.py
index 8c6cc0c..b103fe2 100644
--- a/doris_mcp_server/tools/tools_manager.py
+++ b/doris_mcp_server/tools/tools_manager.py
@@ -24,7 +24,6 @@ import time
from datetime import datetime
from typing import Any
-from mcp.server.mcpserver import MCPServer
from mcp.types import Tool
from ..auth.operation_policy import (
@@ -34,18 +33,20 @@ from ..auth.operation_policy import (
)
from ..utils.adbc_query_tools import DorisADBCQueryTools
from ..utils.analysis_tools import MemoryTracker, SQLAnalyzer, TableAnalyzer
+from ..utils.config import ADBCConfig
from ..utils.data_exploration_tools import DataExplorationTools
from ..utils.data_governance_tools import DataGovernanceTools
from ..utils.data_quality_tools import DataQualityTools
from ..utils.db import DorisConnectionManager
from ..utils.dependency_analysis_tools import DependencyAnalysisTools
-from ..utils.logger import get_logger
+from ..utils.logger import get_audit_logger, get_logger
from ..utils.monitoring_tools import DorisMonitoringTools
from ..utils.performance_analytics_tools import PerformanceAnalyticsTools
from ..utils.query_executor import DorisQueryExecutor
from ..utils.schema_extractor import MetadataExtractor
from ..utils.security import get_current_auth_context
from ..utils.security_analytics_tools import SecurityAnalyticsTools
+from .tool_registry import ToolDefinition, ToolDefinitionRegistry
logger = get_logger(__name__)
@@ -78,6 +79,7 @@ class DorisToolsManager:
# Initialize ADBC query tools
self.adbc_query_tools = DorisADBCQueryTools(connection_manager)
+ self._tool_registry = self._build_tool_registry()
logger.info(
"DorisToolsManager initialized with business logic processors,
v0.5.0 analytics tools, and ADBC query tools"
@@ -112,758 +114,12 @@ class DorisToolsManager:
raise ValueError(f"{name} must be a non-empty list of strings")
return value
- async def register_tools_with_mcp(self, mcp: MCPServer[Any]) -> None:
- """Register all tools to MCP server"""
- logger.info("Starting to register MCP tools")
-
- # SQL query execution tool (supports catalog federation queries)
- @mcp.tool(
- "exec_query",
- description="""[Function Description]: Execute SQL query and
return result command with catalog federation support.
-
-[Parameter Content]:
-
-- sql (string) [Required] - SQL statement to execute. MUST use three-part
naming for all table references: 'catalog_name.db_name.table_name'. For
internal tables use 'internal.db_name.table_name', for external tables use
'catalog_name.db_name.table_name'
-
-- db_name (string) [Optional] - Target database name, defaults to the current
database
-
-- catalog_name (string) [Optional] - Reference catalog name for context,
defaults to current catalog
-
-- max_rows (integer) [Optional] - Maximum number of rows to return, default 100
-
-- timeout (integer) [Optional] - Query timeout in seconds, default 30
-""",
- )
- async def exec_query_tool(
- sql: str,
- db_name: str | None = None,
- catalog_name: str | None = None,
- max_rows: int = 100,
- timeout: int = 30,
- ) -> str:
- """Execute SQL query (supports federation queries)"""
- return await self.call_tool(
- "exec_query",
- {
- "sql": sql,
- "db_name": db_name,
- "catalog_name": catalog_name,
- "max_rows": max_rows,
- "timeout": timeout,
- },
- )
-
- # Get table schema tool
- @mcp.tool(
- "get_table_schema",
- description="""[Function Description]: Get detailed structure
information of the specified table (columns, types, comments, etc.).
-
-[Parameter Content]:
-
-- table_name (string) [Required] - Name of the table to query
-
-- db_name (string) [Optional] - Target database name, defaults to the current
database
-
-- catalog_name (string) [Optional] - Target catalog name for federation
queries, defaults to current catalog
-""",
- )
- async def get_table_schema_tool(
- table_name: str, db_name: str | None = None, catalog_name: str |
None = None
- ) -> str:
- """Get table schema information"""
- return await self.call_tool(
- "get_table_schema",
- {
- "table_name": table_name,
- "db_name": db_name,
- "catalog_name": catalog_name,
- },
- )
-
- # Get database table list tool
- @mcp.tool(
- "get_db_table_list",
- description="""[Function Description]: Get a list of all table
names in the specified database.
-
-[Parameter Content]:
-
-- db_name (string) [Optional] - Target database name, defaults to the current
database
-
-- catalog_name (string) [Optional] - Target catalog name for federation
queries, defaults to current catalog
-""",
- )
- async def get_db_table_list_tool(
- db_name: str | None = None, catalog_name: str | None = None
- ) -> str:
- """Get database table list"""
- return await self.call_tool(
- "get_db_table_list", {"db_name": db_name, "catalog_name":
catalog_name}
- )
-
- # Get database list tool
- @mcp.tool(
- "get_db_list",
- description="""[Function Description]: Get a list of all database
names on the server.
-
-[Parameter Content]:
-
-- catalog_name (string) [Optional] - Target catalog name for federation
queries, defaults to current catalog
-""",
- )
- async def get_db_list_tool(catalog_name: str | None = None) -> str:
- """Get database list"""
- return await self.call_tool("get_db_list", {"catalog_name":
catalog_name})
-
- # Get table comment tool
- @mcp.tool(
- "get_table_comment",
- description="""[Function Description]: Get the comment information
for the specified table.
-
-[Parameter Content]:
-
-- table_name (string) [Required] - Name of the table to query
-
-- db_name (string) [Optional] - Target database name, defaults to the current
database
-
-- catalog_name (string) [Optional] - Target catalog name for federation
queries, defaults to current catalog
-""",
- )
- async def get_table_comment_tool(
- table_name: str, db_name: str | None = None, catalog_name: str |
None = None
- ) -> str:
- """Get table comment"""
- return await self.call_tool(
- "get_table_comment",
- {
- "table_name": table_name,
- "db_name": db_name,
- "catalog_name": catalog_name,
- },
- )
-
- # Get table column comments tool
- @mcp.tool(
- "get_table_column_comments",
- description="""[Function Description]: Get comment information for
all columns in the specified table.
-
-[Parameter Content]:
-
-- table_name (string) [Required] - Name of the table to query
-
-- db_name (string) [Optional] - Target database name, defaults to the current
database
-
-- catalog_name (string) [Optional] - Target catalog name for federation
queries, defaults to current catalog
-""",
- )
- async def get_table_column_comments_tool(
- table_name: str, db_name: str | None = None, catalog_name: str |
None = None
- ) -> str:
- """Get table column comments"""
- return await self.call_tool(
- "get_table_column_comments",
- {
- "table_name": table_name,
- "db_name": db_name,
- "catalog_name": catalog_name,
- },
- )
-
- # Get table indexes tool
- @mcp.tool(
- "get_table_indexes",
- description="""[Function Description]: Get index information for
the specified table.
-
-[Parameter Content]:
-
-- table_name (string) [Required] - Name of the table to query
-
-- db_name (string) [Optional] - Target database name, defaults to the current
database
-
-- catalog_name (string) [Optional] - Target catalog name for federation
queries, defaults to current catalog
-""",
- )
- async def get_table_indexes_tool(
- table_name: str, db_name: str | None = None, catalog_name: str |
None = None
- ) -> str:
- """Get table indexes"""
- return await self.call_tool(
- "get_table_indexes",
- {
- "table_name": table_name,
- "db_name": db_name,
- "catalog_name": catalog_name,
- },
- )
-
- # Get audit logs tool
- @mcp.tool(
- "get_recent_audit_logs",
- description="""[Function Description]: Get audit log records for a
recent period.
-
-[Parameter Content]:
-
-- days (integer) [Optional] - Number of recent days of logs to retrieve,
default is 7
-
-- limit (integer) [Optional] - Maximum number of records to return, default is
100
-""",
- )
- async def get_recent_audit_logs_tool(days: int = 7, limit: int = 100)
-> str:
- """Get audit logs"""
- return await self.call_tool(
- "get_recent_audit_logs", {"days": days, "limit": limit}
- )
-
- # Get catalog list tool
- @mcp.tool(
- "get_catalog_list",
- description="""[Function Description]: Get a list of all catalog
names on the server.
-
-[Parameter Content]:
-
-- random_string (string) [Required] - Unique identifier for the tool call
-""",
- )
- async def get_catalog_list_tool(random_string: str) -> str:
- """Get catalog list"""
- return await self.call_tool(
- "get_catalog_list", {"random_string": random_string}
- )
-
- # SQL Explain tool
- @mcp.tool(
- "get_sql_explain",
- description="""[Function Description]: Get SQL execution plan
using EXPLAIN command based on Doris syntax.
-
-[Parameter Content]:
-
-- sql (string) [Required] - SQL statement to explain
-
-- verbose (boolean) [Optional] - Whether to show verbose information, default
is false
-
-- db_name (string) [Optional] - Target database name, defaults to the current
database
-
-- catalog_name (string) [Optional] - Target catalog name for federation
queries, defaults to current catalog
-""",
- )
- async def get_sql_explain_tool(
- sql: str,
- verbose: bool = False,
- db_name: str | None = None,
- catalog_name: str | None = None,
- ) -> str:
- """Get SQL execution plan"""
- return await self.call_tool(
- "get_sql_explain",
- {
- "sql": sql,
- "verbose": verbose,
- "db_name": db_name,
- "catalog_name": catalog_name,
- },
- )
-
- # SQL Profile tool
- @mcp.tool(
- "get_sql_profile",
- description="""[Function Description]: Get SQL execution profile
by setting trace ID and fetching profile via FE HTTP API.
-
-[Parameter Content]:
-
-- sql (string) [Required] - SQL statement to profile
-
-- db_name (string) [Optional] - Target database name, defaults to the current
database
-
-- catalog_name (string) [Optional] - Target catalog name for federation
queries, defaults to current catalog
-
-- timeout (integer) [Optional] - Query timeout in seconds, default is 30
-""",
- )
- async def get_sql_profile_tool(
- sql: str,
- db_name: str | None = None,
- catalog_name: str | None = None,
- timeout: int = 30,
- ) -> str:
- """Get SQL execution profile"""
- return await self.call_tool(
- "get_sql_profile",
- {
- "sql": sql,
- "db_name": db_name,
- "catalog_name": catalog_name,
- "timeout": timeout,
- },
- )
-
- # Table data size tool
- @mcp.tool(
- "get_table_data_size",
- description="""[Function Description]: Get table data size
information via FE HTTP API.
-
-[Parameter Content]:
-
-- db_name (string) [Optional] - Database name, if not specified returns all
databases
-
-- table_name (string) [Optional] - Table name, if not specified returns all
tables in the database
-
-- single_replica (boolean) [Optional] - Whether to get single replica data
size, default is false
-""",
- )
- async def get_table_data_size_tool(
- db_name: str | None = None,
- table_name: str | None = None,
- single_replica: bool = False,
- ) -> str:
- """Get table data size information"""
- return await self.call_tool(
- "get_table_data_size",
- {
- "db_name": db_name,
- "table_name": table_name,
- "single_replica": single_replica,
- },
- )
-
- # Unified Monitoring Metrics Tool (combines definitions and data)
- @mcp.tool(
- "get_monitoring_metrics",
- description="""[Function Description]: Get comprehensive Doris
monitoring metrics including definitions and/or actual data from FE and BE
nodes.
-
-[Parameter Content]:
-
-- content_type (string) [Optional] - Type of monitoring content to retrieve,
default is "data"
- * "definitions": Only metric definitions and descriptions
- * "data": Only actual metric data from nodes
- * "both": Both definitions and data
-
-- role (string) [Optional] - Node role to monitor, default is "all"
- * "fe": Only FE nodes/metrics
- * "be": Only BE nodes/metrics
- * "all": Both FE and BE nodes/metrics
-
-- monitor_type (string) [Optional] - Type of monitoring metrics, default is
"all"
- * "process": Process monitoring metrics
- * "jvm": JVM monitoring metrics (FE only)
- * "machine": Machine monitoring metrics
- * "all": All monitoring types
-
-- priority (string) [Optional] - Metric priority level, default is "core"
- * "core": Only core essential metrics (10-12 items for production use)
- * "p0": Only P0 (highest priority) metrics
- * "all": All metrics (P0 and non-P0)
-
-- include_raw_metrics (boolean) [Optional] - Whether to include raw detailed
metrics data (can be very large), default is false
-""",
- )
- async def get_monitoring_metrics_tool(
- content_type: str = "data",
- role: str = "all",
- monitor_type: str = "all",
- priority: str = "core",
- include_raw_metrics: bool = False,
- ) -> str:
- """Get comprehensive monitoring metrics (definitions and/or
data)"""
- return await self.call_tool(
- "get_monitoring_metrics",
- {
- "content_type": content_type,
- "role": role,
- "monitor_type": monitor_type,
- "priority": priority,
- "include_raw_metrics": include_raw_metrics,
- },
- )
-
- # Unified Memory Statistics Tool (combines real-time and historical)
- @mcp.tool(
- "get_memory_stats",
- description="""[Function Description]: Get comprehensive memory
statistics from Doris BE nodes, supporting both real-time and historical data.
-
-[Parameter Content]:
-
-- data_type (string) [Optional] - Type of memory data to retrieve, default is
"realtime"
- * "realtime": Real-time memory statistics via Memory Tracker web interface
- * "historical": Historical memory statistics via Bvar interface
- * "both": Both real-time and historical data
-
-- tracker_type (string) [Optional] - Type of memory trackers to retrieve (for
real-time), default is "overview"
- * "overview": Overview type trackers (process memory, tracked memory summary)
- * "global": Global shared memory trackers (cache, metadata)
- * "query": Query-related memory trackers
- * "load": Load-related memory trackers
- * "compaction": Compaction-related memory trackers
- * "all": All memory tracker types
-
-- tracker_names (array) [Optional] - List of specific tracker names for
historical data
- * Example: ["process_resident_memory", "global", "query", "load",
"compaction"]
-
-- time_range (string) [Optional] - Time range for historical data, default is
"1h"
- * "1h": Last 1 hour
- * "6h": Last 6 hours
- * "24h": Last 24 hours
-
-- include_details (boolean) [Optional] - Whether to include detailed tracker
information and definitions, default is true
-""",
- )
- async def get_memory_stats_tool(
- data_type: str = "realtime",
- tracker_type: str = "overview",
- tracker_names: list[str] | None = None,
- time_range: str = "1h",
- include_details: bool = True,
- ) -> str:
- """Get comprehensive memory statistics (real-time and/or
historical)"""
- return await self.call_tool(
- "get_memory_stats",
- {
- "data_type": data_type,
- "tracker_type": tracker_type,
- "tracker_names": tracker_names,
- "time_range": time_range,
- "include_details": include_details,
- },
- )
-
- # ==================== v0.5.0 Advanced Analytics Tools
====================
-
- # 🔄 Unified Data Quality Analysis Tool (New in v0.5.0)
- @mcp.tool(
- "get_table_basic_info",
- description="""[Function Description]: Get basic information about
a table including row count, column count, partitions, and size.
-
-[Parameter Content]:
-
-- table_name (string) [Required] - Name of the table to analyze
-- catalog_name (string) [Optional] - Target catalog name
-- db_name (string) [Optional] - Target database name
-""",
- )
- async def get_table_basic_info_tool(
- table_name: str, catalog_name: str | None = None, db_name: str |
None = None
- ) -> str:
- """Get table basic information"""
- return await self.call_tool(
- "get_table_basic_info",
- {
- "table_name": table_name,
- "catalog_name": catalog_name,
- "db_name": db_name,
- },
- )
-
- @mcp.tool(
- "analyze_columns",
- description="""[Function Description]: Analyze completeness and
distribution of specified columns in a table.
-
-[Parameter Content]:
-
-- table_name (string) [Required] - Name of the table to analyze
-- columns (array) [Required] - List of column names to analyze
-- analysis_types (array) [Optional] - Types of analysis to perform, default is
["both"]
- * "completeness": Only completeness analysis (null rates, non-null counts)
- * "distribution": Only distribution analysis (statistical patterns by data
type)
- * "both": Both completeness and distribution analysis
-- sample_size (integer) [Optional] - Maximum number of rows to sample, default
is 100000
-- catalog_name (string) [Optional] - Target catalog name
-- db_name (string) [Optional] - Target database name
-- detailed_response (boolean) [Optional] - Whether to return detailed response
including raw data, default is false
-""",
- )
- async def analyze_columns_tool(
- table_name: str,
- columns: list[str],
- analysis_types: list[str] | None = None,
- sample_size: int = 100000,
- catalog_name: str | None = None,
- db_name: str | None = None,
- detailed_response: bool = False,
- ) -> str:
- """Analyze table columns"""
- return await self.call_tool(
- "analyze_columns",
- {
- "table_name": table_name,
- "columns": columns,
- "analysis_types": analysis_types or ["both"],
- "sample_size": sample_size,
- "catalog_name": catalog_name,
- "db_name": db_name,
- "detailed_response": detailed_response,
- },
- )
-
- @mcp.tool(
- "analyze_table_storage",
- description="""[Function Description]: Analyze table's physical
distribution and storage information.
-
-[Parameter Content]:
-
-- table_name (string) [Required] - Name of the table to analyze
-- catalog_name (string) [Optional] - Target catalog name
-- db_name (string) [Optional] - Target database name
-- detailed_response (boolean) [Optional] - Whether to return detailed response
including raw data, default is false
-""",
- )
- async def analyze_table_storage_tool(
- table_name: str,
- catalog_name: str | None = None,
- db_name: str | None = None,
- detailed_response: bool = False,
- ) -> str:
- """Analyze table storage"""
- return await self.call_tool(
- "analyze_table_storage",
- {
- "table_name": table_name,
- "catalog_name": catalog_name,
- "db_name": db_name,
- "detailed_response": detailed_response,
- },
- )
-
- @mcp.tool(
- "trace_column_lineage",
- description="""[Function Description]: Trace data lineage for
specified columns through SQL analysis and dependency mapping.
-
-[Parameter Content]:
-
-- target_columns (array) [Required] - List of column specifications in format
"table.column" or "db.table.column"
-- analysis_depth (integer) [Optional] - Maximum depth for lineage tracing,
default is 3
-- include_transformations (boolean) [Optional] - Whether to include
transformation details, default is true
-- catalog_name (string) [Optional] - Target catalog name
-""",
- )
- async def trace_column_lineage_tool(
- target_columns: list[str],
- analysis_depth: int = 3,
- include_transformations: bool = True,
- catalog_name: str | None = None,
- ) -> str:
- """Trace column data lineage"""
- return await self.call_tool(
- "trace_column_lineage",
- {
- "target_columns": target_columns,
- "analysis_depth": analysis_depth,
- "include_transformations": include_transformations,
- "catalog_name": catalog_name,
- },
- )
-
- @mcp.tool(
- "monitor_data_freshness",
- description="""[Function Description]: Monitor data freshness and
staleness patterns for specified tables.
-
-[Parameter Content]:
-
-- table_names (array) [Optional] - List of table names to monitor, if not
specified monitors all tables
-- freshness_threshold_hours (integer) [Optional] - Freshness threshold in
hours, default is 24
-- include_update_patterns (boolean) [Optional] - Whether to include update
pattern analysis, default is true
-- catalog_name (string) [Optional] - Target catalog name
-- db_name (string) [Optional] - Target database name
-""",
- )
- async def monitor_data_freshness_tool(
- table_names: list[str] | None = None,
- freshness_threshold_hours: int = 24,
- include_update_patterns: bool = True,
- catalog_name: str | None = None,
- db_name: str | None = None,
- ) -> str:
- """Monitor data freshness and staleness"""
- return await self.call_tool(
- "monitor_data_freshness",
- {
- "table_names": table_names,
- "freshness_threshold_hours": freshness_threshold_hours,
- "include_update_patterns": include_update_patterns,
- "catalog_name": catalog_name,
- "db_name": db_name,
- },
- )
-
- # Security Analytics Tools
- @mcp.tool(
- "analyze_data_access_patterns",
- description="""[Function Description]: Analyze user data access
patterns, security anomalies, and access behavior.
-
-[Parameter Content]:
-
-- days (integer) [Optional] - Number of days to analyze, default is 7
-- include_system_users (boolean) [Optional] - Whether to include system users
in analysis, default is false
-- min_query_threshold (integer) [Optional] - Minimum queries for user
inclusion, default is 5
-""",
- )
- async def analyze_data_access_patterns_tool(
- days: int = 7,
- include_system_users: bool = False,
- min_query_threshold: int = 5,
- ) -> str:
- """Analyze data access patterns and security insights"""
- return await self.call_tool(
- "analyze_data_access_patterns",
- {
- "days": days,
- "include_system_users": include_system_users,
- "min_query_threshold": min_query_threshold,
- },
- )
-
- # Dependency Analysis Tools
- @mcp.tool(
- "analyze_data_flow_dependencies",
- description="""[Function Description]: Analyze data flow
dependencies and impact relationships between tables.
-
-[Parameter Content]:
-
-- target_table (string) [Optional] - Specific table to analyze, if not
specified analyzes all tables
-- analysis_depth (integer) [Optional] - Maximum depth for dependency
traversal, default is 3
-- include_views (boolean) [Optional] - Whether to include views in analysis,
default is true
-- catalog_name (string) [Optional] - Target catalog name
-- db_name (string) [Optional] - Target database name
-""",
- )
- async def analyze_data_flow_dependencies_tool(
- target_table: str | None = None,
- analysis_depth: int = 3,
- include_views: bool = True,
- catalog_name: str | None = None,
- db_name: str | None = None,
- ) -> str:
- """Analyze data flow dependencies and impact"""
- return await self.call_tool(
- "analyze_data_flow_dependencies",
- {
- "target_table": target_table,
- "analysis_depth": analysis_depth,
- "include_views": include_views,
- "catalog_name": catalog_name,
- "db_name": db_name,
- },
- )
-
- # Performance Analytics Tools
- @mcp.tool(
- "analyze_slow_queries_topn",
- description="""[Function Description]: Analyze top N slowest
queries and identify performance patterns and issues.
-
-[Parameter Content]:
-
-- days (integer) [Optional] - Number of days to analyze, default is 7
-- top_n (integer) [Optional] - Number of top slow queries to return, default
is 20
-- min_execution_time_ms (integer) [Optional] - Minimum execution time
threshold in milliseconds, default is 1000
-- include_patterns (boolean) [Optional] - Whether to include query pattern
analysis, default is true
-""",
- )
- async def analyze_slow_queries_topn_tool(
- days: int = 7,
- top_n: int = 20,
- min_execution_time_ms: int = 1000,
- include_patterns: bool = True,
- ) -> str:
- """Analyze top N slow queries and performance patterns"""
- return await self.call_tool(
- "analyze_slow_queries_topn",
- {
- "days": days,
- "top_n": top_n,
- "min_execution_time_ms": min_execution_time_ms,
- "include_patterns": include_patterns,
- },
- )
-
- @mcp.tool(
- "analyze_resource_growth_curves",
- description="""[Function Description]: Analyze resource growth
patterns and trends for capacity planning.
-
-[Parameter Content]:
-
-- days (integer) [Optional] - Number of days to analyze, default is 30
-- resource_types (array) [Optional] - Types of resources to analyze, default
is ["storage", "query_volume", "user_activity"]
-- include_predictions (boolean) [Optional] - Whether to include growth
predictions, default is false
-- detailed_response (boolean) [Optional] - Whether to return detailed data
including daily breakdowns, default is false
-""",
- )
- async def analyze_resource_growth_curves_tool(
- days: int = 30,
- resource_types: list[str] | None = None,
- include_predictions: bool = False,
- detailed_response: bool = False,
- ) -> str:
- """Analyze resource growth patterns and capacity planning"""
- return await self.call_tool(
- "analyze_resource_growth_curves",
- {
- "days": days,
- "resource_types": resource_types
- or ["storage", "query_volume", "user_activity"],
- "include_predictions": include_predictions,
- "detailed_response": detailed_response,
- },
- )
-
- # ==================== ADBC Query Tools ====================
-
- # ADBC Query Execution Tool
- @mcp.tool(
- "exec_adbc_query",
- description=f"""[Function Description]: Execute SQL query using
ADBC (Arrow Flight SQL) protocol for high-performance data transfer.
-
-[Parameter Content]:
-
-- sql (string) [Required] - SQL statement to execute
-- max_rows (integer) [Optional] - Maximum number of rows to return, default is
{self.connection_manager.config.adbc.default_max_rows}
-- timeout (integer) [Optional] - Query timeout in seconds, default is
{self.connection_manager.config.adbc.default_timeout}
-- return_format (string) [Optional] - Format for returned data, default is
"{self.connection_manager.config.adbc.default_return_format}"
- * "arrow": Return Arrow format with metadata
- * "pandas": Return Pandas DataFrame format
- * "dict": Return dictionary format
-
-[Prerequisites]:
-- Environment variables FE_ARROW_FLIGHT_SQL_PORT and BE_ARROW_FLIGHT_SQL_PORT
must be configured
-- Required Python packages: adbc_driver_manager, adbc_driver_flightsql
-- Arrow Flight SQL services must be running on FE and BE nodes
-""",
- )
- async def exec_adbc_query_tool(
- sql: str,
- max_rows: int | None = None,
- timeout: int | None = None,
- return_format: str | None = None,
- ) -> str:
- """Execute SQL query using ADBC (Arrow Flight SQL) protocol"""
- return await self.call_tool(
- "exec_adbc_query",
- {
- "sql": sql,
- "max_rows": max_rows,
- "timeout": timeout,
- "return_format": return_format,
- },
- )
-
- # ADBC Connection Information Tool
- @mcp.tool(
- "get_adbc_connection_info",
- description="""[Function Description]: Get ADBC (Arrow Flight SQL)
connection information and status.
-
-[Parameter Content]:
-
-No parameters required. Returns connection status, configuration, and
diagnostic information.
-""",
- )
- async def get_adbc_connection_info_tool() -> str:
- """Get ADBC connection information and status"""
- return await self.call_tool("get_adbc_connection_info", {})
-
- logger.info(
- "Successfully registered 25 tools to MCP server (14 basic + 9
advanced analytics + 2 ADBC tools)"
- )
-
- async def list_tools(self) -> list[Tool]:
- """List all available query tools (for stdio mode)"""
+ def _build_tool_registry(self) -> ToolDefinitionRegistry:
+ """Build the immutable tool registry from the MCP tool catalog."""
# Get ADBC configuration defaults
- adbc_config = self.connection_manager.config.adbc
+ connection_manager = getattr(self, "connection_manager", None)
+ config = getattr(connection_manager, "config", None)
+ adbc_config = getattr(config, "adbc", None) or ADBCConfig()
tools = [
Tool(
@@ -1734,100 +990,66 @@ No parameters required. Returns connection status,
configuration, and diagnostic
),
]
- return filter_tools_for_auth_context(get_current_auth_context(), tools)
+ return ToolDefinitionRegistry.from_tools(tools, self)
+
+ @property
+ def tool_registry(self) -> ToolDefinitionRegistry:
+ """Return the registry, constructing it for legacy test fixtures."""
+ registry = getattr(self, "_tool_registry", None)
+ if registry is None:
+ registry = self._build_tool_registry()
+ self._tool_registry = registry
+ return registry
+
+ async def list_tools(self) -> list[Tool]:
+ """List tools visible to the current authorization context."""
+ return filter_tools_for_auth_context(
+ get_current_auth_context(),
+ self.tool_registry.listed_tools(),
+ )
async def call_tool(self, name: str, arguments: dict[str, Any]) -> str:
"""
Call the specified query tool (tool routing and scheduling center)
"""
authorize_operation(get_current_auth_context(), f"tool:{name}")
+ definition: ToolDefinition | None = None
+ start_time = time.time()
try:
- start_time = time.time()
-
- # Tool routing - dispatch requests to corresponding business logic
processors
- if name == "exec_query":
- result = await self._exec_query_tool(arguments)
- elif name == "get_table_schema":
- result = await self._get_table_schema_tool(arguments)
- elif name == "get_db_table_list":
- result = await self._get_db_table_list_tool(arguments)
- elif name == "get_db_list":
- result = await self._get_db_list_tool(arguments)
- elif name == "get_table_comment":
- result = await self._get_table_comment_tool(arguments)
- elif name == "get_table_column_comments":
- result = await self._get_table_column_comments_tool(arguments)
- elif name == "get_table_indexes":
- result = await self._get_table_indexes_tool(arguments)
- elif name == "get_recent_audit_logs":
- result = await self._get_recent_audit_logs_tool(arguments)
- elif name == "get_catalog_list":
- result = await self._get_catalog_list_tool(arguments)
- elif name == "get_sql_explain":
- result = await self._get_sql_explain_tool(arguments)
- elif name == "get_sql_profile":
- result = await self._get_sql_profile_tool(arguments)
- elif name == "get_table_data_size":
- result = await self._get_table_data_size_tool(arguments)
- elif name == "get_monitoring_metrics":
- result = await self._get_monitoring_metrics_tool(arguments)
- elif name == "get_memory_stats":
- result = await self._get_memory_stats_tool(arguments)
- # Legacy support for old tool names (deprecated)
- elif name == "get_monitoring_metrics_info":
- arguments["content_type"] = "definitions"
- result = await self._get_monitoring_metrics_tool(arguments)
- elif name == "get_monitoring_metrics_data":
- arguments["content_type"] = "data"
- result = await self._get_monitoring_metrics_tool(arguments)
- elif name == "get_realtime_memory_stats":
- arguments["data_type"] = "realtime"
- result = await self._get_memory_stats_tool(arguments)
- elif name == "get_historical_memory_stats":
- arguments["data_type"] = "historical"
- result = await self._get_memory_stats_tool(arguments)
- # v0.5.0 Advanced Analytics Tools - Atomic Data Quality Tools
- elif name == "get_table_basic_info":
- result = await self._get_table_basic_info_tool(arguments)
- elif name == "analyze_columns":
- result = await self._analyze_columns_tool(arguments)
- elif name == "analyze_table_storage":
- result = await self._analyze_table_storage_tool(arguments)
- elif name == "trace_column_lineage":
- result = await self._trace_column_lineage_tool(arguments)
- elif name == "monitor_data_freshness":
- result = await self._monitor_data_freshness_tool(arguments)
- elif name == "analyze_data_access_patterns":
- result = await
self._analyze_data_access_patterns_tool(arguments)
- elif name == "analyze_data_flow_dependencies":
- result = await
self._analyze_data_flow_dependencies_tool(arguments)
- elif name == "analyze_slow_queries_topn":
- result = await self._analyze_slow_queries_topn_tool(arguments)
- elif name == "analyze_resource_growth_curves":
- result = await
self._analyze_resource_growth_curves_tool(arguments)
- # ADBC Query Tools
- elif name == "exec_adbc_query":
- result = await self._exec_adbc_query_tool(arguments)
- elif name == "get_adbc_connection_info":
- result = await self._get_adbc_connection_info_tool(arguments)
- else:
- raise ValueError(f"Unknown tool: {name}")
-
+ definition = self.tool_registry.resolve(name)
+ handler = definition.bind_handler(self)
+ prepared_arguments = definition.prepare_arguments(arguments)
+ result = await handler(prepared_arguments)
execution_time = time.time() - start_time
# Add execution information
if isinstance(result, dict):
result["_execution_info"] = {
"tool_name": name,
+ "canonical_tool_name": definition.canonical_name,
+ "audit_event": definition.audit.event_name,
"execution_time": round(execution_time, 3),
"timestamp": datetime.now().isoformat(),
}
+ self._audit_tool_call(
+ definition,
+ arguments=prepared_arguments,
+ status="success",
+ execution_time=execution_time,
+ )
return json.dumps(result, ensure_ascii=False, indent=2)
except OperationAuthorizationError:
raise
except Exception as e:
+ if definition is not None:
+ self._audit_tool_call(
+ definition,
+ arguments=arguments,
+ status="error",
+ execution_time=time.time() - start_time,
+ )
logger.error(
"Tool call failed (%s)",
type(e).__name__,
@@ -1839,6 +1061,31 @@ No parameters required. Returns connection status,
configuration, and diagnostic
}
return json.dumps(error_result, ensure_ascii=False, indent=2)
+ @staticmethod
+ def _audit_tool_call(
+ definition: ToolDefinition,
+ *,
+ arguments: dict[str, Any],
+ status: str,
+ execution_time: float,
+ ) -> None:
+ audit = definition.audit
+ provided_names = sorted(
+ set(arguments).intersection(audit.argument_names)
+ )
+ get_audit_logger().info(
+ "event=%s tool=%s canonical_tool=%s category=%s risk=%s "
+ "status=%s duration_ms=%d argument_names=%s",
+ audit.event_name,
+ definition.name,
+ definition.canonical_name,
+ audit.category,
+ audit.risk,
+ status,
+ round(execution_time * 1000),
+ ",".join(provided_names),
+ )
+
async def _exec_query_tool(self, arguments: dict[str, Any]) -> dict[str,
Any]:
"""SQL query execution tool routing (supports federation queries)"""
sql = self._required_string(arguments, "sql")
diff --git a/doris_mcp_server/utils/config.py b/doris_mcp_server/utils/config.py
index 00dd3dc..5772623 100644
--- a/doris_mcp_server/utils/config.py
+++ b/doris_mcp_server/utils/config.py
@@ -33,6 +33,12 @@ from urllib.parse import urlparse
from dotenv import load_dotenv
from .._version import __version__
+from ..tools.tool_registry import (
+ DORIS_OAUTH_EXPLAIN_TOOL_SET,
+ DORIS_OAUTH_METADATA_TOOL_NAMES,
+ DORIS_OAUTH_METADATA_TOOL_SET,
+ DORIS_OAUTH_QUERY_TOOL_SET,
+)
from .logger import get_logger
from .secret_policy import (
is_static_token_environment_variable,
@@ -46,18 +52,9 @@ class AuthConfigError(ValueError):
"""Raised when authentication configuration is inconsistent."""
-DORIS_OAUTH_METADATA_TOOL_ALLOWLIST_DEFAULT = [
- "get_db_list",
- "get_db_table_list",
- "get_table_schema",
- "get_table_comment",
- "get_table_column_comments",
- "get_table_indexes",
- "get_catalog_list",
-]
-DORIS_OAUTH_METADATA_TOOL_SET =
frozenset(DORIS_OAUTH_METADATA_TOOL_ALLOWLIST_DEFAULT)
-DORIS_OAUTH_QUERY_TOOL_SET = frozenset({"exec_query"})
-DORIS_OAUTH_EXPLAIN_TOOL_SET = frozenset({"get_sql_explain"})
+DORIS_OAUTH_METADATA_TOOL_ALLOWLIST_DEFAULT = list(
+ DORIS_OAUTH_METADATA_TOOL_NAMES
+)
@dataclass(frozen=True)
diff --git a/test/protocol/test_mcp_v2_protocol.py
b/test/protocol/test_mcp_v2_protocol.py
index d6e0e1e..f7941dc 100644
--- a/test/protocol/test_mcp_v2_protocol.py
+++ b/test/protocol/test_mcp_v2_protocol.py
@@ -51,6 +51,7 @@ from doris_mcp_server.protocol import (
)
from test.protocol.schema_validation_server import
create_schema_validation_server
from test.protocol.stdio_capability_server import OneToolManager as
ProfileToolManager
+from test.protocol.tool_registry_server import create_registry_test_server
REQUIRED_EXTENSION = "io.apache.doris/read"
PROFILE_TOOL_NAMES = sorted(
@@ -361,6 +362,64 @@ def modern_tool_headers(name: str) -> dict[str, str]:
}
[email protected]
+async def
test_http_uses_production_tool_registry_for_list_validation_and_dispatch():
+ server = create_registry_test_server()
+ app = server.streamable_http_app(
+ json_response=True,
+ stateless_http=True,
+ host="127.0.0.1",
+ transport_security=create_transport_security("127.0.0.1"),
+ )
+
+ async with (
+ app.router.lifespan_context(app),
+ httpx2.ASGITransport(app) as transport,
+ httpx2.AsyncClient(
+ transport=transport,
+ base_url="http://127.0.0.1:3000";,
+ ) as client,
+ ):
+ listed = await client.post(
+ "/mcp",
+ json=modern_request(1, "tools/list"),
+ headers=modern_headers("tools/list"),
+ )
+ assert listed.status_code == 200
+ tools = {
+ tool["name"]: tool
+ for tool in listed.json()["result"]["tools"]
+ }
+ assert len(tools) == 25
+ assert tools["exec_query"]["inputSchema"]["required"] == ["sql"]
+ assert "get_monitoring_metrics_info" not in tools
+
+ invalid = await client.post(
+ "/mcp",
+ json=modern_tool_request(2, "exec_query", {}),
+ headers=modern_tool_headers("exec_query"),
+ )
+ assert invalid.status_code == 400
+ assert invalid.json()["error"]["code"] == -32602
+
+ called = await client.post(
+ "/mcp",
+ json=modern_tool_request(
+ 3,
+ "exec_query",
+ {"sql": "SELECT 1"},
+ ),
+ headers=modern_tool_headers("exec_query"),
+ )
+ assert called.status_code == 200
+ structured = called.json()["result"]["structuredContent"]
+ assert structured["registry_dispatch"] is True
+ assert structured["sql_length"] == len("SELECT 1")
+ assert structured["_execution_info"]["canonical_tool_name"] == (
+ "exec_query"
+ )
+
+
@pytest.mark.asyncio
async def
test_http_discover_is_stateless_and_unknown_method_does_not_kill_server():
app = create_test_server().streamable_http_app(
@@ -1243,6 +1302,43 @@ async def
test_true_subprocess_stdio_does_not_advertise_or_serve_subscriptions()
assert recovered.result_type == "complete"
[email protected]
+async def test_true_subprocess_stdio_uses_production_tool_registry():
+ server_script = Path(__file__).with_name("tool_registry_server.py")
+ server_params = StdioServerParameters(
+ command=sys.executable,
+ args=[str(server_script)],
+ )
+
+ async with Client(stdio_client(server_params)) as modern:
+ tools = {
+ tool.name: tool
+ for tool in (await modern.list_tools(cache_mode="bypass")).tools
+ }
+ assert len(tools) == 25
+ assert tools["exec_query"].input_schema["required"] == ["sql"]
+ assert "get_monitoring_metrics_info" not in tools
+
+ with pytest.raises(MCPError) as invalid:
+ await modern.call_tool("exec_query", {})
+ assert invalid.value.code == -32602
+
+ called = await modern.call_tool("exec_query", {"sql": "SELECT 1"})
+ assert called.structured_content["registry_dispatch"] is True
+ assert called.structured_content["_execution_info"][
+ "canonical_tool_name"
+ ] == "exec_query"
+
+ async with Client(stdio_client(server_params), mode="legacy") as legacy:
+ tools = {
+ tool.name: tool
+ for tool in (await legacy.list_tools()).tools
+ }
+ assert len(tools) == 25
+ called = await legacy.call_tool("exec_query", {"sql": "SELECT 1"})
+ assert called.structured_content["registry_dispatch"] is True
+
+
@pytest.mark.asyncio
async def test_true_subprocess_stdio_enforces_tool_schemas_for_both_eras():
server_script = Path(__file__).with_name("schema_validation_server.py")
diff --git a/test/protocol/tool_registry_server.py
b/test/protocol/tool_registry_server.py
new file mode 100644
index 0000000..3058305
--- /dev/null
+++ b/test/protocol/tool_registry_server.py
@@ -0,0 +1,95 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Production-registry fixture for HTTP and true subprocess Stdio tests."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from typing import Any
+
+from mcp.server.stdio import stdio_server
+from mcp.types import GetPromptResult, Prompt, Resource
+
+from doris_mcp_server import __version__
+from doris_mcp_server.protocol import create_doris_mcp_server
+from doris_mcp_server.tools.tools_manager import DorisToolsManager
+from doris_mcp_server.utils.config import DorisConfig
+from doris_mcp_server.utils.db import DorisConnectionManager
+
+
+class EmptyResourcesManager:
+ async def list_resources(self) -> list[Resource]:
+ return []
+
+ async def read_resource(self, uri: str) -> str:
+ return json.dumps({"uri": uri})
+
+
+class EmptyPromptsManager:
+ async def list_prompts(self) -> list[Prompt]:
+ return []
+
+ async def get_prompt(
+ self,
+ name: str,
+ arguments: dict[str, Any],
+ ) -> GetPromptResult:
+ del name, arguments
+ raise ValueError("Unknown prompt")
+
+
+class RegistryToolsManager(DorisToolsManager):
+ """Use the production registry while replacing database execution."""
+
+ async def _exec_query_tool(
+ self,
+ arguments: dict[str, Any],
+ ) -> dict[str, Any]:
+ return {
+ "registry_dispatch": True,
+ "sql_length": len(self._required_string(arguments, "sql")),
+ }
+
+
+def create_registry_test_server():
+ config = DorisConfig()
+ connection_manager = DorisConnectionManager(config)
+ return create_doris_mcp_server(
+ resources_manager=EmptyResourcesManager(),
+ tools_manager=RegistryToolsManager(connection_manager),
+ prompts_manager=EmptyPromptsManager(),
+ name="doris-mcp-tool-registry-test",
+ version=__version__,
+ logger=logging.getLogger(__name__),
+ )
+
+
+async def main() -> None:
+ server = create_registry_test_server()
+ async with stdio_server() as (read_stream, write_stream):
+ await server.run(
+ read_stream,
+ write_stream,
+ server.create_initialization_options(),
+ )
+
+
+if __name__ == "__main__":
+ asyncio.run(main())
+
diff --git a/test/tools/test_tool_registry.py b/test/tools/test_tool_registry.py
new file mode 100644
index 0000000..5a5631c
--- /dev/null
+++ b/test/tools/test_tool_registry.py
@@ -0,0 +1,200 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Contract tests for the single Doris MCP tool definition registry."""
+
+from __future__ import annotations
+
+import inspect
+import json
+from pathlib import Path
+from unittest.mock import AsyncMock, Mock, patch
+
+import pytest
+from mcp.types import Tool
+
+from doris_mcp_server.schema_validation import ToolSchemaGuard
+from doris_mcp_server.tools.tool_registry import (
+ DORIS_OAUTH_EXPLAIN_TOOL_SET,
+ DORIS_OAUTH_METADATA_TOOL_SET,
+ DORIS_OAUTH_QUERY_TOOL_SET,
+ LEGACY_TOOL_ALIAS_NAMES,
+ RESTRICTED_TOOL_NAMES,
+ ToolDefinitionRegistry,
+ ToolRegistryError,
+ policy_definition_for_tool,
+)
+from doris_mcp_server.tools.tools_manager import DorisToolsManager
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+
+
[email protected]
+def tools_manager() -> DorisToolsManager:
+ connection_manager = Mock()
+ connection_manager.config.adbc.default_max_rows = 1000
+ connection_manager.config.adbc.default_timeout = 30
+ connection_manager.config.adbc.default_return_format = "dict"
+ return DorisToolsManager(connection_manager)
+
+
+def test_registry_is_the_complete_schema_policy_handler_and_audit_source(
+ tools_manager: DorisToolsManager,
+) -> None:
+ registry = tools_manager.tool_registry
+ advertised = registry.advertised_definitions
+
+ assert len(advertised) == 25
+ assert len(registry.definitions) == 25 + len(LEGACY_TOOL_ALIAS_NAMES)
+ assert len(registry.advertised_names) ==
len(set(registry.advertised_names))
+
+ ToolSchemaGuard().compile_catalog(registry.listed_tools())
+ for definition in registry.definitions:
+ assert definition.bind_handler(tools_manager)
+ assert definition.audit.event_name == (
+ f"mcp.tool.call.{definition.name}"
+ )
+ assert definition.policy == policy_definition_for_tool(definition.name)
+ if definition.tool is not None:
+ properties = definition.tool.input_schema.get("properties", {})
+ assert set(definition.audit.argument_names) == set(properties)
+
+
+def test_registry_policy_classes_cover_every_executable_tool(
+ tools_manager: DorisToolsManager,
+) -> None:
+ names_by_policy = {
+ policy_class: {
+ definition.name
+ for definition in tools_manager.tool_registry.definitions
+ if definition.policy.policy_class == policy_class
+ }
+ for policy_class in ("metadata", "query", "explain", "restricted")
+ }
+
+ assert names_by_policy["metadata"] == DORIS_OAUTH_METADATA_TOOL_SET
+ assert names_by_policy["query"] == DORIS_OAUTH_QUERY_TOOL_SET
+ assert names_by_policy["explain"] == DORIS_OAUTH_EXPLAIN_TOOL_SET
+ assert names_by_policy["restricted"] == RESTRICTED_TOOL_NAMES
+
+
[email protected]
+async def test_registry_dispatch_and_audit_do_not_log_sql_values(
+ tools_manager: DorisToolsManager,
+) -> None:
+ sql = "SELECT 'registry-secret-value'"
+ tools_manager._exec_query_tool = AsyncMock(return_value={"ok": True})
+ audit_logger = Mock()
+
+ with patch(
+ "doris_mcp_server.tools.tools_manager.get_audit_logger",
+ return_value=audit_logger,
+ ):
+ payload = json.loads(
+ await tools_manager.call_tool("exec_query", {"sql": sql})
+ )
+
+ tools_manager._exec_query_tool.assert_awaited_once_with({"sql": sql})
+ assert payload["ok"] is True
+ assert payload["_execution_info"]["canonical_tool_name"] == "exec_query"
+ assert payload["_execution_info"]["audit_event"] == (
+ "mcp.tool.call.exec_query"
+ )
+ assert sql not in repr(audit_logger.info.call_args)
+ assert "sql" in repr(audit_logger.info.call_args)
+ assert tools_manager.tool_registry.resolve(
+ "exec_query"
+ ).audit.sensitive_arguments == ("sql",)
+
+
[email protected]
+async def test_legacy_alias_uses_registry_override_without_mutating_input(
+ tools_manager: DorisToolsManager,
+) -> None:
+ tools_manager._get_monitoring_metrics_tool = AsyncMock(
+ return_value={"ok": True}
+ )
+ arguments = {"role": "fe"}
+
+ payload = json.loads(
+ await tools_manager.call_tool(
+ "get_monitoring_metrics_info",
+ arguments,
+ )
+ )
+
+ assert arguments == {"role": "fe"}
+ tools_manager._get_monitoring_metrics_tool.assert_awaited_once_with(
+ {"role": "fe", "content_type": "definitions"}
+ )
+ assert payload["_execution_info"]["canonical_tool_name"] == (
+ "get_monitoring_metrics"
+ )
+ assert "get_monitoring_metrics_info" not in (
+ tools_manager.tool_registry.advertised_names
+ )
+
+
+def test_registry_rejects_duplicate_unknown_and_missing_handler_definitions(
+ tools_manager: DorisToolsManager,
+) -> None:
+ tool = Tool(
+ name="exec_query",
+ description="Execute a query",
+ input_schema={
+ "type": "object",
+ "properties": {"sql": {"type": "string"}},
+ "required": ["sql"],
+ },
+ )
+
+ with pytest.raises(ToolRegistryError, match="Duplicate tool schema"):
+ ToolDefinitionRegistry.from_tools([tool, tool], Mock())
+
+ with pytest.raises(ToolRegistryError, match="has no policy definition"):
+ ToolDefinitionRegistry.from_tools(
+ [
+ Tool(
+ name="unclassified_tool",
+ description="Unclassified",
+ input_schema={"type": "object", "properties": {}},
+ )
+ ],
+ Mock(),
+ )
+
+ with pytest.raises(ToolRegistryError, match="missing handler"):
+ ToolDefinitionRegistry.from_tools(
+ tools_manager.tool_registry.listed_tools(),
+ object(),
+ )
+
+
+def test_checked_in_tool_catalog_is_generated_from_registry(
+ tools_manager: DorisToolsManager,
+) -> None:
+ checked_in = (REPO_ROOT / "docs" / "tool-registry.md").read_text(
+ encoding="utf-8"
+ )
+ assert checked_in == tools_manager.tool_registry.render_markdown()
+
+
+def test_tools_manager_has_no_parallel_decorator_or_dispatch_registry() ->
None:
+ source = inspect.getsource(DorisToolsManager)
+
+ assert "register_tools_with_mcp" not in source
+ assert "@mcp.tool" not in source
+ assert 'elif name == "' not in source
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
