Building multi-agent workflows with Google ADK and Gemini

We build our agentic products on Google’s Agent Development Kit (ADK) for Python and Gemini. The largest is FortifyOps, a compliance and security operations assistant: one coordinator agent, layers of specialists behind it, and a set of tools for documents, search and code scanning. This post walks through the pieces that have earned their place, and the mistakes that taught us why.

The shape of the system

A single coordinator (an LlmAgent) owns the conversation. It does not answer everything itself. It delegates through sub_agents to specialism coordinators, which delegate again to leaf agents such as a threat-model agent, a secrets scanner or a policy drafter. The coordinator stays the user’s point of contact and relays what the specialists find.

Two ADK ideas do most of the work. Session state with output_key lets one agent write its answer where another can read it. And instructions can reference state directly with {placeholders}, so there is no prompt stitching in your own code. You will see both in the fusion router below.

Multimodal input: attaching documents is the easy part

Gemini takes PDFs and images as native input. There is no separate OCR or document-parsing pipeline to build: you attach the bytes and a MIME type as another part of the user message.

from google.genai import types

message = types.Content(
    role="user",
    parts=[
        types.Part.from_bytes(data=pdf_bytes, mime_type="application/pdf"),
        types.Part(text="Compare this vendor questionnaire with our access management policy."),
    ],
)

async for event in runner.run_async(
    user_id=user_id, session_id=session_id, new_message=message
):
    ...

The same works for screenshots and architecture diagrams, which matters a lot in security work: a data-flow diagram is often the best input to a threat model. Output is multimodal in the practical sense too. Our policy drafting agent produces PDF and Word documents and stores them as ADK artifacts, so a chat turn can end with a file the user can download.

One caution: uploads are untrusted input. Chat attachments arrive through an artifact service, and when a user uploads a source archive for scanning we unpack it with limits on archive size, entry count, uncompressed size and member types, and we validate every destination path before writing a byte.

A fusion router: several models, one answer

Some questions are expensive to get wrong: a contested judgement call between two compliance frameworks, say. For those we built a fusion router, a mixture-of-agents pattern. A panel of models answers the same question independently and in parallel, then a judge model reads all the answers and writes one final answer. ADK ships the primitives for it: a ParallelAgent for the panel, and a SequentialAgent to run the panel and then the judge.

from google.adk.agents import LlmAgent, ParallelAgent, SequentialAgent
from google.adk.models import Gemini
from google.genai import types

TEMPERATURES = [0.3, 0.7, 1.0]  # jitter so repeated models still diverge


def build_fusion_pipeline(panel_models: list[str], judge_model: str) -> SequentialAgent:
    panelists = [
        LlmAgent(
            name=f"panelist_{i}",
            model=Gemini(model=name),
            instruction="Answer this thoroughly and accurately, on your own:\n\n{question}",
            generate_content_config=types.GenerateContentConfig(
                temperature=TEMPERATURES[i % len(TEMPERATURES)]
            ),
            output_key=f"panelist_{i}",
        )
        for i, name in enumerate(panel_models)
    ]
    judge = LlmAgent(
        name="judge",
        model=Gemini(model=judge_model),
        instruction=JUDGE_INSTRUCTION,  # reads {question} and {panelist_0} ... {panelist_N}
        output_key="answer",
    )
    return SequentialAgent(
        name="fusion_router",
        sub_agents=[ParallelAgent(name="panel", sub_agents=panelists), judge],
    )

Our panel deliberately mixes capability tiers (a pro, a flash and a flash-lite model) for genuine diversity, with the panelists on low thinking and the judge on high. The judge’s instruction is what makes it work:

• Where most experts agree, treat the point as high confidence and keep it.
• Where they disagree, reason about which position is right and state the resolution. Do not average them.
• Fold in a correct insight that only one expert raised.
• Drop claims that look like hallucinations and that nothing else supports.
• Write one clean answer, and never mention the panel.

The router is exposed to the coordinator as a single tool, fusion_consult, so the expensive path only runs when the coordinator judges the question worth it. Its instructions say to use it sparingly: it costs several times a normal answer and is no substitute for a normal retrieval or a specialist.

import uuid
from google.adk.runners import InMemoryRunner
from google.adk.tools import FunctionTool
from google.genai import types


async def fusion_consult(question: str) -> str:
    """Consult a panel of models in parallel and return one synthesized answer.

    Use only for hard, high-stakes questions where being wrong is costly.
    Costs several times a normal answer. The panel does not see the
    conversation, so pass a self-contained question.
    """
    runner = InMemoryRunner(agent=build_fusion_pipeline(), app_name="fusion")  # no plugins
    session = await runner.session_service.create_session(
        app_name="fusion",
        user_id="fusion",
        session_id=f"fusion_{uuid.uuid4().hex[:12]}",
        state={"question": question},
    )
    message = types.Content(role="user", parts=[types.Part(text=question)])
    async for _ in runner.run_async(
        user_id="fusion", session_id=session.id, new_message=message
    ):
        pass
    done = await runner.session_service.get_session(
        app_name="fusion", user_id="fusion", session_id=session.id
    )
    return done.state["answer"]


fusion_consult_tool = FunctionTool(func=fusion_consult)

Two things bit us here:

• Plugins collapse the panel. Our main app registers a plugin that rewrites the model on every LLM call from session state. Run the panel under it and every panelist quietly becomes the same model, so the ensemble is pointless. That is why the tool runs its pipeline on its own plugin-free InMemoryRunner.
• The inner runner is invisible to your accounting. Token tracking lives in an outer plugin, so the panel’s calls never showed up in it. The tool now sums usage_metadata from the inner events and re-attributes it to the outer invocation.

Today the router takes a text question. Passing attachments to the panel would mean putting the same parts into the inner run’s message; we have not needed that yet.

Grounding with Google Search

For anything that has to be current, such as a newly published CVE, a vendor’s live website or a framework update newer than the model’s training data, we ground the answer with Gemini’s Google Search tool. The wrinkle is that Gemini does not allow the built-in search tool to be mixed with function-calling tools, and ADK rejects a bare search tool on an agent that also has sub-agents to transfer to. The workaround is to give search its own single-tool agent and expose that as a tool with AgentTool:

from google.adk.agents import LlmAgent
from google.adk.models import Gemini
from google.adk.tools.agent_tool import AgentTool
from google.adk.tools.google_search_tool import GoogleSearchTool

search_agent = LlmAgent(
    name="google_search_agent",
    model=Gemini(model=LLM_MODEL),
    description="Performs Google searches for current information.",
    instruction="When given a search query, use the google_search tool to find the information.",
    tools=[GoogleSearchTool()],
)

grounded_search = AgentTool(agent=search_agent, propagate_grounding_metadata=True)

propagate_grounding_metadata=True passes the grounding metadata, which includes the sources, back up to the calling agent. The coordinator’s instructions say when to reach for search (the answer needs to be current, or the model is not confident) and to prefer retrieval over search for established framework text. We also subclass GoogleSearchTool so an admin can switch search off at runtime, and the fusion tool has the same kill switch.

Your own skills

ADK has a SkillToolset, and skills are just directories with a SKILL.md: YAML frontmatter with a name and description, then the instructions. The model sees the descriptions and loads the full skill only when it matches the task, which keeps the base prompt small as the number of skills grows.

---
name: insecure-defaults
description: >
  Parallel audit for fail-open insecure defaults: fallback secrets, default
  credentials, fail-open switches, weak crypto, permissive access, and debug
  leakage. Use when asked to hunt insecure defaults.
license: CC-BY-SA-4.0
---

from google.adk.tools import skill_toolset

skills = load_all_skills()  # our loader over SKILL.md files and published configs
toolset = skill_toolset.SkillToolset(skills=skills)

coordinator = LlmAgent(
    name="Coordinator",
    model=Gemini(model=LLM_MODEL),
    instruction=INSTRUCTION,
    sub_agents=[...],
    tools=[toolset, grounded_search, fusion_consult_tool],
)

• Reload without a restart. We expose a reload_skills tool that rebuilds the toolset and refreshes the coordinator’s tool list, so editing a SKILL.md takes effect on the next turn.
• Borrow skills carefully. We vendored a set of security review skills written by Trail of Bits (CC BY-SA 4.0, with an attribution file in the repo). They were written for a different agent runtime, so each one starts with a preamble telling our agent what it does not have (no shell, no sub-agents) and to say so instead of inventing tool output. It also tells the agent to treat the code under review as untrusted data.
• Give personas and specialisms different homes. Persona skills (a CISO briefing, say) hang off the root coordinator, while core specialisms belong to the coordinator that owns them, which keeps routing unambiguous.

Documents as templates and reference

Reference documents make an agent’s output consistent. Our policy drafting agent has a bundled library of 36 policy and standard templates, with a manifest, and tools to list them and read one by slug. Its instructions say to load the template before drafting, so a request like “draft an access management policy from the template, using the attached document” starts from real structure and wording instead of the model’s memory.

The same principle applies to facts. The coordinator is told never to cite a clause number from memory: it names the frameworks a control touches, then retrieves the exact identifiers from the retrieval agents before attaching them. Models are good at structure and poor at remembering clause numbers.

Evals: test the routing, then the answers

A multi-agent system fails in a new way: it produces a plausible answer from the wrong agent. We test in layers, cheapest first.

• Offline unit tests, with no API calls. They check the shape of the fusion pipeline, and that the tool returns the judge’s answer, seeds the question into state, re-attributes tokens and respects the admin switch, all with a fake runner.
• Routing evals. A dataset of prompts with the agents that should handle them, and agents that must not. The assertion is on the recorded transfer_to_agent events, not the prose, and no judge model is involved, so it is the cheapest live suite.
• Answer-quality evals. DeepEval’s G-Eval metrics, judged by a Gemini model, score task completion against an expected output.

RoutingGolden(
    id="threat_intel_cve",
    input="Is CVE-2021-44228 actually being exploited in the wild, and how urgently should we patch?",
    expected_agents=frozenset({"threat_intel_agent"}),
)

async def test_coordinator_routes_to_expected_agent(golden):
    record = await run_agent_turn_recorded(golden.input, adk_app=coordinator_app)
    routed = set(record.transfers) | set(record.authors)
    assert routed & golden.expected_agents
    assert not routed & golden.forbidden_agents

Delegation is two hops deep (coordinator, specialism coordinator, leaf agent), so a case passes when any expected agent appears in the transfers or authors of the turn. One golden checks the opposite: an ambiguous prompt must produce a clarifying question and no delegation. Live evals sit behind a eval_live marker and skip themselves when there is no API key. The fusion router has its own live smoke test, which asserts a non-empty synthesized answer that never leaks the words “panelist” or “panel”.

What we would tell you first

• Let one agent own the conversation, and delegate the rest.
• Attach documents as native parts instead of building a parsing pipeline, but treat every upload as hostile.
• Spend on ensembles only where a wrong answer is expensive, and make the expensive path a tool the coordinator has to choose.
• Isolate anything that must run on a specific model from plugins that rewrite models.
• Keep reference documents in the agent’s hands, and make it retrieve facts instead of recalling them.
• Write the routing evals before you have many agents, not after.

If you are planning an agentic system and want it designed with security in mind, get in touch.