feat: Create EventQueue interface and make tap() async.#914
Merged
feat: Create EventQueue interface and make tap() async.#914
Conversation
🧪 Code Coverage (vs
|
| Base | PR | Delta | |
|---|---|---|---|
| src/a2a/server/events/event_consumer.py | 91.80% | 90.20% | 🔴 -1.61% |
| src/a2a/server/events/event_queue.py | 96.19% | 95.51% | 🔴 -0.68% |
| src/a2a/server/events/event_queue_v2.py (new) | — | 88.64% | — |
| Total | 91.59% | 91.50% | 🔴 -0.09% |
Generated by coverage-comment.yml
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This pull request refactors the event handling system by introducing an abstract base class for EventQueue and a new dispatcher-based implementation consisting of EventQueueSource and EventQueueSink. The original implementation is maintained as EventQueueLegacy for backward compatibility, with a factory method in the base class to handle instantiation. Key changes include making the tap() method asynchronous, removing the consume_one method from EventConsumer, and deleting the clear_events functionality. Review feedback suggests increasing the log level for unexpected dispatcher task cancellations to improve production visibility and simplifying the idempotent logic within the sink's closure method.
bartek-w
changed the title
bartek-w
force-pushed
the
bartekw-event-queue-v2
branch
from
9e83e6b to
bf66e13
Compare
guglielmo-san
approved these changes
Mar 31, 2026
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).
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Refactor
EventQueueto introduce a formal interface (ABC) and a more robust v2 implementation (EventQueueSource/EventQueueSink) while maintaining EventQueueLegacy.The previous
EventQueueimplementation (nowEventQueueLegacy) include multiple concurency issue and provide fragile not documented synchronization contracts.New EventQueue/EventQueueSource/EventQueueSink will be used in new version of DefaultRequestHandler.
Fixes #869