如何使用Python boto3调用AWS start_session返回结果执行命令

核心原理 #

你调用start_session接口返回的StreamURL是SSM会话服务的WebSocket长连接端点，TokenValue是身份校验令牌，boto3本身没有内置处理会话流的逻辑，必须依赖AWS官方的Session Manager插件完成加密、流转发、命令交互等逻辑。

前置依赖 #

• 本地环境已安装AWS Session Manager插件
• 运行代码的身份、目标实例都已配置SSM会话所需的IAM权限，实例和数据库之间的网络连通性、数据库访问权限已提前配置完成

两种落地方案 #

方案1：交互式会话（适合手动登录数据库场景） #

如果你需要和会话做交互（比如输入数据库密码登录），可以直接把接口返回的会话参数传递给Session Manager插件建立连接：

import subprocess
import json
import boto3

client = boto3.client('ssm', region_name='替换为你的AWS区域，比如us-east-1')

# 你的原有会话创建逻辑
response = client.start_session(
    Target='替换为你的目标实例ID，比如i-xxxxxx',
    DocumentName='AWS-StartInteractiveCommand',
    Parameters={
        'command': ['mysql -h 替换为你的数据库端点 -u 数据库用户名 -p']
    }
)

# 调用插件建立连接，直接弹出交互式终端
subprocess.run([
    'session-manager-plugin',
    json.dumps(response),
    '替换为你的AWS区域',
    'StartSession',
    json.dumps({"Target": "替换为你的目标实例ID"}),
    'https://ssm.替换为你的AWS区域.amazonaws.com'
])

方案2：非交互式自动化执行（更适合批量自动化场景） #

如果你不需要人工交互，只是要自动执行命令拿结果，建议直接用send_command接口，不需要处理复杂的WebSocket流，使用更简单：

import boto3
import time

client = boto3.client('ssm', region_name='替换为你的AWS区域')

# 下发要执行的命令
run_resp = client.send_command(
    InstanceIds=['替换为你的目标实例ID'],
    DocumentName='AWS-RunShellScript',
    Parameters={
        # 直接写要执行的非交互式数据库命令，比如查询库表
        'commands': ['mysql -h 数据库端点 -u 用户名 -p"数据库密码" -e "show databases;"']
    }
)
command_id = run_resp['Command']['CommandId']
instance_id = '替换为你的目标实例ID'

# 轮询获取命令执行结果
while True:
    result = client.get_command_invocation(
        CommandId=command_id,
        InstanceId=instance_id
    )
    if result['Status'] in ['Success', 'Failed', 'TimedOut']:
        break
    time.sleep(1)

# 输出执行结果
print("命令标准输出：\n", result['StandardOutputContent'])
print("命令错误输出：\n", result['StandardErrorContent'])

注意事项 #

• 涉及数据库密码的场景，建议不要明文写在代码中，可通过AWS Secrets Manager等服务加密存储后动态获取
• 如果命令执行失败，优先检查实例IAM权限、实例到数据库的安全组配置、SSM插件版本是否符合要求

内容的提问来源于stack exchange，提问作者vinstart