Skip to content

feat: Persistent Task Store and Distributed event bus implementation … #759

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hashtekconsulting wants to merge 6 commits into
base: main
Choose a base branch
from
Open

feat: Persistent Task Store and Distributed event bus implementation … #759

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
13fd561
feat: Persistent Task Store and Distributed event bus implementation …
Mar 4, 2026
3627c06
feat: Addressed PR review comments and fixed spelling and linting issues
Mar 4, 2026
52b2d01
feat: Addressed couple more linting and spelling issues to fix the bu…
Mar 4, 2026
30fe2d3
feat: Fixed build issues related to linting and formatting
Mar 4, 2026
55fd5d7
feat: Added instructions file how to leverage the distributed compone…
Mar 4, 2026
37ae801
feat: Addressed build issue by fixing the spelling issues
Mar 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
feat: Addressed couple more linting and spelling issues to fix the bu… 
…ild failures.
  • Loading branch information
Rajesh Ramamoorthy committed Mar 4, 2026
commit 52b2d018b91275e589c731b6c795f75ae2260151
18 changes: 11 additions & 7 deletions src/a2a/server/events/queue_lifecycle_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,8 +83,12 @@ def __post_init__(self) -> None:
'pip install "a2a-sdk[aws]"'
) from exc

if self.session is None:
self.session = aioboto3.Session()
# Resolve the session once into a non-optional attribute so that
# internal methods can call .client() without optional-access errors.
# The public `session` field remains Optional for API convenience.
self._resolved_session: aioboto3.Session = (
self.session or aioboto3.Session()
)

logger.debug(
'QueueLifecycleManager created (instance_id=%s, topic=%s).',
Expand Down Expand Up @@ -152,7 +156,7 @@ async def provision(self) -> QueueProvisionResult:
self.region_name,
)

async with self.session.client(
async with self._resolved_session.client(
'sqs', region_name=self.region_name
) as sqs:
# Step 1: Create the SQS queue.
Expand Down Expand Up @@ -198,7 +202,7 @@ async def provision(self) -> QueueProvisionResult:
# Step 4: Subscribe the SQS queue to the SNS topic.
subscription_arn = ''
try:
async with self.session.client(
async with self._resolved_session.client(
'sns', region_name=self.region_name
) as sns:
sub_resp = await sns.subscribe(
Expand All @@ -216,7 +220,7 @@ async def provision(self) -> QueueProvisionResult:
queue_url,
)
try:
async with self.session.client(
async with self._resolved_session.client(
'sqs', region_name=self.region_name
) as sqs:
await sqs.delete_queue(QueueUrl=queue_url)
Expand Down Expand Up @@ -257,7 +261,7 @@ async def teardown(self) -> None:

# Step 1: Unsubscribe from SNS (best-effort).
try:
async with self.session.client(
async with self._resolved_session.client(
'sns', region_name=self.region_name
) as sns:
await sns.unsubscribe(SubscriptionArn=result.subscription_arn)
Expand All @@ -272,7 +276,7 @@ async def teardown(self) -> None:

# Step 2: Delete the SQS queue.
try:
async with self.session.client(
async with self._resolved_session.client(
'sqs', region_name=self.region_name
) as sqs:
await sqs.delete_queue(QueueUrl=result.queue_url)
Expand Down
2 changes: 1 addition & 1 deletion src/a2a/server/events/sns_queue_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -340,7 +340,7 @@ async def _handle_sqs_message(self, sqs_msg: dict[str, Any]) -> None: # noqa: P
sqs_msg: A single SQS message dictionary from ReceiveMessage.
"""
# Use message ID in warnings to avoid logging potentially sensitive
# message body content (PII in Task/Message payloads).
# message body content, which may include personal user data.
msg_id = sqs_msg.get('MessageId', '<no-id>')
body_str = sqs_msg.get('Body', '{}')

Expand Down
Loading