adk-fluent 0.13.1

Fluent builder API for Google's Agent Development Kit (ADK)

adk-fluent

Fluent builder API for Google's Agent Development Kit (ADK). Reduces agent creation from 22+ lines to 1-3 lines while producing identical native ADK objects.

Table of Contents

• Install
• Quick Start
• Zero to Running
• Why adk-fluent
• Expression Language
• Context Engineering (C Module)
• Common Errors
• Fluent API Reference
• When to Use adk-fluent
• Run with adk web
• Visual Cookbook Runner
• Cookbook
• Performance
• ADK Compatibility
• How It Works
• Features
• AI Coding Skills
• Development

Install

pip install adk-fluent

Autocomplete works immediately -- the package ships with .pyi type stubs for every builder. Type Agent("name"). and your IDE shows all available methods with type hints.

Optional Extras

Install additional capabilities as needed:

pip install adk-fluent[a2a]            # A2A remote agent-to-agent communication
pip install adk-fluent[yaml]           # .to_yaml() / .from_yaml() serialization
pip install adk-fluent[rich]           # Rich terminal output for .explain()
pip install adk-fluent[search]         # BM25-indexed tool discovery (T.search)
pip install adk-fluent[pii]            # PII detection guard (G.pii with Cloud DLP)
pip install adk-fluent[observability]  # OpenTelemetry tracing and metrics
pip install adk-fluent[dev]            # Development tools (pytest, ruff, pyright)
pip install adk-fluent[docs]           # Documentation build (Sphinx, Furo)

Combine extras: pip install adk-fluent[a2a,yaml,rich]

A2UI (Agent-to-UI): The UI namespace for declarative agent UIs ships with the core package -- no extra install needed. The full A2UI toolset (SendA2uiToClientToolset) will be available via pip install adk-fluent[a2ui] when the a2ui-agent package is published. Until then, all UI composition, compilation, and presets work out of the box.

IDE Setup

VS Code -- install the Pylance extension (included in the Python extension pack). Autocomplete and type checking work out of the box.

PyCharm -- works automatically. The .pyi stubs are bundled in the package and PyCharm discovers them on install.

Neovim (LSP) -- use pyright as your language server. Stubs are picked up automatically.

Discover the API

from adk_fluent import Agent

agent = Agent("demo")
agent.  # <- autocomplete shows: .model(), .instruct(), .tool(), .build(), ...

# Typos are caught at definition time, not runtime:
agent.instuction("oops")  # -> AttributeError: 'instuction' is not a recognized field.
                          #    Did you mean: 'instruction'?

# Inspect any builder's current state:
print(agent.model("gemini-2.5-flash").instruct("Help.").explain())
# Agent: demo
#   Config fields: model, instruction

# See everything available:
print(dir(agent))  # All methods including forwarded ADK fields

Quick Start

from adk_fluent import Agent

# Create an agent and get a response -- no Runner, no Session, no boilerplate
agent = Agent("helper", "gemini-2.5-flash").instruct("You are a helpful assistant.")
print(agent.ask("What is the capital of France?"))
# => The capital of France is Paris.

.ask() handles Runner, Session, and cleanup internally. One line to define, one line to run.

Try without an API key — verify the library works using .mock():

from adk_fluent import Agent

agent = Agent("demo", "gemini-2.5-flash").instruct("You are helpful.").mock(["Hello! How can I help?"])
print(agent.ask("Hi"))
# => Hello! How can I help?

For ADK integration, .build() returns the native ADK object:

from adk_fluent import Agent, Pipeline, FanOut, Loop

# Simple agent -- returns a native LlmAgent
agent = Agent("helper", "gemini-2.5-flash").instruct("You are a helpful assistant.").build()

# Pipeline -- sequential agents
pipeline = (
    Pipeline("research")
    .step(Agent("searcher", "gemini-2.5-flash").instruct("Search for information."))
    .step(Agent("writer", "gemini-2.5-flash").instruct("Write a summary."))
    .build()
)

# Fan-out -- parallel agents
fanout = (
    FanOut("parallel_research")
    .branch(Agent("web", "gemini-2.5-flash").instruct("Search the web."))
    .branch(Agent("papers", "gemini-2.5-flash").instruct("Search papers."))
    .build()
)

# Loop -- iterative refinement
loop = (
    Loop("refine")
    .step(Agent("writer", "gemini-2.5-flash").instruct("Write draft."))
    .step(Agent("critic", "gemini-2.5-flash").instruct("Critique."))
    .max_iterations(3)
    .build()
)

Every .build() returns a real ADK object (LlmAgent, SequentialAgent, etc.). Fully compatible with adk web, adk run, and adk deploy.

Two Styles, Same Result

Every workflow can be expressed two ways -- the explicit builder API or the expression operators. Both produce identical ADK objects:

# Explicit builder style — readable, IDE-friendly
pipeline = (
    Pipeline("research")
    .step(Agent("web", "gemini-2.5-flash").instruct("Search web.").outputs("web_data"))
    .step(Agent("analyst", "gemini-2.5-flash").instruct("Analyze {web_data}."))
    .build()
)

# Operator style — compact, composable
pipeline = (
    Agent("web", "gemini-2.5-flash").instruct("Search web.").outputs("web_data")
    >> Agent("analyst", "gemini-2.5-flash").instruct("Analyze {web_data}.")
).build()

The builder style shines for complex multi-step workflows where each step is configured with callbacks, tools, and context. The operator style excels at composing reusable sub-expressions:

# Complex builder-style pipeline with tools and callbacks
pipeline = (
    Pipeline("customer_support")
    .step(
        Agent("classifier", "gemini-2.5-flash")
        .instruct("Classify the customer's intent.")
        .outputs("intent")
        .before_model(log_fn)
    )
    .step(
        Agent("resolver", "gemini-2.5-flash")
        .instruct("Resolve the {intent} issue.")
        .tool(lookup_customer)
        .tool(create_ticket)
        .history("none")
    )
    .step(
        Agent("responder", "gemini-2.5-flash")
        .instruct("Draft a response to the customer.")
        .after_model(audit_fn)
    )
    .build()
)

# Same complexity, composed from reusable parts with operators
classify = Agent("classifier", "gemini-2.5-flash").instruct("Classify intent.").outputs("intent")
resolve = Agent("resolver", "gemini-2.5-flash").instruct("Resolve {intent}.").tool(lookup_customer)
respond = Agent("responder", "gemini-2.5-flash").instruct("Draft response.")

support_pipeline = classify >> resolve >> respond
# Reuse sub-expressions in different pipelines
escalation_pipeline = classify >> Agent("escalate", "gemini-2.5-flash").instruct("Escalate.")

Pipeline Architecture

graph TD
    n1[["customer_support (sequence)"]]
    n2["classifier"]
    n3["resolver"]
    n4["responder"]
    n2 --> n3
    n3 --> n4
    n2 -. "intent" .-> n3

Zero to Running

Fastest: Google AI Studio (free tier)

pip install adk-fluent
export GOOGLE_API_KEY="your-key-from-aistudio.google.com"
python quickstart.py

Get a free API key at aistudio.google.com.

Production: Vertex AI

pip install adk-fluent
export GOOGLE_CLOUD_PROJECT="your-project-id"
export GOOGLE_CLOUD_LOCATION="us-central1"
export GOOGLE_GENAI_USE_VERTEXAI="TRUE"
python quickstart.py

Requires a GCP project with the Vertex AI API enabled. See Vertex AI setup.

Both paths produce the same result -- the quickstart.py file works with either configuration.

Why adk-fluent

Real patterns. Real reduction. Every adk-fluent expression compiles to the same native ADK objects you'd build by hand.

Document Processing Pipeline

A contract review system that extracts terms, analyzes risks, and summarizes.

LangGraph (~35 lines):

class State(TypedDict):
    document: str
    terms: str
    risks: str
    summary: str

def extract(state): ...
def analyze(state): ...
def summarize(state): ...

graph = StateGraph(State)
graph.add_node("extract", extract)
graph.add_node("analyze", analyze)
graph.add_node("summarize", summarize)
graph.add_edge("extract", "analyze")
graph.add_edge("analyze", "summarize")
graph.add_edge(START, "extract")
graph.add_edge("summarize", END)
app = graph.compile()

adk-fluent (1 expression):

pipeline = extractor >> analyst >> summarizer

Full example with native ADK comparison →

Multi-Source Research with Quality Loop

Decompose a query, search 3 sources in parallel, synthesize, and review in a loop until quality threshold is met.

LangGraph (~60 lines):

class ResearchState(TypedDict):
    query: str
    web_results: str
    academic_results: str
    synthesis: str
    quality_score: float

def analyze_query(state): ...
def search_web(state): ...
def search_academic(state): ...
def search_news(state): ...
def synthesize(state): ...
def review_quality(state): ...
def revise(state): ...
def should_continue(state):
    return "revise" if state["quality_score"] < 0.85 else "report"

graph = StateGraph(ResearchState)
graph.add_node("analyze", analyze_query)
graph.add_node("search_web", search_web)
graph.add_node("search_academic", search_academic)
graph.add_node("search_news", search_news)
graph.add_node("synthesize", synthesize)
graph.add_node("review", review_quality)
graph.add_node("revise", revise)
graph.add_node("report", write_report)
graph.add_edge(START, "analyze")
graph.add_edge("analyze", "search_web")
graph.add_edge("analyze", "search_academic")
graph.add_edge("analyze", "search_news")
graph.add_edge("search_web", "synthesize")
graph.add_edge("search_academic", "synthesize")
graph.add_edge("search_news", "synthesize")
graph.add_edge("synthesize", "review")
graph.add_conditional_edges("review", should_continue)
graph.add_edge("report", END)
app = graph.compile()

adk-fluent (1 expression):

research = (
    analyzer
    >> (web | papers | news)
    >> synthesizer
    >> (reviewer >> reviser) * until(lambda s: s["quality_score"] >= 0.85)
    >> writer @ ResearchReport
)

Full example with typed output and state wiring →

Customer Support Triage

Classify customer intent and route to the right specialist.

LangGraph (~45 lines):

class SupportState(TypedDict):
    message: str
    intent: str
    response: str

def classify(state): ...
def handle_billing(state): ...
def handle_technical(state): ...
def handle_general(state): ...
def route_intent(state):
    return state["intent"]

graph = StateGraph(SupportState)
graph.add_node("classify", classify)
graph.add_node("billing", handle_billing)
graph.add_node("technical", handle_technical)
graph.add_node("general", handle_general)
graph.add_edge(START, "classify")
graph.add_conditional_edges(
    "classify", route_intent,
    {"billing": "billing", "technical": "technical", "general": "general"},
)
graph.add_edge("billing", END)
graph.add_edge("technical", END)
graph.add_edge("general", END)
app = graph.compile()

adk-fluent (1 expression):

support = (
    S.capture("message")
    >> classifier
    >> Route("intent")
        .eq("billing", billing)
        .eq("technical", technical)
        .otherwise(general)
)

Full example with escalation gates →

The Pattern

What	LangGraph	Native ADK	adk-fluent
Sequential pipeline	~35 lines	~20 lines	1 expression
Parallel + loop	~60 lines	~45 lines	1 expression
Routing	~45 lines	~35 lines	1 expression
Boilerplate	StateGraph, TypedDict, edges	Agent, Runner, Session	None
Result type	LangGraph graph	ADK agent	ADK agent

For a detailed comparison including CrewAI and native ADK code, see the Framework Comparison guide.

Expression Language

Nine operators compose any agent topology:

Operator	Meaning	ADK Type
a >> b	Sequence	SequentialAgent
a >> fn	Function step	Zero-cost transform
a | b	Parallel	ParallelAgent
a * 3	Loop (fixed)	LoopAgent
a * until(pred)	Loop (conditional)	LoopAgent + checkpoint
a @ Schema	Typed output	output_schema
a // b	Fallback	First-success chain
Route("key").eq(...)	Branch	Deterministic routing
S.pick(...), S.rename(...)	State transforms	Dict operations via >>
C.user_only(), C.none()	Context engineering	Selective Turn History

Eight control loop primitives for agent orchestration:

Primitive	Purpose	ADK Mechanism
tap(fn)	Observe state without mutating	Custom BaseAgent (no LLM)
expect(pred, msg)	Assert state contract	Raises ValueError on failure
.mock(responses)	Bypass LLM for testing	before_model_callback → LlmResponse
.retry_if(pred)	Retry while condition holds	LoopAgent + checkpoint escalate
map_over(key, agent)	Iterate agent over list	Custom BaseAgent loop
.timeout(seconds)	Time-bound execution	asyncio deadline + cancel
gate(pred, msg)	Human-in-the-loop approval	EventActions(escalate=True)
race(a, b, ...)	First-to-finish wins	asyncio.wait(FIRST_COMPLETED)

All operators are immutable -- sub-expressions can be safely reused:

review = agent_a >> agent_b
pipeline_1 = review >> agent_c  # Independent
pipeline_2 = review >> agent_d  # Independent

How Operators Map to Agent Trees

Expression:   a >> (b | c) * 3

Agent tree:
  SequentialAgent
  +-- a (LlmAgent)
  +-- LoopAgent (max_iterations=3)
      +-- ParallelAgent
          +-- b (LlmAgent)
          +-- c (LlmAgent)

Expression:   a >> fn >> Route("key").eq("x", b).eq("y", c)

Agent tree:
  SequentialAgent
  +-- a (LlmAgent)
  +-- fn (FunctionAgent)
  +-- RoutingAgent
      +-- "x" -> b (LlmAgent)
      +-- "y" -> c (LlmAgent)

Expression:   (a | b) >> merge_fn >> writer @ Report // fallback_writer @ Report

Agent tree:
  SequentialAgent
  +-- ParallelAgent
  |   +-- a (LlmAgent)
  |   +-- b (LlmAgent)
  +-- merge_fn (FunctionAgent)
  +-- FallbackAgent
      +-- writer (LlmAgent, output_schema=Report)
      +-- fallback_writer (LlmAgent, output_schema=Report)

Function Steps

Plain Python functions compose with >> as zero-cost workflow nodes (no LLM call):

def merge_research(state):
    return {"research": state["web"] + "\n" + state["papers"]}

pipeline = web_agent >> merge_research >> writer_agent

Typed Output

@ binds a Pydantic schema as the agent's output contract:

from pydantic import BaseModel

class Report(BaseModel):
    title: str
    body: str

agent = Agent("writer").model("gemini-2.5-flash").instruct("Write.") @ Report

Fallback Chains

// tries each agent in order -- first success wins:

answer = (
    Agent("fast").model("gemini-2.0-flash").instruct("Quick answer.")
    // Agent("thorough").model("gemini-2.5-pro").instruct("Detailed answer.")
)

Conditional Loops

* until(pred) loops until a predicate on session state is satisfied:

from adk_fluent import until

loop = (
    Agent("writer").model("gemini-2.5-flash").instruct("Write.").outputs("quality")
    >> Agent("reviewer").model("gemini-2.5-flash").instruct("Review.")
) * until(lambda s: s.get("quality") == "good", max=5)

State Transforms

S factories return dict transforms that compose with >>:

from adk_fluent import S

pipeline = (
    (web_agent | scholar_agent)
    >> S.merge("web", "scholar", into="research")
    >> S.default(confidence=0.0)
    >> S.rename(research="input")
    >> writer_agent
)

Factory	Purpose
S.pick(*keys)	Keep only specified keys
S.drop(*keys)	Remove specified keys
S.rename(**kw)	Rename keys
S.default(**kw)	Fill missing keys
S.merge(*keys, into=)	Combine keys
S.transform(key, fn)	Map a single value
S.compute(**fns)	Derive new keys
S.guard(pred)	Assert invariant
S.log(*keys)	Debug-print

Context Engineering (C Module)

Control exactly what conversation history each agent sees. Prevents prompt pollution in complex DAGs:

from adk_fluent import C

pipeline = (
    Agent("classifier").outputs("intent")
    >> Agent("booker")
        .instruct("Process {intent}")
        .context(C.user_only()) # Booker sees user prompt + {intent}, but NOT classifier text
)

Transform	Purpose
C.user_only()	Include only original user messages
C.none()	No turn history (stateless prompt)
C.window(n=5)	Sliding window of last N turns
C.from_agents("a", "b")	Include user + named agent outputs
C.capture("key")	Snapshot user message into state

Common Errors

Missing required field:

Agent("x").instruct("Hi").build()
# BuilderError: Agent 'x' is missing required field 'model'
#   model: required (not set)

Fix: add .model("gemini-2.5-flash") before .build().

Typo in method name:

Agent("x").modle("gemini-2.5-flash")
# AttributeError: 'modle' is not a recognized field. Did you mean: 'model'?

The typo detector suggests the closest valid field name.

Invalid operator operand:

Agent("a") | "not an agent"
# TypeError: unsupported operand type(s) for |: 'AgentBuilder' and 'str'

Operators work with Agent, Pipeline, FanOut, Loop builders, callables, and built ADK agents.

Template variable at runtime:

# {topic} resolves from session state at runtime, not at definition time
agent = Agent("writer", "gemini-2.5-flash").instruct("Write about {topic}.")
agent.ask("hello")  # {topic} appears literally if not in state

Use .outputs("topic") on a prior agent, or pass initial state via .session().

Full error reference: Error Reference

IR, Backends, and Middleware (v4)

Builders can compile to an intermediate representation (IR) for inspection, testing, and alternative backends:

from adk_fluent import Agent, ExecutionConfig, CompactionConfig

# IR: inspect the agent tree without building
pipeline = Agent("a") >> Agent("b") >> Agent("c")
ir = pipeline.to_ir()  # Returns frozen dataclass tree

# to_app(): compile through IR to a native ADK App
app = pipeline.to_app(config=ExecutionConfig(
    app_name="my_app",
    resumable=True,
    compaction=CompactionConfig(interval=10),
))

# Middleware: app-global cross-cutting behavior
from adk_fluent import Middleware, RetryMiddleware, StructuredLogMiddleware

app = (
    Agent("a") >> Agent("b")
).middleware(RetryMiddleware(max_retries=3)).to_app()

# Data contracts: verify pipeline wiring at build time
from pydantic import BaseModel
from adk_fluent.testing import check_contracts

class Intent(BaseModel):
    category: str
    confidence: float

pipeline = Agent("classifier").produces(Intent) >> Agent("resolver").consumes(Intent)
issues = check_contracts(pipeline.to_ir())  # [] = all good

# Deterministic testing without LLM calls
from adk_fluent.testing import mock_backend, AgentHarness

harness = AgentHarness(pipeline, backend=mock_backend({
    "classifier": {"category": "billing", "confidence": 0.9},
    "resolver": "Ticket #1234 created.",
}))

# Graph visualization
print(pipeline.to_mermaid())  # Mermaid diagram source

# Tool confirmation (human-in-the-loop approval)
agent = Agent("ops").tool(deploy_fn, require_confirmation=True)

# Resource DI (hide infra params from LLM)
agent = Agent("lookup").tool(search_db).inject(db=my_database)

Deterministic Routing

Route on session state without LLM calls:

from adk_fluent import Agent
from adk_fluent._routing import Route

classifier = Agent("classify").model("gemini-2.5-flash").instruct("Classify intent.").outputs("intent")
booker = Agent("booker").model("gemini-2.5-flash").instruct("Book flights.")
info = Agent("info").model("gemini-2.5-flash").instruct("Provide info.")

# Route on exact match — zero LLM calls for routing
pipeline = classifier >> Route("intent").eq("booking", booker).eq("info", info)

# Dict shorthand
pipeline = classifier >> {"booking": booker, "info": info}

Conditional Gating

# Only runs if predicate(state) is truthy
enricher = (
    Agent("enricher")
    .model("gemini-2.5-flash")
    .instruct("Enrich the data.")
    .proceed_if(lambda s: s.get("valid") == "yes")
)

Tap (Observe Without Mutating)

tap(fn) creates a zero-cost observation step. It reads state but never writes back -- perfect for logging, metrics, and debugging:

from adk_fluent import tap

pipeline = (
    writer
    >> tap(lambda s: print("Draft:", s.get("draft", "")[:50]))
    >> reviewer
)

# Also available as a method
pipeline = writer.tap(lambda s: log_metrics(s)) >> reviewer

Expect (State Assertions)

expect(pred, msg) asserts a state contract at a pipeline step. Raises ValueError if the predicate fails:

from adk_fluent import expect

pipeline = (
    writer
    >> expect(lambda s: "draft" in s, "Writer must produce a draft")
    >> reviewer
)

Mock (Testing Without LLM)

.mock(responses) bypasses LLM calls with canned responses. Uses the same before_model_callback mechanism as ADK's ReplayPlugin, but scoped to a single agent:

# List of responses (cycles when exhausted)
agent = Agent("writer").model("gemini-2.5-flash").instruct("Write.").mock(["Draft 1", "Draft 2"])

# Callable for dynamic mocking
agent = Agent("echo").model("gemini-2.5-flash").mock(lambda req: "Mocked response")

Retry If

.retry_if(pred) retries agent execution while the predicate returns True. Thin wrapper over loop_until with inverted logic:

agent = (
    Agent("writer").model("gemini-2.5-flash")
    .instruct("Write a high-quality draft.").outputs("quality")
    .retry_if(lambda s: s.get("quality") != "good", max_retries=3)
)

Map Over

map_over(key, agent) iterates an agent over each item in a state list:

from adk_fluent import map_over

pipeline = (
    fetcher
    >> map_over("documents", summarizer, output_key="summaries")
    >> compiler
)

Timeout

.timeout(seconds) wraps an agent with a time limit. Raises asyncio.TimeoutError if exceeded:

agent = Agent("researcher").model("gemini-2.5-pro").instruct("Deep research.").timeout(60)

Gate (Human-in-the-Loop)

gate(pred, msg) pauses the pipeline for human approval when the condition is met. Uses ADK's escalate mechanism:

from adk_fluent import gate

pipeline = (
    analyzer
    >> gate(lambda s: s.get("risk") == "high", message="Approve high-risk action?")
    >> executor
)

Race (First-to-Finish)

race(a, b, ...) runs agents concurrently and keeps only the first to finish:

from adk_fluent import race

winner = race(
    Agent("fast").model("gemini-2.0-flash").instruct("Quick answer."),
    Agent("thorough").model("gemini-2.5-pro").instruct("Detailed answer."),
)

Full Composition

All operators compose into a single expression:

from pydantic import BaseModel
from adk_fluent import Agent, S, until

class Report(BaseModel):
    title: str
    body: str
    confidence: float

pipeline = (
    (   Agent("web").model("gemini-2.5-flash").instruct("Search web.")
      | Agent("scholar").model("gemini-2.5-flash").instruct("Search papers.")
    )
    >> S.merge("web", "scholar", into="research")
    >> Agent("writer").model("gemini-2.5-flash").instruct("Write.") @ Report
       // Agent("writer_b").model("gemini-2.5-pro").instruct("Write.") @ Report
    >> (
        Agent("critic").model("gemini-2.5-flash").instruct("Score.").outputs("confidence")
        >> Agent("reviser").model("gemini-2.5-flash").instruct("Improve.")
    ) * until(lambda s: s.get("confidence", 0) >= 0.85, max=4)
)

Fluent API Reference

Agent Builder

The Agent builder wraps ADK's LlmAgent. Every method returns self for chaining.

Core Configuration

Method	Alias for	Description
.model(name)	model	LLM model identifier ("gemini-2.5-flash", "gemini-2.5-pro", etc.)
.instruct(text_or_fn)	instruction	System instruction. Accepts a string or Callable[[ReadonlyContext], str]
.describe(text)	description	Agent description (used in delegation and tool descriptions)
.outputs(key)	output_key	Store the agent's final response in session state under this key
.tool(fn)	—	Add a tool function or BaseTool instance. Multiple calls accumulate
.build()	—	Resolve into a native ADK LlmAgent

Prompt & Context Control

Method	Alias for	Description
.instruct(text)	instruction	Dynamic instruction. Supports {variable} placeholders auto-resolved from session state
.instruct(fn)	instruction	Callable receiving ReadonlyContext, returns string. Full programmatic control
.static(content)	static_instruction	Cacheable instruction that never changes. Sent as system instruction for context caching
.history("none")	include_contents	Control conversation history: "default" (full history) or "none" (stateless)
.global_instruct(text)	global_instruction	Instruction inherited by all sub-agents
.inject_context(fn)	—	Prepend dynamic context via before_model_callback. The function receives callback context, returns a string

Template variables in string instructions are auto-resolved from session state:

# {topic} and {style} are replaced at runtime from session state
agent = Agent("writer").instruct("Write about {topic} in a {style} tone.")

This composes naturally with the expression algebra:

pipeline = (
    Agent("classifier").instruct("Classify.").outputs("topic")
    >> S.default(style="professional")
    >> Agent("writer").instruct("Write about {topic} in a {style} tone.")
)

Optional variables use ? suffix ({maybe_key?} returns empty string if missing). Namespaced keys: {app:setting}, {user:pref}, {temp:scratch}.

Prompt Builder

For multi-section prompts, the Prompt builder provides structured composition:

from adk_fluent import Prompt

prompt = (
    Prompt()
    .role("You are a senior code reviewer.")
    .context("The codebase uses Python 3.11 with type hints.")
    .task("Review the code for bugs and security issues.")
    .constraint("Be concise. Max 5 bullet points.")
    .constraint("No false positives.")
    .format("Return markdown with ## sections.")
    .example("Input: x=eval(input()) | Output: - **Critical**: eval() on user input")
)

agent = Agent("reviewer").model("gemini-2.5-flash").instruct(prompt).build()

Sections are emitted in a fixed order (role, context, task, constraints, format, examples) regardless of call order. Prompts are composable and reusable:

base_prompt = Prompt().role("You are a senior engineer.").constraint("Be precise.")

reviewer = Agent("reviewer").instruct(base_prompt + Prompt().task("Review code."))
writer   = Agent("writer").instruct(base_prompt + Prompt().task("Write documentation."))

Method	Description
.role(text)	Agent persona (emitted without header)
.context(text)	Background information
.task(text)	Primary objective
.constraint(text)	Rules to follow (multiple calls accumulate)
.format(text)	Desired output format
.example(text)	Few-shot examples (multiple calls accumulate)
.section(name, text)	Custom named section
.merge(other) / +	Combine two Prompts
.build() / str()	Compile to instruction string

Static Instructions & Context Caching

Split prompts into cacheable and dynamic parts:

agent = (
    Agent("analyst")
    .model("gemini-2.5-flash")
    .static("You are a financial analyst. Here is the 50-page annual report: ...")
    .instruct("Answer the user's question about the report.")
    .build()
)

When .static() is set, the static content goes as a system instruction (eligible for context caching), while .instruct() content goes as user content. This avoids re-processing large static contexts on every turn.

Dynamic Context Injection

Prepend runtime context to every LLM call:

agent = (
    Agent("support")
    .model("gemini-2.5-flash")
    .instruct("Help the customer.")
    .inject_context(lambda ctx: f"Customer: {ctx.state.get('customer_name', 'unknown')}")
    .inject_context(lambda ctx: f"Plan: {ctx.state.get('plan', 'free')}")
)

Each .inject_context() call accumulates. The function receives the callback context and returns a string that gets prepended as content before the LLM processes the request.

Callbacks

All callback methods are additive -- multiple calls accumulate handlers, never replace:

Method	Alias for	Description
.before_model(fn)	before_model_callback	Runs before each LLM call. Receives (callback_context, llm_request)
.after_model(fn)	after_model_callback	Runs after each LLM call. Receives (callback_context, llm_response)
.before_agent(fn)	before_agent_callback	Runs before agent execution
.after_agent(fn)	after_agent_callback	Runs after agent execution
.before_tool(fn)	before_tool_callback	Runs before each tool call
.after_tool(fn)	after_tool_callback	Runs after each tool call
.on_model_error(fn)	on_model_error_callback	Handles LLM errors
.on_tool_error(fn)	on_tool_error_callback	Handles tool errors
.guardrail(fn)	—	Registers fn as both before_model and after_model

Conditional variants append only when the condition is true:

agent = (
    Agent("service")
    .before_model_if(debug_mode, log_fn)
    .after_model_if(audit_enabled, audit_fn)
)

Control Flow

Method	Description
.proceed_if(pred)	Only run this agent if pred(state) is truthy. Uses before_agent_callback
.loop_until(pred, max_iterations=N)	Wrap in a loop that exits when pred(state) is satisfied
.retry_if(pred, max_retries=3)	Retry while pred(state) returns True. Inverse of loop_until
.mock(responses)	Bypass LLM with canned responses (list or callable). For testing
.tap(fn)	Append observation step: self >> tap(fn). Returns Pipeline
.timeout(seconds)	Wrap with time limit. Raises asyncio.TimeoutError on expiry

Delegation (LLM-Driven Routing)

# The coordinator's LLM decides when to delegate
coordinator = (
    Agent("coordinator")
    .model("gemini-2.5-flash")
    .instruct("Route tasks to the right specialist.")
    .delegate(Agent("math").model("gemini-2.5-flash").instruct("Solve math."))
    .delegate(Agent("code").model("gemini-2.5-flash").instruct("Write code."))
    .build()
)

.delegate(agent) wraps the sub-agent in an AgentTool so the coordinator's LLM can invoke it by name.

One-Shot Execution

Method	Description
.ask(prompt)	Send a prompt, get response text. No Runner/Session boilerplate
.ask_async(prompt)	Async version of .ask()
.stream(prompt)	Async generator yielding response text chunks
.events(prompt)	Async generator yielding raw ADK Event objects
.map(prompts, concurrency=5)	Batch execution against multiple prompts
.map_async(prompts, concurrency=5)	Async batch execution
.session()	Create an interactive async with session context manager
.test(prompt, contains=, matches=, equals=)	Smoke test: calls .ask() and asserts output

Cloning and Variants

base = Agent("base").model("gemini-2.5-flash").instruct("Be helpful.")

# Clone — independent deep copy with new name
math_agent = base.clone("math").instruct("Solve math.")

# with_() — immutable variant (original unchanged)
creative = base.with_(name="creative", model="gemini-2.5-pro")

Validation and Introspection

Method	Description
.validate()	Try .build() and raise ValueError with clear message on failure. Returns self
.explain()	Multi-line summary of builder state (config fields, callbacks, lists)
.to_dict() / .to_yaml()	Serialize builder state (inspection only, no round-trip)

Dynamic Field Forwarding

Any ADK LlmAgent field can be set through __getattr__, even without an explicit method:

agent = Agent("x").generate_content_config(my_config)  # Works via forwarding

Misspelled names raise AttributeError with the closest match suggestion.

Workflow Builders

All workflow builders accept both built ADK agents and fluent builders as arguments. Builders are auto-built at .build() time, enabling safe sub-expression reuse.

Pipeline (Sequential)

from adk_fluent import Pipeline, Agent

# Builder style — full control over each step
pipeline = (
    Pipeline("data_processing")
    .step(Agent("extractor", "gemini-2.5-flash").instruct("Extract entities.").outputs("entities"))
    .step(Agent("enricher", "gemini-2.5-flash").instruct("Enrich {entities}.").tool(lookup_db))
    .step(Agent("formatter", "gemini-2.5-flash").instruct("Format output.").history("none"))
    .build()
)

# Operator style — same result
pipeline = (
    Agent("extractor", "gemini-2.5-flash").instruct("Extract entities.").outputs("entities")
    >> Agent("enricher", "gemini-2.5-flash").instruct("Enrich {entities}.").tool(lookup_db)
    >> Agent("formatter", "gemini-2.5-flash").instruct("Format output.").history("none")
).build()

Method	Description
.step(agent)	Append an agent as the next step. Lazy -- built at .build() time
.build()	Resolve into a native ADK SequentialAgent

FanOut (Parallel)

from adk_fluent import FanOut, Agent

# Builder style — named branches with different models
fanout = (
    FanOut("research")
    .branch(Agent("web", "gemini-2.5-flash").instruct("Search the web.").outputs("web_results"))
    .branch(Agent("papers", "gemini-2.5-pro").instruct("Search academic papers.").outputs("paper_results"))
    .branch(Agent("internal", "gemini-2.5-flash").instruct("Search internal docs.").outputs("internal_results"))
    .build()
)

# Operator style
fanout = (
    Agent("web", "gemini-2.5-flash").instruct("Search web.").outputs("web_results")
    | Agent("papers", "gemini-2.5-pro").instruct("Search papers.").outputs("paper_results")
    | Agent("internal", "gemini-2.5-flash").instruct("Search internal docs.").outputs("internal_results")
).build()

Method	Description
.branch(agent)	Add a parallel branch agent. Lazy -- built at .build() time
.build()	Resolve into a native ADK ParallelAgent

Loop

from adk_fluent import Loop, Agent, until

# Builder style — explicit loop configuration
loop = (
    Loop("quality_loop")
    .step(Agent("writer", "gemini-2.5-flash").instruct("Write draft.").outputs("quality"))
    .step(Agent("reviewer", "gemini-2.5-flash").instruct("Review and score."))
    .max_iterations(5)
    .until(lambda s: s.get("quality") == "good")
    .build()
)

# Operator style
loop = (
    Agent("writer", "gemini-2.5-flash").instruct("Write draft.").outputs("quality")
    >> Agent("reviewer", "gemini-2.5-flash").instruct("Review and score.")
) * until(lambda s: s.get("quality") == "good", max=5)

Method	Description
.step(agent)	Append a step agent. Lazy -- built at .build() time
.max_iterations(n)	Set maximum loop iterations
.until(pred)	Set exit predicate. Exits when pred(state) is truthy
.build()	Resolve into a native ADK LoopAgent

Combining Builder and Operator Styles

The styles mix freely. Use builders for complex individual steps and operators for composition:

from adk_fluent import Agent, Pipeline, FanOut, S, until, Prompt

# Define reusable agents with full builder configuration
researcher = (
    Agent("researcher", "gemini-2.5-flash")
    .instruct(Prompt().role("You are a research analyst.").task("Find relevant information."))
    .tool(search_tool)
    .before_model(log_fn)
    .outputs("findings")
)

writer = (
    Agent("writer", "gemini-2.5-pro")
    .instruct("Write a report about {findings}.")
    .static("Company style guide: use formal tone, cite sources...")
    .outputs("draft")
)

reviewer = (
    Agent("reviewer", "gemini-2.5-flash")
    .instruct("Score the draft 1-10 for quality.")
    .outputs("quality_score")
)

# Compose with operators — each sub-expression is reusable
research_phase = (
    FanOut("gather")
    .branch(researcher.clone("web").tool(web_search))
    .branch(researcher.clone("papers").tool(paper_search))
)

pipeline = (
    research_phase
    >> S.merge("web", "papers", into="findings")
    >> writer
    >> (reviewer >> writer) * until(lambda s: int(s.get("quality_score", 0)) >= 8, max=3)
)

Presets

Reusable configuration bundles:

from adk_fluent.presets import Preset

production = Preset(model="gemini-2.5-flash", before_model=log_fn, after_model=audit_fn)

agent = Agent("service").instruct("Handle requests.").use(production).build()

@agent Decorator

from adk_fluent.decorators import agent

@agent("weather_bot", model="gemini-2.5-flash")
def weather_bot():
    """You help with weather queries."""

@weather_bot.tool
def get_weather(city: str) -> str:
    return f"Sunny in {city}"

built = weather_bot.build()

Typed State Keys

from adk_fluent import StateKey

call_count = StateKey("call_count", scope="session", type=int, default=0)

# In callbacks/tools:
current = call_count.get(ctx)
call_count.increment(ctx)

When to Use adk-fluent

Use adk-fluent when you want to:

• Define agents in 1-3 lines instead of 22+
• Compose pipelines, fan-out, loops, and routing with operators (>>, |, *, //)
• Get IDE autocomplete and type checking during development
• Test agents deterministically with .mock() and .test() (no API calls)
• Iterate quickly with .ask() and .stream() (no Runner/Session boilerplate)

Use raw ADK directly when you need to:

• Subclass BaseAgent with custom _run_async_impl logic
• Access ADK internals not exposed through the builder API
• Build framework-level tooling that wraps ADK itself
• Manage Runner/Session lifecycle with fine-grained control beyond .session()

adk-fluent produces native ADK objects. You can mix fluent-built agents with hand-built agents in the same pipeline -- they're the same types.

Run with adk web

Environment Setup

Before running any example, copy the .env.example and fill in your credentials. Two auth paths are supported — pick one:

cd examples
cp .env.example .env

Option A: Gemini API Key (simplest — no GCP project needed):

# Get a free key at https://aistudio.google.com/app/apikey
GOOGLE_API_KEY=your-gemini-api-key-here

Option B: Vertex AI (full GCP — required for some advanced features):

GOOGLE_CLOUD_PROJECT=your-project-id
GOOGLE_CLOUD_LOCATION=us-central1
GOOGLE_GENAI_USE_VERTEXAI=TRUE

Every agent loads these variables automatically via load_dotenv().

Run an Example

cd examples
adk web simple_agent          # Basic agent
adk web weather_agent         # Agent with tools
adk web research_team         # Multi-agent pipeline
adk web real_world_pipeline   # Full expression language
adk web route_branching       # Deterministic routing
adk web delegate_pattern      # LLM-driven delegation
adk web operator_composition  # >> | * operators
adk web function_steps        # >> fn (function nodes)
adk web until_operator        # * until(pred)
adk web typed_output          # @ Schema
adk web fallback_operator     # // fallback
adk web state_transforms      # S.pick, S.rename, ...
adk web full_algebra          # All operators together
adk web tap_observation       # tap() observation steps
adk web mock_testing          # .mock() for testing
adk web race                  # race() first-to-finish

68 runnable examples covering all features. See examples/ for the full list.

Cookbook

68 annotated examples in examples/cookbook/ with side-by-side Native ADK vs Fluent comparisons. Each file is also a runnable test: pytest examples/cookbook/ -v

Start Here

#	Example	What You'll Learn
01	Simple Agent	Create and build your first agent
08	One-Shot Ask	Run an agent with .ask() -- no boilerplate
04	Sequential Pipeline	Chain agents with Pipeline or >>

Core Patterns

#	Example	What You'll Learn
02	Agent with Tools	Attach tool functions
03	Callbacks	before_model, after_model hooks
05	Parallel FanOut	Run agents in parallel with FanOut or |
07	Team Coordinator	LLM-driven delegation with .delegate()
16	Operator Composition	>> | * operators together
17	Route Branching	Deterministic routing with Route
33	State Transforms	S.pick, S.rename, S.merge
37	Mock Testing	Test without LLM calls using .mock()
31	Typed Output	Pydantic schemas with @ Schema
11	Inline Testing	Smoke tests with .test()

All Examples

Full list (66 examples)

#	Example	Feature
01	Simple Agent	Basic agent creation
02	Agent with Tools	Tool registration
03	Callbacks	Additive callback accumulation
04	Sequential Pipeline	Pipeline builder
05	Parallel FanOut	FanOut builder
06	Loop Agent	Loop builder
07	Team Coordinator	Sub-agent delegation
08	One-Shot Ask	.ask() execution
09	Streaming	.stream() execution
10	Cloning	.clone() deep copy
11	Inline Testing	.test() smoke tests
12	Guardrails	.guardrail() shorthand
13	Interactive Session	.session() context manager
14	Dynamic Forwarding	__getattr__ field access
15	Production Runtime	Full agent setup
16	Operator Composition	>> | * operators
17	Route Branching	Deterministic Route
18	Dict Routing	>> dict shorthand
19	Conditional Gating	.proceed_if()
20	Loop Until	.loop_until()
21	StateKey	Typed state descriptors
22	Presets	Preset + .use()
23	With Variants	.with_() immutable copy
24	@agent Decorator	Decorator syntax
25	Validate & Explain	.validate() .explain()
26	Serialization	to_dict / to_yaml
27	Delegate Pattern	.delegate()
28	Real-World Pipeline	Full composition
29	Function Steps	>> fn zero-cost transforms
30	Until Operator	* until(pred) conditional loops
31	Typed Output	@ Schema output contracts
32	Fallback Operator	// first-success chains
33	State Transforms	S.pick, S.rename, S.merge, ...
34	Full Algebra	All operators composed together
35	Tap Observation	tap() pure observation steps
36	Expect Assertions	expect() state contract checks
37	Mock Testing	.mock() bypass LLM for tests
38	Retry If	.retry_if() conditional retry
39	Map Over	map_over() iterate agent over list
40	Timeout	.timeout() time-bound execution
41	Gate Approval	gate() human-in-the-loop
42	Race	race() first-to-finish wins
43+	Advanced	Middleware, DI, schemas, contracts
70	A2UI Basics	UI components, operators, surfaces
71	A2UI Agent Integration	.ui(), T.a2ui(), G.a2ui()
72	A2UI Operators	| (Row), >> (Column) layouts
73	A2UI LLM-Guided	UI.auto(), P.ui_schema()
74	A2UI Pipeline	S.to_ui(), S.from_ui() bridges

Browse by use case on the docs site.

Visual Cookbook Runner

A custom web frontend for interactively running and debugging all cookbook agents, with live A2UI surface rendering.

Quick Start (from scratch)

# 1. Clone and install
git clone https://github.com/vamsiramakrishnan/adk-fluent.git
cd adk-fluent
pip install -e ".[a2a,yaml,rich]"      # or: uv sync --all-extras

# 2. Configure credentials
cp visual/.env.example visual/.env
# Edit visual/.env — add your Gemini API key or Vertex AI credentials

# 3. Generate agent folders from cookbooks (one-time)
just agents                             # or: uv run python scripts/cookbook_to_agents.py --force

# 4. Launch the visual runner
just visual                             # or: uv run uvicorn visual.server:app --port 8099 --reload

Opens at http://localhost:8099 with:

• Left sidebar — all cookbooks, grouped by difficulty
• Center panel — interactive chat with any agent
• Right panel — live A2UI surface rendering + JSON inspector

A2UI Preview (no API key needed)

To browse A2UI component surfaces without any LLM calls:

just a2ui-preview

This exports surfaces from cookbooks 70-74 and opens a static gallery in your browser.

Visual Regression Tests

uv run pytest tests/visual/ -v                      # run tests (no API key needed)
uv run pytest tests/visual/ -v --update-golden       # update golden snapshots

See visual/README.md for full architecture details.

Performance

adk-fluent is a build-time layer. Calling .build() produces a native ADK object -- the same LlmAgent, SequentialAgent, or ParallelAgent you'd construct manually. After .build(), adk-fluent is not in the execution path. There is no runtime wrapper, proxy, or middleware layer injected by the builder itself.

Build overhead: Builder construction adds microseconds of Python dict manipulation per agent. For context, a single Gemini API call takes 500ms-30s.

Verify yourself:

python scripts/benchmark.py

ADK Compatibility

Version Support Policy (N-5 Guarantee)

adk-fluent officially supports the current and previous 5 minor/patch releases of google-adk. Every CI run tests the generated fluent API against all 6 versions in the compatibility matrix.

google-adk	Status	Tested in CI
1.27.x	Current	Yes
1.26.x	Supported	Yes
1.25.x	Supported	Yes
1.24.x	Supported	Yes
1.23.x	Supported	Yes
1.22.x	Supported	Yes
< 1.22	Best-effort	No

How it works: The fluent API surface is generated from the latest ADK (via scanner.py), but the generated builders pass configuration through _safe_build(), which delegates to ADK's own Pydantic models at runtime. When you run against an older ADK:

• Builder methods for fields that exist in your ADK version work normally.
• Builder methods for fields added in a newer ADK will raise a clear BuilderError at .build() time, not silently fail.
• The pyproject.toml floor (google-adk>=1.20.0) is intentionally lower than the N-5 window to avoid hard-blocking users on older versions, but only versions within the N-5 window are actively tested.

Deprecation: When a new ADK version is released, the oldest version in the window drops out. We update the CI matrix, but do not add code to intentionally break older versions. Users on deprecated versions may continue to work but are not covered by CI.

A weekly sync workflow scans for new ADK releases every Monday, regenerates code, runs the full N-5 compatibility matrix, and opens a PR automatically. If you hit an incompatibility, open an issue.

How It Works

adk-fluent is auto-generated from the installed ADK package:

scanner.py ──> manifest.json ──> seed_generator.py ──> seed.toml ──> generator.py ──> Python code
                                      ^
                              seed.manual.toml
                              (hand-crafted extras)

1. Scanner introspects all ADK modules and produces manifest.json
2. Seed Generator classifies classes and produces seed.toml (merged with manual extras)
3. Code Generator emits fluent builders, .pyi type stubs, and test scaffolds

This means adk-fluent automatically stays in sync with ADK updates:

pip install --upgrade google-adk
just all   # Regenerate everything
just test  # Verify

API Reference

Generated API docs are in docs/generated/api/:

• agent.md -- Agent, BaseAgent builders
• workflow.md -- Pipeline, FanOut, Loop
• tool.md -- 40+ tool builders
• service.md -- Session, artifact, memory services
• config.md -- Configuration builders
• plugin.md -- Plugin builders
• runtime.md -- Runner, App builders

Migration guide: docs/generated/migration/from-native-adk.md

Features

• 130+ builders covering agents, tools, configs, services, plugins, planners, executors
• Expression algebra: >> (sequence), | (parallel), * (loop), @ (typed output), // (fallback), >> fn (transforms), S (state ops), Route (branch)
• Prompt builder: structured multi-section prompt composition via Prompt
• Template variables: {key} in instructions auto-resolved from session state
• Context control: .static() for cacheable context, .history("none") for stateless agents, .inject_context() for dynamic preambles
• State transforms: S.pick, S.drop, S.rename, S.default, S.merge, S.transform, S.compute, S.guard
• Full IDE autocomplete via .pyi type stubs
• PEP 561 py.typed marker included -- type checkers recognize the package natively
• Zero-maintenance __getattr__ forwarding for any ADK field
• Callback accumulation: multiple .before_model() calls append, not replace
• Typo detection: misspelled methods raise AttributeError with suggestions
• A2UI (Agent-to-UI): declarative UI composition via UI namespace -- UI.form(), UI.dashboard(), | (Row), >> (Column) operators, compile_surface() to A2UI JSON
• Deterministic routing: Route evaluates predicates against session state (zero LLM calls)
• One-shot execution: .ask(), .stream(), .session(), .map() without Runner boilerplate
• Presets: reusable config bundles via Preset + .use()
• Cloning: .clone() and .with_() for independent variants
• Validation: .validate() catches config errors at definition time
• Serialization: to_dict(), to_yaml(), from_dict(), from_yaml()
• @agent decorator: FastAPI-style agent definition
• Typed state: StateKey with scope, type, and default

AI Coding Skills

adk-fluent ships with Agent Skills that teach AI coding assistants how to use the library. Install them into any compatible tool:

npx skills add vamsiramakrishnan/adk-fluent -y -g

Skill	Description
adk-fluent-cheatsheet	API quick reference — builder methods, operators, namespaces, ADK→fluent mapping
adk-fluent-dev-guide	Development lifecycle — spec-driven workflow, phase-based development, troubleshooting
adk-fluent-eval-guide	Evaluation methodology — E namespace, eval suites, criteria, LLM-as-judge
adk-fluent-deploy-guide	Deployment — Agent Engine, Cloud Run, production middleware, guards
adk-fluent-observe-guide	Observability — M namespace middleware, introspection, Cloud Trace
adk-fluent-scaffold	Project scaffolding — directory structure, templates, ADK CLI integration

Works with Gemini CLI, Claude Code, Cursor, GitHub Copilot, Amp, and more.

Development

Requires: Python 3.11+, just, uv

Container: Open in VS Code or GitHub Codespaces with the included Dev Container for a pre-configured environment.

# Setup
uv venv .venv && source .venv/bin/activate
uv pip install -e ".[dev]"

# Full pipeline: scan -> seed -> generate -> docs
just all

# Run tests (780+ tests)
just test

# Type check hand-written code
just typecheck-core

# Local CI (lint + check-gen + test)
just ci

See CONTRIBUTING.md for the full development guide.

Latest Changes

See CHANGELOG.md for the full release history. Recent highlights:

• v0.11.0 -- A module Phase 2+3 (batch ops, LLM tools, content transforms, ArtifactSchema), auto-generated namespace API docs, DevEx overhaul
• v0.10.0 -- A module Phase 1 (artifact lifecycle), Fallback builder, verb harmonization (agent_tool, guard, loop_while, prepend)
• v0.9.6 -- T module for tool composition, ToolRegistry with BM25-indexed discovery
• v0.9.5 -- Middleware v2 (TraceContext, per-agent scoping, topology hooks), M module, P module, MiddlewareSchema

Publishing

Releases are published automatically to PyPI when a version tag is pushed:

# 1. Bump version in pyproject.toml
# 2. Commit and tag
git tag v0.2.0
git push origin v0.2.0
# 3. CI runs tests -> builds -> publishes to PyPI automatically

TestPyPI publishing is available manually via the GitLab CI web interface.

License

MIT