AnsibleByExample

Ansible SDK: Run Ansible Programmatically from Python

AdvancedAnsible SDKIT automationPython libraryautomation tasks
2024-01-015By Luca Berton

Lesson 1 of 15 in Foundations

Ansible Foundations to Production — full curriculum
  1. 1. Introduction to Ansible and Automation
  2. 2. Installing and Configuring Ansible
  3. 3. Running Ad-Hoc Commands
  4. 4. Writing Your First Playbook
  5. 5. Understanding and Using Modules
  6. 6. Variables and Facts
  7. 7. Conditionals and Loops
  8. 8. Handlers and Idempotency
  9. 9. Structuring Projects with Roles
  10. 10. Inventory Management
  11. 11. Using Ansible Vault for Secrets
  12. 12. Debugging and Troubleshooting
  13. 13. Real-World Automation Examples
  14. 14. Final Project: Deploy a Web Server
  15. 15. Bonus - What's Next with Ansible?
View path overview →

Introduction

While Ansible is typically run from the command line, many automation platforms and custom applications need to execute Ansible programmatically. The Ansible SDK is a lightweight Python library that provides a clean interface to dispatch, monitor, and manage Ansible jobs directly from Python code — no shell commands needed.

Whether you're building a self-service portal, integrating Ansible into a CI/CD pipeline, or creating a custom automation dashboard, the Ansible SDK provides the programmatic bridge.

What is Ansible SDK?

Ansible SDK is a Python library that wraps Ansible's execution engine, providing:

  • Programmatic execution of playbooks, roles, and tasks
  • Asynchronous job management with status monitoring
  • Local execution via Ansible Runner
  • Remote execution via Automation Mesh
  • Native Python data structures for inputs and outputs

Installation

pip install ansible-sdk

Dependencies include ansible-runner and ansible-core.

Architecture

Your Python App
      │
      ▼
  Ansible SDK
      │
      ├── Local Execution ──► Ansible Runner ──► Target Hosts
      │
      └── Remote Execution ──► Automation Mesh Controller ──► Mesh Nodes ──► Target Hosts

Local Execution

The SDK uses Ansible Runner to execute playbooks on the local machine. Runner handles:

  • Pulling execution environments (containers)
  • Running jobs
  • Collecting output and status
  • Reporting results back to the SDK

Remote Execution

For distributed environments, the SDK connects to an Automation Mesh controller node, which distributes work across a mesh of Receptor nodes.

Basic Usage

Run a Playbook

import asyncio
from ansible_sdk import AnsibleJobDef
from ansible_sdk.executors import AnsibleSubprocessJobExecutor

async def run_playbook():
    # Create executor
    executor = AnsibleSubprocessJobExecutor()
    
    # Define the job
    jobdef = AnsibleJobDef(
        data_dir='./my_project',        # Directory containing playbooks
        playbook='site.yml',             # Playbook filename
    )
    
    # Submit and wait for completion
    job_status = await executor.submit_job(jobdef)
    
    # Check results
    print(f"Status: {job_status.status}")
    print(f"RC: {job_status.rc}")
    
    return job_status

# Run
result = asyncio.run(run_playbook())

Project Directory Structure

The data_dir should follow Ansible Runner's project layout:

my_project/
├── project/
│   ├── site.yml          # Your playbook
│   ├── roles/
│   └── group_vars/
├── inventory/
│   └── hosts             # Inventory file
└── env/
    ├── extravars          # Extra variables (JSON)
    └── settings           # Runner settings

Pass Extra Variables

import json

# Write extra vars to the env directory
extravars = {
    'app_version': '2.1.0',
    'environment': 'production',
    'deploy_user': 'appuser',
}

with open('./my_project/env/extravars', 'w') as f:
    json.dump(extravars, f)

# Then run the job as normal
jobdef = AnsibleJobDef(
    data_dir='./my_project',
    playbook='deploy.yml',
)

Specify Inventory

# Write inventory to the inventory directory
inventory = """
[webservers]
web1.example.com
web2.example.com

[dbservers]
db1.example.com
"""

with open('./my_project/inventory/hosts', 'w') as f:
    f.write(inventory)

Advanced Usage

Monitor Job Progress

async def run_with_monitoring():
    executor = AnsibleSubprocessJobExecutor()
    jobdef = AnsibleJobDef(
        data_dir='./my_project',
        playbook='site.yml',
    )
    
    # Submit job
    job_status = await executor.submit_job(jobdef)
    
    # Access events as they occur
    async for event in job_status.events:
        if event.get('event') == 'runner_on_ok':
            host = event['event_data']['host']
            task = event['event_data']['task']
            print(f"✅ {host}: {task}")
        elif event.get('event') == 'runner_on_failed':
            host = event['event_data']['host']
            task = event['event_data']['task']
            print(f"❌ {host}: {task}")
    
    return job_status

Run Ad-Hoc Commands

async def run_adhoc():
    executor = AnsibleSubprocessJobExecutor()
    
    jobdef = AnsibleJobDef(
        data_dir='./my_project',
        module='ansible.builtin.ping',
        module_args='',
        host_pattern='all',
    )
    
    job_status = await executor.submit_job(jobdef)
    return job_status

Error Handling

async def run_with_error_handling():
    executor = AnsibleSubprocessJobExecutor()
    jobdef = AnsibleJobDef(
        data_dir='./my_project',
        playbook='site.yml',
    )
    
    try:
        job_status = await executor.submit_job(jobdef)
        
        if job_status.status == 'successful':
            print("Playbook completed successfully")
        elif job_status.status == 'failed':
            print(f"Playbook failed with rc={job_status.rc}")
            # Access stdout for error details
            for event in job_status.events:
                if event.get('event') == 'runner_on_failed':
                    print(f"Failed task: {event['event_data']}")
        
    except Exception as e:
        print(f"Execution error: {e}")

Integration Patterns

Flask Web Application

from flask import Flask, jsonify, request
import asyncio
from ansible_sdk import AnsibleJobDef
from ansible_sdk.executors import AnsibleSubprocessJobExecutor

app = Flask(__name__)

@app.route('/api/deploy', methods=['POST'])
def deploy():
    data = request.json
    version = data.get('version', 'latest')
    
    async def run_deploy():
        executor = AnsibleSubprocessJobExecutor()
        jobdef = AnsibleJobDef(
            data_dir='./ansible_project',
            playbook='deploy.yml',
        )
        return await executor.submit_job(jobdef)
    
    result = asyncio.run(run_deploy())
    
    return jsonify({
        'status': result.status,
        'rc': result.rc,
    })

CI/CD Pipeline Script

#!/usr/bin/env python3
"""CI/CD deployment script using Ansible SDK."""
import asyncio
import sys
from ansible_sdk import AnsibleJobDef
from ansible_sdk.executors import AnsibleSubprocessJobExecutor

async def deploy(environment: str, version: str) -> int:
    executor = AnsibleSubprocessJobExecutor()
    
    jobdef = AnsibleJobDef(
        data_dir=f'./ansible/{environment}',
        playbook='deploy.yml',
    )
    
    job_status = await executor.submit_job(jobdef)
    
    if job_status.status != 'successful':
        print(f"Deployment failed! RC: {job_status.rc}")
        return 1
    
    print(f"Deployed {version} to {environment} successfully")
    return 0

if __name__ == '__main__':
    env = sys.argv[1] if len(sys.argv) > 1 else 'staging'
    ver = sys.argv[2] if len(sys.argv) > 2 else 'latest'
    sys.exit(asyncio.run(deploy(env, ver)))

Ansible SDK vs Ansible Runner vs CLI

FeatureAnsible SDKAnsible Runneransible-playbook CLI
InterfacePython async APIPython APICommand line
Async supportNativeLimitedNo
Mesh supportYesNoNo
Event streamingYesYesVia callback
Best forApps, platformsScripts, CI/CDManual runs

When to Use Each

  • Ansible SDK: Building automation platforms, web applications, or any software that needs to run Ansible programmatically
  • Ansible Runner: Simple Python scripts that need to run playbooks without full SDK overhead
  • CLI: Manual execution, simple scripts, Cron jobs

Resources

Conclusion

The Ansible SDK bridges the gap between Ansible's powerful automation engine and Python application development. Instead of shelling out to ansible-playbook, you get a clean async Python API with job management, event streaming, and support for both local and distributed execution. For teams building automation platforms, self-service portals, or CI/CD integrations, the SDK provides the programmatic control that the CLI can't offer.

All Tutorials