Doris MCP Server

Instructions

Implementation Reference

• doris_mcp_server/tools/tools_manager.py:114-136 (registration)

  Registers the 'get_table_schema' tool with the MCP server using the @mcp.tool decorator. Defines input parameters: table_name (required), db_name and catalog_name (optional). The function proxies to self.call_tool for execution.

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

• doris_mcp_server/tools/tools_manager.py:807-826 (schema)

  Defines the input schema validation for the 'get_table_schema' tool in the list_tools method, specifying JSON schema with table_name as required string property.

                  name="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
  """,
                  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:1476-1485 (handler)

  Dispatcher handler in ToolsManager.call_tool that routes 'get_table_schema' tool invocations by extracting arguments and delegating to MetadataExtractor.get_table_schema_for_mcp.

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

• doris_mcp_server/utils/schema_extractor.py:1542-1568 (handler)

  Core handler implementation for 'get_table_schema' tool in MetadataExtractor. Validates input, calls get_table_schema_async to fetch schema via DESCRIBE query, formats MCP response with success/error handling.

  async def get_table_schema_for_mcp(
      self, 
      table_name: str, 
      db_name: str = None, 
      catalog_name: str = None
  ) -> Dict[str, Any]:
      """Get detailed schema information for specified table (columns, types, comments, etc.) - MCP interface"""
      logger.info(f"Getting table schema: 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:
          schema = await self.get_table_schema_async(table_name=table_name, db_name=db_name, catalog_name=catalog_name)
          
          if not schema:
              return self._format_response(
                  success=False, 
                  error="Table does not exist or has no columns", 
                  message=f"Unable to get schema for table {catalog_name or 'default'}.{db_name or self.db_name}.{table_name}"
              )
          
          return self._format_response(success=True, result=schema)
      except Exception as e:
          logger.error(f"Failed to get table schema: {str(e)}", exc_info=True)
          return self._format_response(success=False, error=str(e), message="Error occurred while getting table schema")

• doris_mcp_server/utils/schema_extractor.py:1186-1223 (helper)

  Helper function that executes DESCRIBE query (with catalog.db.table support) to retrieve raw table schema and formats it into structured column information list.

  async def get_table_schema_async(self, table_name: str, db_name: str = None, catalog_name: str = None) -> List[Dict[str, Any]]:
      """Asynchronously get table schema information"""
      try:
          # Use async query method
          effective_catalog = catalog_name or self.catalog_name
          
          # Build query statement
          if effective_catalog and effective_catalog != "internal":
              query = f"DESCRIBE `{effective_catalog}`.`{db_name or self.db_name}`.`{table_name}`"
          else:
              query = f"DESCRIBE `{db_name or self.db_name}`.`{table_name}`"
          
          # Execute async query
          result = await self._execute_query_async(query, db_name)
          
          if not result:
              return []
          
          # Process results
          schema = []
          for row in result:
              if isinstance(row, dict):
                  schema.append({
                      'column_name': row.get('Field', ''),
                      'data_type': row.get('Type', ''),
                      'is_nullable': row.get('Null', 'NO') == 'YES',
                      'default_value': row.get('Default', None),
                      'comment': row.get('Comment', ''),
                      'key': row.get('Key', ''),
                      'extra': row.get('Extra', '')
                  })
          
          return schema
          
      except Exception as e:
          logger.error(f"Failed to get table schema: {e}")
          return []