feat: EventQueue - unify implementation between python versions by bartek-w · Pull Request #877 · a2aproject/a2a-python · GitHub
Skip to content

feat: EventQueue - unify implementation between python versions - #877

Merged
bartek-w merged 3 commits into
1.0-devfrom
bartekw-event-queue-close
Mar 20, 2026
Merged

feat: EventQueue - unify implementation between python versions#877
bartek-w merged 3 commits into
1.0-devfrom
bartekw-event-queue-close

Conversation

@bartek-w

@bartek-w bartek-w commented Mar 20, 2026

Copy link
Copy Markdown
Collaborator

Introduced a compatibility layer using the culsans library to backport asyncio.Queue.shutdown functionality to Python versions older than 3.13. Previous implementation was broken (deadlocks and inconsistent behaviour with 3.13 implementation). Culsans library allowed for unified code between versions.

EventConsumer now starts a background task to gracefully wait for queue to finish.

This is one of the steps towards better concurrency model in a2a python sdk.

Fixes #869

@gemini-code-assist

Copy link
Copy Markdown
Contributor

@github-actions

github-actions Bot commented Mar 20, 2026

Copy link
Copy Markdown

🧪 Code Coverage (vs 1.0-dev)

⬇️ Download Full Report

Base PR Delta
src/a2a/server/events/event_consumer.py 90.91% 91.80% 🟢 +0.89%
src/a2a/server/events/event_queue.py 90.24% 95.37% 🟢 +5.13%
Total 90.69% 90.77% 🟢 +0.08%

Generated by coverage-comment.yml

@gemini-code-assist gemini-code-assist Bot left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Code Review

This pull request significantly improves the EventQueue and EventConsumer by introducing graceful shutdown mechanisms and ensuring cross-Python version compatibility for asyncio.QueueShutDown behavior. The refactoring of the close method in EventQueue is well-executed, abstracting away version-specific logic. The addition of culsans for older Python versions and the updated test suite demonstrate a thorough approach to handling these complexities. The new tests adequately cover the graceful closing scenarios and race conditions, which is crucial for robust asynchronous event handling. Additionally, the docstring for the close method has been expanded for better clarity and maintainability.

Comment thread src/a2a/server/events/event_queue.py Outdated
@bartek-w bartek-w changed the title EventQueue: Graceful close. feat: EventQueue graceful close. Mar 20, 2026
@bartek-w
bartek-w force-pushed the bartekw-event-queue-close branch from 334ee08 to ec31c7b Compare March 20, 2026 07:56
@bartek-w

Copy link
Copy Markdown
Collaborator Author

@gemini-code-assist gemini-code-assist Bot left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Code Review

This pull request introduces a significant improvement to the event queue's graceful shutdown mechanism by backporting asyncio.Queue.shutdown functionality using the culsans library. This change unifies the queue behavior across different Python versions, simplifies the EventQueue implementation by removing version-specific logic, and resolves potential deadlocks in the EventConsumer. The addition of comprehensive, parametrized tests for various shutdown scenarios is a notable enhancement to the test suite's robustness. The overall changes are well-implemented and improve both the correctness and maintainability of the concurrency model.

Comment thread tests/server/events/test_event_consumer.py Outdated
@bartek-w
bartek-w force-pushed the bartekw-event-queue-close branch from ec31c7b to 1ac0a1c Compare March 20, 2026 08:02
@bartek-w
bartek-w requested a review from ishymko March 20, 2026 08:02
@bartek-w
bartek-w marked this pull request as ready for review March 20, 2026 08:03
@bartek-w
bartek-w requested a review from a team as a code owner March 20, 2026 08:03
Comment thread src/a2a/server/events/event_consumer.py
Comment thread src/a2a/server/events/event_consumer.py Outdated
Comment thread tests/server/events/test_event_consumer.py
@bartek-w bartek-w changed the title feat: EventQueue graceful close. feat: EventQueue - unify implementation between python versions Mar 20, 2026
@bartek-w
bartek-w force-pushed the bartekw-event-queue-close branch from 4e5e6df to bb3ee7a Compare March 20, 2026 12:10
@bartek-w
bartek-w force-pushed the bartekw-event-queue-close branch from bb3ee7a to d7ecdbd Compare March 20, 2026 12:12
@bartek-w
bartek-w merged commit 7437b88 into 1.0-dev Mar 20, 2026
12 checks passed
@bartek-w
bartek-w deleted the bartekw-event-queue-close branch March 20, 2026 12:16
guglielmo-san pushed a commit that referenced this pull request Apr 10, 2026
🤖 I have created a release *beep* *boop*
---


##
[1.0.0-alpha.1](v1.0.0-alpha.0...v1.0.0-alpha.1)
(2026-04-10)


### ⚠ BREAKING CHANGES

* **client:** make ClientConfig.push_notification_config singular
([#955](#955))
* **client:** reorganize ClientFactory API
([#947](#947))
* **server:** add build_user function to DefaultContextBuilder to allow
A2A user creation customization
([#925](#925))
* **client:** remove `ClientTaskManager` and `Consumers` from client
([#916](#916))
* **server:** migrate from Application wrappers to Starlette route-based
endpoints for rest
([#892](#892))
* **server:** migrate from Application wrappers to Starlette route-based
endpoints for jsonrpc
([#873](#873))

### Features

* A2A Version Header validation on server side.
([#865](#865))
([b261ceb](b261ceb))
* Add GetExtendedAgentCard Support to RequestHandlers
([#919](#919))
([2159140](2159140))
* Add support for more Task Message and Artifact fields in the Vertex
Task Store ([#936](#936))
([605fa49](605fa49))
* Create EventQueue interface and make tap() async.
([#914](#914))
([9ccf99c](9ccf99c)),
closes [#869](#869)
* EventQueue - unify implementation between python versions
([#877](#877))
([7437b88](7437b88)),
closes [#869](#869)
* EventQueue is now a simple interface with single enqueue_event method.
([#944](#944))
([f0e1d74](f0e1d74))
* Implementation of DefaultRequestHandlerV2
([#933](#933))
([462eb3c](462eb3c)),
closes [#869](#869)
* InMemoryTaskStore creates a copy of Task by default to make it
consistent with database task stores
([#887](#887))
([8c65e84](8c65e84)),
closes [#869](#869)
* merge metadata of new and old artifact when append=True
([#945](#945))
([cc094aa](cc094aa))
* **server:** add async context manager support to EventQueue
([#743](#743))
([f68b22f](f68b22f))
* **server:** validate presence according to `google.api.field_behavior`
annotations
([#870](#870))
([4586c3e](4586c3e))
* Simplify ActiveTask.subscribe()
([#958](#958))
([62e5e59](62e5e59))
* Support AgentExectuor enqueue of a Task object.
([#960](#960))
([12ce017](12ce017))
* Support Message-only simplified execution without creating Task
([#956](#956))
([354fdfb](354fdfb))
* Unhandled exception in AgentExecutor marks task as failed
([#943](#943))
([4fc6b54](4fc6b54))


### Bug Fixes

* Add `packaging` to base dependencies
([#897](#897))
([7a9aec7](7a9aec7))
* **client:** do not mutate SendMessageRequest in
BaseClient.send_message
([#949](#949))
([94537c3](94537c3))
* fix `athrow()` RuntimeError on streaming responses
([#912](#912))
([ca7edc3](ca7edc3))
* fix docstrings related to `CallContextBuilder` args in constructors
and make ServerCallContext mandatory in `compat` folder
([#907](#907))
([9cade9b](9cade9b))
* fix error handling for gRPC and SSE streaming
([#879](#879))
([2b323d0](2b323d0))
* fix JSONRPC error handling
([#957](#957))
([6c807d5](6c807d5))
* fix REST error handling
([#893](#893))
([405be3f](405be3f))
* handle SSE errors occurred after stream started
([#894](#894))
([3a68d8f](3a68d8f))
* remove the use of deprecated types from VertexTaskStore
([#889](#889))
([6d49122](6d49122))
* Remove unconditional SQLAlchemy dependency from SDK core
([#898](#898))
([ab762f0](ab762f0)),
closes [#883](#883)
* remove unused import and request for FastAPI in pyproject
([#934](#934))
([fe5de77](fe5de77))
* replace stale entry in a2a.types.__all__ with actual import name
([#902](#902))
([05cd5e9](05cd5e9))
* wrong method name for ExtendedAgentCard endpoint in JsonRpc compat
version ([#931](#931))
([5d22186](5d22186))


### Documentation

* add Database Migration Documentation
([#864](#864))
([fd12dff](fd12dff))


### Miscellaneous Chores

* release 1.0.0-alpha.1
([a61f6d4](a61f6d4))


### Code Refactoring

* **client:** make ClientConfig.push_notification_config singular
([#955](#955))
([be4c5ff](be4c5ff))
* **client:** remove `ClientTaskManager` and `Consumers` from client
([#916](#916))
([97058bb](97058bb)),
closes [#734](#734)
* **client:** reorganize ClientFactory API
([#947](#947))
([01b3b2c](01b3b2c))
* **server:** add build_user function to DefaultContextBuilder to allow
A2A user creation customization
([#925](#925))
([2648c5e](2648c5e))
* **server:** migrate from Application wrappers to Starlette route-based
endpoints for jsonrpc
([#873](#873))
([734d062](734d062))
* **server:** migrate from Application wrappers to Starlette route-based
endpoints for rest
([#892](#892))
([4be2064](4be2064))

---
This PR was generated with [Release
Please](https://github.com/googleapis/release-please). See
[documentation](https://github.com/googleapis/release-please#release-please).
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants