Doris MCP Server

Instructions

Implementation Reference

• doris_mcp_server/utils/schema_extractor.py:1340-1369 (handler)

  Core handler function that executes the SQL query to fetch column names and comments from information_schema.columns and formats them into a dictionary.

  async def get_column_comments_async(self, table_name: str, db_name: str = None, catalog_name: str = None) -> Dict[str, str]:
      """Async version: get comments for all columns in a table."""
      try:
          effective_db = db_name or self.db_name
          effective_catalog = catalog_name or self.catalog_name
  
          query = f"""
          SELECT 
              COLUMN_NAME, 
              COLUMN_COMMENT 
          FROM 
              information_schema.columns 
          WHERE 
              TABLE_SCHEMA = '{effective_db}' 
              AND TABLE_NAME = '{table_name}'
          ORDER BY 
              ORDINAL_POSITION
          """
  
          rows = await self._execute_query_with_catalog_async(query, effective_db, effective_catalog)
          comments: Dict[str, str] = {}
          for col in rows or []:
              name = col.get("COLUMN_NAME", "")
              if name:
                  comments[name] = col.get("COLUMN_COMMENT", "") or ""
          return comments
      except Exception as e:
          logger.error(f"Failed to get column comments asynchronously: {e}")
          return {}

• doris_mcp_server/utils/schema_extractor.py:1614-1632 (handler)

  MCP-specific wrapper method that calls the core get_column_comments_async handler and formats the response.

  async def get_table_column_comments_for_mcp(
      self, 
      table_name: str, 
      db_name: str = None, 
      catalog_name: str = None
  ) -> Dict[str, Any]:
      """Get comment information for all columns in specified table - MCP interface"""
      logger.info(f"Getting table column comments: Table: {table_name}, DB: {db_name}, Catalog: {catalog_name}")
      
      if not table_name:
          return self._format_response(success=False, error="Missing table_name parameter")
      
      try:
          comments = await self.get_column_comments_async(table_name=table_name, db_name=db_name, catalog_name=catalog_name)
          return self._format_response(success=True, result=comments)
      except Exception as e:
          logger.error(f"Failed to get table column comments: {str(e)}", exc_info=True)
          return self._format_response(success=False, error=str(e), message="Error occurred while getting table column comments")

• doris_mcp_server/tools/tools_manager.py:1513-1523 (handler)

  Routing handler in tools_manager that delegates the tool execution to MetadataExtractor.

  async def _get_table_column_comments_tool(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
      """Get table column comments tool routing"""
      table_name = arguments.get("table_name")
      db_name = arguments.get("db_name")
      catalog_name = arguments.get("catalog_name")
      
      # Delegate to metadata extractor for processing
      return await self.metadata_extractor.get_table_column_comments_for_mcp(
          table_name, db_name, catalog_name
      )

• doris_mcp_server/tools/tools_manager.py:198-220 (registration)

  MCP tool registration decorator and wrapper function that routes to the internal call_tool dispatcher.

          # 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, catalog_name: str = 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
              })

• doris_mcp_server/tools/tools_manager.py:883-904 (schema)

  Tool schema definition including input schema validation for the get_table_column_comments tool.

              Tool(
                  name="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
  """,
                  inputSchema={
                      "type": "object",
                      "properties": {
                          "table_name": {"type": "string", "description": "Table name"},
                          "db_name": {"type": "string", "description": "Database name"},
                          "catalog_name": {"type": "string", "description": "Catalog name"},
                      },
                      "required": ["table_name"],
                  },
              ),