Jobs Service - Kubiya

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

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

Cron Scheduling

Schedule recurring tasks with cron expressions

Webhook Triggers

Trigger jobs via secure webhook endpoints

Manual Execution

Manually trigger jobs on-demand

Execution History

Track job execution history and status

List Jobs

Retrieve all jobs with optional filtering:

# 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:

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:

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:

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:

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 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

For webhook jobs, save the webhook_url and webhook_secret - the secret is only shown once at creation time.

Update Job

Update an existing job (partial update):

# 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

Updating cron_schedule will recreate the Temporal Schedule, which may briefly pause execution.

Delete Job

Delete a job and its Temporal Schedule:

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:

# 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:

# 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

For cron jobs, enabling creates the Temporal Schedule if it doesn’t exist, and disabling pauses it.

Get Execution History

Retrieve execution history for a job:

# 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:

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

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

Complete Example

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

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:

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:

# 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"

Use crontab.guru to validate and understand cron expressions.

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/	Get job by ID
`create()`	POST /jobs	Create new job
`update()`	PATCH /jobs/	Update job
`delete()`	DELETE /jobs/	Delete job
`trigger()`	POST /jobs//trigger	Manually trigger
`enable()`	POST /jobs//enable	Enable job
`disable()`	POST /jobs//disable	Disable job
`get_executions()`	GET /jobs//executions	Get execution history
`trigger_webhook()`	POST /webhooks/	Trigger via webhook

Next Steps

Agents Service

Create agents for job execution

Teams Service

Use teams in jobs

Workers Service

Configure worker queues

Environments Service

Manage execution environments

Getting Started

Control Plane API

Control Plane Client

Context Graph & Memory

Client SDK

Workflow DSL

Advanced

Resources

​Quick Start

​Features

Cron Scheduling

Webhook Triggers

Manual Execution

Execution History

​List Jobs

​Get Job

​Create Job

​Cron Job

​Webhook Job

​Manual Job

​Update Job

​Delete Job

​Job Execution

​Manual Trigger

​Enable/Disable Job

​Get Execution History

​Webhook Triggers

​Trigger via Webhook

​Complete Example

​Error Handling

​Cron Schedule Examples

​Best Practices

​Job Design

​Scheduling

​Webhook Security

​Monitoring

​API Reference

​Next Steps

Agents Service

Teams Service

Workers Service

Environments Service

Quick Start

Features

List Jobs

Get Job

Create Job

Cron Job

Webhook Job

Manual Job

Update Job

Delete Job

Job Execution

Manual Trigger

Enable/Disable Job

Get Execution History

Webhook Triggers

Trigger via Webhook

Complete Example

Error Handling

Cron Schedule Examples

Best Practices

Job Design

Scheduling

Webhook Security

Monitoring

API Reference

Next Steps