Consistent Ansible Python API and CLI with container and process isolation runtime capabilities
npx @tessl/cli install tessl/pypi-ansible-runner@2.4.0Ansible Runner is a comprehensive tool and Python library that provides a stable, consistent interface abstraction to Ansible automation platform. It functions as a standalone command-line tool, container image interface, and importable Python module, designed to facilitate seamless integration with Ansible in various deployment scenarios. The library offers process and container isolation runtime capabilities, supports programmatic execution of Ansible playbooks and ad-hoc commands, provides comprehensive event streaming and callback mechanisms for monitoring execution progress, and includes advanced features like artifact collection, job isolation, and result processing.
pip install ansible-runnerimport ansible_runnerCommon imports for specific functionality:
from ansible_runner import run, run_async, run_command
from ansible_runner import Runner, RunnerConfig
from ansible_runner import AnsibleRunnerException, ConfigurationErrorimport ansible_runner
# Simple playbook execution
result = ansible_runner.run(
private_data_dir='/tmp/demo',
playbook='playbook.yml',
inventory='inventory',
extravars={'key': 'value'}
)
print(f"Status: {result.status}")
print(f"Return code: {result.rc}")
# Process events
for event in result.events:
if event['event'] == 'playbook_on_play_start':
print(f"Starting play: {event['event_data']['play']}")
# Simple command execution
result = ansible_runner.run_command(
executable_cmd='ansible-playbook',
cmdline_args=['playbook.yml', '-i', 'inventory'],
private_data_dir='/tmp/demo'
)
# Async execution with callbacks
def status_handler(data, runner_config):
print(f"Status: {data['status']}")
def event_handler(data):
print(f"Event: {data['event']}")
return True # Continue processing
result = ansible_runner.run_async(
private_data_dir='/tmp/demo',
playbook='playbook.yml',
status_handler=status_handler,
event_handler=event_handler
)
# Wait for completion
result.wait()Ansible Runner provides multiple layers of abstraction:
run, run_async, run_command) for common use casesThis architecture enables flexible deployment patterns from simple script integration to complex orchestration platforms with distributed execution capabilities.
Execute Ansible playbooks with comprehensive configuration options, event handling, and artifact collection. Supports both synchronous and asynchronous execution modes with full callback support.
def run(
private_data_dir: str = None,
playbook: str = None,
inventory: str = None,
**kwargs
) -> Runner: ...
def run_async(
private_data_dir: str = None,
playbook: str = None,
inventory: str = None,
**kwargs
) -> Runner: ...Execute Ansible command-line tools (ansible-playbook, ansible, ansible-inventory, etc.) programmatically with full argument support and output capture.
def run_command(
executable_cmd: str,
cmdline_args: list = None,
**kwargs
) -> Runner: ...
def run_command_async(
executable_cmd: str,
cmdline_args: list = None,
**kwargs
) -> Runner: ...Discover, document, and manage Ansible plugins and roles programmatically. Provides access to plugin documentation, role specifications, and inventory operations.
def get_plugin_docs(
plugin_names: list,
plugin_type: str = None,
response_format: str = None,
**kwargs
) -> Runner: ...
def get_role_list(
collection: str = None,
playbook_dir: str = None,
**kwargs
) -> Runner: ...
def get_role_argspec(
role: str,
collection: str = None,
**kwargs
) -> Runner: ...Core classes for advanced use cases requiring fine-grained control over execution parameters, event handling, and process management.
class Runner:
def __init__(
self,
config,
cancel_callback=None,
event_handler=None,
**kwargs
): ...
class RunnerConfig:
def __init__(self, **kwargs): ...Configuration and Runner Classes
Components for distributed execution across multiple processes or containers with real-time event streaming and coordination.
class Transmitter:
def __init__(self, private_data_dir: str, **kwargs): ...
class Worker:
def __init__(self, private_data_dir: str, **kwargs): ...
class Processor:
def __init__(self, private_data_dir: str, **kwargs): ...Streaming and Distributed Execution
Helper functions for artifact management, cleanup operations, output handling, and system integration.
def cleanup_folder(folder: str) -> bool: ...
def register_for_cleanup(folder: str) -> None: ...
class Bunch:
def __init__(self, **kwargs): ...class AnsibleRunnerException(Exception):
"""Generic Runner Error"""
class ConfigurationError(AnsibleRunnerException):
"""Misconfiguration of Runner"""
class CallbackError(AnsibleRunnerException):
"""Exception occurred in Callback"""