核心思想是使用 Ansible 提供的 Python API 库，直接在 Python 代码中创建和执行 Ansible 的 "Playbook" 或 "Ad-hoc Command"。

我们将主要介绍官方推荐的 ansible-runner 库，因为它是最稳定、功能最全的集成方式，也会简要介绍更底层的 ansible 库。

使用 ansible-runner (推荐)

ansible-runner 是一个独立的工具和 Python 库，它提供了一个稳定、可靠的接口来运行 Ansible Playbook 和 Ad-hoc 命令，它负责处理执行环境、状态管理、日志记录和结果返回，是生产环境集成的首选。

安装

你需要安装 ansible-runner 和 ansible。

pip install ansible-runner
pip install ansible

基本概念

• Private Data Dir: 一个临时目录，ansible-runner 会在这里生成执行所需的文件，如 inventory, envvars, ssh_key 等，执行完成后，你可以选择保留或删除它。
• Playbook Path: 你的 Ansible Playbook 文件的路径。
• Inventory Path: 你的 Ansible Inventory 文件的路径。
• extravars: 一个字典，用于向 Playbook 传递额外的变量，类似于 ansible-playbook 的 --extra-vars 参数。

示例：执行一个简单的 Playbook

假设你有一个简单的 Playbook 文件 site.yml，用于在远程节点上创建一个文件。

site.yml

---
- name: Create a file using API
  hosts: my_servers
  gather_facts: no
  tasks:
    - name: Create a file with some content
      ansible.builtin.file:
        path: /tmp/api_created_file.txt
        state: touch
        mode: '0644'

以及一个 Inventory 文件 inventory.ini，定义你的目标主机。

inventory.ini

[my_servers]
server1.example.com
server2.example.com
[my_servers:vars]
ansible_user=your_ssh_user
ansible_ssh_private_key_file=/path/to/your/ssh_key

Python 脚本 (run_playbook.py)

import ansible_runner
# 定义 Playbook 和 Inventory 的路径
playbook_path = './site.yml'
inventory_path = './inventory.ini'
# 定义要传递给 Playbook 的额外变量
extra_vars = {
    'message': 'This file was created by the Python API!'
}
# 调用 ansible_runner 运行 Playbook
# private_data_dir 是一个临时工作目录
r = ansible_runner.run(
    private_data_dir='./runner',  # 工作目录
    playbook=playbook_path,
    inventory=inventory_path,
    extravars=extra_vars
)
# 检查执行状态
if r.status == 'successful':
    print("Playbook 执行成功!")
    print(f"返回码: {r.rc}")
    # 获取标准输出
    print("标准输出:")
    for event in r.events:
        if event['event'] == 'runner_on_ok':
            print(f"  - {event['event_data']['host']}: {event['event_data']['res']['stdout']}")
else:
    print(f"Playbook 执行失败! 状态: {r.status}")
    print(f"返回码: {r.rc}")
    # 打印错误信息
    print("错误信息:")
    print(r.stdout.read())
    print(r.stderr.read())
# private_data_dir 目录包含了执行的所有日志和结果
# 你可以在这里找到详细的执行记录
print(f"执行结果保存在: {r.private_data_dir}")

如何运行：

1. 将 site.yml, inventory.ini, 和 run_playbook.py 放在同一个目录下。
2. 确保你的 SSH 密钥可以无密码登录到 server1.example.com 和 server2.example.com。
3. 执行 python run_playbook.py。

执行后,你会看到 runner 目录被创建，里面包含了 env, inventory, artifacts 等子目录，artifacts 中有详细的日志。

使用底层的 ansible 库

这种方法更直接,让你可以像写 Python 代码一样构建 Ansible 的执行逻辑，但它更复杂，需要对 Ansible 内部有更多了解，且稳定性可能不如 ansible-runner。

安装

pip install ansible

示例：执行一个 Ad-hoc 命令

这个例子将直接在 Python 中构建一个 Ansible 的 "Play"，并执行它。

Python 脚本 (run_adhoc.py)

import ansible.inventory
import ansible.parsing.dataloader
import ansible.playbook
from ansible.vars.manager import VariableManager
from ansible.executor.task_queue_manager import TaskQueueManager
from ansible.plugins.callback import CallbackBase
# 自定义回调类，用于捕获执行结果
class ResultCallback(CallbackBase):
    def __init__(self, *args, **kwargs):
        super(ResultCallback, self).__init__(*args, **kwargs)
        self.host_ok = {}
        self.host_unreachable = {}
        self.host_failed = {}
    def v2_runner_on_ok(self, result, **kwargs):
        """当任务成功时调用"""
        self.host_ok[result._host.get_name()] = result
        print(f"成功: 主机 {result._host.get_name()}, 结果: {result._result}")
    def v2_runner_on_unreachable(self, result, **kwargs):
        """当主机无法连接时调用"""
        self.host_unreachable[result._host.get_name()] = result
        print(f"无法连接: 主机 {result._host.get_name()}, 错误: {result._result['msg']}")
    def v2_runner_on_failed(self, result, **kwargs):
        """当任务失败时调用"""
        self.host_failed[result._host.get_name()] = result
        print(f"失败: 主机 {result._host.get_name()}, 错误: {result._result['msg']}")
# --- 1. 初始化 Ansible 核心组件 ---
# 加载器，用于解析 YAML/JSON 文件
loader = ansible.parsing.dataloader.DataLoader()
# Inventory 对象，管理主机清单
inventory = ansible.inventory.Inventory(
    loader=loader,
    variable_manager=VariableManager(),
    host_list=['server1.example.com', 'server2.example.com'], # 可以直接传入主机列表
    # 或者从文件加载: host_list='./inventory.ini'
)
# 变量管理器
variable_manager = VariableManager(loader=loader, inventory=inventory)
# --- 2. 创建 Play ---
# 创建一个 Play 对象，相当于 Playbook 中的一个 play
play_source = dict(
    name="执行 shell 命令",
    hosts=['my_servers'], # 使用 inventory 中定义的组名
    gather_facts='no',
    tasks=[
        dict(action=dict(module='shell', args='echo "Hello from Python API"'), register='shell_out'),
        dict(action=dict(module='debug', args=dict(msg='{{ shell_out.stdout }}')))
    ]
)
# --- 3. 执行 Play ---
# 创建回调实例
callback = ResultCallback()
# 创建任务队列管理器 (TQM)
# 需要传入 loader, inventory, variable_manager, 和一个插件列表 (如回调插件)
tqm = None
try:
    tqm = TaskQueueManager(
        inventory=inventory,
        variable_manager=variable_manager,
        loader=loader,
        passwords={}, # 如果需要 SSH 密码或 sudo 密码，可以在这里提供
        stdout_callback=callback, # 使用我们的自定义回调
        # 你还可以指定其他插件，如连接插件
        # connection='smart', 
        # forks=10
    )
    # 执行 Play
    tqm.run(play_source)
finally:
    # 确保在最后清理 TQM
    if tqm is not None:
        tqm.cleanup()
# --- 4. 输出最终结果 ---
print("\n--- 最终结果 ---")
print(f"成功的主机: {list(callback.host_ok.keys())}")
print(f"失败的主机: {list(callback.host_failed.keys())}")
print(f"无法连接的主机: {list(callback.host_unreachable.keys())}")

注意：

• 这种方法需要你手动管理所有 Ansible 的核心组件 (Loader, Inventory, VariableManager, TaskQueueManager)。
• 连接和认证信息（如 SSH 密钥）需要配置在 Inventory 文件中，或者通过 VariableManager 传递。
• 这种方式非常灵活,但代码量也更大，更容易出错。

总结与对比

对于绝大多数用户,强烈推荐使用 ansible-runner，它更简单、更可靠，并且是 Ansible 官方推荐的集成方式。

只有在需要实现非常复杂的、动态的自动化逻辑，且不希望依赖 Playbook 文件时，才考虑使用底层的 ansible 库。