create_job
Set up a new Databricks job to execute a notebook, with serverless mode as default, and define parameters, timeout, or cluster configurations as needed.
Instructions
Create a new Databricks job to run a notebook (uses serverless by default)
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| cluster_id | No | ||
| job_name | Yes | ||
| notebook_path | Yes | ||
| parameters | No | ||
| timeout_seconds | No | ||
| use_serverless | No |
Implementation Reference
- The primary MCP tool implementation for 'create_job'. This handler is registered via @mcp.tool() decorator, constructs a multi-task job configuration optimized for serverless compute, and delegates the API call to the jobs module.@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)Helper function in the jobs API module that performs the core HTTP POST request to Databricks Jobs API endpoint /api/2.0/jobs/create.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)