a2aproject / a2aproject/a2a-python
[Feat]: Add streaming-aware helper for artifact updates
- Dominant language
- Python
- Stars
- 2.1k
- Forks
- 496
- Avg merge
- 4d 17h
- Merged PRs (30d)
- 12
Description
### Problem
When implementing streaming text output via `TaskArtifactUpdateEvent`, the current `new_text_artifact` utility generates a **fresh UUID** for `artifact_id` on every call. This makes it impossible to use `append=True` correctly, since append semantics require a stable `artifact_id` across chunks.
The sample [`travel_planner_agent`](https://github.com/a2aproject/a2a-samples/tree/main/samples/python/agents/travel_planner_agent) demonstrates this problem — it calls `new_text_artifact` in a loop:
```python
async for event in self.agent.stream(query):
message = TaskArtifactUpdateEvent(
contextId=context.context_id,
taskId=context.task_id,
artifact=new_text_artifact(
name='current_result',
text=event['content'],
),
)
await event_queue.enqueue_event(message)
```
Each iteration produces a new artifact_id, so clients cannot merge chunks into a single artifact. This results in N separate artifact cards in the UI instead of one progressively streamed response.
### Describe the solution you'd like
Add a stateful streaming helper to a2a.utils that:
1. Generates a stable artifact_id once on construction
2. Provides an append(text) method that returns a `TaskArtifactUpdateEvent` with `append=True`, `last_chunk=False`
3. Provides a finalize() method that returns a TaskArtifactUpdateEvent with append=True, last_chunk=True
Example API:
```python
from a2a.utils import ArtifactStreamer
streamer = ArtifactStreamer(context_id, task_id, name="response")
async for chunk in llm.stream(prompt):
await event_queue.enqueue_event(streamer.append(chunk))
await event_queue.enqueue_event(streamer.finalize())
```
This would live in a2a/utils/artifact.py alongside the existing new_text_artifact and new_artifact helpers.
### Describe alternatives you've considered
_No response_
### Additional context
- The [travel_planner_agent](https://github.com/a2aproject/a2a-samples/tree/main/samples/python/agents/travel_planner_agent) sample should also be updated to use this helper once available.
- The A2A spec defines append on [`TaskArtifactUpdateEvent`](https://a2a-protocol.org/latest/specification/#422-taskartifactupdateevent) as: "If true, the content of this artifact should be appended to a previously sent artifact with the same ID.". So the current `new_text_artifact` usage in a streaming loop is a misuse of the spec's intent.
### Code of Conduct
- [x] I agree to follow this project's Code of Conduct
Contributor guide
Assessment
This issue has not been assessed yet.