[FEATURE] Async reading from S3 Session Manager List Message

[Unshure]

Problem Statement

I would like Strands to list messages asynchronously from S3 to improve its performance.

Proposed Solution

We will spawn a thread and start an event_loop in that thread. This threaded event loop will call s3 asynchronously for reading the objects it needs to from s3.

• Code for spawning a thread for async activities: https://github.com/strands-agents/sdk-python/blob/main/src/strands/_async.py
• Tracking issue: Add parallel reading support to S3SessionManager.list_messages() #1186
• Similar implementation that called s3 async, but did not spawn a thread: feat: implement concurrent message reading for session managers #897

Use Case

Call aws boto s3 get object asynchronously

Alternatives Solutions

N/A

---

Implementation Requirements

Based on clarification discussion and repository analysis:

Technical Approach

• Target File: src/strands/session/s3_session_manager.py
• Target Method: list_messages() (lines 259-300)
• Pattern: Use threaded async event loop (similar to existing run_async() utility)
• Performance Bottleneck: Current sequential S3 reads (lines 292-295)

Dependencies

• Add: aiobotocore (https://github.com/aio-libs/aiobotocore)
  • Add to pyproject.toml dependencies
  • Use aiobotocore for async S3 operations instead of aioboto3

Scope

• ONLY convert list_messages() to async
• DO NOT convert other S3 operations (read_message, write_s3_object, delete_session, etc.)
• Maintain both async and sync S3 clients in S3SessionManager

Implementation Details

1. Add aiobotocore Client

• Create an async S3 client using aiobotocore alongside existing sync boto3 client
• Initialize in __init__() method
• Ensure proper cleanup/resource management

2. Implement Async list_messages()

• Create async version of message reading logic
• Use asyncio.gather() or similar to read multiple messages concurrently
• Maintain message ordering (sort by message index)
• Use existing run_async() utility from src/strands/_async.py to execute async code

3. Integration Pattern

# Use run_async() to spawn thread and run async event loop
from strands._async import run_async
...
# Pseudocode example
def list_messages(self, session_id, agent_id, limit=None, offset=0, **kwargs):

    
    async def async_list_messages():
        # 1. List message keys (can stay sync or convert to async)
        # 2. Sort and paginate keys
        # 3. Read messages concurrently using asyncio.gather()
        # 4. Return ordered list
        pass
    
    return run_async(async_list_messages)

Configuration

• Default Behavior: Async reading is the DEFAULT behavior
• No configuration parameters needed: All list_messages calls will use async
• Backward Compatible: External API remains synchronous (async is internal implementation)

Error Handling

• Fail Fast: If ANY S3 message read fails, raise an error
• Stop the entire operation on first failure
• Propagate errors to caller with appropriate error messages
• Use existing SessionException for error reporting

Testing Strategy

• Unit Tests: Existing tests in tests/strands/session/test_s3_session_manager.py should work
  • Tests use moto for AWS mocking
  • Tests should be agnostic to underlying sync vs async client
  • Verify tests still pass with async implementation
• Integration Tests: Already exist, should continue to work
• No new tests required: Existing test coverage should be sufficient

Files to Modify

1. src/strands/session/s3_session_manager.py

   • Add aiobotocore client initialization
   • Modify list_messages() method
   • Add async helper method for concurrent message reading

2. pyproject.toml

   • Add aiobotocore to dependencies

Acceptance Criteria

• aiobotocore added to project dependencies
• Async S3 client created and managed in S3SessionManager
• list_messages() uses async reading with threaded event loop via run_async()
• Messages are read concurrently from S3
• Message ordering is preserved (sorted by message index)
• Error handling: fail fast on any read error
• All existing unit tests pass (tests/strands/session/test_s3_session_manager.py)
• All existing integration tests pass
• External API remains synchronous (no breaking changes)
• Code follows existing patterns and style guidelines

Related Work

• PR Add parallel reading support to S3SessionManager.list_messages() #1186: Will be closed in favor of this approach
• PR feat: implement concurrent message reading for session managers #897: Referenced as similar implementation
• Existing Utility: src/strands/_async.py - run_async() function to be reused

Implementation Notes

• Reuse existing run_async() utility as-is (no modifications needed)
• Don't worry about performance benchmarks or metrics
• Focus on correctness and maintaining existing test coverage
• Single cohesive implementation (no sub-task breakdown)

[Unshure]

/strands

[github-actions]

Repository Analysis & Clarifying Questions

Repository Analysis

I've analyzed the repository structure and current implementation to understand how this feature would integrate:

Current Implementation

• File: src/strands/session/s3_session_manager.py
• Target Method: list_messages() (lines 259-300)
• Current Behavior: Sequential S3 reads using synchronous boto3 client
• Performance Bottleneck: Lines 292-295 read each message object one at a time

Existing Async Utilities

• File: src/strands/_async.py
• Function: run_async() - Executes async code in separate thread using ThreadPoolExecutor
• Pattern: Isolates async execution to avoid event loop conflicts

Dependencies

• Current: boto3>=1.26.0,<2.0.0 (synchronous only)
• No async S3: aioboto3 not currently included

Related Work

• PR Add parallel reading support to S3SessionManager.list_messages() #1186 (OPEN): Proposes parallel reading using ThreadPoolExecutor with sync boto3
• PR feat: implement concurrent message reading for session managers #897 (MERGED per API, but code shows sequential): Mentioned as similar implementation

Testing

• Unit Tests: tests/strands/session/test_s3_session_manager.py
• Mocking: Uses moto library for AWS service mocks
• Coverage: Tests exist for list_messages (lines 260-307)

---

Clarifying Questions

To ensure this feature is well-defined and implementable, I have some questions:

1. Relationship to PR #1186

PR #1186 is currently open and proposes parallel reading using ThreadPoolExecutor with synchronous boto3. This issue proposes a different approach using a threaded event loop with async boto3 (aioboto3).

• Are these competing implementations, or should they coexist?
• If competing, what are the criteria for choosing this async approach over PR Add parallel reading support to S3SessionManager.list_messages() #1186's ThreadPoolExecutor approach?
• Should PR Add parallel reading support to S3SessionManager.list_messages() #1186 be closed if this issue moves forward?

2. Dependency Management

Implementing async S3 calls will likely require aioboto3 as a new dependency.

• Should aioboto3 be added as a core dependency or an optional dependency?
• Are there concerns about adding this dependency (package size, maintenance, compatibility)?
• Should we use aioboto3 or another async boto3 solution?

3. Scope of Implementation

The issue mentions "list messages asynchronously" but S3SessionManager has multiple S3 operations.

• Should ONLY list_messages() be converted to async, or should other methods be converted too?
  • _read_s3_object() (used by read_message, read_agent, read_session)
  • _write_s3_object() (used by create/update operations)
  • delete_session() (batch delete operations)
• If only list_messages(), should we maintain both sync and async S3 clients?

4. Configuration and Backward Compatibility

• Should async reading be opt-in or the default behavior?
• Should there be a configuration parameter (e.g., use_async=True) for users to choose?
• How should this interact with existing code that expects synchronous behavior?
• Should there be performance tuning parameters (e.g., max concurrent reads)?

5. Testing Strategy

• Does moto (the AWS mocking library used in tests) support aioboto3?
• If not, what's the testing strategy for async S3 operations?
• Should integration tests be added for this feature?

6. Error Handling

• How should async errors be handled and surfaced to users?
• Should individual message read failures stop the entire operation or just log/skip?
• What retry logic should be implemented for transient S3 errors?

7. Performance Expectations

• What's the expected performance improvement compared to sequential reading?
• Are there specific benchmarks or use cases driving this feature?
• Should performance benchmarks be included in the implementation?

8. Integration with Existing run_async() Utility

• Can the existing run_async() function in src/strands/_async.py be reused as-is?
• Does it need modifications to support this use case?
• Or should a new async utility be created specifically for S3 operations?

9. Task Breakdown

Given the complexity and multiple design decisions needed:

• Should this be broken down into smaller sub-tasks?
  • Sub-task 1: Add aioboto3 dependency and basic async S3 client setup
  • Sub-task 2: Implement async list_messages()
  • Sub-task 3: Add tests and documentation
  • Sub-task 4: (Optional) Convert other S3 operations to async
• Or should it be implemented as a single cohesive feature?

---

Please respond to these questions when you have a chance. Based on your answers, I'll update the issue description with a comprehensive implementation specification.

[Unshure]

/strands

1. I will close that PR. The correct path forward is to spawn a thread, and run an async event loop on that thread
2. Can you instead use aiobotocore: https://github.com/aio-libs/aiobotocore/tree/master
3. Only list_messages. And can you maintain an async and sync client for now?
4. async read should be the default behavior. The rest of the code is sync, so you can spawn a thread and run the async code in that thread. Look for usage of this function in the sdk for reference on how this should work: https://github.com/strands-agents/sdk-python/blob/main/src/strands/_async.py
5. Look at the existing tests, they should work regardless of the underlying client. Integ tests should already exist
6. If any s3 message reads fail, you should raise an error
7. Dont worry about performace expectations
8. yes, reuse it as-is
9. Single task implementation

[github-actions]

Task Refinement Complete

Summary

I've updated the issue description with comprehensive implementation requirements based on our discussion. The feature is now ready for implementation.

What Was Accomplished

1. Repository Analysis

• Identified target file and method: S3SessionManager.list_messages() (lines 259-300)
• Located existing async utility pattern: src/strands/_async.py
• Confirmed current sequential reading bottleneck
• Reviewed testing infrastructure (moto-based unit tests)

2. Clarification & Design Decisions

• Approach: Threaded async event loop using run_async()
• Dependency: Use aiobotocore (not aioboto3)
• Scope: Only list_messages() method
• Configuration: Async is default, external API stays synchronous
• Error Handling: Fail fast on any read error
• Testing: Existing tests should work as-is

3. Implementation Specification

The updated issue description now includes:

• Technical approach and architecture
• Dependency requirements
• Detailed implementation steps
• Code integration pattern with pseudocode
• Error handling strategy
• Testing strategy
• Acceptance criteria checklist
• Files to modify

Ready for Implementation

This issue is now ready for implementation as a single cohesive feature. All ambiguities have been resolved, and the implementation path is clear.

Key Implementation Steps:

1. Add aiobotocore to pyproject.toml
2. Create async S3 client in S3SessionManager.__init__()
3. Implement async message reading in list_messages() using run_async()
4. Ensure error handling fails fast on any read failure
5. Verify all existing tests pass

Notes:

• PR Add parallel reading support to S3SessionManager.list_messages() #1186 will be closed in favor of this approach
• No breaking changes to external API
• Reuses existing run_async() utility without modification

[Unshure]

/strands implement

[github-actions]

Unable to create pull request via API. You can create it manually by clicking here.