Hacktoberfest 2026: the issues maintainers tagged for October, open and beginner-friendly. Browse Hacktoberfest issues

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

Open
#833 2 comments 0 reactions 1 assignee View on GitHub

Nobody has claimed this yet.

Assessment

Difficulty
3/5
Estimated time
1-2 days
Newbie friendliness
55/100
Issue type
Feature
Clarity
Clearly specified
Activity status
Stale
Tech stack
python

Research direction

The helper should be added to a2a/utils/artifact.py. Start by examining the existing new_text_artifact function and the TaskArtifactUpdateEvent class. The new ArtifactStreamer needs to generate a stable UUID, manage state, and produce events with the correct append and last_chunk flags. Update the travel_planner_agent sample to use the new helper as a final step.

Written by the indexing model from the issue text.

Description

component: core help wanted
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 demonstrates this problem — it calls new_text_artifact in a loop:

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:

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 sample should also be updated to use this helper once available.
  • The A2A spec defines append on 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
  • I agree to follow this project's Code of Conduct
Dominant language
Python
Stars
2.2k
Forks
496
Avg merge
2d 6h
Merged PRs (30d)
19

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

More from a2aproject/a2a-python

All issues in a2aproject/a2a-python

Similar issues

More Python issues

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.