a2aproject / a2aproject/a2a-python

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

Abierto
#833 2 comentarios 0 reacciones 1 asignado Asignado a @kevinlu310 Ver en GitHub
component: core help wanted
Lenguaje dominante
Python
Estrellas
2.1k
Forks
496
Merge medio
4 d 17 h
PR fusionados (30 d)
12

Descripción

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

Guía de contribución

Abrir la guía de contribución

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.