ADK Streaming API

This page describes how ADK agents stream output to A2A clients through partial events and artifacts.

1. Partial Events — Real-time Text Streaming

Yield ADK events with partial=True to stream text chunks to the client in real time. Each chunk is forwarded as a transitory STREAM_DELTA artifact update:

from google.adk.events import Event
from google.adk.agents import BaseAgent
from google.genai import types


class MyAgent(BaseAgent):
    async def _run_async_impl(self, ctx):
        # Stream text in chunks
        for chunk in ["Hello", " world", "!"]:
            yield Event(
                author=self.name,
                content=types.Content(
                    role="model",
                    parts=[types.Part(text=chunk)],
                ),
                partial=True,
            )

        # Final non-partial event closes the stream
        yield Event(
            author=self.name,
            content=types.Content(
                role="model",
                parts=[types.Part(text="Hello world!")],
            ),
            partial=False,
        )

The STREAM_DELTA artifact is transitory — it is not persisted to the Task’s durable state. Partial and non-partial ADK event content may populate the SDK-managed response buffer for the current turn. If that buffer stays empty, Aion falls back to a2a_outbox and then framework-native fallback rules. See Message Mapping for outbound precedence rules.

2. Non-partial Events

When a non-partial ADK event is yielded:

The open STREAM_DELTA stream is closed (if streaming was active).
Content parts are emitted as a TaskStatusUpdateEvent(working, message=...).
Artifacts declared in event.actions.artifact_delta are loaded and forwarded as TaskArtifactUpdateEvent.

3. File and Data Artifacts

Use ctx.artifact_service to save file or data artifacts. Saved artifacts are automatically forwarded to the client as TaskArtifactUpdateEvent:

from google.adk.agents import BaseAgent
from google.adk.events import Event, EventActions
from google.genai import types


class MyAgent(BaseAgent):
    async def _run_async_impl(self, ctx):
        pdf_bytes = b"..."  # your PDF content

        artifact = types.Part(
            inline_data=types.Blob(mime_type="application/pdf", data=pdf_bytes)
        )
        version = await ctx.artifact_service.save_artifact(
            app_name=ctx.app_name,
            user_id=ctx.user_id,
            session_id=ctx.session.id,
            filename="report.pdf",
            artifact=artifact,
        )

        # Declare the saved artifact in the event so it is forwarded to the client
        yield Event(
            author=self.name,
            actions=EventActions(artifact_delta={"report.pdf": version}),
        )

Each entry in artifact_delta maps a filename to its saved version. Aion Server automatically forwards each declared artifact to the client as a TaskArtifactUpdateEvent.

4. Artifact Namespaces

The filename prefix controls artifact scope:

Prefix	Scope
`user:...`	User-scoped — shared across all sessions for this user
(none)	Session-scoped — private to the current context

Examples:

# Session-scoped artifact (default)
await ctx.artifact_service.save_artifact(..., filename="report.pdf", ...)

# User-scoped artifact (persists across sessions)
await ctx.artifact_service.save_artifact(..., filename="user:profile.json", ...)

For artifact storage backend configuration, see Artifact Storage.

Artifact Builders — url_artifact, file_artifact, data_artifact reference
Card — Card builder API reference

Python SDK

LangGraph

Google ADK

1. Partial Events — Real-time Text Streaming

2. Non-partial Events

3. File and Data Artifacts

4. Artifact Namespaces

​1. Partial Events — Real-time Text Streaming

​2. Non-partial Events

​3. File and Data Artifacts

​4. Artifact Namespaces

​Related Pages

1. Partial Events — Real-time Text Streaming

2. Non-partial Events

3. File and Data Artifacts

4. Artifact Namespaces

Related Pages