Jobs Service

The Jobs Service provides comprehensive management of scheduled and webhook-triggered jobs through the Kubiya platform. Jobs allow you to automate recurring tasks, set up cron schedules, and trigger executions via webhooks.

Jobs: detailed guide

This page focuses on the job-specific parts of the SDK: creation, scheduling, triggers, and execution management. If you haven’t set up a ControlPlaneClient yet, see the Client Quick Start at Client Overview for initialization and authentication instructions.

Create a scheduled job example

The following example demonstrates creating a cron-scheduled job using the Control Plane SDK:

from kubiya.control_plane_client import ControlPlaneClient

client = ControlPlaneClient()

job_data = {
    "name": "Daily Infrastructure Check",
    "enabled": True,
    "trigger_type": "cron",
    "cron_schedule": "0 9 * * *",  # Every day at 9 AM
    "cron_timezone": "UTC",
    "planning_mode": "predefined_agent",
    "entity_type": "agent",
    "entity_id": "12b68fe6-1c5d-4f96-8770-b65b50feb277",
    "prompt_template": "Check the health of all infrastructure components and report any issues.",
    "executor_type": "specific_queue",
    "worker_queue_name": "e54ecc24-a818-40b6-b554-51005319b293"
}

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

Core Operations

List Jobs

Retrieve a list of jobs with optional filtering:

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

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

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

for job in jobs:
    print(f"Job: {job['name']} - Trigger: {job.get('trigger_type')} - Enabled: {job.get('enabled')}")

Parameters:

Parameter	Type	Default	Description
`enabled`	bool	None	Filter by enabled status
`trigger_type`	str	None	Filter by trigger type: `cron`, `webhook`, `manual`

Get Job Details

Fetch a single job’s details by ID:

job = client.jobs.get("<job_uuid>")
print(job)

Create Job

Create a new job with various trigger configurations:

# Cron-triggered job
cron_job = client.jobs.create({
    "name": "Weekly Report",
    "enabled": True,
    "trigger_type": "cron",
    "cron_schedule": "0 8 * * MON",  # Every Monday at 8 AM
    "cron_timezone": "America/New_York",
    "planning_mode": "predefined_agent",
    "entity_type": "agent",
    "entity_id": "agent-uuid",
    "prompt_template": "Generate the weekly performance report for {{department}}.",
    "executor_type": "specific_queue",
    "worker_queue_name": "worker-queue-uuid"
})

# Webhook-triggered job
webhook_job = client.jobs.create({
    "name": "Deployment Notifier",
    "enabled": True,
    "trigger_type": "webhook",
    "planning_mode": "predefined_team",
    "entity_type": "team",
    "entity_id": "team-uuid",
    "prompt_template": "Process deployment notification for {{service}} version {{version}}.",
    "executor_type": "specific_queue",
    "worker_queue_name": "worker-queue-uuid"
})
print(f"Webhook URL: {webhook_job['webhook_url']}")

# Manual-triggered job
manual_job = client.jobs.create({
    "name": "On-Demand Analysis",
    "enabled": True,
    "trigger_type": "manual",
    "planning_mode": "on_the_fly",
    "prompt_template": "Analyze {{target}} and provide recommendations.",
    "executor_type": "specific_queue",
    "worker_queue_name": "worker-queue-uuid"
})

Job configuration fields:

Field	Type	Required	Description
`name`	str	Yes	Job name
`enabled`	bool	Yes	Whether job is enabled
`trigger_type`	str	Yes	`cron`, `webhook`, or `manual`
`cron_schedule`	str	Cron only	Cron expression (e.g., `0 9 * * *`)
`cron_timezone`	str	No	Timezone (default: `UTC`)
`planning_mode`	str	Yes	`on_the_fly`, `predefined_agent`, `predefined_team`, `predefined_workflow`
`entity_type`	str	Predefined	`agent` or `team`
`entity_id`	str	Predefined	Entity UUID
`prompt_template`	str	Yes	Prompt template with `{{variable}}` placeholders
`executor_type`	str	Yes	`auto`, `specific_queue`, or `environment`
`worker_queue_name`	str	specific_queue	Worker queue UUID (required when `executor_type` is `specific_queue`)
`environment_name`	str	No	Environment name for execution
`system_prompt`	str	No	System prompt override
`config`	dict	No	Additional configuration
`description`	str	No	Job description

Update Job

Update an existing job’s configuration (partial update):

update_data = {
    "description": "Updated job description",
    "cron_schedule": "0 10 * * *",  # Changed to 10 AM
    "enabled": True
}

updated = client.jobs.update("<job_uuid>", update_data)
print(f"Updated job: {updated['name']}")

Updating cron_schedule will recreate the Temporal Schedule.

Delete Job

Permanently remove a job and its Temporal Schedule:

client.jobs.delete("<job_uuid>")

Job Control

Enable Job

Enable a job and unpause its Temporal Schedule:

result = client.jobs.enable("<job_uuid>")
print(f"Job enabled: {result['enabled']}")

For cron jobs, this will create the Temporal Schedule if it doesn’t exist.

Disable Job

Disable a job and pause its Temporal Schedule:

result = client.jobs.disable("<job_uuid>")
print(f"Job disabled: {result['enabled']}")

Triggering Jobs

Manual Trigger

Manually trigger a job execution with optional parameters:

result = client.jobs.trigger(
    "<job_uuid>",
    parameters={"department": "engineering", "date": "2024-01-15"},
    config_override={"timeout": 300}
)

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

Trigger parameters:

Parameter	Type	Description
`parameters`	dict	Variables to substitute in prompt template
`config_override`	dict	Configuration overrides for this execution

Webhook Trigger

Trigger a job via webhook (requires HMAC signature authentication):

import hmac
import hashlib
import json

webhook_path = "unique-webhook-path"
webhook_secret = "your-webhook-secret"

payload = {
    "parameters": {"service": "api-gateway", "version": "2.1.0"},
    "metadata": {"triggered_by": "ci-pipeline"}
}

# Generate HMAC signature
payload_bytes = json.dumps(payload).encode()
signature = hmac.new(
    webhook_secret.encode(),
    payload_bytes,
    hashlib.sha256
).hexdigest()

result = client.jobs.trigger_webhook(
    webhook_path=webhook_path,
    payload=payload,
    signature=signature
)
print(f"Webhook triggered: {result}")

Execution History

Get Executions

Retrieve execution history for a job:

history = client.jobs.get_executions(
    "<job_uuid>",
    limit=50,
    offset=0
)

print(f"Total executions: {history['total_count']}")
for execution in history['executions']:
    print(f"  {execution['execution_id']}: {execution['status']} - {execution.get('duration_ms', 'N/A')}ms")

Execution history fields:

Field	Description
`execution_id`	Execution UUID
`trigger_type`	Type of trigger (cron, webhook, manual)
`status`	Execution status
`started_at`	Start timestamp
`completed_at`	Completion timestamp
`duration_ms`	Duration in milliseconds
`error_message`	Error message if failed
`trigger_metadata`	Trigger metadata

Error Handling

The Jobs Service raises JobError for API errors:

from kubiya.resources.exceptions import JobError

try:
    job = client.jobs.get("invalid-job-id")
except JobError as e:
    print(f"Job operation failed: {e}")

Best Practices

Use descriptive job names that reflect the purpose
Set appropriate timezones for cron schedules
Use prompt templates with placeholders for flexible executions
Store webhook secrets securely and rotate them periodically
Monitor execution history for failed jobs
Disable jobs instead of deleting them when temporarily not needed
Use config_override for one-off execution variations

Common Patterns

Create and Monitor a Scheduled Job

# Create the job
job = client.jobs.create({
    "name": "Hourly Metrics Collection",
    "enabled": True,
    "trigger_type": "cron",
    "cron_schedule": "0 * * * *",  # Every hour
    "cron_timezone": "UTC",
    "planning_mode": "predefined_agent",
    "entity_type": "agent",
    "entity_id": "metrics-agent-uuid",
    "prompt_template": "Collect and aggregate metrics from all services.",
    "executor_type": "specific_queue",
    "worker_queue_name": "worker-queue-uuid"
})

print(f"Job created: {job['id']}")
print(f"Next execution at: {job['next_execution_at']}")

# Check execution history after some time
history = client.jobs.get_executions(job["id"], limit=5)
print(f"Total executions: {job['total_executions']}")
print(f"Successful: {job['successful_executions']}")
print(f"Failed: {job['failed_executions']}")

Set Up Webhook Integration

# Create webhook job
job = client.jobs.create({
    "name": "GitHub Webhook Handler",
    "enabled": True,
    "trigger_type": "webhook",
    "planning_mode": "predefined_agent",
    "entity_type": "agent",
    "entity_id": "github-agent-uuid",
    "prompt_template": "Process GitHub event: {{event_type}} for repository {{repo}}.",
    "executor_type": "specific_queue",
    "worker_queue_name": "worker-queue-uuid"
})

# The webhook_url and webhook_secret in the response can be configured in GitHub
print(f"Configure this webhook URL in GitHub: {job['webhook_url']}")
print(f"Use this secret for signature verification: {job['webhook_secret']}")

Getting Started

Control Plane API

Control Plane Client

Context Graph & Memory

Client SDK

Workflow DSL

Advanced

Resources

Jobs: detailed guide

Create a scheduled job example

Core Operations

List Jobs

Get Job Details

Create Job

Update Job

Delete Job

Job Control

Enable Job

Disable Job

Triggering Jobs

Manual Trigger

Webhook Trigger

Execution History

Get Executions

Error Handling

Best Practices

Common Patterns

Create and Monitor a Scheduled Job

Set Up Webhook Integration

Getting Started

Control Plane API

Control Plane Client

Context Graph & Memory

Client SDK

Workflow DSL

Advanced

Resources

​Jobs: detailed guide

​Create a scheduled job example

​Core Operations

​List Jobs

​Get Job Details

​Create Job

​Update Job

​Delete Job

​Job Control

​Enable Job

​Disable Job

​Triggering Jobs

​Manual Trigger

​Webhook Trigger

​Execution History

​Get Executions

​Error Handling

​Best Practices

​Common Patterns

​Create and Monitor a Scheduled Job

​Set Up Webhook Integration

Jobs: detailed guide

Create a scheduled job example

Core Operations

List Jobs

Get Job Details

Create Job

Update Job

Delete Job

Job Control

Enable Job

Disable Job

Triggering Jobs

Manual Trigger

Webhook Trigger

Execution History

Get Executions

Error Handling

Best Practices

Common Patterns

Create and Monitor a Scheduled Job

Set Up Webhook Integration