Python Bindings

Async Python bindings via PyO3 with full type hints.

Table of contents

1. Installation
2. Quick Start
3. Session Manager
4. Context Manager
5. Interactive Shell
6. Multiple Sessions
7. Session Types
8. Type Hints

---

Installation

pip install aws-ssm-bridge

---

Quick Start

import asyncio
from aws_ssm_bridge import SessionManager

async def main():
    # Create session manager
    manager = await SessionManager.new()
    
    # Use context manager for automatic cleanup
    async with await manager.start_session(target="i-0123456789abcdef0") as session:
        await session.send(b"whoami\n")
        
        async for chunk in await session.output():
            print(chunk.decode(), end="")

asyncio.run(main())

---

Session Manager

The SessionManager handles AWS authentication and session lifecycle:

import asyncio
from aws_ssm_bridge import SessionManager

async def main():
    # Create manager (loads AWS credentials from default chain)
    manager = await SessionManager.new()
    
    # Start session with options
    session = await manager.start_session(
        target="i-0123456789abcdef0",
        region="us-west-2",                    # Optional
        session_type="standard_stream",        # Optional
        reason="Debugging issue #123",         # Optional (for audit)
    )
    
    # Wait for session to be ready
    if await session.wait_for_ready(timeout_secs=30.0):
        await session.send(b"hostname\n")
        async for chunk in await session.output():
            print(chunk.decode(), end="")
    
    # Always terminate when done
    await session.terminate()

asyncio.run(main())

---

Context Manager

Use the async context manager for automatic cleanup:

async with await manager.start_session(target="i-xxx") as session:
    await session.send(b"ls -la\n")
    async for chunk in await session.output():
        print(chunk.decode(), end="")
# Session automatically terminated on exit

---

Interactive Shell

Full terminal support with raw mode, resize, and signal handling:

import asyncio
from aws_ssm_bridge import InteractiveShell, InteractiveConfig, run_shell

async def main():
    # Option 1: Quick one-liner
    await run_shell("i-0123456789abcdef0")
    
    # Option 2: Full control
    config = InteractiveConfig(
        show_banner=True,
        send_initial_size=True,
        forward_signals=True,
    )
    shell = InteractiveShell(config)
    await shell.connect("i-0123456789abcdef0")
    await shell.run()  # Blocks until Ctrl+D

asyncio.run(main())

Features:

• Raw terminal mode (no echo, immediate input)
• Terminal resize detection (SIGWINCH)
• Signal forwarding (Ctrl+C, Ctrl+D, Ctrl+Z)
• Automatic terminal restoration on exit/crash

---

Multiple Sessions

Run commands on multiple instances concurrently:

import asyncio
from aws_ssm_bridge import SessionManager

async def run_command(manager, instance_id: str, command: str) -> str:
    async with await manager.start_session(target=instance_id) as session:
        await session.send(f"{command}\n".encode())
        output = []
        async for chunk in await session.output():
            output.append(chunk.decode())
            if len(output) > 10:
                break
        return "".join(output)

async def main():
    manager = await SessionManager.new()
    instances = ["i-aaa", "i-bbb", "i-ccc"]
    results = await asyncio.gather(*[
        run_command(manager, i, "hostname") for i in instances
    ], return_exceptions=True)
    
    for inst, result in zip(instances, results):
        if isinstance(result, Exception):
            print(f"{inst}: ERROR - {result}")
        else:
            print(f"{inst}: {result.strip()}")

asyncio.run(main())

---

Session Types

AWS SSM Session Manager supports 3 session types:

# Shell session (default) - standard interactive shell
shell = await manager.start_session(
    target="i-xxx",
    session_type="standard_stream",
)

# Port forwarding session (AWS-StartPortForwardingSession)
port = await manager.start_session(
    target="i-xxx",
    session_type="port",
    parameters={"portNumber": ["3306"], "localPortNumber": ["13306"]},
)

# Interactive commands (AWS-StartInteractiveCommand)
cmd = await manager.start_session(
    target="i-xxx",
    session_type="interactive_commands",
    parameters={"command": ["top"]},
)

For custom documents like AWS-StartSSHSession or AWS-StartPortForwardingSessionToRemoteHost:

# SSH over Session Manager
ssh = await manager.start_session(
    target="i-xxx",
    document_name="AWS-StartSSHSession",
    parameters={"portNumber": ["22"]},
)

# Port forward through bastion to RDS
rds = await manager.start_session(
    target="i-bastion",
    document_name="AWS-StartPortForwardingSessionToRemoteHost",
    parameters={
        "host": ["mydb.cluster-xxx.us-east-1.rds.amazonaws.com"],
        "portNumber": ["3306"],
        "localPortNumber": ["13306"],
    },
)

---

Type Hints

Full .pyi stubs are included for IDE autocomplete:

from aws_ssm_bridge import SessionManager, Session, OutputStream

async def example(manager: SessionManager) -> None:
    session: Session = await manager.start_session(target="i-xxx")
    stream: OutputStream = await session.output()
    
    async for chunk in stream:
        data: bytes = chunk
        print(data.decode())