google-agents-cli-adk-code

# ADK Go Workflow API

> Requires `google.golang.org/adk/v2 >= v2.0.0`, which is where the `workflow` package and
> `agent/workflowagent` first ship.

The graph runtime in **`google.golang.org/adk/v2/workflow`**, for the cases the linear
`sequentialagent` / `parallelagent` / `loopagent` composition in `adk-go.md` cannot express:
conditional routing, fan-out with a fan-in barrier, per-node retries or timeouts, or a workflow
that pauses for human input and resumes later. For a plain chain of agents, those workflow
*agents* are simpler.

Read `adk-go.md` first for the base API.

**Official docs:** [Workflows overview](https://adk.dev/workflows/index.md) ·
[pkg.go.dev/workflow](https://pkg.go.dev/google.golang.org/adk/v2/workflow) ·
[runnable examples](https://github.com/google/adk-go/tree/main/examples/workflow)

## 1. Core concepts

A `Workflow` is a graph-based agent: nodes do work, edges define flow, `workflow.Start` is the entry point.

```go
import (
	"strings"

	"google.golang.org/adk/v2/agent"
	"google.golang.org/adk/v2/agent/workflowagent"
	"google.golang.org/adk/v2/workflow"
)

cfg := workflow.NodeConfig{RetryConfig: workflow.DefaultRetryConfig()}

upper := workflow.NewFunctionNode("upper",
	func(_ agent.Context, in string) (string, error) {
		return strings.ToUpper(in), nil
	}, cfg)

suffix := workflow.NewFunctionNode("suffix",
	func(_ agent.Context, in string) (string, error) {
		return in + " IS AWESOME!", nil
	}, cfg)

rootAgent, err := workflowagent.New(workflowagent.Config{
	Name:        "simple_sequence_workflow",
	Description: "Uppercases a string and appends a suffix",
	Edges:       workflow.Chain(workflow.Start, upper, suffix),
})
```

The first node receives the user's message as its input. `workflowagent.New` returns a plain
`agent.Agent`.

## Node kinds

Every constructor takes a `NodeConfig` (retries, timeout — see *Retries, timeouts, errors*) and
returns a value you drop into an `Edge`.

### `NewFunctionNode` — a typed transform

```go
upper := workflow.NewFunctionNode("upper",
	func(_ agent.Context, in string) (string, error) {
		return strings.ToUpper(in), nil
	}, workflow.NodeConfig{})
```

The return value becomes `Event.Output` and arrives as the successor's typed input.

### `NewEmittingFunctionNode` — emit events, then return

Use it when the node must produce something *besides* its return value: user-visible progress, a
state delta, a routing tag, or a HITL prompt.

```go
progress := workflow.NewEmittingFunctionNode("progress",
	func(ctx agent.Context, in string, emit func(*session.Event) error) (any, error) {
		ev := session.NewEvent(ctx, ctx.InvocationID())
		// Content renders in the UI; Output is what the next node receives.
		ev.Content = genai.NewContentFromText("working…", genai.RoleModel)
		if err := emit(ev); err != nil {
			return nil, err
		}
		// Returning a non-nil value emits a terminal event carrying it as Output.
		// Returning nil instead suppresses that terminal event entirely — which is
		// what you want when an event you already emitted carries the output.
		return in, nil
	}, workflow.NodeConfig{})
```

**What a function node's return value becomes:** a `*session.Event` is yielded as-is (this is how
you set `Event.Routes`), a `*genai.Content` becomes `event.Content`, and anything else becomes
`event.Output`.

### `NewAgentNode` — wrap an agent

```go
drafter, err := workflow.NewAgentNode(draftAgent, workflow.NodeConfig{})
// NewAgentNodeTyped[In, Out](draftAgent, cfg) instead reflects In/Out into JSON
// schemas, so the agent's input and reply are validated against your structs.
```

An `LlmAgent` with unset `Mode` defaults to single-turn here. Register the wrapped agent in
`workflowagent.Config.SubAgents` so event authors resolve.

### `NewToolNode` — call a tool as a step

```go
lookup, err := workflow.NewToolNode(weatherTool, workflow.NodeConfig{})
// NewNamedToolNode("weather_step", weatherTool, cfg) to override the node name,
// which otherwise comes from the tool.
```

### `NewWorkflowNode` — nest a sub-workflow

```go
sub, err := workflow.NewWorkflowNode("subflow", workflow.Chain(workflow.Start, upper, suffix))
```

`NewJoinNode`, `NewDynamicNode` and `NewParallelWorker` are the concurrency kinds — see
*Parallelism and fan-in*.

## Edges and routing

An edge with no `Route` always fires. A routed edge fires only when the source node tagged its
event with a matching value:

```go
edges := []workflow.Edge{
	{From: workflow.Start, To: triage},
	{From: triage, To: answer, Route: workflow.StringRoute("question")},
	{From: triage, To: escalate, Route: workflow.IntRoute(2)},
	{From: triage, To: archive, Route: workflow.BoolRoute(false)},
	{From: triage, To: urgent, Route: workflow.MultiRoute[string]{"p0", "p1"}},
	{From: triage, To: fallback, Route: workflow.Default},
}
```

Every route is compared as a string against the entries in `Event.Routes`, so `IntRoute(2)`
matches the entry `"2"` and `BoolRoute(false)` matches `"false"`. Any type with a
`Matches(*session.Event) bool` method works as a route.

**A node signals its branch by setting `Event.Routes []string`** on an event it emits:

```go
func classifyAndRoute(ctx agent.Context, msg string, emit func(*session.Event) error) (any, error) {
	ev := session.NewEvent(ctx, ctx.InvocationID())
	ev.Routes = []string{classify(msg)} // e.g. "question"
	ev.Output = msg                     // feeds the successor's typed input
	if err := emit(ev); err != nil {
		return nil, err
	}
	return nil, nil // nil output suppresses the default terminal event
}

triage := workflow.NewEmittingFunctionNode("triage", classifyAndRoute, workflow.NodeConfig{})
```

### Building edge sets

`EdgeBuilder` is the readable way to express fan-out and fan-in:

```go
eb := workflow.NewEdgeBuilder()
eb.Add(workflow.Start, triage)
eb.AddRoute(triage, answer, workflow.StringRoute("question"))
eb.AddRoutes(triage, map[string]workflow.Node{ // shorthand for several StringRoutes
	"statement":   comment,
	"exclamation": react,
})
eb.AddFanOut(workflow.Start, alpha, beta, gamma) // one source, many targets
eb.AddFanIn(gather, alpha, beta, gamma)          // target FIRST, then its predecessors
edges := eb.Build()
```

`workflow.Chain(a, b, c)` builds the unconditional edges of a straight line, and returns nil for
fewer than two nodes. `workflow.Concat(...)` flattens `Edge` values and `[]Edge` slices into one
slice — handy for gluing a chain onto a routed fan-out — and silently drops anything else.

### Dispatch rules

Which successors fire is decided per activation, and the edge set is the clearest place to see it:

```go
edges := []workflow.Edge{
	{From: workflow.Start, To: triage},

	// Unconditional: always fires, whatever the node tagged.
	{From: triage, To: audit},

	// Concrete routes: fire only on a matching tag in Event.Routes.
	{From: triage, To: answer, Route: workflow.StringRoute("question")},
	{From: triage, To: urgent, Route: workflow.MultiRoute[string]{"p0", "p1"}},

	// Default fires when no *concrete* route matched. The unconditional edge
	// above does NOT count as a match, so a run tagged "other" fires both
	// `audit` and `fallback` — and if both then end the graph with output, the
	// run fails with ErrMultipleTerminalOutputs. Converge them on a JoinNode.
	{From: triage, To: fallback, Route: workflow.Default},
}

// Two edges to the same target within one activation are deduplicated.
// If every outgoing edge is routed and none match, with no Default, the graph
// silently dead-ends here — treated as a deliberate decision, not an error.
```

Inside the node body the corresponding limits are one routing event and one output per
activation:

```go
func routeOnce(ctx agent.Context, msg string, emit func(*session.Event) error) (any, error) {
	ev := session.NewEvent(ctx, ctx.InvocationID())
	ev.Routes = []string{"question"} // a second event with Routes -> ErrMultipleRoutingEvents
	ev.Output = msg                  // a second event with Output -> ErrMultipleOutputs
	if err := emit(ev); err != nil {
		return nil, err
	}
	return nil, nil
}
```

## Parallelism and fan-in

Three separate mechanisms — pick deliberately.

### (a) Graph fan-out to a `JoinNode`

Multiple edges from one source run concurrently. The join activates exactly once, after every
declared predecessor completes, and its input is a `map[string]any` keyed by predecessor node name.

```go
alpha := workflow.NewFunctionNode("alpha",
	func(_ agent.Context, _ string) (string, error) { return "A", nil }, workflow.NodeConfig{})
beta := workflow.NewFunctionNode("beta",
	func(_ agent.Context, _ string) (string, error) { return "B", nil }, workflow.NodeConfig{})

gather := workflow.NewJoinNode("gather")
format := workflow.NewFunctionNode("format_summaries",
	func(_ agent.Context, gathered map[string]any) (string, error) {
		return fmt.Sprintf("alpha=%v beta=%v", gathered["alpha"], gathered["beta"]), nil
	}, workflow.NodeConfig{})

eb := workflow.NewEdgeBuilder()
eb.AddFanOut(workflow.Start, alpha, beta)
eb.AddFanIn(gather, alpha, beta) // target FIRST, then its predecessors
eb.Add(gather, format)

joinAgent, err := workflowagent.New(workflowagent.Config{
	Name: "fan_in", Edges: eb.Build(),
}) // one turn yields "alpha=A beta=B"
```

A predecessor that completed with no output contributes `nil` — the barrier counts completions,
not outputs. **Never route conditionally into a `JoinNode`**: the barrier waits for every declared
predecessor, and a route-skipped one never fires, so the graph hangs.

### (b) `ParallelWorker` — map one node over a slice

One goroutine per item, bounded by the worker's own concurrency limit. It yields a single event
whose `Output` is a positionally ordered `[]any`.

```go
// The wrapped node handles ONE item; the worker runs it over the whole slice.
score := workflow.NewFunctionNode("score",
	func(_ agent.Context, city string) (string, error) {
		return city + ":ok", nil
	}, workflow.NodeConfig{})

fanout, err := workflow.NewParallelWorker("score_all", score, 4, workflow.NodeConfig{})
if err != nil {
	return nil, err
}

split := workflow.NewFunctionNode("split",
	func(_ agent.Context, in string) ([]string, error) {
		return strings.Fields(in), nil // the slice the worker maps over
	}, workflow.NodeConfig{})

join := workflow.NewFunctionNode("join_results",
	func(_ agent.Context, results []any) (string, error) {
		parts := make([]string, len(results))
		for i, r := range results {
			parts[i] = fmt.Sprint(r) // positional: input order preserved
		}
		return strings.Join(parts, " "), nil
	}, workflow.NodeConfig{})

workerAgent, err := workflowagent.New(workflowagent.Config{
	Name:  "parallel_worker",
	Edges: workflow.Chain(workflow.Start, split, fanout, join),
}) // "paris rome" yields "paris:ok rome:ok"
```

Fail-fast: the first non-retryable error cancels in-flight workers. The wrapped node must not
carry a `RetryConfig` — retries belong to the worker, applied per item.

### (c) Dynamic nodes — orchestrate in Go

A dynamic node calls `RunNode` per child instead of declaring a static edge set, so the fan-out
can depend on the input. Children run concurrently if you launch them concurrently.

```go
hello := workflow.NewFunctionNode("hello_node",
	func(_ agent.Context, name string) (string, error) {
		return "Hello " + name, nil
	}, workflow.NodeConfig{})

orchestrator := workflow.NewDynamicNode[string, string]("orchestrate",
	func(ctx agent.Context, in string, _ func(*session.Event) error) (string, error) {
		// Decide at run time how many children to run, and with what input.
		var out []string
		for _, name := range strings.Fields(in) {
			got, err := workflow.RunNode[string](ctx, hello, name)
			if err != nil {
				return "", err
			}
			out = append(out, got)
		}
		return strings.Join(out, "; "), nil
	}, workflow.NodeConfig{})

dynAgent, err := workflowagent.New(workflowagent.Config{
	Name:  "dynamic",
	Edges: workflow.Chain(workflow.Start, orchestrator),
}) // "ada grace" yields "Hello ada; Hello grace"
```

`WithMaxConcurrency(n)` caps graph-scheduled nodes per run; nodes over the cap queue FIFO. It
deliberately does **not** apply to dynamic sub-nodes invoked via `RunNode`, which are awaited
inline by the parent — gating them would deadlock.

## Passing data between nodes

**Edge-passed values are `any`.** Typing is per-node generics, not a typed graph: nothing checks
at compile time that one node's output matches the next node's input. The engine reconciles them
at run time — it type-asserts the incoming `any` to `IN`, and on failure falls back to a JSON
round-trip.

That fallback is what lets an untyped payload land in a typed parameter. Here the producer knows
nothing about `Itinerary`, and it still arrives populated:

```go
type Itinerary struct {
	Origin string `json:"origin"`
	Days   int    `json:"days"`
}

produce := workflow.NewFunctionNode("produce",
	func(_ agent.Context, _ string) (map[string]any, error) {
		return map[string]any{"origin": "SFO", "days": 3}, nil
	}, workflow.NodeConfig{})

consume := workflow.NewFunctionNode("consume",
	func(_ agent.Context, in Itinerary) (string, error) {
		return fmt.Sprintf("origin=%s days=%d", in.Origin, in.Days), nil // origin=SFO days=3
	}, workflow.NodeConfig{})
```

**Session state** is the other channel. Write via `Event.Actions.StateDelta`:

```go
ev := session.NewEvent(ctx, ctx.InvocationID())
ev.Actions.StateDelta = map[string]any{"progress": "halfway"}
if err := emit(ev); err != nil {
	return nil, err
}
```

Read it back through the context:

```go
lang, err := ctx.Session().State().Get("progress")
```

`Input`, `Output` and interrupt payloads must be **JSON-encodable** to survive pause/resume.
For binary data, store an artifact and carry its URI.

## Retries, timeouts, errors

`NodeConfig` carries the per-node execution policy:

```go
retry := workflow.DefaultRetryConfig() // 5 attempts, 1s → 60s, factor 2.0, jitter 1.0
retry.MaxAttempts = 3
retry.ShouldRetry = func(err error) bool { return !errors.Is(err, errFatal) }

cfg := workflow.NodeConfig{
	RetryConfig: retry,
	Timeout:     30 * time.Second, // bounds a single activation

	// RerunOnResume: &rerun, // &true re-runs the node after HITL; nil/&false hands off
	// WaitForOutput: &wait,  // park until every predecessor has triggered this node
	// EmitsOwnSpan:  true,   // the node emits its own tracing span
}
```

`DefaultRetryConfig()` does not retry `ErrInputValidation` — the input is deterministic per
activation, so retrying cannot help.

**Construction-time validation** is strict, so a malformed graph fails at `New` rather than
mid-run:

1. `Start` must exist and have no incoming edges (`ErrNoStartNode`, `ErrNodePointsToStart`)
2. All nodes must be reachable from `Start` (`ErrNodesNotReachable`)
3. No duplicate node names (`ErrDuplicateNodeName`), and no duplicate `(From, To)` pairs
   regardless of route (`ErrDuplicateEdge`) — use `MultiRoute` instead
4. At most one `Default` route per node (`ErrMultipleDefaultRoutes`)
5. No unconditional cycles — a cycle needs at least one routed edge (`ErrUnconditionalCycle`)
6. Two or more unconditional incoming edges on a non-`JoinNode` target is rejected
   (`ErrUnsupportedFanIn`): "use a JoinNode to converge branches"

## Human-in-the-loop

HITL pauses a run to ask the user something and resumes it on a later turn. 

**Two resume modes**, chosen by `NodeConfig.RerunOnResume`.

**Handoff** (`nil` or `&false`) — the reply is routed to the asker's *successors* as their
input, and the asker does not re-run:

```go
ask := workflow.NewEmittingFunctionNode[any, any]("ask_name",
	func(ctx agent.Context, _ any, emit func(*session.Event) error) (any, error) {
		if err := emit(workflow.NewRequestInputEvent(ctx, session.RequestInput{
			InterruptID: "ask_name-" + uuid.NewString(),
			Message:     "What's your name?",
		})); err != nil {
			return nil, err
		}
		return nil, workflow.ErrNodeInterrupted
	},
	workflow.NodeConfig{},
)

greet := workflow.NewFunctionNode("greet",
	func(_ agent.Context, name string) (string, error) {
		return fmt.Sprintf("Hello, %s!", name), nil
	}, workflow.NodeConfig{})

edges := workflow.Chain(workflow.Start, ask, greet) // the reply becomes greet's input
```

**Re-entry** (`&true`) — the asker re-runs from scratch. `ResumeOrRequestInput` collapses both
passes into one call: it asks and pauses the first time, and returns the human's reply the
second:

```go
rerun := true
greet := workflow.NewEmittingFunctionNode[any, any]("greet",
	func(nc agent.Context, _ any, emit func(*session.Event) error) (any, error) {
		reply, err := workflow.ResumeOrRequestInput(nc, emit, session.RequestInput{
			// Stable across this run's re-entry, unique per run.
			InterruptID: "ask_name-" + nc.InvocationID(),
			Message:     "What's your name?",
		})
		if err != nil {
			return nil, err
		}
		name, _ := reply.(string)
		return fmt.Sprintf("Hello, %s!", name), nil
	},
	workflow.NodeConfig{RerunOnResume: &rerun},
)
```

`workflowagent` wires resumption automatically: on each turn it inspects inbound content for
`adk_request_input` function responses, rebuilds the run state from session history, and
dispatches to `Workflow.Resume` instead of `Workflow.Run`. Run state lives in `session.State`,
not on the agent, so one agent instance safely serves many concurrent sessions.

A resume consumes the pending request. Submitting the same response twice does not re-run the
node, but the second call yields `workflow.ErrNothingToResume`.
Handoff successors are matched with no event, so a successor reachable only via a *concrete*
route does **not** fire on resume — only unconditional edges and `Default`.

## See also

- `adk-go.md` — the base ADK Go API (agents, tools, models, sessions, runner)

Upstream examples: [`examples/workflow/`](https://github.com/google/adk-go/tree/main/examples/workflow)
— `basic`, `routing/string`, `hitl_simple`, `hitl_rerun`, `dynamic/basic`, `complex`.

# ADK Go API reference

> Reflects `google.golang.org/adk/v2 v2.1.0`, the version the `adk_go` template pins. If a symbol
> here is missing, check your `go.mod` before assuming the page is wrong.

## 1. Core Concepts & Project Structure

### Essential Primitives

*   **`Agent`**: The core intelligent unit. Built with `llmagent.New` (LLM-driven) or `agent.New` (custom `Run` function).
*   **`Tool`**: Callable capability given to an agent.
*   **`Session`**: A stateful conversation thread with history (`Events()`) and short-term memory (`State()`).
*   **`State`**: Key-value store within a `Session` for transient conversation data.
*   **`Runner`**: The execution engine; drives an agent and yields its event stream.
*   **`Event`**: Atomic unit of communication; carries content and side-effect `Actions`.

### Scaffolded Project Layout

```text
my-agent/
├── app/
│   ├── agent.go             // NewRootAgent(ctx) (agent.Agent, error) — model, instruction, tools
│   └── agent_test.go
├── appinfo/                 // template-owned: serves {prefix}/apps/{app}/app-info for agents-cli eval
├── e2e/
│   ├── integration/         // server_e2e_test.go — drives a running server
│   └── load_test/
├── deployment/terraform/    // only with a deployment target; absent in prototype mode
├── main.go                  // telemetry, launcher.Config, launcher wiring
├── appurl.go                // resolveAppURL() — the base URL in the A2A agent card
├── Dockerfile
├── Makefile
├── agents-cli-manifest.yaml // language: go, agent_directory: app, create_params
├── .env / .env.example
├── .golangci.yml
└── go.mod                   // module <project-name>
```

`app/agent_test.go` tests the sample `GetWeather` tool. When you replace the sample tools in
`app/agent.go`, rewrite those tests for your tools in the same change: `go build` and
`agents-cli build` skip `_test.go` files, so only `go test ./...` shows the breakage.

## 2. Agents

### Basic Setup

```go
import (
	"context"

	"google.golang.org/genai"

	"google.golang.org/adk/v2/agent"
	"google.golang.org/adk/v2/agent/llmagent"
	"google.golang.org/adk/v2/model/gemini"
	"google.golang.org/adk/v2/tool"
	"google.golang.org/adk/v2/tool/functiontool"
)

type WeatherArgs struct {
	City string `json:"city" jsonschema:"City name to look up"`
}
type WeatherResult struct {
	Report string `json:"report"`
}

func GetWeather(ctx agent.Context, in WeatherArgs) (WeatherResult, error) {
	return WeatherResult{Report: "sunny in " + in.City}, nil
}

func NewRootAgent(ctx context.Context) (agent.Agent, error) {
	model, err := gemini.NewModel(ctx, "gemini-3.8-flash", &genai.ClientConfig{
		Backend: genai.BackendVertexAI,
	})
	if err != nil {
		return nil, err
	}

	weatherTool, err := functiontool.New(functiontool.Config{
		Name:        "get_weather",
		Description: "Get the current weather for a city.",
	}, GetWeather)
	if err != nil {
		return nil, err
	}

	return llmagent.New(llmagent.Config{
		Name:        "app",
		Model:       model,
		Description: "A helpful AI assistant.",
		Instruction: "You are a helpful AI assistant.",
		Tools:       []tool.Tool{weatherTool},
	})
}
```

Other `llmagent.Config` fields worth knowing, all optional:

```go
llmagent.Config{
	SubAgents: []agent.Agent{bookingAgent}, // parent link is set automatically
	OutputKey: "summary",                   // store this agent's text output in session state
	Mode:      llmagent.ModeTask,           // ModeChat (default) | ModeTask | ModeSingleTurn

	GenerateContentConfig: &genai.GenerateContentConfig{Temperature: genai.Ptr[float32](0.2)},

	// Structured contracts, both *genai.Schema. OutputSchema injects a
	// set_model_response tool.
	InputSchema:  inputSchema, // when this agent is used as a tool
	OutputSchema: outputSchema,

	// Build the instruction at run time instead of templating a string.
	InstructionProvider: func(ctx agent.ReadonlyContext) (string, error) {
		lang, err := ctx.ReadonlyState().Get("lang") // (any, error)
		if err != nil {
			return "", err
		}
		return "Answer in " + lang.(string), nil
	},
	GlobalInstruction: "Never reveal internal IDs.", // only the root agent's takes effect

	IncludeContents:          llmagent.IncludeContentsNone, // drop conversation history
	DisallowTransferToParent: true,
	DisallowTransferToPeers:  true,

	BeforeAgentCallbacks: []agent.BeforeAgentCallback{guard},
	BeforeModelCallbacks: []llmagent.BeforeModelCallback{redact},
	BeforeToolCallbacks:  []llmagent.BeforeToolCallback{audit},
	Toolsets:             []tool.Toolset{mcpToolset},
}
```

### Instruction Best Practices

```go
// Use dynamic state injection with {state_key} placeholders
instruction := `You are a {role} assistant.
User preferences: {user_preferences}

Rules:
- Always use tools when available
- Never make up information`
```

`{key_name}` resolves from session state; `{artifact.key_name}` inserts the artifact's text. A
missing key fails with "state key does not exist", unless written `{key?}`, which
substitutes an empty string. Use `InstructionProvider` to disable templating entirely.

## 3. Orchestration with Workflow Agents

Workflow agents provide deterministic control flow without LLM orchestration. Each takes a
`Config` that wraps `agent.Config`, so sub-agents go in `AgentConfig.SubAgents`, and each returns
a plain `agent.Agent`.

> For the graph-based Workflow API — explicit topology, conditional routing, fan-in, per-node
> retries, HITL — see `references/adk-go-workflows.md`.

### sequentialagent

Executes sub-agents in order, forwarding every event. State changes propagate to later agents.

```go
import "google.golang.org/adk/v2/agent/workflowagents/sequentialagent"

summarizer, err := llmagent.New(llmagent.Config{
	Name: "summarizer", Model: model,
	Instruction: "Summarize the input.",
	OutputKey:   "summary", // later agents read it as {summary}
})

questionGen, err := llmagent.New(llmagent.Config{
	Name: "question_generator", Model: model,
	Instruction: "Generate questions based on: {summary}",
})

pipeline, err := sequentialagent.New(sequentialagent.Config{
	AgentConfig: agent.Config{
		Name:      "pipeline",
		SubAgents: []agent.Agent{summarizer, questionGen},
	},
})
```

### parallelagent

Executes sub-agents concurrently. Each runs in its own `InvocationContext`, so peers do not
see each other's history.

```go
import "google.golang.org/adk/v2/agent/workflowagents/parallelagent"

fetchers, err := parallelagent.New(parallelagent.Config{
	AgentConfig: agent.Config{
		Name:      "fetchers",
		SubAgents: []agent.Agent{fetchA, fetchB}, // OutputKey: "data_a" / "data_b"
	},
})

merger, err := llmagent.New(llmagent.Config{
	Name: "merger", Model: model,
	Instruction: "Combine data_a: {data_a} and data_b: {data_b}",
})

full, err := sequentialagent.New(sequentialagent.Config{
	AgentConfig: agent.Config{Name: "full_pipeline", SubAgents: []agent.Agent{fetchers, merger}},
})
```

### loopagent

Loops the sub-agent.

```go
import (
	"google.golang.org/adk/v2/agent/workflowagents/loopagent"
	"google.golang.org/adk/v2/tool/exitlooptool"
)

exit, err := exitlooptool.New()

refiner, err := llmagent.New(llmagent.Config{
	Name: "refiner", Model: model,
	Instruction: "Improve the draft. Call exit_loop when it is good enough.",
	Tools:       []tool.Tool{exit},
})

refinementLoop, err := loopagent.New(loopagent.Config{
	AgentConfig:   agent.Config{Name: "refinement_loop", SubAgents: []agent.Agent{refiner}},
	MaxIterations: 5,
})
```

### Custom agents

```go
router, err := agent.New(agent.Config{
	Name: "conditional_router",
	Run: func(ictx agent.InvocationContext) iter.Seq2[*session.Event, error] {
		return func(yield func(*session.Event, error) bool) {
			tier, _ := ictx.Session().State().Get("user_type")

			chosen := regularAgent
			if tier == "premium" {
				chosen = premiumAgent
			}
			for ev, err := range chosen.Run(ictx) {
				if !yield(ev, err) {
					return
				}
			}
		}
	},
})
```

## 4. Tools

### Function tools

A tool is a Go function plus a name and description. The argument and result types drive the
schema the model sees:

```go
import "google.golang.org/adk/v2/tool/functiontool"

type weatherIn struct {
	City string `json:"city" jsonschema:"City name to look up"`
}
type weatherOut struct {
	Report string `json:"report"`
}

func getWeather(ctx agent.Context, in weatherIn) (weatherOut, error) {
	return weatherOut{Report: "sunny in " + in.City}, nil
}

weatherTool, err := functiontool.New(functiontool.Config{
	Name:        "get_weather",
	Description: "Get the current weather for a city.",
}, getWeather)
```

The remaining `functiontool.Config` fields, all optional:

```go
functiontool.Config{
	// Both *jsonschema.Schema here.
	InputSchema:   argsSchema,   // override the schema reflected from TArgs
	OutputSchema:  resultSchema, // override the schema reflected from TResults
	IsLongRunning: true,         // the tool returns before its work finishes
}
```

### `Context`

Passed to each tool. It embeds `context.Context`.

```go
func bookFlight(ctx agent.Context, in bookIn) (bookOut, error) {
	// Session state. Get returns ErrStateKeyNotExist when the key is absent.
	city, err := ctx.State().Get("home_city")
	if err != nil {
		return bookOut{}, err
	}
	ctx.State().Set("last_booking", in.FlightID)
	_ = city

	hits, err := ctx.SearchMemory(ctx, "past trips to "+in.Dest) // long-term memory
	if err != nil {
		return bookOut{}, err
	}

	ctx.Actions().SkipSummarization = true // control the run loop
	_ = ctx.Artifacts()                    // read/write artifacts
	_ = ctx.InvocationID()                 // identifiers: also UserID, SessionID, AppName, AgentName, Branch

	timeoutCtx, cancel := ctx.WithAgentTimeout(30 * time.Second)
	defer cancel()
	_ = timeoutCtx

	return bookOut{Confirmed: len(hits.Memories) > 0}, nil
}
```

### Built-in tools

| Import (`google.golang.org/adk/v2/tool/...`) | Constructor | Tool name |
|---|---|---|
| `geminitool` | `geminitool.GoogleSearch{}` | `google_search` |
| `geminitool` | `New(name, description string, t *genai.Tool) tool.Tool` | caller-chosen — wraps any `genai.Tool` |
| `exitlooptool` | `New() (tool.Tool, error)` | `exit_loop` |
| `loadmemorytool` | `New()` | `load_memory` |
| `preloadmemorytool` | `New()` | `preload_memory` |
| `loadartifactstool` | `New()` | `load_artifacts` |
| `agenttool` | `New(a agent.Agent, cfg *Config) tool.Tool` | the wrapped agent's name |
| `exampletool` | `New(config ExampleToolConfig)` | few-shot examples |
| `mcptoolset` | `New(cfg Config) (tool.Toolset, error)` | from the MCP server |
| `skilltoolset` | `New(ctx, cfg Config) (*SkillToolset, error)` | includes `load_skill` |


### Tool confirmation

Gate a tool behind human approval before it runs — statically, or per call.

```go
// Always ask.
deleteTool, err := functiontool.New(functiontool.Config{
	Name: "delete_record", Description: "Deletes a record.",
	RequireConfirmation: true,
}, deleteRecord)

// Ask only for large transfers.
transferTool, err := functiontool.New(functiontool.Config{
	Name: "transfer_money", Description: "Transfers money.",
	RequireConfirmationProvider: func(in transferIn) bool { return in.Amount > 1000 },
}, transferMoney)
```

### Human-in-the-loop from a tool

```go
func transferMoney(ctx agent.Context, in transferIn) (transferOut, error) {
	if in.Amount > 1000 {
		if err := ctx.RequestConfirmation(
			fmt.Sprintf("Approve transfer of $%.2f to %s?", in.Amount, in.Payee),
			in, // payload the UI renders alongside the prompt
		); err != nil {
			return transferOut{}, err
		}
		return transferOut{Status: "awaiting_approval"}, nil
	}
	return transferOut{Status: "sent"}, nil
}
```

The reply arrives as a `toolconfirmation.ToolConfirmation`
(`google.golang.org/adk/v2/tool/toolconfirmation`).

### MCP tools

Connect to an MCP server and expose its tools as a toolset. Set `Endpoint` for a remote
streamable-HTTP server, or supply your own `Transport` (for example a stdio command) instead.

```go
import (
	"google.golang.org/adk/v2/tool/mcptoolset"
	"google.golang.org/adk/v2/auth"
)

mcpTools, err := mcptoolset.New(mcptoolset.Config{
	Endpoint: "https://my-mcp-server.example.com/mcp",
	Auth:     auth.ADC(), // see Tool authentication below
})

rootAgent, err := llmagent.New(llmagent.Config{
	Name: "assistant", Model: model,
	Toolsets: []tool.Toolset{
		tool.FilterToolset(mcpTools, tool.AllowedToolsPredicate([]string{"list_directory", "read_file"})),
	},
})
```

`mcptoolset.Config` also takes `RequireConfirmation` / `RequireConfirmationProvider`, which apply
to every tool in the set.

### Tool authentication

`auth.CredentialProvider` supplies a credential to every outgoing request of an authenticated
toolset. The `auth` package ships the common providers.

| Auth type | Provider |
|---|---|
| Static bearer token | `auth.StaticToken(token)` |
| API key | `auth.APIKey(name, value)` |
| Application Default Credentials | `auth.ADC(scopes...)` |
| Service account | `auth.ServiceAccount(auth.ServiceAccountConfig{...})` |
| Any `oauth2.TokenSource` | `auth.TokenSourceProvider(ts)` |
| Custom | `auth.ProviderFunc(func(ctx) (auth.Credential, error))` |

```go
mcpTools, err := mcptoolset.New(mcptoolset.Config{
	Endpoint: "https://my-mcp-server.example.com/mcp",
	Auth:     auth.ADC("https://www.googleapis.com/auth/cloud-platform"),
})
```

There is no OpenAPI toolset in ADK Go — wrap the calls you need as function tools.

## 5. Models

Only the constructor differs between providers; agents, tools, the runner and the launcher are
identical downstream. Pick one and assign it to `llmagent.Config.Model`.

### Google Gemini (default)

```go
import "google.golang.org/adk/v2/model/gemini"

// Gemini platform (prod) — set GOOGLE_CLOUD_PROJECT and GOOGLE_CLOUD_LOCATION.
model, err := gemini.NewModel(ctx, "gemini-3.8-flash", &genai.ClientConfig{
	Backend: genai.BackendVertexAI,
})
```

```go
// AI Studio (dev) — set an API key instead.
model, err := gemini.NewModel(ctx, "gemini-3.8-flash", &genai.ClientConfig{
	APIKey: os.Getenv("GOOGLE_API_KEY"),
})
```

### OpenAI and OpenAI-compatible endpoints

```go
import "google.golang.org/adk/v2/model/openaimodel"

model, err := openaimodel.NewModel(ctx, "gpt-5", &openaimodel.ClientConfig{
	APIKey:  os.Getenv("OPENAI_API_KEY"),
	BaseURL: os.Getenv("OPENAI_BASE_URL"),
})
```

### Apigee proxy

```go
import "google.golang.org/adk/v2/model/apigee"

// The name must carry the apigee/ prefix.
model, err := apigee.NewModel(ctx, "apigee/gemini-3.8-flash", apigee.WithProxyURL(proxyURL))
```

> Name the variable something other than `model` in a file that also refers to the `model`
> package — a callback signature such as `func(agent.Context, *model.LLMRequest)` will not resolve
> if a variable shadows it.

Provider guides: [Anthropic](https://adk.dev/agents/models/anthropic/index.md), [Ollama](https://adk.dev/agents/models/ollama/index.md), [vLLM](https://adk.dev/agents/models/vllm/index.md), [LiteLLM](https://adk.dev/agents/models/litellm/index.md)

## 6. Sessions, state, memory, artifacts

A **session** is one conversation thread: its event history plus a key-value **state** store.
**Memory** is the long-term store that outlives any single session, and **artifacts** hold binary
data too large for state.

| Need | Solution |
|---|---|
| Within one conversation (task data, form state) | Session state — see [State prefixes](#state-prefixes) below |
| Across conversations (remember preferences, learn over time) | Memory — see [Memory](#memory) below |
| Binary blobs (PDFs, images, audio) | Artifacts — `artifact.InMemoryService()` or `artifact/gcsartifact` |

Services are passed once, to the runner or `launcher.Config`, and the framework uses them for the
whole run:

```go
config := &launcher.Config{
	AgentLoader:     agent.NewSingleLoader(rootAgent),
	SessionService:  session.InMemoryService(),
	ArtifactService: artifact.InMemoryService(),
	MemoryService:   memory.InMemoryService(),
}
```

Session service implementations: `session.InMemoryService()` (dev), `session/database` (sqlite),
`session/vertexai`.

### Reading and writing state from a tool

```go
func logActivity(ctx agent.Context, in Params) (Result, error) {
	var log []LogEntry
	if v, err := ctx.State().Get("activity_log"); err == nil {
		log, _ = v.([]LogEntry)
	}
	log = append(log, LogEntry{Message: in.Message})
	if err := ctx.State().Set("activity_log", log); err != nil {
		return Result{}, err
	}
	return Result{Status: "logged"}, nil
}
```

### State prefixes

```go
const (
	KeyPrefixApp  = "app:"  // shared across all users and sessions
	KeyPrefixTemp = "temp:" // current invocation only; stripped on AppendEvent
	KeyPrefixUser = "user:" // tied to user_id across that user's sessions
)
```

An unprefixed key is session-scoped.

### Memory

From inside a tool use `ctx.SearchMemory(ctx, query)`; app/user scoping is
already bound.

#### `memory.InMemoryService()` (dev)

```go
import "google.golang.org/adk/v2/memory"

memSvc := memory.InMemoryService()

err := memSvc.AddSessionToMemory(ctx, sess)
resp, err := memSvc.SearchMemory(ctx, &memory.SearchRequest{
	AppName: "my-agent", UserID: "u1", Query: "past trips",
})
```

#### `memory/vertexai` — Memory Bank (production)

```go
import memoryvertexai "google.golang.org/adk/v2/memory/vertexai"

memSvc, err := memoryvertexai.NewService(ctx, &memoryvertexai.ServiceConfig{
	AgentEngineData:               engineData, // project, location, Agent Runtime ID
	StateKeySessionLastUpdateTime: "last_memory_update",
	WaitForCompletion:             false,
})
```

To surface memories to the model, add `preloadmemorytool.New()` (injects them at the start of each
turn) or `loadmemorytool.New()` (the model calls it on demand).

Set `StateKeySessionLastUpdateTime` to generate memories only from events newer
than that timestamp, leave it empty to use the whole session.

## 7. Callbacks and plugins

Short-circuit the model when a cached answer will do, and redact tool arguments before they run:

```go
rootAgent, err := llmagent.New(llmagent.Config{
	Name:  "assistant",
	Model: llm,

	BeforeModelCallbacks: []llmagent.BeforeModelCallback{
		func(ctx agent.Context, req *model.LLMRequest) (*model.LLMResponse, error) {
			if cached, ok := lookupCache(req); ok {
				return cached, nil // returning non-nil replaces the model call
			}
			return nil, nil // nil, nil continues to the model
		},
	},

	BeforeToolCallbacks: []llmagent.BeforeToolCallback{
		func(ctx agent.Context, t tool.Tool, args map[string]any) (map[string]any, error) {
			delete(args, "ssn") // mutate in place, then continue
			return nil, nil
		},
	},
})
```

Other callbacks:

- **Model and tool callbacks are `llmagent` only**: `BeforeModelCallbacks`, `AfterModelCallbacks`,
  `OnModelErrorCallbacks`, `BeforeToolCallbacks`, `AfterToolCallbacks`, `OnToolErrorCallbacks`.
  All follow the shape above.
- **Agent lifecycle callbacks work on every agent kind**: `agent.BeforeAgentCallback` and
  `agent.AfterAgentCallback` take just an `agent.Context` and return `(*genai.Content, error)`.

Within a group, callbacks run **in the order provided**; the first to return a non-nil result or
error stops the rest and replaces the underlying call. To mutate tool args and still run the tool,
edit `args` in place and return `(nil, nil)`.

### Plugins

Global callback hooks across all agents/tools/LLMs. Use for cross-cutting concerns (logging, guardrails); use callbacks for per-agent logic.

```go
import "google.golang.org/adk/v2/plugin"

logging, err := plugin.New(plugin.Config{
	Name: "audit",
	OnUserMessageCallback: func(ictx agent.InvocationContext, msg *genai.Content) (*genai.Content, error) {
		log.Printf("user: %v", msg)
		return nil, nil // nil content leaves the message unchanged
	},
	AfterToolCallback: func(ctx agent.Context, t tool.Tool, args, result map[string]any, err error) (map[string]any, error) {
		log.Printf("tool %s -> %v", t.Name(), result)
		return nil, nil
	},
	CloseFunc: func() error { return nil },
})

r, err := runner.New(runner.Config{
	AppName:        "my-agent",
	Agent:          rootAgent,
	SessionService: session.InMemoryService(),
	PluginConfig:   runner.PluginConfig{Plugins: []*plugin.Plugin{logging}},
})
```

Bundled plugins: `plugin/loggingplugin`, `plugin/functioncallmodifier`, `plugin/retryandreflect`.

## 8. Serving the agent over HTTP

### Launcher

Serves the standard surface, driven by command-line keywords. This is what the scaffolded template
does.

```go
config := &launcher.Config{
	AgentLoader:    agent.NewSingleLoader(rootAgent),
	SessionService: session.InMemoryService(),
}

l := universal.NewLauncher(
	console.NewLauncher(),
	web.NewLauncher(
		webui.NewLauncher(), a2a.NewLauncher(), pubsub.NewLauncher(),
		eventarc.NewLauncher(), api.NewLauncher(),
	),
)
if err := l.Execute(ctx, config, os.Args[1:]); err != nil {
	log.Fatalf("Run failed: %v\n\n%s", err, l.CommandLineSyntax())
}
```

The `api` sub-launcher serves under `/api` by default, but the scaffolded template
overrides this with `-path_prefix /`.

### Adding custom routes

```go
restServer, _ := adkrest.NewServer(adkrest.ServerConfig{
	AgentLoader:     agent.NewSingleLoader(rootAgent),
	SessionService:  session.InMemoryService(),
	SSEWriteTimeout: 120 * time.Second,
})

mux := http.NewServeMux()
mux.HandleFunc("POST /health", myHealthHandler) // your routes
mux.Handle("/", restServer)                     // ADK at the root
http.ListenAndServe(":8080", mux)
```

These are layers, not alternatives: the `api` sub-launcher's `SetupSubrouters` calls
`adkrest.NewServer` itself and mounts the result on the launcher's shared router
(`cmd/launcher/web/api/api.go`). Call `adkrest.NewServer` yourself when you need to own the mux, as
above; use the launcher when its command-line surface is enough.

## 9. A2A protocol

A2A is built into the scaffolded template via `a2a.NewLauncher()`.

To consume a remote A2A agent, use `agent/remoteagent/v2` — the result is an ordinary
`agent.Agent`. The unversioned `agent/remoteagent` is deprecated in favour of it:

```go
// The package is named remoteagent even at the /v2 path.
import remoteagent "google.golang.org/adk/v2/agent/remoteagent/v2"

remote, err := remoteagent.NewA2A(remoteagent.A2AConfig{
	Name:        "remote_agent",
	Description: "...",
	// Set exactly one of AgentCardProvider or AgentCard. The provider re-resolves
	// the card on every invocation; NewAgentCardProvider takes a URL or a file path.
	AgentCardProvider: remoteagent.NewAgentCardProvider(
		"http://remote-host:8080/.well-known/agent-card.json",
	),
})
```

## 10. Event-driven / ambient agents

Ambient agents process events (Pub/Sub, Eventarc, schedules) rather than chat turns. ADK Go
ships trigger sub-launchers for both sources:

```go
web.NewLauncher(
	pubsub.NewLauncher(),   // google.golang.org/adk/v2/cmd/launcher/web/triggers/pubsub
	eventarc.NewLauncher(), // google.golang.org/adk/v2/cmd/launcher/web/triggers/eventarc
	api.NewLauncher(),
)
```

## See also

- [ADK Documentation](https://adk.dev/llms.txt)
- [ADK Go examples](https://github.com/google/adk-go/tree/main/examples)

# ADK Live and Voice Agents

Real-time voice and video agents built on ADK's **Gemini Live API Toolkit**
(`Runner.run_live` + `LiveRequestQueue`). Use this for low-latency spoken
conversation and interruption ("barge-in") — not for turn-based
request/response agents (use a normal `Agent` with `run_async` for those).

**Official docs:** [ADK Live Docs](https://adk.dev/live/index.md) ·
[Configuration](https://adk.dev/live/configuration/index.md) ·
[Tools](https://adk.dev/live/tools/index.md) ·
[Get started](https://adk.dev/live/get-started/index.md)

---

## 1. When to use Live

| Use `run_live` (Live) | Use `run_async` (normal) |
|-----------------------|--------------------------|
| Voice/phone assistants, live captioning | Chatbots, tools, batch/RAG |
| User can interrupt mid-response | One turn completes before the next |
| Continuous audio/video input | Discrete text/multimodal messages |
| Latency-critical spoken UX | Latency-tolerant |

> **Live agents can't be served over A2A, and can't be published to Gemini.**
> Both are request/response with no Live transport. This has
> consequences for a scaffolded project beyond the model line — see
> "Converting a scaffolded project to Live" next.

## 2. Converting a scaffolded project to Live

There is no Live scaffold template. Scaffold `adk` normally, then make these
edits **together** — they are all consequences of one decision, and stopping
after the first leaves a project whose `uv run pytest` fails.

| # | File | Edit |
|---|------|------|
| 1 | `app/agent.py` | Point `MODEL` at a Live model and, on Gemini platform, add `client_kwargs={"location": …}` to the `Gemini(...)` the scaffold already emits (see "Models"). No `.env` or Terraform change. |
| 2 | `app/fast_api_app.py` | Delete the `attach_a2a_routes` call and its imports; delete `app/app_utils/a2a.py` and the `a2a-sdk` dependency. `/run_live` comes from `get_fast_api_app` and is untouched. |
| 3 | `tests/integration/test_agent.py` | Rewrite onto `run_live` (below). |
| 4 | `tests/integration/test_server_e2e.py` | Repoint the readiness probe, drop the A2A and SSE tests, add a `/run_live` test (below). |
| 5 | — | Drop `agents-cli publish gemini` from the plan; the Gemini app has no Live transport to register against. |

Steps 3 and 4 are not optional cleanup. Both scaffolded integration tests drive
the **non-live** path, and a Live model doesn't degrade to text — it rejects it,
so `runner.run()` and `/run_sse` fail outright. Rewrite the tests; don't delete
them.

### `tests/integration/test_agent.py`

Same shape as the scaffolded test, but driven by `run_live` and asserting on the
transcript rather than on text parts:

```python
import asyncio

from google.adk.agents.live_request_queue import LiveRequestQueue
from google.adk.agents.run_config import RunConfig
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

from app.agent import root_agent

APP_NAME = "test"
USER_ID = "test_user"


def test_agent_live_stream() -> None:
    """The agent answers over run_live and returns audio plus a transcript."""

    async def _turn() -> tuple[str, int]:
        session_service = InMemorySessionService()
        # The session must exist before run_live, or it raises "Session not found".
        session = await session_service.create_session(
            app_name=APP_NAME, user_id=USER_ID
        )
        runner = Runner(
            agent=root_agent, session_service=session_service, app_name=APP_NAME
        )
        queue = LiveRequestQueue()
        queue.send_content(
            types.Content(role="user", parts=[types.Part(text="Why is the sky blue?")])
        )

        transcript, audio_chunks = "", 0
        try:
            async for event in runner.run_live(
                user_id=USER_ID,
                session_id=session.id,
                live_request_queue=queue,
                # Transcription is on by default — see "RunConfig".
                run_config=RunConfig(response_modalities=["AUDIO"]),
            ):
                # Consume the finished aggregate; partials would duplicate text.
                if (
                    event.output_transcription
                    and event.output_transcription.finished
                    and event.output_transcription.text
                ):
                    transcript += event.output_transcription.text
                if event.content and event.content.parts:
                    audio_chunks += sum(
                        1 for part in event.content.parts if part.inline_data
                    )
                if event.turn_complete:
                    break
        finally:
            queue.close()
        return transcript, audio_chunks

    transcript, audio_chunks = asyncio.run(_turn())
    assert transcript.strip(), "Expected a non-empty output transcript"
    assert audio_chunks > 0, "Expected the agent to return audio"
```

### `tests/integration/test_server_e2e.py`

Three edits, and the first is the one that's easy to miss:

- **Repoint the readiness probe.** `wait_for_server()` polls `AGENT_CARD_URL`,
  which step 2 deleted — use `f"{BASE_URL}/list-apps"` instead. Miss this and
  the probe raises `NameError`, which the surrounding `except RequestException`
  does **not** catch, so every test in the file errors before the server is even
  contacted. (`/list-apps` is a sound readiness signal: uvicorn serves no
  request until the lifespan startup completes.)
- **Delete `test_adk_run_sse`, `test_a2a_chat_stream`, and `test_agent_card`**,
  and add the `/run_live` test below. Those three are the whole file, so the
  new test replaces them rather than joining them.
- **On the `agent_runtime` target only**, also delete
  `test_reasoning_engine_stream` — `async_stream_query` is the non-live path
  too.

```python
import asyncio
import json
import uuid

import requests
from websockets.asyncio.client import connect

APP_NAME = "app"  # your agent_directory
BASE_URL = "http://127.0.0.1:8000"
HEADERS = {"Content-Type": "application/json"}


def test_adk_run_live(server_fixture) -> None:
    """Test the native ADK Live route (/run_live) end to end."""
    user_id = f"user_{uuid.uuid4()}"
    # /run_live closes with 1002 if the session doesn't already exist.
    session_response = requests.post(
        f"{BASE_URL}/apps/{APP_NAME}/users/{user_id}/sessions",
        headers=HEADERS,
        json={},
        timeout=60,
    )
    assert session_response.status_code == 200
    session_id = session_response.json()["id"]

    ws_url = (
        f"ws://127.0.0.1:8000/run_live?app_name={APP_NAME}&user_id={user_id}"
        f"&session_id={session_id}&modalities=AUDIO"
    )

    async def _turn() -> list[dict]:
        events: list[dict] = []
        async with connect(ws_url) as websocket:
            await websocket.send(
                json.dumps(
                    {"content": {"role": "user", "parts": [{"text": "Hi!"}]}}
                )
            )
            while True:
                raw = await asyncio.wait_for(websocket.recv(), timeout=60)
                event = json.loads(raw)
                events.append(event)
                # A tool round-trip emits an extra turnComplete before the
                # answer; a tool-using agent must keep reading past the first.
                if event.get("turnComplete"):
                    break
            await websocket.send(json.dumps({"close": True}))
        return events

    events = asyncio.run(_turn())
    assert events, "No events received from /run_live"
    assert any(
        (transcription := event.get("outputTranscription"))
        and transcription.get("finished")
        and transcription.get("text")
        for event in events
    ), "Expected a finished output transcription"
```

Frame details: the local server sends serialized events as **text** frames (see
"Events"), so keys are camelCase and absent means unset — `.get()` is the right
accessor, not truthiness on a guaranteed key. Only Agent Runtime in Gemini platform also delivers
event JSON as binary frames — see "Deployed URL shapes". `websockets` is
already available via `google-genai`.

## 3. Models

Every supported Live model is **native audio** — audio in, audio out, end to
end, with no intermediate text-to-speech stage.

| Model | Gemini platform | AI Studio | Use for |
|---|---|---|---|
| `gemini-3.8-live` | Supported | Supported | **Default.** Low-latency dialogue, interleaved reasoning, no reasoning-induced pauses. |
| `gemini-3.8-live-extended-thinking` | Supported | Supported | Complex, multi-step problem solving mid-conversation. Runs background reasoning and async tool calls while it keeps streaming audio, trading latency for depth. |
| `gemini-live-2.5-flash-native-audio` | Supported | — | Legacy. Pre-3.8 behavior. |

### What 3.8 changed

| Area | 3.8 behavior |
|---|---|
| Affective dialog | **Removed from the API.** `enable_affective_dialog` now errors. Delete it. |
| Proactive audio | **Permanently on.** `proactive_audio=False` errors. |
| Function calling | Async (`NON_BLOCKING`) is the default. `gemini-3.8-live` still accepts `BLOCKING`; `-extended-thinking` rejects it with a hard error. |
| Function scheduling | `SILENT` / `WHEN_IDLE` / `INTERRUPTED` work on `gemini-3.8-live`, and are unsupported on `-extended-thinking`. |
| Thinking | `gemini-3.8-live` takes no `thinking_level`; omit it. `-extended-thinking` takes `thinking_config` with `low`, `medium`, or `high` (not `MINIMAL`). |

> **`-extended-thinking` breaks the `turnComplete` assumption.** It keeps
> reasoning and calling tools after `turnComplete: true`, so that frame no
> longer means idle. Read `interaction_status`: `IN_PROGRESS` means more output
> is coming, `IDLE` ends the turn. A client that stops at `turnComplete`
> truncates the answer. `--mode adk_live` already handles this.

Being native audio has consequences the rest of this page depends on:

- **AUDIO-only output.** `response_modalities=["TEXT"]` is rejected — send
  `["AUDIO"]` and read the transcripts (see "Events").
- **Language is inferred**, not set — see "Voice and language".
- **Proactive audio is always on** in 3.8 (see "RunConfig").

Keep the model **and its region** env-overridable — Live model IDs change
often, and the two are coupled, since changing the model can force a different
region:

```python
import os
from google.adk.agents import Agent
from google.adk.models import Gemini

MODEL = os.getenv("LIVE_MODEL", "gemini-3.8-live")
# Gemini platform only — pins the model's region (see "Region" below). The Gemini API
# rejects `location` outright, so leave LIVE_LOCATION empty on AI Studio.
LIVE_LOCATION = os.getenv("LIVE_LOCATION", "us-central1")

root_agent = Agent(
    name="live_assistant",
    model=Gemini(
        model=MODEL,
        client_kwargs={"location": LIVE_LOCATION} if LIVE_LOCATION else None,
    ),
    instruction="You are a helpful voice assistant. Keep replies concise.",
)
```

Confirm the current ID before shipping against
`https://ai.google.dev/gemini-api/docs/live` (AI Studio) or
`https://cloud.google.com/vertex-ai/generative-ai/docs/live-api` (Gemini platform). Load
`.env` **before** importing the agent module, or `os.getenv` runs first and
silently uses the fallback.

**Region.** On Gemini platform, Live models are served from a **regional** endpoint such
as `us-central1`, **not** the `global` endpoint — so a valid model ID still
fails under the wrong location. Confirm yours in the model × region matrix at
[Google model endpoint locations](https://docs.cloud.google.com/gemini-enterprise-agent-platform/resources/locations#google-models).

**Pin the region on the model, not in the environment.** `client_kwargs` is a
passthrough to the `genai.Client` constructor, and an explicit `location` there
beats `GOOGLE_CLOUD_LOCATION` — as in the example above. Leave it out and the
model falls back to `GOOGLE_CLOUD_LOCATION`, so an exported `global` (a common
default) silently sends Live to an endpoint that doesn't serve it. Pinning is
also the smaller change: `GOOGLE_CLOUD_LOCATION` is shared with sessions,
telemetry, and eval, and `agents-cli deploy` sets it to `global` on Agent
Runtime — so an env-only fix works locally and then breaks once deployed.

Two things to know about `client_kwargs`:

- **It is Gemini platform only.** On AI Studio, a `location` raises `ValueError: Gemini
  API does not support project/location.` The Live client is built lazily, so
  this surfaces on connect and looks like a transport failure. Keep the
  `if LIVE_LOCATION else None` guard if the same code runs on both backends.
- **Pass only `location` / `project`.** Its keys are merged *over* the ones ADK
  builds, so passing `http_options` replaces the object carrying Live's
  `api_version` and tracking headers.

> Don't reach for the `api_client` subclass override shown in ADK's `Gemini`
> docstring — Live reads a different client (`_live_api_client`), so that
> pattern silently does nothing here. `client_kwargs` is applied to both.

## 4. The agent

A Live agent is a normal `Agent` — the difference is how you *run* it, plus the
voice it speaks with and how you write for the ear rather than the eye.

```python
from google.adk.agents import Agent
from google.adk.models import Gemini
from google.adk.tools import google_search

root_agent = Agent(
    name="live_assistant",
    model=Gemini(
        model="gemini-3.8-live",
        # Gemini platform only — see "Models"; omit on AI Studio.
        client_kwargs={"location": "us-central1"},
    ),
    instruction="You are a helpful voice assistant. Keep replies concise.",
    tools=[google_search],
)
```

### Writing the instruction for speech

The instruction is where most voice-agent quality comes from, and a chat
agent's instruction ported unchanged is the usual cause of a bad first demo.
Every reply is **spoken**, so:

- **Ban visual formatting.** Markdown, bullets, tables, emoji, and URLs get
  read aloud literally. Say so explicitly — models default to formatting.
- **Cap turn length.** Ask for a sentence or two, then a question. Long
  monologues make the user talk over the agent to get a word in.
- **Speak numbers, not glyphs.** Dates, currency, phone numbers, and IDs need
  spoken forms ("March third", "four one five…"), and long codes want
  grouping and pauses.
- **Confirm before acting on what it misheard.** Speech recognition mangles
  names, addresses, and account numbers; tell the agent to read critical
  values back rather than proceed on its best guess.
- **Cover the pauses.** Tell it to say what it's doing before a slow tool
  ("let me look that up"). Silence reads as a dropped call — see "Tools and
  latency" for the non-blocking patterns behind this.

Language choice is instruction-driven too — see "Voice and language".

### Voice and language (`speech_config`)

Set the voice **per agent** by passing a configured `Gemini` instance — the
usual choice, because it lets each agent in a multi-agent system have its own
voice:

```python
from google.adk.agents import Agent
from google.adk.models import Gemini
from google.genai import types

support_llm = Gemini(
    model="gemini-3.8-live",
    client_kwargs={"location": "us-central1"},  # Gemini platform only — see "Models"
    speech_config=types.SpeechConfig(
        voice_config=types.VoiceConfig(
            prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Aoede")
        ),
        language_code="en-US",
    ),
)

support_agent = Agent(name="support", model=support_llm, instruction="...")
```

Or set one voice for the whole session via `RunConfig(speech_config=...)`.

Precedence runs agent-level (`Gemini(speech_config=…)`) → `RunConfig` → the
Live API default, so a per-agent voice always wins. Set a session default in
`RunConfig` and override it for the agents that need their own voice.

`language_code` (e.g. `"en-US"`, `"ja-JP"`) is a hint, not a switch: native
audio models may ignore it and infer the language from the conversation, so
say what language to speak in the agent's instruction when it matters. An
unsupported voice fails at **connection time**, so test your voice on the
target platform early.

## 5. Driving a turn: `run_live` + `LiveRequestQueue`

`run_live()` is an async generator; `LiveRequestQueue` is the upstream channel.
They run concurrently — you send input while receiving output.

```python
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.agents.live_request_queue import LiveRequestQueue
from google.adk.agents.run_config import RunConfig
from google.genai import types

session_service = InMemorySessionService()
runner = Runner(app_name="app", agent=root_agent, session_service=session_service)

async def one_turn(user_id: str, session_id: str, text: str):
    # Create the session first, or run_live raises "Session not found".
    queue = LiveRequestQueue()
    run_config = RunConfig(
        # One modality per session, fixed at start.
        response_modalities=["AUDIO"],
        input_audio_transcription=types.AudioTranscriptionConfig(),
        output_audio_transcription=types.AudioTranscriptionConfig(),
    )

    # Upstream: send the user's turn.
    queue.send_content(types.Content(parts=[types.Part(text=text)]))

    # Downstream: consume events until the turn completes.
    async for event in runner.run_live(
        user_id=user_id, session_id=session_id,
        live_request_queue=queue, run_config=run_config,
    ):
        if event.content and event.content.parts:
            for part in event.content.parts:
                if part.text:
                    print(event.author, part.text)
        if event.turn_complete:
            break
    queue.close()  # always close — never reuse a queue across sessions
```

Response modality constrains **output only**: you can always send text, audio,
or video input regardless of what the model replies with.

### Sending input
- **Text:** `queue.send_content(types.Content(parts=[types.Part(text=...)]))`
- **Audio (realtime):** `queue.send_realtime(types.Blob(mime_type="audio/pcm;rate=16000", data=pcm_bytes))`
- **Video frame:** `queue.send_realtime(types.Blob(mime_type="image/jpeg", data=frame_bytes))`
- **Close:** `queue.close()` (ends the session)

### Audio format and chunking

Input is **16-bit PCM, 16 kHz, mono**; output is **24 kHz**. ADK does no format
conversion — wrong rates produce garbled audio or a `1007` close. Stream
continuously in **50–100 ms** chunks rather than waiting for replies;
`LiveRequestQueue` forwards each chunk without batching.

## 6. RunConfig

| Field | Purpose |
|-------|---------|
| `response_modalities=["AUDIO"]` | One per session, fixed at start. Defaults to `["AUDIO"]`. Every supported Live model is native audio, so `["TEXT"]` is rejected. |
| `speech_config` | Voice + language — see "Voice and language". Overridden by agent-level config. |
| `input_audio_transcription` / `output_audio_transcription` | `types.AudioTranscriptionConfig()` — text transcripts of speech. **Default-on** via `default_factory`, and ADK's built-in `/run_live` doesn't override them, so transcripts flow with no config. Set to `None` to disable. |
| `realtime_input_config` | Turn-taking / VAD (below). |
| `proactivity` / `enable_affective_dialog` | Legacy models only; both error on 3.8 (below). |
| `session_resumption` / `context_window_compression` | See "Session lifetime and limits". |
| `save_live_blob` | Persist audio/video artifacts for debugging or compliance (default `False`). |

Full options: [Configuration](https://adk.dev/live/configuration/index.md).

### Turn-taking and VAD

Voice activity detection is **enabled by default**: just stream audio
continuously and the API decides when the user started and stopped talking,
including handling interruptions.

Disable it for push-to-talk, noisy/cross-talk environments, or when your client
does its own detection:

```python
run_config = RunConfig(
    response_modalities=["AUDIO"],
    realtime_input_config=types.RealtimeInputConfig(
        automatic_activity_detection=types.AutomaticActivityDetection(
            disabled=True
        )
    ),
)
```

With VAD off you **must** delimit each turn yourself:
`queue.send_activity_start()` before the first audio chunk,
`queue.send_activity_end()` after the last. The model only processes audio
between those signals, so mistimed signals cause dropped or ignored speech.

### Proactivity and affective dialog

On 3.8 neither is a knob: proactive audio is permanently on, and
`enable_affective_dialog` errors. Pass neither. Proactive audio is
**probabilistic**, so where predictability matters more than warmth, constrain
it in the instruction.

Legacy `gemini-live-2.5-flash-native-audio` takes both as opt-in config:

```python
run_config = RunConfig(
    response_modalities=["AUDIO"],
    proactivity=types.ProactivityConfig(proactive_audio=True),
    enable_affective_dialog=True,
)
```

### Multi-agent and workflow Live

Live is not limited to one agent. Four shapes work:

| Shape | How a stage ends | Use for |
|-------|------------------|---------|
| Single `Agent` | — | Most voice agents |
| Coordinator + `sub_agents` | `transfer_to_agent` | Route by intent to a specialist |
| `SequentialAgent` pipeline | `task_completed` | Fixed stages: research → write → review |
| Graph `Workflow` | stage completes its task | Explicit topology, typed handoffs |

`transfer_to_agent` hands the conversation to a specialist and the stream
continues in the **same** `run_live()` loop. `task_completed` ends one stage of
a `SequentialAgent`; it does *not* exit your loop — only the last agent
completing does. Use **one** `LiveRequestQueue` for the whole conversation;
never create one per agent.

A graph `Workflow` runs as a voice conversation with no structural change — the
only Live-specific part is the model on each stage. `mode='task'` is what makes
a stage run its own turn-taking loop and complete before handing off, which is
what you want for a staged voice call:

```python
from google.adk.agents import Agent
from google.adk.workflow import START, Workflow

greeter = Agent(model=LIVE_MODEL, name="greeter", mode="task",
                instruction="Greet the caller and confirm their name.")
verifier = Agent(model=LIVE_MODEL, name="verifier", mode="task",
                 instruction="Verify their date of birth.")

root_agent = Workflow(name="intake", edges=[(START, greeter), (greeter, verifier)])
```

Typed handoffs between stages (`output_schema`) and the rest of the graph API
are in `references/adk-python-workflows.md`. Working example:
[live_workflow sample](https://github.com/google/adk-python/tree/main/contributing/samples/live/live_workflow)
— which also evaluates the workflow in-process with ADK's audio user simulator.
To evaluate a Workflow over the wire instead, serve it and run
`agents-cli eval generate --mode adk_live`: the server drives `run_live`, so graph roots
need no special handling on the client.

Whenever an agent has `sub_agents`, ADK **forces audio transcription on** even
if you set it to `None` — agent transfer needs text context to pass along. Plan
on receiving transcription events. Give each agent its own voice (see "Voice
and language") so transitions are audible to the user.

> Built-in grounding tools (`google_search`, `google_maps_grounding`) are
> model-internal and don't mix with other tools on the same agent — a
> constraint that bites conversational voice roots especially hard. Put them on
> a grounding sub-agent instead; see `references/adk-python.md`, "Tools".

## 7. Events (`run_live` output)

Same `Event` model as `run_async`, plus Live flags. **Live emits control,
transcription, and terminator frames as their own content-less events** — a
consumer must not assume every frame carries `content`. Usage metadata and
voice-activity frames are content-less too; skip them.

- `event.turn_complete` — standalone, content-less frame; the model finished
  this turn (enable user input).
- `event.interrupted` — user barged in; may be standalone (no content) or
  attached to a final text response. Flush any buffered audio you haven't
  played yet, or the user hears stale speech over their own.
- `event.input_transcription` / `event.output_transcription` — speech text,
  each on its own content-less frame. Streamed as partial chunks
  (`finished=False`, `partial=True`) then a final aggregate (`finished=True`,
  `partial=False`); ADK also flushes the aggregate on turn_complete/interrupted.
  **Consume the finished aggregate; skip partials** to avoid duplicating text.
  Null-check twice: the object may exist with empty `.text`.
- `event.content.parts[].inline_data` — raw audio bytes (AUDIO modality).
- `event.author` — agent name for model output; `"user"` for input transcripts.
- `event.error_code` / `event.error_message` — failures (e.g. `SAFETY`, `MAX_TOKENS`).

Serialize for the wire with `event.model_dump_json(exclude_none=True, by_alias=True)`.
Full event handling (interruption, tool calls, error decision table):
[Events](https://adk.dev/live/events/index.md).

## 8. Session lifetime and limits

**ADK `Session` vs Live session.** The ADK `Session` (from `SessionService`) is
durable conversation storage that survives restarts. The Live session is
ephemeral: created when `run_live()` starts, destroyed when the queue closes.
Each new `run_live()` re-seeds a fresh Live session from the ADK `Session`
history — which is why conversations survive across calls.

**Default duration limits**, absent compression:

| | Gemini Live API | Agent Runtime |
|---|---|---|
| Audio-only session | 15 min | 10 min |
| Audio + video session | 2 min | 10 min |
| Connection | ~10 min (ADK reconnects transparently) | — |

**Session resumption** keeps a session alive across the connection timeout.
ADK caches the resumption handle and reconnects on its own — you write no
reconnection code. Enable it by default in production:

```python
run_config = RunConfig(
    response_modalities=["AUDIO"],
    session_resumption=types.SessionResumptionConfig(),
)
```

**Context window compression removes the session duration limit entirely** and
keeps long conversations inside the model's context window. Enable it when
calls can outlast the table above:

```python
run_config = RunConfig(
    response_modalities=["AUDIO"],
    session_resumption=types.SessionResumptionConfig(),
    context_window_compression=types.ContextWindowCompressionConfig(
        trigger_tokens=100_000,  # ~78% of a 128k context
        sliding_window=types.SlidingWindow(target_tokens=80_000),
    ),
)
```

Trade-off: older turns are summarized rather than kept verbatim, so skip it
when precise recall of early conversation matters.

## 9. Serving and testing a Live agent

Live agents are reached over a **WebSocket** at `/run_live`, not `/run_sse` —
and there is **no server code to write**. `get_fast_api_app` serves that route
on every target, and the scaffold's `fast_api_app.py` already wires it up.
Work up this ladder and stop at the first rung that does the job.

**1. Talk to it in the playground.** `agents-cli playground` starts `adk web`,
whose dev UI has mic and camera controls wired to `/run_live`. This is the
first thing to do after pointing a project at a Live model — a real voice
conversation, no client code, no flags. The UI drives session defaults (AUDIO,
VAD on) and sends no session flags, so what you're validating here is the
agent itself: instruction, voice, tools, and barge-in.

**2. Script a turn.** `agents-cli run "<prompt>" --mode adk_live` runs one turn
and prints the transcript; `agents-cli eval run --mode adk_live` plays a
dataset's user turns over one persistent socket and grades the transcripts
(see `/google-agents-cli-eval`). Both connect with `modalities=AUDIO` and use a
**text-in, audio-out** turn model — one text `content` frame per user turn.
There is no flag to point them at a custom WebSocket route; see
`agents-cli run --help` and `agents-cli eval generate --help`.

**3. Write a *client*, not a server.** A browser UI, phone bridge, or device
client connects to the same built-in endpoint:
`ws://host/run_live?app_name=<app>&user_id=<uid>&session_id=<sid>&modalities=AUDIO`
— create the session over HTTP first. Upstream frames are `LiveRequest` JSON
(snake_case): `{"content": {...}}` for a discrete turn, `{"blob": {...}}` for
realtime mic audio, `{"close": true}` to end; with VAD off, bracket a streamed
turn as `{"activity_start": {}}` → one `{"blob": {...}}` per chunk →
`{"activity_end": {}}`. `LiveRequest` has no `realtime_input` field and does
not forbid extras, so that key is dropped silently and the audio never lands.
Downstream frames are serialized ADK `Event`s (binary on some targets, so
decode-as-JSON before treating bytes as audio).

**4. Only then, edit `fast_api_app.py`.** The file already exists — you are
adding a handler, not creating the server. Reach for this only when you need
something the rungs above can't express: a `RunConfig` field with no query
param (`context_window_compression`, disabling transcription,
`tool_thread_pool_config`), custom auth on the WebSocket handshake, or your own
binary wire protocol. Bridge your `@app.websocket("/ws")` route to `run_live` +
`LiveRequestQueue` (see
[Build a custom server](https://adk.dev/live/custom-server/index.md)); `agents-cli` boots
`fast_api_app.py` in preference to `adk api_server` for local serving.

> Voice selection is **not** a reason to write a handler — set it per agent
> with `Gemini(speech_config=…)`, as in "Voice and language".

**Where each decision lives.** What you build into the agent — model and its
region, voice, instruction, tools, sub-agents — ships with the deployment and
behaves the same for every caller. Per-session behavior does not: `RunConfig`
is an argument to `run_live`, and `App` has nowhere to hold one, so the client
opening the socket chooses it. On the built-in endpoint that means query
params (`modalities`, `explicit_vad_signal` for push-to-talk, and the rest of
section 6's fields that it exposes); rung 4 is for the fields it doesn't.
Design accordingly: anything that must hold for every caller belongs in agent
code, not in a URL.

### Deployed URL shapes (every target serves the same `/run_live`)

`get_fast_api_app` serves `/run_live` on a local server, Cloud Run, GKE, and
Agent Runtime alike — only the base URL differs. `agents-cli run`/`eval`
auto-detect these; the shapes are here for hand-rolled clients.

- **Local / Cloud Run / GKE:** `wss://<host>/run_live?app_name=…&user_id=…&session_id=…`
  (sessions over `https://<host>/apps/{APP}/users/{USER}/sessions`).
- **Agent Runtime** proxies the container behind an `/api` passthrough with a
  separate WebSocket ingress:
  - **WebSocket:** `wss://{LOC}-aiplatform.googleapis.com/reasoningEngines/ws/internal/{RESOURCE}/api/run_live?app_name=…&user_id=…&session_id=…`
  - **Session create (HTTP):** `POST https://{LOC}-aiplatform.googleapis.com/reasoningEngines/v1/{RESOURCE}/api/apps/{APP}/users/{USER}/sessions`
  - Auth is a standard `Authorization: Bearer <ADC token>` header on the handshake.
  - Only the WebSocket uses the separate `ws/internal` ingress; HTTP goes
    through the same `reasoningEngines/v1/{RESOURCE}/api` passthrough as every
    other container route (see `/google-agents-cli-deploy`,
    `references/agent-runtime.md`).
  - The managed `AdkApp` + `bidi_stream_query` path is a different mechanism and
    is **not** used for ADK Live agents.

## 10. Tools and latency

Live raises the stakes on tool latency: in a spoken conversation a pause over a
second reads as a dropped call, where the same pause in chat is invisible.
**Prefer non-blocking tools for anything that isn't reliably fast** — but only
when a tool can actually outrun the conversation; tools that return quickly
need no special handling.

| Expected tool latency | Pattern |
|----------------------|---------|
| Under ~500 ms | Plain `async def` tool — the pause is conversationally invisible |
| Seconds, bounded | `NON_BLOCKING` + `response_scheduling` |
| Open-ended / continuous | Streaming tool (`AsyncGenerator`) |
| Sync/blocking library | `RunConfig.tool_thread_pool_config` |

**Non-blocking tools.** ADK runs the tool in a background task and returns a
"pending" ack immediately so the model keeps talking, then feeds the real
result back into the live queue:

```python
from google.adk.tools import FunctionTool
from google.genai import types

my_tool = FunctionTool(my_async_fn)  # my_async_fn is `async def`
my_tool.response_scheduling = types.FunctionResponseScheduling.WHEN_IDLE
#   WHEN_IDLE  → speak the result when the caller is idle (typical)
#   SILENT     → apply the result without speaking (e.g. a state update)
#   INTERRUPT  → speak the result as soon as it's ready
```

`-extended-thinking` runs every tool non-blocking and rejects
`response_scheduling`; omit that line there.

Inside such a tool you can drive a multi-step **graph `Workflow`**
(`google.adk.workflow`) via `await tool_context.run_node(workflow,
node_input=...)`. Always `await` it directly — never wrap in
`asyncio.create_task`. See `references/adk-python-workflows.md`.

**Streaming tools.** An `async def` tool that `yield`s streams intermediate
results back into the conversation (e.g. monitor a price or a video feed). A
tool parameter named `input_stream: LiveRequestQueue` is a reserved hook ADK
fills with inbound realtime frames. You must also supply a
`stop_streaming(function_name: str)` tool so the model can end the stream. See
[Tools](https://adk.dev/live/tools/index.md).

**Blocking I/O.** For unavoidably blocking tools,
`RunConfig.tool_thread_pool_config` offloads them to a thread pool so the event
loop — and therefore barge-in — stays responsive.

## 11. Vision input (camera and screen share)

Images and video are both sent as individual **JPEG frames** on the same
socket — `send_realtime` in-process, a `blob` frame over the wire. There is no
video codec involved.

- **Frame rate:** 1 FPS recommended maximum.
- **Resolution:** 768×768 recommended.

Good for "look at this" moments: a user pointing a camera at a product, sharing
a screen for troubleshooting, or a device sending periodic stills. Note that
audio + video sessions have a much tighter default duration limit than
audio-only on Gemini Live API (see "Session lifetime and limits").

## 12. Deploy and evaluate

Deploy with `agents-cli deploy`, adding `--timeout 3600` on Cloud Run so a
session outlives the 300s default (`/google-agents-cli-deploy`). Evaluate with
`agents-cli eval run --mode adk_live` (`/google-agents-cli-eval`).

# ADK Workflow API Cheatsheet

> Requires `google-adk >= 2.0.0`. This page documents the Python graph API; ADK Go has its own — see `references/adk-go-workflows.md`.
> Requires **Python >= 3.11**. A `Workflow` can also run as a Live (voice) agent with no structural change — give the stage agents a Live model; see `references/adk-python-live.md`.

**Official docs:** [Workflows overview](https://adk.dev/workflows/index.md) ·
[Graph routes](https://adk.dev/graphs/routes/index.md) ·
[Collaboration](https://adk.dev/workflows/collaboration/index.md) ·
[Data handling](https://adk.dev/graphs/data-handling/index.md) ·
[Dynamic workflows](https://adk.dev/graphs/dynamic/index.md) ·
[Human-in-the-loop](https://adk.dev/graphs/human-input/index.md)

## 1. Core Concepts

A `Workflow` is a graph-based agent: nodes do work, edges define flow, `START` is the entry point.

```python
from google.adk.workflow import Workflow

def greet(node_input: str) -> str:
  return f"Hello, {node_input}!"

root_agent = Workflow(
    name="greeter",
    edges=[('START', greet)],
)
```

Three building blocks: **Nodes** (functions, LLM agents, tools), **Edges** (connections with optional route conditions), **START** (built-in entry receiving user input).

### Workflow Constructor

```python
root_agent = Workflow(
    name="my_workflow",
    edges=[...],                # Edge definitions (or use graph= instead)
    description="",             # Agent description
    input_schema=None,          # Pydantic model for input validation
    output_schema=None,         # Pydantic model for the workflow's output
    state_schema=None,          # Pydantic model for state validation
    rerun_on_resume=True,       # Rerun workflow on resume (default: True)
    max_concurrency=None,       # Limit parallel node execution (None = no limit)
    retry_config=None,          # Default RetryConfig applied to nodes
    timeout=None,               # Whole-workflow timeout in seconds
    wait_for_output=False,      # Wait for dynamically scheduled child output
)
```

---

## 2. Node Types

Any "NodeLike" is accepted in edges and auto-wrapped:

| Python Object | Wrapped As | Default `rerun_on_resume` |
|--------------|-----------|------------------------|
| Function/callable | `FunctionNode` | `False` |
| `LlmAgent` | Internal `_LlmAgentWrapper` | `True` |
| Other `BaseAgent` | Internal `AgentNode` | `False` |
| `BaseTool` | Internal `_ToolNode` | `False` |
| `BaseNode` subclass | Used as-is | Per subclass |

> **Auto-wrapping is the recommended approach.** Place functions, agents, and tools directly in edges — the framework wraps them automatically. You do not need to import or use internal wrapper classes directly.

---

## 3. Function Nodes

Most common node type. Parameter resolution:

| Parameter | Source |
|-----------|--------|
| `ctx` | Workflow `Context` object |
| `node_input` | Output from predecessor node |
| Any other name | `ctx.state[param_name]` |

```python
from google.adk.agents.context import Context

def process(ctx: Context, node_input: Any, user_name: str) -> str:
  # node_input = predecessor output; user_name = ctx.state['user_name']
  # START outputs types.Content (not str) unless input_schema is set
  return f"{user_name}: {node_input}"
```

### Return Types

- **Value** -> wrapped in `Event(output=value)`, triggers downstream
- **`None`** -> no event emitted, no downstream trigger
- **`Event`** -> used directly (for routing or state updates)
- **Generator** -> yield multiple events; only the last with `output` triggers downstream

```python
from google.adk.events.event import Event

def classify(node_input: str):
  if "urgent" in node_input:
    return Event(output=node_input, route="urgent")
  return Event(output=node_input, route="normal", state={"processed": True})
```

### Auto Type Conversion

FunctionNode auto-converts `dict` inputs to Pydantic models based on type hints. Works for `list[Model]` and `dict[str, Model]` too.

### `node_input` Type by Predecessor

| Predecessor | `node_input` Type |
|-------------|-------------------|
| Function returning `str`/`dict` | `str`/`dict` |
| Function returning `Event(output=X)` | type of `X` |
| `LlmAgent` (no `output_schema`) | `types.Content` |
| `LlmAgent` (with `output_schema`) | `dict` |
| `JoinNode` | `dict[str, Any]` (keyed by predecessor names) |
| `ParallelWorker` | `list` |
| `START` (no `input_schema`) | `types.Content` |
| `START` (with `input_schema`) | parsed schema type |

### @node Decorator & Explicit FunctionNode

```python
from google.adk.workflow import node, FunctionNode, RetryConfig

@node
def my_func(node_input: str) -> str:
  return node_input

@node(name="custom", rerun_on_resume=True)
async def my_async(node_input: str) -> str:
  return node_input

# Explicit FunctionNode for full control (func is keyword-only)
fn = FunctionNode(
    func=my_func,
    retry_config=RetryConfig(max_attempts=3),
    timeout=30.0,                          # Seconds before timeout
    parameter_binding='state',             # 'state' (default) or 'node_input'
    auth_config=None,                      # Requires rerun_on_resume=True
    state_schema=None,                     # Pydantic model for state validation
)
```

---

## 4. Edge Patterns

```python
# Sequential chain
edges = [('START', a), (a, b), (b, c)]

# Conditional routing (node returns Event with route=)
edges = [
    ('START', classifier),
    (classifier, success_handler, "success"),
    (classifier, error_handler, "error"),
    (classifier, fallback_handler, '__DEFAULT__'),  # Fallback route
]

# Fan-out (parallel branches)
edges = [('START', (branch_a, branch_b, branch_c))]

# Fan-in with JoinNode
from google.adk.workflow import JoinNode
join = JoinNode(name="merge")
edges = [((branch_a, branch_b), join), (join, final)]
# JoinNode output: {"branch_a": output_a, "branch_b": output_b}

# Looping (must have at least one routed edge — unconditional cycles rejected)
edges = [
    ('START', process),
    (process, check),
    (check, process, "continue"),
    (check, finish, "exit"),
]
```

Route values: `str`, `bool`, `int`. Multi-route fan-out: `return Event(output=x, route=["a", "b"])`. Edge matching multiple routes: `(node, target, ["route_x", "route_y"])`.

---

## 5. LLM Agent Nodes

Use `google.adk.agents.LlmAgent` in workflow edges — auto-wrapped internally, emits `Event(output=...)` for downstream data passing.

```python
from google.adk.agents import LlmAgent
from pydantic import BaseModel

class DraftOutput(BaseModel):
  title: str
  content: str

writer = LlmAgent(
    name="writer",
    model="gemini-3.8-flash",
    instruction="Write a draft based on the user's request.",
    output_schema=DraftOutput,  # Always set for structured output
    output_key="draft",         # Also store in state['draft']
)

agent = Workflow(
    name="pipeline",
    edges=[('START', writer), (writer, process_draft)],
)
```

**Always use `output_schema`** (Pydantic model) on LLM agents in workflows. Without it, output is `types.Content` which may cause type errors in downstream function nodes or serialization failures with JoinNode/database sessions.

---

## 6. Parallel Processing

### ParallelWorker — process list items concurrently

```python
from google.adk.workflow import node

@node(parallel_worker=True)
def process_item(node_input: int) -> int:
  return node_input * 2

# Input: [1, 2, 3] -> Output: [2, 4, 6]
agent = Workflow(
    name="parallel",
    edges=[('START', split_input), (split_input, process_item), (process_item, collect)],
)
```

Workers named `{parent_name}@{index}`. Input must be a list. Output is a list in same order.

### Fan-Out / Fan-In — diamond pattern

```python
from google.adk.workflow import JoinNode
join = JoinNode(name="merge")
edges = [
    ('START', splitter),
    (splitter, (branch_a, branch_b)),
    ((branch_a, branch_b), join),
    (join, combiner),  # combiner receives {"branch_a": ..., "branch_b": ...}
]
```

---

## 7. Human-in-the-Loop (HITL)

HITL is a general ADK feature (app-level `ResumabilityConfig`, the `request_input` tool, `resume_inputs`) — see `adk-python.md` "Human-in-the-Loop". Inside a workflow it works per node: a node yields `RequestInput` and reads replies from `ctx.resume_inputs` (keyed by `interrupt_id`).

```python
from google.adk.events.request_input import RequestInput

async def multi_step(ctx: Context, node_input: str):
  if not ctx.resume_inputs:
    yield RequestInput(interrupt_id="ask_name", message="Name?")
    return
  if "ask_email" not in ctx.resume_inputs:
    yield RequestInput(interrupt_id="ask_email", message="Email?")
    return
  yield Event(output={"name": ctx.resume_inputs["ask_name"],
                      "email": ctx.resume_inputs["ask_email"]})
```

**Node resume behavior:** `rerun_on_resume=False` (default FunctionNode) → the user's response becomes the node output; `rerun_on_resume=True` (default LlmAgent) → the node reruns with `ctx.resume_inputs` populated. **In loops:** use a unique `interrupt_id` per iteration (e.g. `f'review_{count}'`) to avoid infinite restarts.

---

## 8. State & Events

### Context Properties

```python
from google.adk.agents.context import Context

def my_node(ctx: Context, node_input: str) -> str:
  ctx.state.get("key", "default")   # Read state
  ctx.session.id                     # Session ID
  ctx.node_path                      # "Workflow/node_name"
  ctx.node                           # Current node
  ctx.run_id                         # Current execution ID
  ctx.attempt_count                  # 1 on first attempt (1-based)
  ctx.resume_inputs                  # HITL resume data (dict keyed by interrupt_id)
  ctx.interrupt_ids                  # Active interrupt IDs
  ctx.output                         # Node's result value (settable)
  ctx.route                          # Routing value (settable)
  return "result"
```

### Dynamic Node Scheduling

```python
async def orchestrator(ctx: Context, node_input: list) -> list:
  results = []
  for i, item in enumerate(node_input):
    result = await ctx.run_node(process_item, node_input=item)
    results.append(result)
  return results
```

`ctx.run_node()` requires `rerun_on_resume=True` on the calling node. Use `use_as_output=True` to delegate the node's output to the dynamic child.

### Event Fields

| Field | Type | Description |
|-------|------|-------------|
| `output` | `Any` | Output data for downstream nodes (must be JSON-serializable) |
| `route` | `str\|bool\|int\|list` | Routing signal for conditional edges (sets `actions.route`) |
| `state` | `dict` | State delta to apply (sets `actions.state_delta`) |
| `content` | `types.Content` | Content for web UI display |
| `message` | `ContentUnion` | Alias for content (auto-converted) |

### State: Prefer Event over ctx.state

```python
# Preferred — persisted in event history, replayable
def save(node_input: str):
  return Event(output=node_input, state={"key": node_input})

# Avoid — side effect, may be lost on replay
def save(ctx: Context, node_input: str) -> str:
  ctx.state["key"] = node_input
  return node_input
```

### Data Serialization Rules

- `Event.output` must be JSON-serializable. BaseModel returns auto-converted via `model_dump()`.
- `output_key` stores dicts (not BaseModel instances) — `validate_schema()` -> `model_dump()`.
- `ctx.state.get(key)` returns a dict. Use `MyModel(**data)` to reconstruct typed access.

---

## 9. Retry Configuration

```python
from google.adk.workflow import FunctionNode, RetryConfig

node = FunctionNode(
    func=flaky_call,           # func is keyword-only
    retry_config=RetryConfig(
        max_attempts=5,        # Default: None (treated as 5); 1 = no retry
        initial_delay=1.0,     # Seconds before first retry
        max_delay=60.0,        # Max seconds between retries
        backoff_factor=2.0,    # Delay multiplier per attempt
        jitter=1.0,            # Randomness factor (0.0 = none)
        exceptions=None,       # Exception types to retry (None = all)
    ),
)
```

Delay formula: `min(initial_delay * backoff_factor^attempt, max_delay) * (1 + random(0, jitter))`

---

## 10. Testing

> **Note:** The testing utilities below (`testing_utils`, `InMemoryRunner`) are internal to the ADK repository. They are not part of the public `google-adk` package. For your own tests, use `App` + `InMemoryRunner` from `google.adk.runners` or write a custom test harness.

```python
import pytest
from google.adk.workflow import Workflow
from google.adk.apps import App
from google.adk.runners import InMemoryRunner
from google.genai import types

@pytest.mark.asyncio
async def test_workflow():
  def step(node_input: str) -> str:
    return "done"

  agent = Workflow(name="test", edges=[('START', step)])
  app = App(name="test_app", root_agent=agent)
  runner = InMemoryRunner(app=app)

  session = await runner.session_service.create_session(
      app_name="test_app", user_id="test_user"
  )
  async for event in runner.run_async(
      user_id="test_user",
      session_id=session.id,
      new_message=types.Content(role="user", parts=[types.Part.from_text(text="hello")]),
  ):
    if event.output is not None:
      assert event.output == "done"
```

---

## 11. Import Paths

### Workflow Core

| Component | Import |
|-----------|--------|
| `Workflow` | `from google.adk.workflow import Workflow` |
| `Edge` | `from google.adk.workflow import Edge` |
| `FunctionNode` | `from google.adk.workflow import FunctionNode` |
| `JoinNode` | `from google.adk.workflow import JoinNode` |
| `BaseNode`, `START` | `from google.adk.workflow import BaseNode, START` |
| `Node` (subclassable) | `from google.adk.workflow import Node` |
| `@node` decorator | `from google.adk.workflow import node` |
| `RetryConfig` | `from google.adk.workflow import RetryConfig` |
| `NodeTimeoutError` | `from google.adk.workflow import NodeTimeoutError` |
| `DEFAULT_ROUTE` | `from google.adk.workflow import DEFAULT_ROUTE` |

### Workflow Nodes (auto-wrapped)

Nodes are auto-wrapped when placed in edges. You do not need to import wrapper classes.

| Python Object | How to Use |
|--------------|-----------|
| `LlmAgent` | `from google.adk.agents import LlmAgent` — place directly in edges |
| Function/callable | Use as-is or wrap with `@node` decorator for options |
| `BaseTool` | Place directly in edges |
| `BaseAgent` subclass | Place directly in edges |

### Events & Context

| Component | Import |
|-----------|--------|
| `Event` | `from google.adk.events.event import Event` |
| `RequestInput` | `from google.adk.events.request_input import RequestInput` |
| `Context` | `from google.adk.agents.context import Context` |

### LLM Agent

| Component | Import |
|-----------|--------|
| `LlmAgent` | `from google.adk.agents import LlmAgent` |

### App & Resumability

| Component | Import |
|-----------|--------|
| `App` | `from google.adk.apps import App` |
| `ResumabilityConfig` | `from google.adk.apps import ResumabilityConfig` |

---

## 12. Best Practices

### Use Pydantic Models, Not Raw Dicts

Always define `BaseModel` classes for node I/O, LLM `output_schema`, and structured data:

```python
# Wrong: raw dicts
def lookup(node_input: dict[str, Any]) -> dict[str, Any]:
  return {"cost": 500}

# Correct: typed schemas
class FlightInfo(BaseModel):
  cost: int
  details: str

def lookup(node_input: Itinerary) -> FlightInfo:
  return FlightInfo(cost=500, details="Economy")
```

### Emit Content Events for Web UI

`event.output` is internal — only `event.content` renders in the ADK web UI:

```python
from google.genai import types

def final_output(node_input: str):
  yield Event(content=types.Content(role='model', parts=[types.Part.from_text(text=node_input)]))
  yield Event(output=node_input)
```

LLM agents emit content events automatically. Add them explicitly for function nodes with user-facing results.

### Agent Directory Convention

```
my_workflow/
  __init__.py    # from . import agent
  agent.py       # root_agent = Workflow(...)
```

### Advanced Patterns

- **Nested workflows**: A `Workflow` can be used as a node in another workflow
- **Dynamic node scheduling**: Use `await ctx.run_node(func, node_input=item)` at runtime (requires `rerun_on_resume=True`)
- **Custom Node subclass**: Subclass `Node`, implement `run_node_impl(*, ctx, node_input)` -> `AsyncGenerator`. Supports `parallel_worker=True` flag.
- **Custom BaseNode**: Subclass `BaseNode`, implement `_run_impl(*, ctx, node_input)` -> `AsyncGenerator`

### Graph Validation Rules

1. START must exist and have no incoming edges
2. All non-START nodes must be reachable
3. No duplicate node names or edges
4. At most one `__DEFAULT__` route per node
5. No unconditional cycles (cycles need at least one routed edge)

---

## Further Reading

- [Workflows overview](https://adk.dev/workflows/index.md)
- [Graph routes & conditional edges](https://adk.dev/graphs/routes/index.md)
- [Agent collaboration & task mode](https://adk.dev/workflows/collaboration/index.md)
- [Data handling & state](https://adk.dev/graphs/data-handling/index.md)
- [Dynamic workflows](https://adk.dev/graphs/dynamic/index.md)
- [Human-in-the-loop](https://adk.dev/graphs/human-input/index.md)
- [App class (workflow container)](https://adk.dev/apps/index.md)

# ADK Python Cheatsheet

## 1. Core Concepts & Project Structure

### Essential Primitives

*   **`Agent`**: The core intelligent unit. Can be `LlmAgent` (LLM-driven) or `BaseAgent` (custom/workflow).
*   **`Tool`**: Callable function providing external capabilities (`FunctionTool`, `AgentTool`, etc.).
*   **`Session`**: A stateful conversation thread with history (`events`) and short-term memory (`state`).
*   **`State`**: Key-value dictionary within a `Session` for transient conversation data.
*   **`Runner`**: The execution engine; orchestrates agent activity and event flow.
*   **`Event`**: Atomic unit of communication; carries content and side-effect `actions`.

### Standard Project Layout

```text
your_project_root/
├── <agent_name>/ or app/    # Agent code directory
│   ├── __init__.py
│   ├── agent.py            # Contains root_agent definition
│   ├── tools.py            # Custom tool functions
│   └── .env                # Environment variables
├── tests/
│   ├── eval/
│   │   ├── eval_config.yaml    # Eval criteria and thresholds
│   │   └── datasets/           # Eval datasets (JSON)
│   ├── integration/
│   └── unit/
└── pyproject.toml or requirements.txt
```

---

## 2. Agent Definitions (`LlmAgent`)

### Basic Setup

```python
from google.adk.agents import Agent

def get_weather(city: str) -> dict:
    """Returns weather for a city."""
    return {"status": "success", "weather": "sunny", "temp": 72}

my_agent = Agent(
    name="weather_agent",
    model="gemini-3.8-flash",
    instruction="You help users check the weather. Use the get_weather tool.",
    description="Provides weather information.",  # Important for multi-agent delegation
    tools=[get_weather]
)
```

### Key Configuration Options

```python
from google.genai import types as genai_types
from google.adk.agents import Agent

agent = Agent(
    name="my_agent",
    model="gemini-3.8-flash",
    instruction="Your instructions here. Use {state_key} for dynamic injection.",
    description="Description for delegation.",

    # LLM generation parameters
    generate_content_config=genai_types.GenerateContentConfig(
        temperature=0.2,
        max_output_tokens=1024,
    ),

    # Save final output to state
    output_key="agent_response",

    # Control history sent to LLM
    include_contents='default',  # 'default' or 'none'

    # Delegation control
    disallow_transfer_to_parent=False,
    disallow_transfer_to_peers=False,

    # Sub-agents for delegation
    sub_agents=[specialist_agent],

    # Tools
    tools=[my_tool],

    # Callbacks
    before_agent_callback=my_callback,
    after_agent_callback=my_callback,
    before_model_callback=my_callback,
    after_model_callback=my_callback,
    before_tool_callback=my_callback,
    after_tool_callback=my_callback,
)
```

### Structured Output with Pydantic

> **Warning**: Using `output_schema` disables tool calling and delegation.

```python
from pydantic import BaseModel, Field
from typing import Literal

class Evaluation(BaseModel):
    grade: Literal["pass", "fail"] = Field(description="The evaluation result.")
    comment: str = Field(description="Explanation of the grade.")

evaluator = Agent(
    name="evaluator",
    model="gemini-3.8-flash",
    instruction="Evaluate the input and provide structured feedback.",
    output_schema=Evaluation,
    output_key="evaluation_result",
)
```

### Instruction Best Practices

```python
# Use dynamic state injection with {state_key} placeholders
instruction = """
You are a {role} assistant.
User preferences: {user_preferences}

Rules:
- Always use tools when available
- Never make up information
"""
```

---

## 3. Orchestration with Workflow Agents

Workflow agents provide deterministic control flow without LLM orchestration.

> These are `BaseAgent`-family composites (`SequentialAgent`, `ParallelAgent`, `LoopAgent`). For the new graph-based Workflow API introduced in ADK 2.0, see `references/adk-python-workflows.md`.

### SequentialAgent

Executes sub-agents in order. State changes propagate to subsequent agents.

```python
from google.adk.agents import SequentialAgent, Agent

summarizer = Agent(
    name="summarizer",
    model="gemini-3.8-flash",
    instruction="Summarize the input.",
    output_key="summary"
)

question_gen = Agent(
    name="question_generator",
    model="gemini-3.8-flash",
    instruction="Generate questions based on: {summary}"
)

pipeline = SequentialAgent(
    name="pipeline",
    sub_agents=[summarizer, question_gen],
)
```

### ParallelAgent

Executes sub-agents concurrently. Use distinct `output_key`s to avoid race conditions.

```python
from google.adk.agents import ParallelAgent, SequentialAgent, Agent

fetch_a = Agent(name="fetch_a", ..., output_key="data_a")
fetch_b = Agent(name="fetch_b", ..., output_key="data_b")

merger = Agent(
    name="merger",
    instruction="Combine data_a: {data_a} and data_b: {data_b}"
)

pipeline = SequentialAgent(
    name="full_pipeline",
    sub_agents=[
        ParallelAgent(name="fetchers", sub_agents=[fetch_a, fetch_b]),
        merger
    ]
)
```

### LoopAgent

Repeats sub-agents until `max_iterations` or an event with `escalate=True`.

```python
from google.adk.agents import LoopAgent

refinement_loop = LoopAgent(
    name="refinement_loop",
    sub_agents=[evaluator, refiner, escalation_checker],
    max_iterations=5,
)
```

For a production LoopAgent with EscalationChecker, BuiltInPlanner, and grounding citations, look it up in the topic index in `references/samples.md`.

---

## 4. Multi-Agent Systems & Communication

### Communication Methods

1.  **Shared State**: Agents read/write `session.state`. Use `output_key` for convenience.

2.  **LLM Delegation**: Agent transfers control to a sub-agent based on reasoning.
    ```python
    coordinator = Agent(
        name="coordinator",
        instruction="Route to sales_agent for sales, support_agent for help.",
        sub_agents=[sales_agent, support_agent],
    )
    ```

3.  **AgentTool**: Invoke another agent as a tool (parent stays in control).
    ```python
    from google.adk.tools import AgentTool

    root = Agent(
        name="root",
        tools=[AgentTool(specialist_agent)],
    )
    ```

4.  **Task Delegation (ADK 2.0)**: Set `mode` on a sub-agent for structured, schema-typed delegation — the coordinator gets a `request_task_{name}` tool; the sub-agent returns typed output via the auto-injected `finish_task` tool.
    ```python
    from pydantic import BaseModel

    class ResearchOutput(BaseModel):
        summary: str

    researcher = Agent(
        name="researcher",
        model="gemini-3.8-flash",
        mode="task",                        # 'chat' (default) | 'task' | 'single_turn'
        output_schema=ResearchOutput,
        description="Researches a topic.",  # required for delegation
        instruction="Research the topic, then call finish_task.",
    )
    coordinator = Agent(name="coordinator", model="gemini-3.8-flash", sub_agents=[researcher])
    ```
    Modes: `task` (multi-turn, structured I/O) · `single_turn` (autonomous, no user turn). Sub-agents need a `description`; default I/O schemas (`goal`/`background` in, `result` out) are used if none set. Disabled inside graph `Workflow`s.

---

## 5. Building Custom Agents (`BaseAgent`)

For custom orchestration logic beyond workflow agents.

```python
from google.adk.agents import BaseAgent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.events import Event, EventActions
from typing import AsyncGenerator

class ConditionalRouter(BaseAgent):
    async def _run_async_impl(
        self, ctx: InvocationContext
    ) -> AsyncGenerator[Event, None]:
        # Read state
        user_type = ctx.session.state.get("user_type", "regular")

        # Custom routing logic
        if user_type == "premium":
            agent = self.premium_agent
        else:
            agent = self.regular_agent

        # Run selected agent
        async for event in agent.run_async(ctx):
            yield event

class EscalationChecker(BaseAgent):
    """Stops a LoopAgent when condition is met."""
    async def _run_async_impl(
        self, ctx: InvocationContext
    ) -> AsyncGenerator[Event, None]:
        result = ctx.session.state.get("evaluation")
        if result and result.get("grade") == "pass":
            yield Event(author=self.name, actions=EventActions(escalate=True))
        else:
            yield Event(author=self.name)
```

---

## 6. Models Configuration

### Google Gemini (Default)

```python
# AI Studio (dev): in the project .env, comment the GOOGLE_* lines and
# uncomment GEMINI_API_KEY (GOOGLE_API_KEY is also accepted).

# Gemini platform (prod)
# Set: GOOGLE_CLOUD_PROJECT, GOOGLE_CLOUD_LOCATION, GOOGLE_GENAI_USE_ENTERPRISE=True

agent = Agent(model="gemini-3.8-flash", ...)
```

### Other Models via LiteLLM

```python
from google.adk.models.lite_llm import LiteLlm

agent = Agent(model=LiteLlm(model="openai/gpt-4o"), ...)
agent = Agent(model=LiteLlm(model="anthropic/claude-sonnet-4-20250514"), ...)
agent = Agent(model=LiteLlm(model="ollama_chat/llama3:instruct"), ...)
```

### Gemini platform native models

```python
from google.adk.models import Gemini

# Gemini hosted on Gemini platform (set GOOGLE_GENAI_USE_ENTERPRISE=True)
agent = Agent(model=Gemini(model="gemini-3.8-flash"), ...)
```

Provider guides: [Anthropic](https://adk.dev/agents/models/anthropic/index.md), [Ollama](https://adk.dev/agents/models/ollama/index.md), [vLLM](https://adk.dev/agents/models/vllm/index.md), [LiteLLM](https://adk.dev/agents/models/litellm/index.md)

---

## 7. Tools: The Agent's Capabilities

### Function Tool Basics

```python
from google.adk.tools import ToolContext

def search_database(
    query: str,
    limit: int,
    tool_context: ToolContext  # Optional, for state access
) -> dict:
    """Searches the database for records matching the query.

    Args:
        query: The search query string.
        limit: Maximum number of results to return.

    Returns:
        dict with 'status' and 'results' keys.
    """
    # Access state if needed
    user_id = tool_context.state.get("user_id")

    # Tool logic here
    results = db.search(query, limit=limit, user=user_id)

    return {"status": "success", "results": results}
```

**Tool Rules:**
- Use clear docstrings (sent to LLM)
- Type hints required, NO default values
- Return a dict (JSON-serializable)
- Don't mention `tool_context` in docstring

### ToolContext Capabilities

```python
async def my_tool(query: str, tool_context: ToolContext) -> dict:
    # Read/write state
    tool_context.state["key"] = "value"

    # Trigger escalation (stops LoopAgent)
    tool_context.actions.escalate = True

    # Artifacts — see Artifacts section below for full API
    await tool_context.save_artifact("file.txt", part)

    # Memory search
    results = await tool_context.search_memory("query")

    return {"status": "success"}
```

### Built-in Tools

```python
from google.adk.tools import google_search
from google.adk.tools import VertexAiSearchTool
from google.adk.tools.load_web_page import load_web_page
from google.adk.code_executors import BuiltInCodeExecutor

# Google Search grounding
agent = Agent(tools=[google_search], ...)

# Grounding with Search in Gemini platform (your own data)
agent = Agent(tools=[VertexAiSearchTool(data_store_id="projects/P/locations/L/collections/default_collection/dataStores/DS")], ...)

# Web page loading
agent = Agent(tools=[load_web_page], ...)

# Code execution (model-internal)
agent = Agent(code_executor=BuiltInCodeExecutor(), ...)

# Managed sandbox (Code Interpreter in Gemini platform). For a per-user sandbox an agent works
# in across sessions, this primitive is not it — see the topic index in references/samples.md
# from google.adk.code_executors import VertexAiCodeExecutor
# agent = Agent(code_executor=VertexAiCodeExecutor(optimize_data_file=True, stateful=True), ...)

```

> **`google_search` is model-internal grounding, not a regular tool.** Mixing it with FunctionTools disables Automatic Function Calling (AFC) for all tools. If you need search alongside custom tools, consider a sub-agent architecture or a custom search function — see the [deep-search sample](https://github.com/google/adk-samples/tree/main/core/python/deep-search) for a working pattern. For eval implications, see the eval guide's `builtin-tools-eval` reference.

### Tool Confirmation

```python
from google.adk.tools import FunctionTool

# Simple confirmation
sensitive_tool = FunctionTool(delete_record, require_confirmation=True)

# Conditional confirmation
def needs_approval(amount: float, **kwargs) -> bool:
    return amount > 1000

transfer_tool = FunctionTool(transfer_money, require_confirmation=needs_approval)
```

### Human-in-the-Loop (pause & resume)

Pause a run to ask the user something, then resume. This is a general runtime feature (not workflow-specific). Enable resumption at the app level:

```python
from google.adk.apps import App, ResumabilityConfig

app = App(name="my_app", root_agent=root_agent,
          resumability_config=ResumabilityConfig(is_resumable=True))
```

- **Let the model ask:** add the built-in `request_input` tool (`from google.adk.tools import request_input`) to `tools=` — the model calls it when it needs clarification.
- **Approval gate inside a tool:** `tool_context.request_confirmation(hint="Approve this transfer?")`, or `FunctionTool(fn, require_confirmation=...)` (above).
- **Custom long-running tool:** wrap a function with `LongRunningFunctionTool(fn)` to pause until an external result arrives.

The user's reply is read from `ctx.resume_inputs` (available on `ToolContext` and `CallbackContext`). Inside graph workflows the same mechanism is node-based — see `adk-python-workflows.md` §7.

### Tool Authentication

| Auth Type | Pattern |
|-----------|---------|
| API Key | `token_to_scheme_credential("apikey", "query", "apikey", "KEY")` → `auth_scheme, auth_credential` |
| Service Account | `service_account_dict_to_scheme_credential(config, scopes=[...])` → `auth_scheme, auth_credential` |
| OAuth2 / OIDC | `AuthCredential(auth_type=AuthCredentialTypes.OAUTH2, oauth2=OAuth2Auth(client_id=..., client_secret=...))` |
| Custom FunctionTool | `tool_context.request_credential(AuthConfig(...))` to initiate, `tool_context.get_auth_response(AuthConfig(...))` to retrieve |

Helpers: `from google.adk.tools.openapi_tool.auth.auth_helpers import token_to_scheme_credential, service_account_dict_to_scheme_credential`. Pass `auth_scheme` + `auth_credential` to `OpenAPIToolset(...)`. [Full docs](https://adk.dev/tools-custom/authentication/)

### OpenAPI Tools

```python
from google.adk.tools.openapi_tool.openapi_spec_parser.openapi_toolset import OpenAPIToolset

toolset = OpenAPIToolset(spec_str=open("openapi.json").read(), spec_str_type="json")
agent = Agent(name="api_agent", tools=[toolset], ...)
```

Pass `auth_scheme` + `auth_credential` from the auth helpers above for authenticated APIs. Tool names derive from `operationId` (snake_case, max 60 chars). [Full docs](https://adk.dev/tools-custom/openapi-tools/index.md)

### MCP Tools

Connect to MCP servers to use external tools (needs the `mcp` extra: scaffolded projects ship `google-adk[gcp,otel-gcp]`, so add `mcp` and re-sync). Use `StdioConnectionParams` for local dev, `StreamableHTTPConnectionParams` for remote HTTP servers.

```python
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams, StreamableHTTPConnectionParams
from mcp import StdioServerParameters

# Local MCP server via stdio
agent = Agent(
    name="my_agent",
    tools=[
        McpToolset(
            connection_params=StdioConnectionParams(
                server_params=StdioServerParameters(
                    command="npx",
                    args=["-y", "@modelcontextprotocol/server-filesystem", "/absolute/path"],
                ),
            ),
            tool_filter=["list_directory", "read_file"],  # optional: restrict exposed tools
        )
    ],
    ...
)

# Remote MCP server, e.g. Cloud Run with --no-allow-unauthenticated. ID tokens
# expire in ~1h, so mint per call via `header_provider` (ADK calls it on every
# tool call); a static `headers` dict goes stale. Audience = root, not /mcp.
from google.auth.transport.requests import Request
from google.oauth2.id_token import fetch_id_token

McpToolset(
    connection_params=StreamableHTTPConnectionParams(url=f"{MCP_SERVER_URL}/mcp"),
    header_provider=lambda ctx: {
        "Authorization": f"Bearer {fetch_id_token(Request(), MCP_SERVER_URL)}"
    },
)
```

**Gotchas:**
- Paths must be absolute, not relative.
- Agent definition must be synchronous (not async) for deployment.
- Node.js/npx required for npm-based MCP servers — add to Dockerfile if containerizing.

---

## 8. Context, State, and Memory

| Need | Solution |
|---|---|
| Within one conversation (task data, form state) | Session state — see [State Prefixes](#state-prefixes) and [Session Service Options](#session-service-options) below |
| Across conversations (remember interactions, learn over time) | Memory Bank — see [Memory](#memory-long-term-knowledge) below |

### State Prefixes

```python
# Session-specific (default)
state["booking_step"] = 2

# User-persistent (across sessions)
state["user:preferred_language"] = "en"

# App-wide (all users)
state["app:total_queries"] = 1000

# Temporary (current invocation only)
state["temp:intermediate_result"] = data
```

### Session Service Options

```python
from google.adk.sessions import InMemorySessionService
# For dev: InMemorySessionService()
# For prod: VertexAiSessionService(), DatabaseSessionService()
```

### Session Rewind

Roll back a session to the state before a specific invocation (useful for debugging or user-initiated undo):

```python
from google.adk.runners import InMemoryRunner

runner = InMemoryRunner(agent=root_agent, app_name="my_app")

# Rewind to state before a given invocation
await runner.rewind_async(
    user_id=user_id,
    session_id=session.id,
    rewind_before_invocation_id=invocation_id,  # exclusive: state before this call
)
```

> **Note**: Restores session-level state and artifacts only; app/user-scoped state is unaffected.

### Artifacts (File Storage)

Store and retrieve binary data (PDFs, images, audio) scoped to session or user:

```python
from google.adk.artifacts import InMemoryArtifactService, GcsArtifactService
from google.genai import types

# Configure runner with artifact service
runner = Runner(
    agent=root_agent,
    app_name="app",
    session_service=session_service,
    artifact_service=InMemoryArtifactService(),  # or GcsArtifactService(bucket_name="my-bucket")
)

# In a tool or callback:
async def save_file(data: bytes, tool_context: ToolContext) -> dict:
    part = types.Part(inline_data=types.Blob(mime_type="application/pdf", data=data))
    version = await tool_context.save_artifact("report.pdf", part)       # session-scoped
    await tool_context.save_artifact("user:profile.png", part)           # user-scoped
    artifact = await tool_context.load_artifact("report.pdf")            # latest version
    artifact_v0 = await tool_context.load_artifact("report.pdf", version=0)
    names = await tool_context.list_artifacts()
    return {"status": "saved", "version": version}
```

**Namespace prefixes:** plain name = session-scoped · `"user:"` = persistent across sessions

### Memory (Long-term Knowledge)

#### InMemoryMemoryService (Dev)

In-memory implementation for local development. Memories don't persist across restarts.

```python
from google.adk.memory import InMemoryMemoryService

memory_service = InMemoryMemoryService()
# Add session to memory after conversation
await memory_service.add_session_to_memory(session)
# Search later
results = await memory_service.search_memory(app_name=app_name, user_id=user_id, query="query")
```

#### Memory Bank (Long-term Memory)

Managed cross-session memory that persists user preferences, remembers facts across sessions, and learns from conversations over time. See the [`cross-session-memory` recipe](https://github.com/google/adk-samples/tree/main/core/python/cross-session-memory) for a complete implementation.

```python
from google.adk.agents.callback_context import CallbackContext
from google.adk.tools.preload_memory_tool import PreloadMemoryTool

# PreloadMemoryTool retrieves memories at the start of each turn and injects
# them into the system instruction. Alternative: LoadMemoryTool() — the model
# calls it on-demand when it decides memories are needed.
root_agent = Agent(
    ...,
    tools=[PreloadMemoryTool()],
    after_agent_callback=generate_memories_callback,
)

# Alternative: callback_context.add_events_to_memory(events=...) to send only
# a subset of events, which is better for incremental processing.
async def generate_memories_callback(callback_context: CallbackContext):
    """Sends the session's events to Memory Bank for memory generation."""
    await callback_context.add_session_to_memory()
    return None
```

### Context Caching

Cache large context windows (system prompt + docs) to reduce latency and cost. Transparent to agent code.

```python
from google.adk.apps import App
from google.adk.agents.context_cache_config import ContextCacheConfig

app = App(
    name="my_app",
    root_agent=root_agent,
    context_cache_config=ContextCacheConfig(
        min_tokens=2048,     # only cache if context exceeds this
        ttl_seconds=1800,    # cache lifetime (default 1800)
        cache_intervals=10,  # re-cache every N invocations
    ),
)
```

### Context Compaction

Prevent context overflow on long sessions by compacting older events into summaries. Use **token-based** compaction: it triggers on actual prompt-token volume, so it handles unpredictable inputs (pasted code, large tool results) better than a fixed turn count.

```python
from google.adk.apps import App
from google.adk.apps.app import EventsCompactionConfig
from google.adk.apps.llm_event_summarizer import LlmEventSummarizer
from google.adk.models import Gemini

app = App(
    name="my_app",
    root_agent=root_agent,
    events_compaction_config=EventsCompactionConfig(
        token_threshold=32000,   # compact once prompt tokens reach this
        event_retention_size=5,  # keep the last 5 raw events un-compacted
        # Optional: custom summarizer model
        summarizer=LlmEventSummarizer(llm=Gemini(model="gemini-3.8-flash")),
    ),
)
```

### App Name

The `App(name=...)` parameter **must match the agent directory name** (default: `app`). A mismatch causes "Session not found" errors during evaluation because the runner infers the app name from the directory path.

```python
# CORRECT — matches the "app" directory
app = App(name="app", root_agent=root_agent)

# WRONG — causes eval failures
app = App(name="my_custom_agent", root_agent=root_agent)
```

---

## 9. Callbacks

### Callback Types

```python
from google.adk.agents.callback_context import CallbackContext
from google.adk.models.llm_request import LlmRequest
from google.adk.models.llm_response import LlmResponse
from google.adk.tools import BaseTool, ToolContext
from google.genai import types as genai_types

# Callbacks are invoked by keyword — parameter names must match exactly.

# Agent lifecycle
async def before_agent_callback(callback_context: CallbackContext) -> None:
    callback_context.state["started"] = True

async def after_agent_callback(callback_context: CallbackContext) -> genai_types.Content | None:
    # Return None to continue, or Content to override
    return None

# Model interaction
async def before_model_callback(callback_context: CallbackContext, llm_request: LlmRequest) -> LlmResponse | None:
    # Return None to continue, or LlmResponse to skip model call
    return None

async def after_model_callback(callback_context: CallbackContext, llm_response: LlmResponse) -> LlmResponse | None:
    # Return None to continue, or modified LlmResponse
    return None

# Tool execution
async def before_tool_callback(tool: BaseTool, args: dict, tool_context: ToolContext) -> dict | None:
    # Return None to continue, or dict to skip tool and use as result
    return None

async def after_tool_callback(tool: BaseTool, args: dict, tool_context: ToolContext, tool_response: dict) -> dict | None:
    # Return None to continue, or modified dict
    return None
```

### Common Pattern

```python
# Initialize state before agent runs
async def init_state(callback_context: CallbackContext) -> None:
    if "preferences" not in callback_context.state:
        callback_context.state["preferences"] = {}

agent = Agent(before_agent_callback=init_state, ...)
```

---

## 10. Plugins

Global callback hooks across all agents/tools/LLMs. Use for cross-cutting concerns (logging, guardrails); use callbacks for per-agent logic.

```python
from google.adk.plugins.base_plugin import BasePlugin
from google.adk.apps import App

class MyPlugin(BasePlugin):
    async def before_model_callback(self, *, callback_context, llm_request):
        return None  # return None to observe, return value to intervene

# Register via App — plugins run BEFORE agent-level callbacks
app = App(name="my_app", root_agent=root_agent, plugins=[MyPlugin()])
runner = Runner(app=app, session_service=...)
```

Built-in plugins: `ReflectAndRetryToolPlugin` (retry failed tools), `BigQueryAgentAnalyticsPlugin` (log to BQ), `ContextFilterPlugin` (reduce context size), `GlobalInstructionPlugin` (shared system prompt), `SaveFilesAsArtifactsPlugin`, `LoggingPlugin`, `DebugLoggingPlugin`, `MultimodalToolResultsPlugin`.

Hooks: `before/after_agent_callback`, `before/after_model_callback`, `before/after_tool_callback`, `on_model_error_callback`, `on_tool_error_callback`, `on_user_message_callback`, `before/after_run_callback`, `on_event_callback`. [Full docs](https://adk.dev/plugins/index.md)

### Safety Guardrails

Use `before_model_callback` to filter input or `after_model_callback` to filter output. Return `None` to pass through, or return a modified `LlmResponse` to block/replace. Evaluate with the `safety` metric. [Full docs](https://adk.dev/safety/index.md)

---

## 11. A2A Protocol

Requires `pip install google-adk[a2a]`.

```python
# Expose an agent as an A2A service
# Prefer scaffolding over manual code — scaffold a normal `adk` agent; A2A is built in (see /google-agents-cli-scaffold)
from google.adk.a2a.utils.agent_to_a2a import to_a2a
from a2a.types import AgentCard
to_a2a(root_agent, port=8001)

# Consume a remote A2A agent
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent, AGENT_CARD_WELL_KNOWN_PATH
remote = RemoteA2aAgent(
    name="remote_agent",
    description="...",
    agent_card=f"http://remote-host:8001{AGENT_CARD_WELL_KNOWN_PATH}",
)
```

### A2UI

Agents can return declarative UI via [a2ui](https://github.com/google/A2UI) (cards, forms, charts; rendered client-side over A2A) instead of plain text. Public preview; current release v0.9.1 (v1.0 release candidate). Docs: https://github.com/google/A2UI/tree/main/docs · ADK guide: https://adk.dev/integrations/a2ui/index.md

```python
# pip install a2ui-agent-sdk
from a2ui.core.schema.manager import A2uiSchemaManager
from a2ui.basic_catalog.provider import BasicCatalog
from a2ui.a2a import create_a2ui_part, parse_response_to_parts

# 1. Build the system prompt from a component catalog
manager = A2uiSchemaManager(...)        # loads catalog(s) + few-shot examples
instruction = manager.generate_system_prompt(...)

# 2. Use it as the agent instruction
root_agent = Agent(name="ui_agent", model="gemini-3.8-flash", instruction=instruction)

# 3. Validate the model's JSON output, then wrap as an A2A DataPart
#    (MIME application/a2ui+json) via a2ui.a2a before streaming to the client.
```

Runnable samples: https://github.com/google/A2UI/tree/main/samples/agent/adk

---

## 12. Event-Driven / Ambient Agents

Ambient agents process events (Pub/Sub, Eventarc, schedules) autonomously. ADK provides built-in trigger endpoints that handle payload decoding, session creation, concurrency, and retries.

> **Deployment:** `trigger_sources` registers `/apps/{app}/trigger/*` on the standard FastAPI app, so it works on **all** targets. On **Cloud Run** / **GKE** the endpoints are public HTTP routes you point a Pub/Sub push subscription or Eventarc trigger at. On **Agent Runtime in Gemini platform** the same routes are reachable through its `/api` passthrough (`https://{location}-aiplatform.googleapis.com/reasoningEngines/v1/{resource}/api/apps/{app}/trigger/pubsub`). The scaffolded `fast_api_app.py` does not pass `trigger_sources` by default — add it to enable these endpoints.

```python
from google.adk.cli.fast_api import get_fast_api_app

app = get_fast_api_app(
    agents_dir=AGENTS_DIR,
    web=False,
    trigger_sources=["pubsub", "eventarc"],  # enables /apps/{app}/trigger/pubsub and /apps/{app}/trigger/eventarc
)
```

```bash
# CLI equivalent for local dev
adk api_server --trigger_sources "pubsub,eventarc" path/to/your/agent
```

Trigger endpoints handle: base64 decoding, CloudEvent parsing, per-event session creation (UUID), concurrency semaphore, and exponential backoff on transient errors.

| Setting | Default | Environment Variable |
|---------|---------|----------------------|
| Max concurrent invocations | 10 | `ADK_TRIGGER_MAX_CONCURRENT` |
| Max retry attempts | 3 | `ADK_TRIGGER_MAX_RETRIES` |
| Base backoff delay | 1.0s | `ADK_TRIGGER_RETRY_BASE_DELAY` |
| Max backoff delay | 30.0s | `ADK_TRIGGER_RETRY_MAX_DELAY` |

Sessions are ephemeral by default (`InMemorySessionService`); use `DatabaseSessionService` for audit trails. Pub/Sub and Eventarc have a 10-minute processing limit. For non-GCP sources, use `adk api_server --auto_create_session` with the `/run` endpoint instead.

**Scheduled / cron execution:** Use Cloud Scheduler to publish to a Pub/Sub topic on a cron schedule, then connect the topic to the agent's `/apps/{app}/trigger/pubsub` endpoint. This is how you implement "run daily at 8 PM" — no custom scheduling code needed.

Since ambient agents have no interactive user, route outputs via structured logging (JSON stdout → Cloud Logging → Cloud Monitoring alerts), Pub/Sub, or tool-based integrations (email, Jira, Slack).

**Before implementing an ambient agent, clone and study the production sample** — it covers trigger wiring, middleware, structured logging, and Terraform. Look it up in the topic index in `references/samples.md`. [Full docs](https://adk.dev/runtime/ambient-agents/).

---

## 13. Managed Agents (server-hosted, first-party)

> **Requires ADK ≥ 2.4.0.** `ManagedAgent` connects to Google's first-party, server-hosted agents (e.g. the Antigravity agent) via the Managed Agents API: reasoning, tools, and execution all run in Google's managed environment, so there's no local sandbox to provision. It's a `BaseAgent`, so a standard `Runner` runs it like any other agent.

### When to use it

- **Managed agent** — powerful out-of-the-box capabilities (server-side web search, code execution) without operating the environment yourself. Trade-off: predefined server-side toolset, no client-side tools, runs only in the managed environment.
- **`LlmAgent` (§2)** — when you need control over the model, instructions, custom/MCP tools, or where execution happens.

### Setup

Two backends — satisfy the prerequisites for whichever you use, then supply an `agent_id`:
- **Gemini API:** set `GEMINI_API_KEY`. Use an out-of-the-box id (e.g. `antigravity-preview-05-2026`) or create your own (see below).
- **Gemini platform:** authenticate with ADC (`gcloud auth application-default login`). The Managed Agents API is served only from the `global` location, and `ManagedAgent` enforces it.

### Create & use

```python
from google import genai
from google.adk.agents import ManagedAgent
from google.adk.tools import google_search

# Create your own agent (google-genai SDK, NOT ADK — ManagedAgent has no create()).
# Get-or-create keeps it idempotent; or skip entirely and use an out-of-the-box id like "antigravity-preview-05-2026".
client = genai.Client()
if "researcher" not in {a.id for a in (client.agents.list().agents or [])}:  # id must be unique, no gemini-/google-/... prefixes
    client.agents.create(
        id="researcher", base_agent="antigravity-preview-05-2026",
        system_instruction="Answer with fresh, grounded info from the web.",
    )

# Connect + use. A ManagedAgent is a BaseAgent: set it as root_agent, drop it in a
# workflow, or wrap it as AgentTool. Only server-side tools are allowed.
managed = ManagedAgent(
    name="researcher", agent_id="researcher",
    environment={"type": "remote"},        # tools run in the managed sandbox
    tools=[google_search],                 # or types.Tool(code_execution=types.ToolCodeExecution())
)
```

### Limits

- **Client-side tools raise `NotImplementedError`:** Python functions/callables and client-side MCP (`McpToolset`). Server-side tools work — ADK built-ins, raw `types.Tool` configs, and server-side remote MCP via `RemoteMcpServer`.
- **Backends differ:** the Gemini API and Gemini platform behave slightly differently today — test against your target backend.

Docs: [Gemini API agents](https://ai.google.dev/gemini-api/docs/agents) · [Gemini platform managed agents](https://docs.cloud.google.com/gemini-enterprise-agent-platform/build/managed-agents) · [Interactions API](https://ai.google.dev/gemini-api/docs/interactions-overview) · [building custom agents](https://ai.google.dev/gemini-api/docs/custom-agents). Samples: [basic](https://github.com/google/adk-python/tree/main/contributing/samples/managed_agent/basic), [code execution](https://github.com/google/adk-python/tree/main/contributing/samples/managed_agent/code_execution).

---

## Quick Reference

### Running Agents Programmatically

```python
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

session_service = InMemorySessionService()
await session_service.create_session(app_name="app", user_id="user", session_id="s1")
runner = Runner(agent=my_agent, app_name="app", session_service=session_service)

async for event in runner.run_async(
    user_id="user", session_id="s1",
    new_message=types.Content(role="user", parts=[types.Part.from_text(text="Hello!")]),
):
    if event.is_final_response():
        print(event.content.parts[0].text)
```

### Running a Live (Voice) Agent

> Live agents use `run_live` + `LiveRequestQueue`, not `run_async`. For models,
> voice config, transcription, and serving see `references/adk-python-live.md`.

```python
from google.adk.agents.live_request_queue import LiveRequestQueue
from google.adk.agents.run_config import RunConfig
from google.genai import types

queue = LiveRequestQueue()  # the session must already exist
queue.send_content(types.Content(parts=[types.Part(text="Hello!")]))

async for event in runner.run_live(
    user_id="user", session_id="s1", live_request_queue=queue,
    run_config=RunConfig(response_modalities=["AUDIO"]),
):
    if event.turn_complete:
        break
queue.close()
```

### ADK Built-in Tool Imports (Precision Required)

```python
# CORRECT - imports the tool instance
from google.adk.tools.load_web_page import load_web_page

# WRONG - imports the module, not the tool
from google.adk.tools import load_web_page
```

Pass the imported tool directly to `tools=[load_web_page]`, not `tools=[load_web_page.load_web_page]`.

### Factory Functions for Sub-agents

Use factory functions (not module-level instances) to avoid "agent already has a parent" errors. Always **call** the factory — passing the function reference fails with `ValidationError: Input should be a valid dictionary or instance of BaseAgent`.

```python
def create_researcher():
    return Agent(name="researcher", ...)

root_agent = SequentialAgent(
    sub_agents=[create_researcher(), create_analyst()],  # call the functions!
    ...
)
```

Data flows between sequential sub-agents via conversation history and `output_key` state.

### Further Reading

- [ADK Documentation](https://adk.dev/llms.txt)
- [ADK Samples](https://github.com/google/adk-samples)
- `references/samples.md` — topic index of the reference recipes, and how to clone one

---

## Inspecting ADK Source Code

When you need to look up ADK internals, inspect the installed package directly:

```bash
# Find the ADK package location (use "uv run python" if using uv)
python -c "import google.adk; print(google.adk.__path__[0])"
```

### ADK Package Directory Map

```
google/adk/
├── agents/           # Agent types (LlmAgent, BaseAgent, SequentialAgent, etc.)
├── tools/            # Tool implementations (FunctionTool, google_search, etc.)
├── sessions/         # Session services (InMemory, Database, VertexAI)
├── memory/           # Memory services
├── runners.py        # Runner and execution engine
├── events/           # Event types and actions
├── models/           # Model integrations (Gemini, LiteLLM, etc.)
├── code_executors/   # Code execution (BuiltInCodeExecutor, etc.)
├── evaluation/       # Eval framework (criteria, evaluators, etc.)
├── cli/              # ADK CLI internals (used by agents-cli playground, eval, etc.)
├── flows/            # LLM flow implementations
├── artifacts/        # Artifact services
└── auth/             # Authentication helpers
```

Use Glob/Grep/Read on the installed package to find exact implementations, method signatures, and configuration options.

For the full ADK documentation index, use `curl https://adk.dev/llms.txt`.

# Reference recipes

Recipes live in [google/adk-samples](https://github.com/google/adk-samples). **`core/python/`** is the
curated tier — canonical ADK patterns maintained by the agents-cli team.

**Reading this page is not studying a recipe.** Every `core/` recipe ships an **`AGENTS.md`** — intent,
a ranked "study in this order" file tour, what to copy as-is versus what is recipe-specific,
and the gotchas. Clone the recipe and read its `AGENTS.md` before you implement anything it covers.
Until you have opened it you are answering from memory. Hand-writing a Docker or E2B sandbox wrapper,
a skill loader, a moderation callback or a memory store — for a capability the index lists — means
you stopped at the name.

**Study and adapt — don't scaffold from a recipe.**

```bash
[ -d /tmp/adk-samples ] || git clone --filter=blob:none --depth 1 --sparse \
  https://github.com/google/adk-samples /tmp/adk-samples
cd /tmp/adk-samples
git sparse-checkout add core/python/<recipe>
cat core/python/<recipe>/AGENTS.md
```

## Topic → recipe

Capabilities below are **not** scaffold flags — they come from studying a recipe and adapting it.

| You need | Study |
|---|---|
| Retrieval / search over your own documents (RAG) | `rag-agent-search` (managed ingestion) · `rag-vector-search` (custom chunking + embeddings) |
| Running shell commands or Python on a user's behalf; a sandboxed, isolated or per-user environment or workspace | `long-horizon-harness` |
| Agent-loadable skills — `SKILL.md` folders discovered at runtime, rebound mid-session, promoted and demoted from memory | `long-horizon-harness` |
| Long-running autonomy — works across days, resumes, unattended, compacts context | `long-horizon-harness` |
| Approval gate, escalation or human sign-off before a risky, high-value or irreversible action (human-in-the-loop) | `long-horizon-harness` (durable, mid-turn) · `ambient-expense-agent` (workflow pause) · `deep-search` (plan approval) |
| Memory across conversations | `cross-session-memory` (the primitive) · `long-horizon-harness` (self-improvement loop built on it) |
| Blocking harmful content or risky calls — moderation in one place, covering a coordinator and every sub-agent without editing them | `safety-plugins` (runner-wide plugins) · `long-horizon-harness` (per-tool guard chain + exfil detection) |
| Per-user credentials the model must never see | `long-horizon-harness` |
| OAuth user consent to act on a user's data | `oauth-user-consent-flow` |
| Sub-agent delegation with isolated context windows | `long-horizon-harness` |
| No chat interface — records or messages land on a queue and are processed automatically; event-driven, scheduled, batch or headless worker | `ambient-expense-agent` (Pub/Sub queue consumer) · `long-horizon-harness` (routines + scheduler) |
| Iterative research with cited sources | `deep-search` |
| Generating images or video — product photography, a model wearing the item (virtual try-on), 360° spins, background replacement — and MCP toolsets | `genmedia-for-commerce` |
| A2A interop, incl. Gemini app client quirks | `long-horizon-harness` |
| Monitoring a deployed agent's quality from its production traces — scheduled evaluation, recurring-failure tracking, root-cause analysis | `ambient-quality-agent` |

In Phase 1, clone the recipes named above and read `/tmp/adk-samples/core/python/<recipe>/AGENTS.md`
before you write any code. During Phase 0, naming them in the spec is enough — the clone waits for
approval. A bare how-question has no spec to wait for: clone before you answer it.

## The recipes

These ten are the **complete** set of `core/` python recipes. If a capability isn't listed here,
there is no core recipe for it — don't guess at a plausible name (`core/python/code-execution` and
`core/python/human-in-the-loop` do not exist). Check `contrib/` or build it yourself.

- **`long-horizon-harness`** — a complete agent *harness*: per-user sandbox, runtime-discovered `SKILL.md`
  skills, cross-session memory with a self-improvement loop, layered tool guardrails, sub-agent delegation
  with durable HITL, and per-user secrets. Its `AGENTS.md` maps each interface to the real function
  that implements it, so lift one pattern without adopting the whole harness.
  - Key files: `AGENTS.md`, `horizon/agent.py`, `horizon/fast_api_app.py`, `docs/architecture.md`, `docs/quickstart.md`
  - Keywords: harness, sandbox, shell execution, code execution, isolated environment, long-horizon, long-running, multi-day, autonomous, resumable, compaction, guardrails, exfil, egress, approval gate, human-in-the-loop, HITL, per-user secrets, credentials, sub-agents, delegation, self-improving, memory bank, routines, scheduler, a2a, skills, model routing
- **`rag-agent-search`** — managed document search via Search in Gemini platform (Discovery Engine) with a
  fully-managed GCS Data Connector: drop files in a bucket, no ingestion code to maintain.
  - Key files: `AGENTS.md`, `app/agent.py`, `infra/terraform/agent_platform_search.tf`, `infra/terraform/scripts/setup_data_connector.py`
  - Keywords: RAG, document search, Discovery Engine, Search in Gemini platform, Agent Platform Search, managed ingestion, GCS data connector, PDF, HTML, grounding
- **`rag-vector-search`** — RAG with Vector Search 2.0 in Gemini platform and a KFP ingestion pipeline (chunking +
  BigQuery staging; embeddings auto-generated server-side).
  - Key files: `AGENTS.md`, `app/agent.py`, `data_ingestion/data_ingestion_pipeline/pipeline.py`, `infra/terraform/scripts/setup_vector_search_collection.py`
  - Keywords: RAG, retrieval, vector search, embeddings, similarity search, ScaNN, semantic search, document Q&A, ingestion pipeline, chunking
- **`cross-session-memory`** — remembers user preferences and facts across sessions via Memory Bank in
  Gemini platform: written after each turn, recalled at the start of a later one.
  - Key files: `AGENTS.md`, `app/app_utils/memory_config.py`, `app/agent.py`, `app/fast_api_app.py`
  - Keywords: memory, cross-session, recall, remember, preferences, Memory Bank, PreloadMemoryTool
- **`oauth-user-consent-flow`** — reads a user's Google Drive on their behalf behind an OAuth 2.0 consent
  flow; the same code path works in local ADK Web and in production in the Gemini app.
  - Key files: `AGENTS.md`, `app/auths.py`, `app/tools.py`, `tools/register_oauth.py`
  - Keywords: OAuth, user consent, authentication, Google Drive, Workspace, Agent Runtime, Gemini app, Gemini Enterprise
- **`ambient-expense-agent`** — no chat loop: Pub/Sub events drive a graph-based `Workflow`, business
  rules stay in code, and only high-value cases reach an LLM that pauses for human approval.
  - Key files: `AGENTS.md`, `expense_agent/agent.py`, `expense_agent/fast_api_app.py`, `terraform/pubsub.tf`
  - Keywords: ambient, event-driven, scheduled, cron, Pub/Sub, workflow, human-in-the-loop, approval, alerts, no UI
- **`deep-search`** — research agent that plans (with an approval step), loops search → critique → refine
  until a quality bar is met, then writes a report with inline citations.
  - Key files: `AGENTS.md`, `app/agent.py`, `app/config.py`, `frontend/src/App.tsx`
  - Keywords: research, citations, iterative, critique, grounding, multi-agent, human-in-the-loop, web search, report
- **`safety-plugins`** — runner-wide safety guardrails as ADK `BasePlugin`s: attached to the `Runner`, they
  wrap every agent and sub-agent beneath it and keep harmful content out of session state.
  - Key files: `AGENTS.md`, `safety_plugins/plugins/model_armor.py`, `safety_plugins/plugins/agent_as_a_judge.py`, `safety_plugins/main.py`
  - Keywords: safety, guardrails, harmful content, moderation, content filtering, Model Armor, LLM-as-a-judge, session poisoning, plugins, runner-wide, applies to all sub-agents
- **`genmedia-for-commerce`** — full-stack multi-agent retail media: virtual try-on, 360° product spins and
  background swaps, orchestrated through an MCP tool server and Veo pipelines.
  - Key files: `AGENTS.md`, `genmedia4commerce/mcp_server/server.py`, `genmedia4commerce/workflows/shared/vector_search.py`, `genmedia4commerce/agent.py`
  - Keywords: MCP, media, image generation, video generation, product photography, on-model imagery, virtual try-on, 360° spin, background replacement, Veo, retail, e-commerce, catalogue, full-stack, React, Gemini app, Gemini Enterprise
- **`ambient-quality-agent`** — AQuA: an ADK agent that runs beside the agent it observes, evaluates its
  production conversations from BigQuery telemetry on a schedule or after each deploy, groups failures into
  deduplicated insights, and diagnoses root causes against the agent's source. Ships as an agents-cli extension.
  - Key files: `README.md`, `src/ambient_quality_agent/core/workflow.py`, `src/ambient_quality_agent/core/investigation/job_scheduling.py`, `src/ambient_quality_agent/tools/insights/clustering.py`, `extension/README.md`
  - Keywords: quality monitoring, production evaluation, observability, traces, telemetry, BigQuery Agent Analytics, LLM-as-a-judge, session review, custom metrics, insights, regressions, root-cause analysis, scheduled investigation, Workflow, agents-cli extension, dashboard

**Nothing above matches?** [`contrib/`](https://github.com/google/adk-samples/tree/main/contrib) holds
community- and partner-contributed recipes — broader in scope (complete solutions, not isolated patterns)
and not curated by the agents-cli team, so there is no guaranteed `AGENTS.md` to guide the lift. Check
there when you need something specific `core/` doesn't cover.

---
name: google-agents-cli-adk-code
description: >
  This skill should be used when the user wants to "write agent code",
  "build an agent with ADK", "add a tool", "create a callback", "define an agent",
  "use state management" — in a project that needs ADK (Agent Development Kit) API
  patterns and code examples.
  It provides a quick reference for agent types, tool definitions, orchestration
  patterns, callbacks, state management, the graph Workflow API, and reference
  recipes to study.
  Do NOT use for scaffolding (use google-agents-cli-scaffold) or deployment
  (use google-agents-cli-deploy).
metadata:
  author: Google
  license: Apache-2.0
  version: 1.11.0
  requires:
    bins:
      - agents-cli
    install: "uv tool install google-agents-cli"
---

# ADK Code Reference

Activate `/google-agents-cli-workflow` first for required development phases and scaffolding steps.

## 1. Study Recipes (No Project Needed)

**Python — read the topic index in `references/samples.md` before answering "how do I build X".**
Worked implementations exist for: sandboxed/per-user code execution, agent-loadable `SKILL.md`
skills, cross-session memory, approval gates before risky actions, tool guardrails, per-user
credentials, and scheduled/event-driven runs.

**Go — read the upstream [`examples/`](https://github.com/google/adk-go/tree/main/examples).**
clone the repo and read the one matching the capability you need before implementing it.

## 2. Prerequisites for Writing Code

Do NOT write agent code until a project is scaffolded.

1. Verify project: run `agents-cli info` (proceed if config exists).
2. New project: run `agents-cli scaffold create <name> --agent adk --deployment-target <target>` — use `--agent adk_go` for Go.
3. Existing code: run `agents-cli scaffold enhance .`.

## Quick Reference — Most Common Patterns

See `references/adk-python.md` and `references/adk-go.md` for Python and Go code examples, respectively.

---

## References

Use the cheatsheets for common patterns. For deep knowledge, fetch the docs index or inspect the
installed package.

| Reference | Language | When to read |
|------|------|-------------|
| `references/samples.md` | Python | **Topic-indexed catalog of ADK reference recipes.** Read in workflow Phase 1 — before scaffolding and before writing code — maps a capability to the recipe that implements it. |
| `references/adk-python.md` | Python | Core ADK API: `Agent`, tools, callbacks, plugins, state, artifacts, multi-agent systems, `SequentialAgent` / `ParallelAgent` / `LoopAgent`, custom `BaseAgent`, `ManagedAgent` (server-hosted first-party agents), A2A protocol, A2UI. Default for most agents. |
| `references/adk-python-workflows.md` | Python | Graph-based Workflow API (ADK Python 2.0): nodes, edges, fan-out/fan-in, HITL, parallel processing. Use when you need explicit graph topology. |
| `references/adk-python-live.md` | Python | Live and voice agents: `Runner.run_live`, `LiveRequestQueue`, model and regional availability, voice/`speech_config`, VAD, session limits, `/run_live` serving. Use for real-time voice or video agents. |
| `references/adk-go.md` | Go | Core ADK Go API: `llmagent`, tools, callbacks, plugins, state, artifacts, multi-agent systems, sequential/parallel/loop agents, custom agents, the runner, serving over HTTP, A2A protocol, and ambient triggers. Default for most agents. |
| `references/adk-go-workflows.md` | Go | Graph-based Workflow API: nodes, edges, fan-out/fan-in, HITL, parallel processing. Use when you need explicit graph topology. |
| [`examples/`](https://github.com/google/adk-go/tree/main/examples) | Go | Runnable upstream programs — the closest thing to a recipe catalog for Go. |
| `curl https://adk.dev/llms.txt` | Python | Docs index (every page title + URL). Fetch it, then `WebFetch` the specific page for anything beyond the cheatsheets. |
| Installed ADK package | Python | Exact signatures and symbols — inspect the source (see "Inspecting ADK Source Code" in `references/adk-python.md`). |

## Related Skills

- `/google-agents-cli-workflow` — Development workflow, coding guidelines, and operational rules
- `/google-agents-cli-scaffold` — Project creation and enhancement with `agents-cli scaffold create` / `scaffold enhance`
- `/google-agents-cli-eval` — Evaluation methodology, dataset schema, and the eval-fix loop
- `/google-agents-cli-deploy` — Deployment targets, CI/CD pipelines, and production workflows