Databricks MCP Server

Instructions

Implementation Reference

• src/server/simple_databricks_mcp_server.py:29-39 (handler)

  MCP tool handler for 'list_clusters': decorated with @mcp.tool(), logs the action, calls clusters.list_clusters() helper, returns JSON result or error.

  @mcp.tool()
  async def list_clusters() -> str:
      """List all Databricks clusters"""
      logger.info("Listing clusters")
      try:
          result = await clusters.list_clusters()
          return json.dumps(result)
      except Exception as e:
          logger.error(f"Error listing clusters: {str(e)}")
          return json.dumps({"error": str(e)})

• src/api/clusters.py:48-60 (helper)

  Core helper function that performs the Databricks API GET request to /api/2.0/clusters/list to fetch clusters.

  async def list_clusters() -> Dict[str, Any]:
      """
      List all Databricks clusters.
      
      Returns:
          Response containing a list of clusters
          
      Raises:
          DatabricksAPIError: If the API request fails
      """
      logger.info("Listing all clusters")
      return make_api_request("GET", "/api/2.0/clusters/list")

• src/server/simple_databricks_mcp_server.py:29-29 (registration)

  The @mcp.tool() decorator registers the list_clusters function as an MCP tool.

  @mcp.tool()