Databricks MCP Server

Instructions

Implementation Reference

• src/server/simple_databricks_mcp_server.py:370-412 (handler)

  MCP tool handler for create_job: constructs job configuration for a notebook task (serverless or existing cluster) and delegates to the jobs API wrapper. This is the primary entrypoint for the MCP 'create_job' tool.

  @mcp.tool()
  async def create_job(
      job_name: str,
      notebook_path: str,
      timeout_seconds: int = 3600,
      parameters: Optional[dict] = None,
      cluster_id: Optional[str] = None,
      use_serverless: bool = True
  ) -> str:
      """Create a new Databricks job to run a notebook (uses serverless by default)"""
      logger.info(f"Creating job: {job_name}")
      try:
          task_config = {
              "task_key": "main_task",
              "notebook_task": {
                  "notebook_path": notebook_path,
                  "base_parameters": parameters or {}
              },
              "timeout_seconds": timeout_seconds
          }
          
          # Configure compute: serverless vs cluster  
          if use_serverless:
              # For serverless compute, simply don't specify any cluster configuration
              # Databricks will automatically use serverless compute
              pass
          elif cluster_id:
              task_config["existing_cluster_id"] = cluster_id
          else:
              raise ValueError("Must specify either use_serverless=True or provide cluster_id")
              
          job_config = {
              "name": job_name,
              "tasks": [task_config],
              "format": "MULTI_TASK"
          }
          
          result = await jobs.create_job(job_config)
          return json.dumps(result)
      except Exception as e:
          logger.error(f"Error creating job: {str(e)}")
          return json.dumps({"error": str(e)})

• src/api/jobs.py:14-28 (helper)

  Underlying helper function that performs the actual Databricks API call to create a job via POST /api/2.0/jobs/create. Called by the MCP handler.

  async def create_job(job_config: Dict[str, Any]) -> Dict[str, Any]:
      """
      Create a new Databricks job.
      
      Args:
          job_config: Job configuration
          
      Returns:
          Response containing the job ID
          
      Raises:
          DatabricksAPIError: If the API request fails
      """
      logger.info("Creating new job")
      return make_api_request("POST", "/api/2.0/jobs/create", data=job_config)