Ansible Has a Free API: Here's How to Use It for Infrastructure Automation

#ansible #automation #devops #infrastructure

Ansible doesn't have a traditional REST API, but AWX (the open-source version of Ansible Tower) provides a complete API for managing inventories, running playbooks, and orchestrating infrastructure at scale.

Why Use AWX/Ansible API?

Trigger playbooks from your CI/CD pipeline
Schedule automated maintenance tasks
Track job execution history and audit logs
Manage inventories and credentials programmatically

Getting Started with AWX API

export AWX_URL="http://localhost:8052/api/v2"
export AWX_TOKEN="your-oauth-token"

# List all job templates
curl -s -H "Authorization: Bearer $AWX_TOKEN" \
  "$AWX_URL/job_templates/" | jq '.results[] | {id: .id, name: .name, project: .summary_fields.project.name}'

# Launch a job template
curl -s -H "Authorization: Bearer $AWX_TOKEN" \
  -X POST "$AWX_URL/job_templates/1/launch/" \
  -H "Content-Type: application/json" \
  -d '{"extra_vars": {"target_env": "production", "version": "2.1.0"}}' | jq '{job: .job, status: .status}'

Python Client

import requests
import time

class AWXClient:
    def __init__(self, url, token):
        self.url = f"{url}/api/v2"
        self.headers = {'Authorization': f'Bearer {token}', 'Content-Type': 'application/json'}

    def list_templates(self):
        resp = requests.get(f"{self.url}/job_templates/", headers=self.headers)
        return resp.json()['results']

    def launch_job(self, template_id, extra_vars=None):
        payload = {}
        if extra_vars:
            payload['extra_vars'] = extra_vars
        resp = requests.post(f"{self.url}/job_templates/{template_id}/launch/", json=payload, headers=self.headers)
        return resp.json()

    def get_job_status(self, job_id):
        resp = requests.get(f"{self.url}/jobs/{job_id}/", headers=self.headers)
        return resp.json()

    def wait_for_job(self, job_id, timeout=300):
        start = time.time()
        while time.time() - start < timeout:
            job = self.get_job_status(job_id)
            status = job['status']
            if status in ('successful', 'failed', 'error', 'canceled'):
                return job
            time.sleep(5)
        raise TimeoutError(f"Job {job_id} did not complete within {timeout}s")

    def get_job_output(self, job_id):
        resp = requests.get(f"{self.url}/jobs/{job_id}/stdout/?format=txt", headers=self.headers)
        return resp.text

# Usage
awx = AWXClient('http://localhost:8052', 'your-token')

# List available playbooks
for template in awx.list_templates():
    print(f"{template['id']:4d} | {template['name']}")

# Deploy to production
job = awx.launch_job(1, extra_vars={'version': '2.1.0', 'region': 'us-east-1'})
print(f"Job {job['job']} launched!")

# Wait for completion
result = awx.wait_for_job(job['job'])
print(f"Status: {result['status']}")
print(awx.get_job_output(job['job']))

Ansible Runner (No AWX Needed)

import ansible_runner

# Run a playbook directly from Python
result = ansible_runner.run(
    private_data_dir='/path/to/project',
    playbook='deploy.yml',
    extravars={'target_env': 'staging', 'app_version': '2.0.0'}
)

print(f"Status: {result.status}")
print(f"RC: {result.rc}")

# Get host results
for host, events in result.stats['processed'].items():
    print(f"  {host}: ok={events}")

# Get event log
for event in result.events:
    if event['event'] == 'runner_on_ok':
        task = event['event_data'].get('task', 'unknown')
        host = event['event_data'].get('host', 'unknown')
        print(f"  OK: {host} -> {task}")

CI/CD Integration

def deploy_with_ansible(awx, environment, version, dry_run=False):
    template_map = {
        'staging': 10,
        'production': 11,
        'canary': 12
    }

    template_id = template_map.get(environment)
    if not template_id:
        raise ValueError(f"Unknown environment: {environment}")

    extra_vars = {
        'app_version': version,
        'dry_run': dry_run,
        'deployer': 'ci-pipeline'
    }

    job = awx.launch_job(template_id, extra_vars=extra_vars)
    job_id = job['job']
    print(f"Deployment started: job #{job_id}")

    result = awx.wait_for_job(job_id, timeout=600)

    if result['status'] == 'successful':
        print(f"Deployment successful! Version {version} deployed to {environment}")
        return True
    else:
        output = awx.get_job_output(job_id)
        print(f"Deployment FAILED!\n{output[-500:]}")
        return False

# Blue-green deployment
def blue_green_deploy(awx, version):
    print("Step 1: Deploy to canary...")
    if not deploy_with_ansible(awx, 'canary', version):
        return False

    print("Step 2: Run smoke tests...")
    time.sleep(30)  # Wait for canary to stabilize

    print("Step 3: Deploy to production...")
    return deploy_with_ansible(awx, 'production', version)

Inventory Management

# List inventories
curl -s -H "Authorization: Bearer $AWX_TOKEN" \
  "$AWX_URL/inventories/" | jq '.results[] | {id: .id, name: .name, hosts: .total_hosts}'

# Add a host to inventory
curl -s -H "Authorization: Bearer $AWX_TOKEN" \
  -X POST "$AWX_URL/inventories/1/hosts/" \
  -H "Content-Type: application/json" \
  -d '{"name": "web-server-04", "variables": "{\"ansible_host\": \"10.0.1.4\", \"role\": \"webserver\"}"}'

Real-World Use Case

A DevOps team managed 2,000+ servers across 3 data centers. They built a self-service portal using AWX API — developers select their deployment target, version, and configuration, and the system runs the appropriate Ansible playbook. Deployment time dropped from 2 hours (manual SSH) to 8 minutes (automated), and human errors dropped to zero.