Of course! Using the Ansible API with Python is a powerful way to automate your infrastructure directly within your Python applications, scripts, or services. It gives you programmatic control over Ansible's capabilities without needing to run ansible-playbook from the command line.

Here's a comprehensive guide covering the main ways to use the Ansible API with Python, from the simplest to the most advanced.

Core Concepts

Before diving into the code, it's essential to understand two key components:

1. Ansible Inventory: This is the source of truth for your managed nodes. It's a file (or a data structure in Python) that lists the hosts you want to manage and the groups they belong to. It can also contain variables for those hosts and groups.
2. Ansible Playbook: This is a YAML file that defines a "play." A play maps a group of hosts to a set of tasks. Tasks are the individual actions you want to run (e.g., install a package, copy a file, run a command).

The Python API allows you to programmatically create these components in memory and execute them.

Method 1: The High-Level AnsibleRunner (Recommended for Beginners)

This is the simplest and most direct way to run an existing playbook. It's essentially a Python wrapper around the ansible-playbook command-line tool. It's perfect for when you have a playbook ready and just want to trigger it from a script.

Installation: You'll need the ansible-runner package.

pip install ansible-runner

Python Script (run_playbook.py):

import ansible_runner
# Define the path to your playbook
playbook_path = '/path/to/your/playbook.yml'
# Run the playbook
# The 'private_data_dir' is where ansible-runner will store logs, artifacts, etc.
# It's good practice to create a temporary directory for this.
r = ansible_runner.run(
    private_data_dir='/tmp/ansible_runner',
    playbook=playbook_path,
    inventory='/path/to/your/inventory.ini', # Optional: if not in playbook
    extravars={'some_variable': 'value'}    # Optional: pass variables
)
# Check the result
print("Status: %s" % r.status)
print("RC: %s" % r.rc)
# Print the stdout from the hosts
for event in r.events:
    if event['event'] == 'runner_on_ok':
        print(f"Host: {event['event_data']['host']}, stdout: {event['event_data']['stdout']}")
# You can also access the full JSON output
# print(r.stdout)

How it works:

• ansible_runner.run() executes the playbook.
• It returns a Runner object that you can query for status, return codes, and detailed events.
• This method is great for its simplicity but doesn't give you fine-grained control over the Ansible execution loop.

Method 2: The Low-Level AnsibleAPI (Full Control)

This method gives you direct access to Ansible's core execution engine. You can build and run plays entirely in Python without needing any pre-written YAML files. This is more complex but offers the most flexibility.

Installation: You need the ansible package itself.

pip install ansible

Example 1: Running a Single Task

This is the most basic use case: running one command on a set of hosts.

from ansible.parsing.dataloader import DataLoader
from ansible.inventory.manager import InventoryManager
from ansible.vars.manager import VariableManager
from ansible.playbook.play import Play
from ansible.executor.task_queue_manager import TaskQueueManager
from ansible.plugins.callback import CallbackBase
import ansible.constants as C
# --- 1. SETUP CALLBACK (Optional but useful for output) ---
class ResultCallback(CallbackBase):
    def v2_runner_on_ok(self, result, **kwargs):
        host = result._host
        print(f"SUCCESS [{host.name}]: {result._result}")
    def v2_runner_on_failed(self, result, **kwargs):
        host = result._host
        print(f"FAILED [{host.name}]: {result._result}")
# --- 2. INITIALIZE ANSIBLE COMPONENTS ---
# DataLoader loads data sources like inventory and vars
loader = DataLoader()
# InventoryManager manages the inventory
inventory = InventoryManager(loader=loader, sources='localhost,') # Run on local machine
# VariableManager manages variables for hosts and groups
variable_manager = VariableManager(loader=loader, inventory=inventory)
# --- 3. DEFINE THE PLAY ---
# We'll create a simple play to ping a host
play_source = dict(
    name="Demo Play",
    hosts='localhost', # The hosts to target
    gather_facts='no', # Don't gather facts for this simple example
    tasks=[
        dict(action=dict(module='ping', ping_timeout=5))
    ]
)
# Create a Play object
play = Play().load(play_source, variable_manager=variable_manager, loader=loader)
# --- 4. EXECUTE THE PLAY ---
# Create a callback instance
callback = ResultCallback()
# Create a TaskQueueManager to execute the play
# Note: This is a synchronous call and will block until the play is finished
tqm = None
try:
    tqm = TaskQueueManager(
        inventory=inventory,
        variable_manager=variable_manager,
        loader=loader,
        passwords={}, # Not needed for this example
        stdout_callback=callback, # Use our custom callback
        # You can add other plugins here, e.g., for connection type
        # connection='local' 
    )
    result = tqm.run(play)
finally:
    if tqm:
        tqm.cleanup()
print("Play execution finished.")

Example 2: Running a More Complex Play with Variables

This example shows how to create a play that uses a variable and executes a more complex task.

# (Reuse the imports and ResultCallback from the previous example)
# --- 1. SETUP ---
loader = DataLoader()
inventory = InventoryManager(loader=loader, sources='localhost,')
variable_manager = VariableManager(loader=loader, inventory=inventory)
# --- 2. DEFINE THE PLAY WITH VARIABLES ---
# Define variables to be passed to the play
extra_vars = {'my_message': 'Hello from the Ansible API!'}
play_source = dict(
    name="Complex Demo Play",
    hosts='localhost',
    gather_facts='no',
    vars=extra_vars, # Pass variables here
    tasks=[
        dict(action=dict(module='debug', msg='{{ my_message }}')),
        dict(action=dict(module='file', path='/tmp/test_file.txt', state='touch'))
    ]
)
play = Play().load(play_source, variable_manager=variable_manager, loader=loader)
# --- 3. EXECUTE THE PLAY ---
callback = ResultCallback()
tqm = None
try:
    tqm = TaskQueueManager(
        inventory=inventory,
        variable_manager=variable_manager,
        loader=loader,
        passwords={},
        stdout_callback=callback,
        # connection='local' # Use 'local' to run on the control node
    )
    result = tqm.run(play)
finally:
    if tqm:
        tqm.cleanup()
print("Complex play execution finished.")

Method 3: Using AnsibleCollectionLoader (For Collections)

If you are using Ansible Collections extensively, you might need to load them directly. The AnsibleCollectionLoader helps with this.

from ansible.parsing.dataloader import DataLoader
from ansible.inventory.manager import InventoryManager
from ansible.vars.manager import VariableManager
from ansible.playbook.play import Play
from ansible.executor.task_queue_manager import TaskQueueManager
from ansible.plugins.callback import CallbackBase
from ansible_collections.my_namespace.my_collection.plugins.modules import my_module # Example
# The setup is similar to the low-level API, but you ensure your
# collection is in the ANSIBLE_COLLECTIONS_PATHS or in the expected location.
# The loader will automatically find modules from the collections.

Key Differences & When to Use What

Important Considerations

1. Concurrency: By default, Ansible runs tasks sequentially. To run tasks in parallel (async), you need to use the async: and poll: keywords within your playbook or task definition.
2. Error Handling: In the low-level API, you are responsible for checking the return status from the TaskQueueManager and handling exceptions. The ResultCallback is the standard way to capture per-task success/failure.
3. Idempotency: The core principle of Ansible is that running a playbook multiple times should have the same effect as running it once. Your Python code should respect this. Don't force a change unless necessary.
4. Credentials: For managing remote hosts, you'll need to handle SSH credentials (passwords, private keys) securely. The passwords dictionary in TaskQueueManager is one way, but environment variables or SSH agents are often better practices. Using the connection: local plugin is great for controlling the machine running the Python script.