Databricks MCP Server

Instructions

Implementation Reference

• src/server/simple_databricks_mcp_server.py:264-280 (handler)

  MCP tool handler for 'export_notebook', decorated with @mcp.tool(). Delegates to notebooks.export_notebook API, trims large content for readability, and returns JSON response.

  @mcp.tool()
  async def export_notebook(path: str, format: str = "JUPYTER") -> str:
      """Export a notebook from the workspace"""
      logger.info(f"Exporting notebook: {path} in format: {format}")
      try:
          result = await notebooks.export_notebook(path, format)
          
          # For notebooks, we might want to trim the response for readability
          content = result.get("content", "")
          if len(content) > 1000:
              summary = f"{content[:1000]}... [content truncated, total length: {len(content)} characters]"
              result["content"] = summary
          
          return json.dumps(result)
      except Exception as e:
          logger.error(f"Error exporting notebook: {str(e)}")
          return json.dumps({"error": str(e)})

• src/api/notebooks.py:57-90 (helper)

  Helper function implementing the Databricks Workspace Export API call (/api/2.0/workspace/export). Handles base64 decoding for SOURCE/JUPYTER formats.

  async def export_notebook(
      path: str,
      format: str = "SOURCE",
  ) -> Dict[str, Any]:
      """
      Export a notebook from the workspace.
      
      Args:
          path: The path of the notebook to export
          format: The format to export (SOURCE, HTML, JUPYTER, DBC)
          
      Returns:
          Response containing the notebook content
          
      Raises:
          DatabricksAPIError: If the API request fails
      """
      logger.info(f"Exporting notebook from path: {path}")
      
      params = {
          "path": path,
          "format": format,
      }
      
      response = make_api_request("GET", "/api/2.0/workspace/export", params=params)
      
      # Optionally decode base64 content
      if "content" in response and format in ["SOURCE", "JUPYTER"]:
          try:
              response["decoded_content"] = base64.b64decode(response["content"]).decode("utf-8")
          except Exception as e:
              logger.warning(f"Failed to decode notebook content: {str(e)}")
              
      return response