Databricks MCP Server

Instructions

Implementation Reference

• src/server/simple_databricks_mcp_server.py:348-368 (handler)

  MCP tool handler and registration for 'create_notebook'. Wraps the notebooks.import_notebook function to create notebooks in Databricks workspace via API.

  @mcp.tool()
  async def create_notebook(
      path: str,
      content: str,
      language: str = "PYTHON",
      overwrite: bool = False
  ) -> str:
      """Create a new notebook in the Databricks workspace"""
      logger.info(f"Creating notebook at path: {path}")
      try:
          result = await notebooks.import_notebook(
              path=path,
              content=content,
              format="SOURCE",
              language=language.upper(),
              overwrite=overwrite
          )
          return json.dumps(result)
      except Exception as e:
          logger.error(f"Error creating notebook: {str(e)}")
          return json.dumps({"error": str(e)})

• src/api/notebooks.py:15-54 (helper)

  Core helper function that performs the actual Databricks API call to import/create a notebook using the Workspace Import API.

  async def import_notebook(
      path: str,
      content: str,
      format: str = "SOURCE",
      language: Optional[str] = None,
      overwrite: bool = False,
  ) -> Dict[str, Any]:
      """
      Import a notebook into the workspace.
      
      Args:
          path: The path where the notebook should be stored
          content: The content of the notebook (base64 encoded)
          format: The format of the notebook (SOURCE, HTML, JUPYTER, DBC)
          language: The language of the notebook (SCALA, PYTHON, SQL, R)
          overwrite: Whether to overwrite an existing notebook
          
      Returns:
          Empty response on success
          
      Raises:
          DatabricksAPIError: If the API request fails
      """
      logger.info(f"Importing notebook to path: {path}")
      
      # Ensure content is base64 encoded
      if not is_base64(content):
          content = base64.b64encode(content.encode("utf-8")).decode("utf-8")
      
      import_data = {
          "path": path,
          "format": format,
          "content": content,
          "overwrite": overwrite,
      }
      
      if language:
          import_data["language"] = language
          
      return make_api_request("POST", "/api/2.0/workspace/import", data=import_data)