a2aproject / a2aproject/a2a-python

[Feat]: Add streaming-aware helper for artifact updates

Aperta
#833 2 commenti 0 reazioni 1 assegnatario Assegnata a @kevinlu310 Vedi su GitHub
component: core help wanted
Lingua principale
Python
Stelle
2.1k
Fork
496
Merge medio
4g 17h
PR unite (30g)
12

Descrizione

### 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

Guida per i contributori

Apri la guida per i contributori

Valutazione

Questa issue non è ancora stata valutata.

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.