Databricks MCP Server

Instructions

Implementation Reference

• src/server/simple_databricks_mcp_server.py:240-251 (handler)

  MCP tool handler for run_job: registers the tool, handles input parameters, calls the jobs API wrapper, serializes response to JSON, and manages errors.

  @mcp.tool()
  async def run_job(job_id: str, notebook_params: Optional[Dict[str, Any]] = None) -> str:
      """Run a Databricks job"""
      logger.info(f"Running job: {job_id}")
      try:
          if notebook_params is None:
              notebook_params = {}
          result = await jobs.run_job(job_id, notebook_params)
          return json.dumps(result)
      except Exception as e:
          logger.error(f"Error running job: {str(e)}")
          return json.dumps({"error": str(e)})

• src/api/jobs.py:31-51 (helper)

  Core API wrapper that constructs the request parameters and calls Databricks /api/2.0/jobs/run-now endpoint to trigger the job execution.

  async def run_job(job_id: int, notebook_params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
      """
      Run a job now.
      
      Args:
          job_id: ID of the job to run
          notebook_params: Optional parameters for the notebook
          
      Returns:
          Response containing the run ID
          
      Raises:
          DatabricksAPIError: If the API request fails
      """
      logger.info(f"Running job: {job_id}")
      
      run_params = {"job_id": job_id}
      if notebook_params:
          run_params["notebook_params"] = notebook_params
          
      return make_api_request("POST", "/api/2.0/jobs/run-now", data=run_params)

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

  FastMCP @mcp.tool() decorator registers the run_job function as an MCP tool with auto-generated schema from signature and docstring.

  @mcp.tool()