a2aproject / a2aproject/a2a-python

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

Đang mở
#833 2 bình luận 0 reaction 1 người được giao Được giao cho @kevinlu310 Xem trên GitHub
component: core help wanted
Ngôn ngữ chính
Python
Star
2.1k
Fork
496
Merge trung bình
4 ngày 17 giờ
Pull request đã merge (30 ngày)
12

Mô tả

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

Hướng dẫn đóng góp

Mở hướng dẫn đóng góp

Đánh giá

Issue này chưa được đánh giá.

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.