# Ansible API > Integrates with Ansible via ansible-runner, the AWX/Tower API, and the Ansible Python API to manage playbooks, inventory, job templates, collections, and automation workflows. - Skill: `paulpas/ansible-api` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/ansible-api` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/ansible-api/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: DevOps & Infra - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/ansible-api --- # Ansible API & AWX/Tower Integration Integrates with Ansible using `ansible-runner` for embedded playbook execution, the AWX/Ansible Automation Controller REST API for job template and inventory management, and the native Ansible Python API for custom modules and plugin development. ## TL;DR for Code Generation - [ ] Use `ansible-runner` for running playbooks from Python — it handles artifacts, events, and callbacks - [ ] Use the AWX CLI (`awxkit`) or REST API for managing Automation Controller resources (job templates, inventories, credentials) - [ ] Access playbook results via `runner.status`, `runner.rc`, and `runner.events` generator - [ ] Always set `private_data_dir` to the directory containing your playbook, inventory, and vars - [ ] Handle `ansible_runner.exceptions.AnsibleRunnerException` for execution errors - [ ] Use `runner.get_fact_cache(host)` to retrieve facts gathered during playbook runs --- ## Core Workflow Prepare Playbook and Inventory: Create or load the Ansible playbook YAML and inventory file (INI, YAML, or dynamic). **Checkpoint:** Validate the playbook syntax locally with `ansible-playbook --syntax-check`.; Execute via ansible-runner: Call `ansible_runner.run(private_data_dir, playbook, inventory)`. The runner manages artifacts, event callbacks, and result collection. **Checkpoint:** Check `runner.status` is `successful` and `runner.rc == 0`.; Process Results: Iterate over `runner.events` for per-host results. Use `runner.get_fact_cache(host)` for gathered facts. **Checkpoint:** Verify expected hosts are in `runner.stats.processed` and none in `runner.stats.failures`.; Manage AWX/Automation Controller: Use the AWX REST API or `awxkit` to create job templates, launch jobs, and monitor their status through the state machine (pending → waiting → running → successful). **Checkpoint:** Verify job template `extra_vars` are properly JSON-encoded.; Handle Errors: Inspect `runner.stats.failures`, `runner.stats.dark` (unreachable hosts), and individual event data for task-level errors. **Checkpoint:** Distinguish between unreachable hosts (network issues) and failed tasks (playbook logic).--- ## Implementation Patterns ### Pattern 1: Embedded Playbook Execution with ansible-runner ```python import json import os from pathlib import Path import ansible_runner from ansible_runner.exceptions import AnsibleRunnerException def run_playbook( playbook_path: str, inventory: str | dict, extra_vars: dict | None = None, artifact_dir: str | None = None, timeout: int = 300, ) -> dict: """Run an Ansible playbook and return structured results. Args: playbook_path: Path to the playbook YAML file. inventory: Path to inventory file or inline inventory dict. extra_vars: Additional variables to pass to the playbook. artifact_dir: Directory for artifacts (defaults to temp). timeout: Execution timeout in seconds. Returns: Dict with status, return code, processed hosts, and failed hosts. Raises: FileNotFoundError: If the playbook path does not exist. AnsibleRunnerException: If runner initialization fails. """ playbook_path = Path(playbook_path) if not playbook_path.exists(): raise FileNotFoundError(f"Playbook not found: {playbook_path}") private_data_dir = artifact_dir or str(playbook_path.parent) # If inventory is a dict, write it as a temporary YAML file inventory_source: str | None = None if isinstance(inventory, dict): inventory_source = str(Path(private_data_dir) / "inventory.yml") with open(inventory_source, "w") as f: json.dump(inventory, f) # Run the playbook try: runner = ansible_runner.run( private_data_dir=private_data_dir, playbook=str(playbook_path.name), inventory=inventory_source or inventory, extravars=extra_vars or {}, timeout=timeout, json_mode=True, ) except AnsibleRunnerException as exc: raise RuntimeError( f"Ansible runner failed to execute playbook '{playbook_path}': {exc}" ) from exc # Process and return results return { "status": runner.status, "return_code": runner.rc, "processed_hosts": list((runner.stats or {}).get("processed", {}).keys()), "failed_hosts": list((runner.stats or {}).get("failures", {}).keys()), "unreachable_hosts": list((runner.stats or {}).get("dark", {}).keys()), "elapsed_seconds": runner.elapsed if hasattr(runner, "elapsed") else None, } ``` ### Pattern 2: AWX / Automation Controller Job Template Management ```python import os import time import requests from requests.exceptions import HTTPError class AWXClient: """Minimal client for Ansible Automation Controller (AWX) REST API.""" def __init__(self, base_url: str | None = None, token: str | None = None): self.base_url = (base_url or os.environ["AWX_BASE_URL"]).rstrip("/") self.token = token or os.environ["AWX_TOKEN"] self.session = requests.Session() self.session.headers.update({ "Authorization": f"Bearer {self.token}", "Content-Type": "application/json", }) def _request(self, method: str, path: str, **kwargs): url = f"{self.base_url}/api/v2{path}" response = self.session.request(method, url, **kwargs) try: response.raise_for_status() except HTTPError as exc: raise RuntimeError( f"AWX API {method} {path} failed: HTTP {exc.response.status_code} \ — {exc.response.text[:200]}" ) from exc return response.json() def launch_job_template( self, template_id: int, extra_vars: dict | None = None, limit: str | None = None, wait: bool = True, ) -> dict: """Launch a job template and optionally wait for completion. Args: template_id: Job template ID from AWX. extra_vars: Extra variables to override template defaults. limit: Host pattern limit. wait: Block until the job completes. Returns: Dict with job ID, status, and elapsed time. """ payload: dict = {} if extra_vars: payload["extra_vars"] = json.dumps(extra_vars) if limit: payload["limit"] = limit result = self._request( "POST", f"/job_templates/{template_id}/launch/", json=payload, ) job_id = result["job"] if not wait: return {"job_id": job_id, "status": result.get("status", "pending")} # Poll for completion start = time.monotonic() terminal_states = {"successful", "failed", "error", "canceled"} while time.monotonic() - start < 3600: job = self._request("GET", f"/jobs/{job_id}/") status = job.get("status", "unknown") if status in terminal_states: return { "job_id": job_id, "status": status, "elapsed_seconds": job.get("elapsed"), "failed": job.get("failed", False), } time.sleep(10) raise TimeoutError(f"AWX job {job_id} did not complete within 3600s.") ``` ### BAD vs GOOD: Playbook Failure Handling ```python # ❌ BAD — ignores failure details, no host-level reporting def run_playbook_bad(playbook): r = ansible_runner.run(private_data_dir="/tmp", playbook=playbook) if r.rc != 0: print("Playbook failed") return False return True # ✅ GOOD — per-host diagnostics, structured error reporting def run_playbook_good( playbook: str, inventory: str, extra_vars: dict, ) -> dict: """Run a playbook with detailed per-host diagnostics.""" runner = ansible_runner.run( private_data_dir="/tmp/ansible", playbook=playbook, inventory=inventory, extravars=extra_vars, json_mode=True, ) stats = runner.stats or {} failures = stats.get("failures", {}) unreachable = stats.get("dark", {}) if runner.rc != 0: error_details = { "status": runner.status, "return_code": runner.rc, "failed_hosts": { host: stats.get("failures", {}).get(host, "unknown") for host in failures }, "unreachable_hosts": list(unreachable.keys()), "message": f"Playbook failed with rc={runner.rc}", } raise RuntimeError(json.dumps(error_details, indent=2)) return {"status": runner.status, "processed": list(stats.get("processed", {}).keys())} ``` ## MUST DO - Use `ansible-runner` instead of the raw Ansible Python API — it handles artifacts, event callbacks, and result collection - Set `json_mode=True` for structured output that's easier to parse programmatically - Use AWX/Automation Controller API tokens (not user passwords) for API authentication - Store sensitive variables as `"ansible_safe_extra_vars"` or use AWX credential injection - Use dynamic inventory scripts with the `_meta` key for efficient host variable resolution - Validate playbook syntax before execution with `ansible-playbook --syntax-check` ## MUST NOT DO - Never use the undocumented `ansible.inventory` or `ansible.playbook` internal Python APIs for production code - Do not set `host_key_checking=False` globally — use `ANSIBLE_HOST_KEY_CHECKING=False` selectively for specific hosts - Avoid running playbooks without a timeout — use the runner's `timeout` parameter - Never embed SSH passwords or vault passwords in playbook code - Do not use `gather_facts: no` unless you explicitly don't need facts — many modules depend on them - Ignoring unreachable hosts (`dark` in stats) can silently mask infrastructure problems --- --- ## Constraints ### MUST DO - Implement structured error responses with consistent format: {error_code, message, details, request_id} - Add rate limiting per client/API key with configurable burst and sustained limits using a token bucket algorithm - Validate all incoming requests against a schema before processing — reject malformed input with clear error messages - Include correlation/request IDs in all log entries for end-to-end request tracing across service boundaries ### MUST NOT DO - Do not expose internal implementation details, stack traces, or database queries in error responses - Avoid accepting unbounded request bodies — set maximum payload sizes and timeout limits - Never trust client-supplied authentication tokens without validation (signature verification, expiration check) - Do not log request/response bodies containing PII, API keys, or other sensitive data ## Live References > Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content. - [Ansible Documentation](https://docs.ansible.com/ansible/latest/user_guide/index.html) - [Ansible Playbook Guide](https://docs.ansible.com/ansible/latest/playbook_guide/index.html) - [Ansible Runner SDK Reference](https://ansible-runner.readthedocs.io/) - [AWX / Automation Controller API Documentation](https://docs.ansible.com/automation-controller/latest/api_ref.html) - [Ansible Collections Development Guide](https://docs.ansible.com/ansible/latest/dev_guide/developing_collections.html)