Doris MCP Server

Instructions

Implementation Reference

• doris_mcp_server/tools/tools_manager.py:175-196 (registration)

  MCP tool registration for 'get_table_comment' including decorator, description, parameters, and stub handler that delegates to internal call_tool method.

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

• doris_mcp_server/tools/tools_manager.py:861-882 (schema)

  Input schema definition for 'get_table_comment' tool used in stdio mode listing.

              Tool(
                  name="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
  """,
                  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"],
                  },
              ),

• doris_mcp_server/tools/tools_manager.py:1502-1512 (handler)

  Tool routing handler in DorisToolsManager that delegates 'get_table_comment' execution to MetadataExtractor.get_table_comment_for_mcp.

  async def _get_table_comment_tool(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
      """Get table comment 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_comment_for_mcp(
          table_name, db_name, catalog_name
      )

• doris_mcp_server/utils/schema_extractor.py:1595-1613 (handler)

  MCP-specific wrapper method in MetadataExtractor that calls get_table_comment_async and formats response for MCP tools.

  async def get_table_comment_for_mcp(
      self, 
      table_name: str, 
      db_name: str = None, 
      catalog_name: str = None
  ) -> Dict[str, Any]:
      """Get comment information for specified table - MCP interface"""
      logger.info(f"Getting table comment: 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:
          comment = await self.get_table_comment_async(table_name=table_name, db_name=db_name, catalog_name=catalog_name)
          return self._format_response(success=True, result=comment)
      except Exception as e:
          logger.error(f"Failed to get table comment: {str(e)}", exc_info=True)
          return self._format_response(success=False, error=str(e), message="Error occurred while getting table comment")

• doris_mcp_server/utils/schema_extractor.py:1316-1339 (handler)

  Core handler function that executes the SQL query against information_schema.tables to retrieve the table comment, with catalog support.

  async def get_table_comment_async(self, table_name: str, db_name: str = None, catalog_name: str = None) -> str:
      """Async version: get the comment for a table."""
      try:
          effective_db = db_name or self.db_name
          effective_catalog = catalog_name or self.catalog_name
  
          query = f"""
          SELECT 
              TABLE_COMMENT 
          FROM 
              information_schema.tables 
          WHERE 
              TABLE_SCHEMA = '{effective_db}' 
              AND TABLE_NAME = '{table_name}'
          """
  
          result = await self._execute_query_with_catalog_async(query, effective_db, effective_catalog)
          if not result or not result[0]:
              return ""
          return result[0].get("TABLE_COMMENT", "") or ""
      except Exception as e:
          logger.error(f"Failed to get table comment asynchronously: {e}")
          return ""