a2aproject/a2a-python

[Feat]: Add extensions support to TaskUpdater.add_artifact

marco-sciatta opened this issue · 0 comments

Is your feature request related to a problem? Please describe.

According to the spec and SDK docs, Artifact has an extensions field to indicate which extensions contributed to a specific artifact.

However, TaskUpdater.add_artifact(...) does not accept a parameter to populate this field on the generated Artifact.

There is no argument on TaskUpdater.add_artifact to set it.

Is it possibile to provide a way to set artifact.extensions explicitly when creating the artifact e.g.:

If there is a canonical way to set artifact.extensions today that I’m missing, please point me to it.
Thanks!

Describe the solution you'd like

a possible solution could be something like that:

add support to pass extensions as arguments

    async def add_artifact(  # noqa: PLR0913
        self,
        parts: list[Part],
        artifact_id: str | None = None,
        name: str | None = None,
        metadata: dict[str, Any] | None = None,
        append: bool | None = None,
        last_chunk: bool | None = None,
        extensions: List[str] | None = None
    ) -> None:
        if not artifact_id:
            artifact_id = str(uuid.uuid4())

        await self.event_queue.enqueue_event(
            TaskArtifactUpdateEvent(
                task_id=self.task_id,
                context_id=self.context_id,
                artifact=Artifact(
                    artifact_id=artifact_id,
                    name=name,
                    parts=parts,
                    metadata=metadata,
                    extensions=extensions
                ),
                append=append,
                last_chunk=last_chunk,
            )
        )

so it can be used like this in the executor

await updater.add_artifact(
    parts,
    name="agent_result",
    append=False,
    last_chunk=True,
    metadata=artifact_metadata,
    extensions=["https://example.com/ext/example/v1"],
)

Describe alternatives you've considered

At the moment I could not find any alternative way to continue to use the TaskUpdater class, the only way i found is to use queue directly in the executor, bypassing the task updater like this:

        if not artifact_id:
            artifact_id = str(uuid.uuid4())

        await self.event_queue.enqueue_event(
            TaskArtifactUpdateEvent(
                task_id=self.task_id,
                context_id=self.context_id,
                artifact=Artifact(
                    artifact_id=artifact_id,
                    name=name,
                    parts=parts,
                    metadata=metadata, 
                    extensions=["extension..."]
                ),
                append=append,
                last_chunk=last_chunk,
            )
        )

Additional context

No response

Code of Conduct

  • I agree to follow this project's Code of Conduct