> ## Documentation Index
> Fetch the complete documentation index at: https://docs.kubiya.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Jobs Service

> Schedule and automate tasks with cron jobs and webhooks

The Jobs Service enables you to schedule recurring tasks, create webhook-triggered automations, and manually trigger job executions. Jobs can execute predefined agents, teams, or workflows at specified intervals or on-demand.

## Quick Start

```python theme={null}
from kubiya import ControlPlaneClient

client = ControlPlaneClient()

# List all jobs
jobs = client.jobs.list()

# Create a cron job
job_data = {
    "name": "Daily Health Check",
    "trigger_type": "cron",
    "cron_schedule": "0 9 * * *",  # Every day at 9 AM
    "planning_mode": "predefined_agent",
    "entity_type": "agent",
    "entity_id": "agent-uuid",
    "prompt_template": "Check the health of all production services"
}

job = client.jobs.create(job_data)
print(f"Created job with webhook URL: {job.get('webhook_url', 'N/A')}")
```

## Features

<CardGroup cols={2}>
  <Card title="Cron Scheduling" icon="calendar-days">
    Schedule recurring tasks with cron expressions
  </Card>

  <Card title="Webhook Triggers" icon="webhook">
    Trigger jobs via secure webhook endpoints
  </Card>

  <Card title="Manual Execution" icon="hand-pointer">
    Manually trigger jobs on-demand
  </Card>

  <Card title="Execution History" icon="clock-rotate-left">
    Track job execution history and status
  </Card>
</CardGroup>

## List Jobs

Retrieve all jobs with optional filtering:

```python theme={null}
# List all jobs
jobs = client.jobs.list()

# Filter by enabled status
enabled_jobs = client.jobs.list(enabled=True)
disabled_jobs = client.jobs.list(enabled=False)

# Filter by trigger type
cron_jobs = client.jobs.list(trigger_type="cron")
webhook_jobs = client.jobs.list(trigger_type="webhook")
manual_jobs = client.jobs.list(trigger_type="manual")
```

**Parameters:**

* `enabled` (bool, optional): Filter by enabled status (True/False)
* `trigger_type` (str, optional): Filter by trigger type: 'cron', 'webhook', or 'manual'

**Returns:** List of job dictionaries

## Get Job

Retrieve a specific job by ID:

```python theme={null}
job = client.jobs.get(job_id="job_550e8400-e29b-41d4-a716-446655440000")

print(f"Job: {job['name']}")
print(f"Trigger Type: {job['trigger_type']}")
print(f"Enabled: {job['enabled']}")

if job['trigger_type'] == 'cron':
    print(f"Schedule: {job['cron_schedule']}")
elif job['trigger_type'] == 'webhook':
    print(f"Webhook URL: {job['webhook_url']}")
```

**Parameters:**

* `job_id` (str, required): Job ID (format: `job_<uuid>`)

**Returns:** Dictionary containing job details

## Create Job

### Cron Job

Create a job that runs on a schedule:

```python theme={null}
job_data = {
    "name": "Nightly Backup",
    "description": "Backup production database",
    "trigger_type": "cron",
    "cron_schedule": "0 2 * * *",  # Daily at 2 AM
    "cron_timezone": "America/New_York",
    "planning_mode": "predefined_agent",
    "entity_type": "agent",
    "entity_id": "backup-agent-uuid",
    "prompt_template": "Perform full database backup",
    "executor_type": "specific_queue",
    "worker_queue_name": "production",
    "enabled": True
}

job = client.jobs.create(job_data)
print(f"Cron job created: {job['id']}")
```

### Webhook Job

Create a job triggered via webhook:

```python theme={null}
job_data = {
    "name": "Deploy on Push",
    "description": "Deploy application when code is pushed",
    "trigger_type": "webhook",
    "planning_mode": "predefined_team",
    "entity_type": "team",
    "entity_id": "deployment-team-uuid",
    "prompt_template": "Deploy branch {{branch_name}} to {{environment}}",
    "executor_type": "environment",
    "environment_name": "production",
    "enabled": True
}

job = client.jobs.create(job_data)
print(f"Webhook URL: {job['webhook_url']}")
print(f"Webhook Secret: {job['webhook_secret']}")
```

### Manual Job

Create a job for manual triggering only:

```python theme={null}
job_data = {
    "name": "Emergency Rollback",
    "description": "Rollback to previous version",
    "trigger_type": "manual",
    "planning_mode": "predefined_workflow",
    "prompt_template": "Rollback service {{service_name}} to version {{version}}",
    "executor_type": "auto",
    "enabled": True
}

job = client.jobs.create(job_data)
print(f"Manual job created: {job['id']}")
```

**Parameters:**

* `job_data` (dict, required): Job configuration
  * `name` (str, required): Job name
  * `trigger_type` (str, required): 'cron', 'webhook', or 'manual'
  * `cron_schedule` (str): Cron expression (required for cron trigger)
  * `cron_timezone` (str): Timezone for cron (default: 'UTC')
  * `planning_mode` (str): 'on\_the\_fly', 'predefined\_agent', 'predefined\_team', or 'predefined\_workflow'
  * `entity_type` (str): 'agent' or 'team' (for predefined modes)
  * `entity_id` (str): Entity UUID (for predefined modes)
  * `prompt_template` (str): Prompt with {{variable}} placeholders
  * `system_prompt` (str, optional): System prompt override
  * `executor_type` (str): 'auto', 'specific\_queue', or 'environment'
  * `worker_queue_name` (str): Worker queue name (for specific\_queue)
  * `environment_name` (str): Environment name (for environment executor)
  * `config` (dict, optional): Additional configuration
  * `execution_environment` (dict, optional): Execution environment config
  * `enabled` (bool): Whether job is enabled (default: True)
  * `description` (str, optional): Job description

**Returns:** Dictionary containing created job with webhook\_url if applicable

<Info>
  For webhook jobs, save the `webhook_url` and `webhook_secret` - the secret is only shown once at creation time.
</Info>

## Update Job

Update an existing job (partial update):

```python theme={null}
# Update cron schedule
updated_job = client.jobs.update(
    job_id="job_uuid",
    job_data={
        "cron_schedule": "0 10 * * *",  # Change to 10 AM
        "cron_timezone": "UTC"
    }
)

# Update prompt template
updated_job = client.jobs.update(
    job_id="job_uuid",
    job_data={
        "prompt_template": "Updated prompt with {{new_variable}}"
    }
)

# Disable job
updated_job = client.jobs.update(
    job_id="job_uuid",
    job_data={
        "enabled": False
    }
)
```

**Parameters:**

* `job_id` (str, required): Job ID
* `job_data` (dict, required): Fields to update
  * `name` (str): Job name
  * `description` (str): Job description
  * `enabled` (bool): Whether job is enabled
  * `cron_schedule` (str): Cron expression
  * `cron_timezone` (str): Timezone
  * `planning_mode` (str): Planning mode
  * `entity_type` (str): Entity type
  * `entity_id` (str): Entity ID
  * `prompt_template` (str): Prompt template
  * `system_prompt` (str): System prompt
  * `executor_type` (str): Executor type
  * `worker_queue_name` (str): Worker queue name
  * `environment_name` (str): Environment name
  * `config` (dict): Configuration
  * `execution_environment` (dict): Execution environment

**Returns:** Dictionary containing updated job details

<Warning>
  Updating `cron_schedule` will recreate the Temporal Schedule, which may briefly pause execution.
</Warning>

## Delete Job

Delete a job and its Temporal Schedule:

```python theme={null}
client.jobs.delete(job_id="job_uuid")
print("Job deleted successfully")
```

**Parameters:**

* `job_id` (str, required): Job ID

**Returns:** None (204 No Content on success)

## Job Execution

### Manual Trigger

Manually trigger a job execution:

```python theme={null}
# Trigger with parameters
result = client.jobs.trigger(
    job_id="job_uuid",
    parameters={
        "branch_name": "main",
        "environment": "production"
    }
)

print(f"Workflow ID: {result['workflow_id']}")
print(f"Execution ID: {result['execution_id']}")
print(f"Status: {result['status']}")

# Trigger with config override
result = client.jobs.trigger(
    job_id="job_uuid",
    parameters={"service": "api"},
    config_override={
        "timeout": 7200
    }
)
```

**Parameters:**

* `job_id` (str, required): Job ID
* `parameters` (dict, optional): Parameters to substitute in prompt template
* `config_override` (dict, optional): Config overrides for this execution

**Returns:** Dictionary with workflow\_id, execution\_id, and status

### Enable/Disable Job

Enable or disable a job:

```python theme={null}
# Enable a job
enabled_job = client.jobs.enable(job_id="job_uuid")
print(f"Job enabled: {enabled_job['enabled']}")

# Disable a job
disabled_job = client.jobs.disable(job_id="job_uuid")
print(f"Job disabled: {not disabled_job['enabled']}")
```

**Parameters:**

* `job_id` (str, required): Job ID

**Returns:** Dictionary containing updated job details

<Tip>
  For cron jobs, enabling creates the Temporal Schedule if it doesn't exist, and disabling pauses it.
</Tip>

### Get Execution History

Retrieve execution history for a job:

```python theme={null}
# Get recent executions
history = client.jobs.get_executions(
    job_id="job_uuid",
    limit=50,
    offset=0
)

print(f"Total executions: {history['total_count']}")

for execution in history['executions']:
    print(f"Execution {execution['execution_id']}:")
    print(f"  Status: {execution['status']}")
    print(f"  Started: {execution['started_at']}")
    print(f"  Duration: {execution['duration_ms']}ms")
    if execution.get('error_message'):
        print(f"  Error: {execution['error_message']}")
```

**Parameters:**

* `job_id` (str, required): Job ID
* `limit` (int): Max executions to return (default: 50)
* `offset` (int): Number to skip (default: 0)

**Returns:** Dictionary with total\_count and executions list

## Webhook Triggers

### Trigger via Webhook

Trigger a job using its webhook endpoint:

```python theme={null}
import hmac
import hashlib
import json

# Prepare payload
payload = {
    "parameters": {
        "branch_name": "feature-x",
        "environment": "staging"
    },
    "metadata": {
        "source": "github",
        "commit": "abc123"
    }
}

# Calculate HMAC signature
webhook_secret = "your-webhook-secret"
payload_bytes = json.dumps(payload).encode('utf-8')
signature = hmac.new(
    webhook_secret.encode('utf-8'),
    payload_bytes,
    hashlib.sha256
).hexdigest()

# Trigger webhook
result = client.jobs.trigger_webhook(
    webhook_path="unique-webhook-path",
    payload=payload,
    signature=signature
)

print(f"Webhook triggered: {result['workflow_id']}")
```

**Parameters:**

* `webhook_path` (str, required): Unique webhook identifier from webhook URL
* `payload` (dict, required): Webhook payload
  * `parameters` (dict): Parameters for prompt template
  * `config_override` (dict, optional): Config overrides
  * `metadata` (dict, optional): Additional metadata
* `signature` (str, required): HMAC-SHA256 signature (hex)

**Returns:** Dictionary with workflow\_id, execution\_id, and status

<Warning>
  Always validate webhook signatures to prevent unauthorized job executions. Never expose webhook secrets in client-side code.
</Warning>

## Complete Example

Here's a complete example of creating and managing a scheduled job:

```python theme={null}
from kubiya import ControlPlaneClient

client = ControlPlaneClient()

# 1. Create a cron job
job_data = {
    "name": "Weekly Security Scan",
    "description": "Scan all repositories for vulnerabilities",
    "trigger_type": "cron",
    "cron_schedule": "0 0 * * 0",  # Every Sunday at midnight
    "cron_timezone": "UTC",
    "planning_mode": "predefined_team",
    "entity_type": "team",
    "entity_id": "security-team-uuid",
    "prompt_template": "Scan all repositories and report vulnerabilities",
    "executor_type": "specific_queue",
    "worker_queue_name": "security",
    "enabled": True
}

job = client.jobs.create(job_data)
print(f"Created job: {job['id']}")

# 2. Manually trigger a test run
result = client.jobs.trigger(
    job_id=job['id'],
    parameters={}
)
print(f"Test execution started: {result['execution_id']}")

# 3. Check execution history
import time
time.sleep(5)  # Wait for execution to complete

history = client.jobs.get_executions(job_id=job['id'], limit=1)
if history['executions']:
    latest = history['executions'][0]
    print(f"Latest execution status: {latest['status']}")

# 4. Temporarily disable job
client.jobs.disable(job_id=job['id'])
print("Job disabled")

# 5. Re-enable job
client.jobs.enable(job_id=job['id'])
print("Job re-enabled")
```

## Error Handling

Handle job-related errors:

```python theme={null}
from kubiya.resources.exceptions import JobError

try:
    job = client.jobs.get(job_id="invalid-id")
except JobError as e:
    print(f"Job error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")
```

## Cron Schedule Examples

Common cron schedule patterns:

```python theme={null}
# Every day at 9 AM
"0 9 * * *"

# Every hour
"0 * * * *"

# Every 15 minutes
"*/15 * * * *"

# Every Monday at 8 AM
"0 8 * * 1"

# First day of every month at midnight
"0 0 1 * *"

# Every weekday at 5 PM
"0 17 * * 1-5"

# Every Sunday at 2 AM
"0 2 * * 0"
```

<Tip>
  Use [crontab.guru](https://crontab.guru) to validate and understand cron expressions.
</Tip>

## Best Practices

### Job Design

* **Idempotent Jobs**: Design jobs to be safely re-runnable
* **Clear Naming**: Use descriptive names indicating purpose and frequency
* **Reasonable Timeouts**: Set appropriate execution timeouts
* **Error Handling**: Include error handling in prompt templates

### Scheduling

* **Timezone Awareness**: Always specify timezone explicitly
* **Off-Peak Hours**: Schedule heavy jobs during low-traffic periods
* **Avoid Overlap**: Ensure jobs don't overlap if they access shared resources
* **Test Manually First**: Test with manual trigger before enabling cron

### Webhook Security

* **Always Validate Signatures**: Never trust unsigned webhooks
* **Secure Secret Storage**: Store webhook secrets securely (env vars, secrets manager)
* **Rate Limiting**: Implement rate limiting on webhook endpoints
* **Audit Logs**: Monitor webhook trigger patterns

### Monitoring

* **Check Execution History**: Regularly review execution status
* **Set Up Alerts**: Alert on consecutive failures
* **Track Duration**: Monitor job duration trends
* **Disable Failed Jobs**: Temporarily disable repeatedly failing jobs

## API Reference

| Method              | Endpoint Pattern              | Description           |
| ------------------- | ----------------------------- | --------------------- |
| `list()`            | GET /jobs                     | List all jobs         |
| `get()`             | GET /jobs/{job_id}            | Get job by ID         |
| `create()`          | POST /jobs                    | Create new job        |
| `update()`          | PATCH /jobs/{job_id}          | Update job            |
| `delete()`          | DELETE /jobs/{job_id}         | Delete job            |
| `trigger()`         | POST /jobs/{job_id}/trigger   | Manually trigger      |
| `enable()`          | POST /jobs/{job_id}/enable    | Enable job            |
| `disable()`         | POST /jobs/{job_id}/disable   | Disable job           |
| `get_executions()`  | GET /jobs/{job_id}/executions | Get execution history |
| `trigger_webhook()` | POST /webhooks/{webhook_path} | Trigger via webhook   |

## Next Steps

<CardGroup cols={2}>
  <Card title="Agents Service" icon="robot" href="/sdk/control-plane-agents">
    Create agents for job execution
  </Card>

  <Card title="Teams Service" icon="users" href="/sdk/control-plane-teams">
    Use teams in jobs
  </Card>

  <Card title="Workers Service" icon="server" href="/sdk/control-plane-workers">
    Configure worker queues
  </Card>

  <Card title="Environments Service" icon="layer-group" href="/sdk/control-plane-environments">
    Manage execution environments
  </Card>
</CardGroup>