custom-agent-scaffold

custom-agent-scaffold

Build a custom ADK 2.0 agent by subclassing BaseAgent. Use when an LlmAgent is overkill (deterministic transformations, API calls, data validation) or when you need full control over the event stream.

When to use

• Pure-Python step inside an otherwise LLM-driven workflow
• Wrapping a non-LLM service (search index, vector DB, calculator)
• Custom event emission for telemetry
• Validation/guardrail steps that should run without an LLM call

Template

from typing import AsyncIterator
from google.adk.agents import BaseAgent
from google.adk.events import Event
from google.adk.agents.invocation_context import InvocationContext


class CsvExportAgent(BaseAgent):
    """Reads state['rows'], writes CSV to state['csv_path'], emits one Event."""

    name: str = "csv_exporter"

    async def _run_async_impl(
        self, ctx: InvocationContext
    ) -> AsyncIterator[Event]:
        rows = ctx.session.state.get("rows", [])
        if not rows:
            yield Event(
                author=self.name,
                content="No rows to export.",
            )
            return

        import csv
        from pathlib import Path
        out = Path("/tmp") / f"{ctx.session.id}.csv"
        with out.open("w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=rows[0].keys())
            writer.writeheader()
            writer.writerows(rows)

        ctx.session.state["csv_path"] = str(out)
        yield Event(
            author=self.name,
            content=f"Exported {len(rows)} rows to {out}",
        )

Wire into a workflow

from google.adk.agents import LlmAgent, SequentialAgent

extractor = LlmAgent(
    name="extractor",
    model="gemini-2.5-flash",
    instruction="Extract rows from the user input into state['rows'].",
    output_key="rows",
)

root_agent = SequentialAgent(
    name="extract_then_export",
    sub_agents=[extractor, CsvExportAgent()],
)

Validation

• Subclass must define name (str class attr or constructor)
• _run_async_impl must be async def and return AsyncIterator[Event]
• Yield at least one Event so the run UI shows the step
• Mutate ctx.session.state for downstream agents to consume