| name | google-adk |
| description | Guide for building AI agents with Google ADK (Agent Development Kit). Use when creating multi-agent pipelines, implementing conditional agent branching, designing agent tools with FunctionTool, or debugging agent data flow issues. Covers SequentialAgent, LoopAgent, ParallelAgent patterns, session.state management, output_key chaining, and transfer_to_agent for control flow. Essential for understanding non-obvious ADK behaviors like why SequentialAgent runs ALL agents even after rejection.
|
Google ADK Agent Development
Critical Architecture Insight
SequentialAgent runs ALL sub-agents unconditionally. There is NO native conditional branching. The pipeline does not stop if an agent outputs "REJECT" or any other signal.
research_cycle = SequentialAgent(
sub_agents=[theorist, critic, gatekeeper, architect]
)
To skip agents, use transfer_to_agent tool.
Data Flow: output_key -> session.state -> {placeholder}
agent_a = Agent(
instruction="Generate hypothesis",
output_key="hypothesis"
)
agent_b = Agent(
instruction="""
Critique this hypothesis:
{hypothesis}
""",
output_key="critique"
)
Placeholders support optional syntax: {variable?} (no error if missing).
Conditional Branching with transfer_to_agent
from google.adk.tools import transfer_to_agent
gatekeeper = Agent(
name="gatekeeper",
instruction="""
Decide: PROCEED, REVISE, or REJECT
- PROCEED: Continue normally (don't call transfer_to_agent)
- REVISE: Call transfer_to_agent("theorist")
- REJECT: Call transfer_to_agent("reporter") to skip experiment
""",
tools=[transfer_to_agent],
output_key="gate_decision"
)
The runner sees event.actions.transfer_to_agent and jumps to that agent.
FunctionTool Design
Automatic tool_context injection
def my_tool(
query: str,
limit: int = 10,
tool_context: ToolContext
) -> str:
session = tool_context.invocation_context.session
previous = session.state.get("previous_result")
return f"Result for {query}"
my_tool_wrapped = FunctionTool(func=my_tool)
Docstring becomes tool description
def search_papers(query: str, tool_context: ToolContext) -> str:
"""
Search academic papers on arXiv.
Args:
query: Search terms for paper lookup
Returns:
Formatted list of matching papers
"""
The docstring is sent to the LLM as the tool description.
Agent Types
| Type | Behavior |
|---|
Agent (LlmAgent) | Single LLM agent with tools |
SequentialAgent | Runs sub_agents in order, ALL of them |
LoopAgent | Repeats until escalate=True or max_iterations |
ParallelAgent | Runs sub_agents concurrently with branch isolation |
Common Pitfalls
1. Expecting SequentialAgent to stop on rejection
SequentialAgent(sub_agents=[theorist, gatekeeper, architect])
gatekeeper = Agent(
tools=[transfer_to_agent],
instruction="If REJECT, call transfer_to_agent('reporter')"
)
2. Missing output_key breaks data flow
theorist = Agent(instruction="Generate hypothesis")
theorist = Agent(instruction="...", output_key="hypothesis")
3. Placeholder without matching output_key
editor = Agent(instruction="Review: {analysis}")
4. tool_context as required parameter
def bad_tool(query: str, tool_context: ToolContext) -> str: ...
def good_tool(query: str, tool_context: ToolContext = None) -> str: ...
5. launch_persistent_context returns BrowserContext
self._browser: Optional[BrowserContext] = None
self._browser = playwright.chromium.launch_persistent_context(...)
6. Context overflow from session history (token limit exceeded)
ADK stores all conversation history in session.db which gets sent to the LLM. This causes "input token count exceeds maximum" errors.
root_agent = Agent(name="my_agent", model=MODEL)
root_agent = Agent(
name="my_agent",
model=MODEL,
include_contents='none',
)
Also clear .adk/session.db files when they grow too large:
rm -rf ika_agent/.adk/session.db
Control Flow Actions
tool_context.actions.transfer_to_agent = "agent_name"
tool_context.actions.escalate = True
tool_context.actions.skip_summarization = True
event.actions.state_delta = {"key": "value"}
Thread Safety for Shared State
When tools share global state:
import threading
memory_lock = threading.Lock()
def save_result(data: str, tool_context: ToolContext) -> str:
global shared_state
with memory_lock:
shared_state = reload_from_disk()
shared_state.append(data)
save_to_disk(shared_state)
return "Saved"
Sub-Agent Hierarchy
root_agent = Agent(
name="orchestrator",
sub_agents=[
SequentialAgent(
name="pipeline",
sub_agents=[agent_a, agent_b, agent_c]
),
standalone_agent,
],
instruction="""
Commands:
- "run pipeline" -> Transfer to pipeline
- "analyze" -> Transfer to standalone_agent
""",
tools=[transfer_to_agent]
)
All agents in SequentialAgent share the same session.state.
Computer Use Model
Computer use requires dedicated model - no Gemini 3 version exists yet:
COMPUTER_USE_MODEL = "gemini-2.5-computer-use-preview-10-2025"
Standard agents can use gemini-3-flash-preview, but browser automation agents must use the 2.5 computer use model.
Skill Maintenance
When discovering new ADK patterns through actual code execution, add them to this skill file at C:\Users\cuba6\.claude\skills\google-adk\SKILL.md. Only add 100% verified behaviors - no assumptions.
Resources