Wire a Multi-Agent Pipeline with ADK

Orchestrate 3 AI agents with typed Pydantic contracts to generate WCAG fixes.

Introduction

30 Second Summary

Accessibility audits are one of those tasks everyone agrees matter, but nobody wants to do manually. You scan a page, get a wall of violations, then spend hours prioritizing them and writing the actual code fixes. What if you could point three AI agents at a URL and get back prioritized violations with ready-to-paste HTML fixes?

In this project, you will wire a multi-agent pipeline using ADK Workflows that orchestrates three stages: a scanner agent, a triage agent, and a fix-generation agent. You will type a URL and watch all three collaborate in sequence to produce structured, validated accessibility fixes.

What You'll Build

You will type a URL into the ADK web interface, hit enter, and watch three agents hand off data to each other in real time. The scanner opens a headless browser, the triage agent prioritizes violations with WCAG citations, and the fix-gen agent writes actual HTML code fixes you can paste into a codebase.

By the end of this project, you'll have:

A 3-stage pipeline you can point at any URL to get prioritized violations and ready-to-use HTML fixes in a single command.
Structured Pydantic schemas (TriageReport, FixGenReport) that validate every piece of data flowing between agents, so downstream code can trust the shape without parsing free text.
A fix-generation agent powered by Gemini 2.5 Pro that writes real HTML remediation code citing the exact WCAG success criterion each fix addresses.
Secret Mission: Add conditional routing that skips the expensive Pro model when a site has zero violations.

Are there any prerequisites?

This project builds on your work from Project 1 (Containerize a Python Scanner with Docker) and Project 2 (Build Your First ADK Agent with Vertex AI). You should have your accessagent-scanner/ directory, a GCP project with Vertex AI enabled, and Application Default Credentials configured.

Before We Start

Before diving into code, let's lock in what you're about to build and why it matters.

You're refactoring from a single agent calling tools into a multi-agent Workflow where each pipeline stage is its own node with typed data flowing between them.

Create the Pipeline Directory

Your triage_agent/ from Project 2 uses a single-agent-with-tools pattern. That works when one agent handles everything, but a multi-stage pipeline needs each stage isolated into its own module with clear boundaries.

The Workflow pattern in ADK organizes a pipeline into separate files for data schemas, tool functions, and agent definitions. In this step, you'll scaffold a new pipeline/ directory and write the scanner function that kicks off the entire chain.

In this step, get ready to:

Create the pipeline/ directory with all required module files.
Write the scan_url function as a Workflow-compatible FunctionNode.
Verify the module imports correctly from the project root.

Create the pipeline directory structure

The Workflow needs a Python package with dedicated modules. You'll create the directory and all five files now, then fill in the important ones.

In your VS Code terminal (which should be in the accessagent-scanner/ directory with your .venv active), create the pipeline directory and its files by running these commands:

mkdir pipeline
touch pipeline/__init__.py pipeline/.env pipeline/tools.py pipeline/schemas.py pipeline/agent.py

Confirm the files exist by running:

ls pipeline/

You should see five files listed: __init__.py, .env, tools.py, schemas.py, and agent.py.

Why separate modules?

Each file has one job. schemas.py defines the data contracts between agents. tools.py holds the scanner function. agent.py wires agents and the Workflow together.

You'll fill schemas.py and agent.py in later steps. For now they stay empty as placeholders.

Open pipeline/.env in VS Code and paste the following configuration:

GOOGLE_GENAI_USE_ENTERPRISE=True
GOOGLE_CLOUD_PROJECT=[[GCP_PROJECT="your-gcp-project-id"]]
GOOGLE_CLOUD_LOCATION=us-central1

What does this configure?

GOOGLE_GENAI_USE_ENTERPRISE=True tells ADK to route requests through Vertex AI (your GCP project) instead of the consumer Gemini API.

Replace the project ID with the same GCP project you used in Project 2. The region us-central1 is where Vertex AI processes your requests.

Open pipeline/__init__.py and paste this single line:

from . import agent

What does this do?

This tells Python that pipeline/ is a package and registers the agent module so ADK can discover root_agent inside it. Even though agent.py is empty right now, importing an empty module is valid Python.

Save both files.

Write the scan_url function

The scanner is the first node in the pipeline. In the Workflow pattern, it runs as a FunctionNode. ADK auto-converts the user's message into the type hinted in the function signature.

Open pipeline/tools.py and paste the imports and browser-scanning logic:

from axe_playwright_python.sync_playwright import Axe
from playwright.sync_api import sync_playwright


def scan_url(node_input: str) -> dict:
    """Scans a URL for WCAG accessibility violations using axe-core.

    Opens the URL in a headless browser, runs the axe-core engine,
    and returns structured violation data for downstream agents.
    """
    url = str(node_input).strip()
    axe = Axe()
    with sync_playwright() as pw:
        browser = pw.chromium.launch(headless=True)
        page = browser.new_page()
        page.goto(url, wait_until="networkidle")
        results = axe.run(page)
        browser.close()

What does this code do?

node_input: str tells ADK to extract the user's text message and pass it as a plain string. When this function sits after START in the edge list, it receives whatever URL the user typed.
Axe() initializes the axe-core accessibility engine. Playwright opens a headless Chromium browser, loads the page, and axe scans it for WCAG violations.
The function returns a dict which ADK automatically passes to the next node in the Workflow edge list.

Still in pipeline/tools.py, add the violations-processing logic directly below browser.close() (inside the function, but outside the with block):

    violations = []
    for v in results.response.get("violations", []):
        wcag_tags = [t for t in v.get("tags", []) if t.startswith("wcag")]
        violations.append({
            "rule_id": v["id"],
            "impact": v.get("impact", "unknown"),
            "description": v.get("description", ""),
            "help_url": v.get("helpUrl", ""),
            "wcag_tags": wcag_tags,
            "affected_elements": len(v.get("nodes", [])),
            "nodes_html": [n.get("html", "") for n in v.get("nodes", [])[:3]],
        })

    return {
        "url": url,
        "total_violations": len(violations),
        "violations": violations,
    }

What does this code do?

The loop iterates over each violation axe-core found. For each one, it extracts the rule ID, impact level, WCAG tags, affected element count, and up to 3 HTML snippets.
wcag_tags filters for only the WCAG-related tags (e.g. wcag2a, wcag21aa), ignoring other metadata tags axe attaches.
The returned dict contains the URL, total violation count, and the full list of structured violations. This dict flows to the next Workflow node automatically.

Save pipeline/tools.py.

✔️ Awesome, I've got everything!

Great. Make sure you've saved tools.py with both chunks in place.

ⓧ I'd like to double check the full code

Your pipeline/tools.py should contain two sections in order:

The imports and browser-scanning chunk (from from axe_playwright_python... through browser.close()).
The violations-processing chunk (from violations = [] through the return statement).

Scroll up and confirm both chunks are present, with correct indentation (4 spaces for code inside the function).

Verify the module imports correctly

Before moving on, confirm Python can find and import your new module from the project root.

Before you run this, what do you expect to see? The import will load pipeline/__init__.py (which imports the empty agent.py), then pull scan_url from tools.py.
Test the import by running this command:

python -c "from pipeline.tools import scan_url; print('OK')"

You should see OK printed with no errors.

Seeing a ModuleNotFoundError?

Make sure your terminal is in the accessagent-scanner/ directory (one level above pipeline/). Run pwd to check.
Confirm your virtual environment is active. You should see (.venv) at the start of your terminal prompt.
Check that pipeline/__init__.py exists. Without it, Python does not recognise the folder as a package.

Still stuck? Help me debug this ModuleNotFoundError when importing from pipeline.tools

Your pipeline scaffold is in place and the scanner function imports cleanly. Next, you'll wire this function together with the triage agent into a working two-stage Workflow.

Build the Scanner-Triage Workflow

Your scanner tool is imported and ready. Now it's time to connect it to an AI agent that can interpret the raw violation data.

A raw scan dumps violations with no prioritization or WCAG citations. In this step, you'll wire scan_url and a triage agent into an ADK Workflow, proving the orchestration works before adding the full 3-stage pipeline.

In this step, get ready to:

Define a triage agent that prioritizes violations and cites WCAG criteria.
Wire the scanner and triage agent into a Workflow using edge sequence shorthand.
Run the pipeline and see a human-readable prioritized report in the ADK web UI.

Define the triage agent

The triage agent receives the raw scan output and transforms it into a prioritized accessibility report. It uses Gemini 2.5 Flash for fast, cost-efficient classification.

Open pipeline/agent.py in VS Code. The file is currently empty.
Add the triage agent definition by pasting this code:

from google.adk import Agent, Workflow

from .tools import scan_url


triage_agent = Agent(
    model="gemini-2.5-flash",
    name="triage_agent",
    instruction="""You are an accessibility expert specializing in WCAG 2.1/2.2 AA compliance.

You receive raw scan results from the axe-core accessibility engine as JSON input. Your job is to:
1. Summarize the site's accessibility health in 1-2 sentences.
2. For each violation, assign a priority: P0 for critical, P1 for serious, P2 for moderate, P3 for minor.
3. Cite the specific WCAG success criterion from the wcag_tags field.
4. For each violation, provide a brief actionable fix suggestion.

Return ALL violations from the input, prioritized and enriched with fix suggestions.""",
    output_key="triage_report",
)

What does this code do?

from google.adk import Agent, Workflow imports the two ADK classes you need: Agent for LLM-powered nodes and Workflow for graph-based orchestration.
model="gemini-2.5-flash" selects the Flash model ($0.30/1M input tokens) for cost-efficient triage classification.
instruction tells the agent exactly what to do with incoming scan data: prioritize, cite WCAG criteria, and suggest fixes.
output_key="triage_report" stores the agent's output in session state under the key triage_report, making it available to downstream agents.

Save pipeline/agent.py.
Verify the agent definition has no syntax errors by running this command:

python -c "from pipeline.agent import triage_agent; print(triage_agent.name)"

You should see triage_agent printed in the terminal.

Seeing an ImportError?

Make sure your virtual environment is active. You should see (.venv) in your terminal prompt.
Check that you're running the command from the accessagent-scanner/ directory, not from inside pipeline/.
Confirm pipeline/__init__.py contains from . import agent.

Still stuck? Help me debug this ImportError when importing triage_agent from pipeline.agent

Wire the Workflow with edge sequence shorthand

A Workflow connects nodes through edges that define the execution order. ADK's sequence shorthand lets you declare the full path in a single tuple instead of listing individual pairs.

In pipeline/agent.py, add the following code below the triage_agent definition:

root_agent = Workflow(
    name="pipeline",
    edges=[("START", scan_url, triage_agent)],
)

What does this code do?

Workflow is the ADK class for graph-based orchestration. It executes nodes in the order defined by edges.
The tuple ("START", scan_url, triage_agent) is sequence shorthand. It means: start with scan_url, then pass the output to triage_agent.
The variable must be named root_agent because ADK's web UI and CLI look for this exact name when discovering agents.

Save pipeline/agent.py.
Confirm the Workflow was created successfully by running:

python -c "from pipeline.agent import root_agent; print(root_agent.name)"

You should see pipeline printed in the terminal.

Seeing a NameError for scan_url?

Check that the import line from .tools import scan_url is at the top of your agent.py file.
Confirm pipeline/tools.py exists and defines scan_url.

Still stuck? Help me fix this NameError when creating the Workflow with scan_url

✔️ Awesome, I've got everything!

Great. Make sure you've saved pipeline/agent.py and both import checks passed.

ⓧ I'd like to double check the full code

Here is the complete pipeline/agent.py file at this point:

from google.adk import Agent, Workflow

from .tools import scan_url


triage_agent = Agent(
    model="gemini-2.5-flash",
    name="triage_agent",
    instruction="""You are an accessibility expert specializing in WCAG 2.1/2.2 AA compliance.

You receive raw scan results from the axe-core accessibility engine as JSON input. Your job is to:
1. Summarize the site's accessibility health in 1-2 sentences.
2. For each violation, assign a priority: P0 for critical, P1 for serious, P2 for moderate, P3 for minor.
3. Cite the specific WCAG success criterion from the wcag_tags field.
4. For each violation, provide a brief actionable fix suggestion.

Return ALL violations from the input, prioritized and enriched with fix suggestions.""",
    output_key="triage_report",
)

root_agent = Workflow(
    name="pipeline",
    edges=[("START", scan_url, triage_agent)],
)

Run the pipeline in the ADK web UI

Time to see the full Workflow in action. You'll launch the ADK dev server, send a URL to the pipeline, and watch the scanner and triage agent collaborate.

Start the ADK web UI by running:

adk web .

What does this command do?

adk web . starts a local development server at http://localhost:8000. The . tells ADK to discover agent directories in the current folder. It will find your pipeline/ directory because it has a __init__.py importing agent and an agent.py defining root_agent.

Open http://localhost:8000 in your browser.
In the top-left dropdown, select pipeline from the agent list.

Before you type a URL, what do you think will happen? The scanner will open a headless browser, run axe-core, and pass the raw violations to the triage agent. The triage agent will then prioritize them and add WCAG citations.

Type a URL into the chat input (for example, https://example.com) and press Enter.

What should I see?

The scan may take 10-20 seconds while the headless browser loads the page. After that, the triage agent responds with a prioritized accessibility report in the chat.

The report includes a 1-2 sentence health summary, each violation with a priority (P0-P3), WCAG success criterion citations, and a brief fix suggestion. If the site is clean (like example.com), you may see zero or very few violations.

Try a site with known issues

For a more interesting report, scan a complex public site. Sites with many interactive elements tend to surface more violations for the triage agent to prioritize.

Agent not showing in the dropdown?

Make sure you ran adk web . from the accessagent-scanner/ directory (the parent of pipeline/).
Confirm pipeline/__init__.py contains from . import agent and pipeline/agent.py defines root_agent.
Stop the server with Ctrl+C and restart it. ADK caches agent discovery on startup.

Still stuck? Help me figure out why my pipeline agent doesn't appear in adk web

Your scanner and triage agent are collaborating through a Workflow. Next up, you'll add a fix-generation agent to the pipeline and discover why free-text output between agents causes problems.

Add the Fix-Gen Agent

Your 2-stage pipeline scans a URL and produces a prioritized triage report. That report tells a human exactly what is wrong and in what order to fix it.

Now let's add a third stage: a fix-generation agent powered by Gemini 2.5 Pro that reads the triage output and writes concrete HTML code fixes for each violation.

In this step, get ready to:

Define a fix-gen agent that generates HTML remediation code from triage data.
Wire the fix-gen agent into the Workflow as a third stage.
Run the full 3-stage pipeline and examine the fix-gen output quality.

Define the fix-gen agent

The fix-gen agent uses gemini-2.5-pro because generating correct HTML code requires deeper reasoning than classification or prioritization.

In VS Code, open pipeline/agent.py.
Find the root_agent definition at the bottom of the file. Add the following code above it:

fix_gen_agent = Agent(
    model="gemini-2.5-pro",
    name="fix_gen_agent",
    instruction="""You are a senior frontend developer specializing in web accessibility remediation.

You receive a triage report of WCAG violations as input. For the top 5 highest-priority violations (P0 first, then P1, then P2), generate a concrete HTML code fix.

For each fix:
1. Identify the specific WCAG success criterion being violated (e.g. 1.1.1 Non-text Content).
2. Show the original problematic HTML from the violation data.
3. Write the corrected HTML that resolves the violation.
4. Explain briefly why the fix works, referencing the WCAG criterion.

Generate real, usable HTML fixes. If the report shows zero violations, return an empty fixes list with a congratulatory summary.""",
    output_key="fix_gen_report",
)

What does this code do?

This defines a new LLM agent node for the Workflow.

gemini-2.5-pro is chosen for its stronger code-generation reasoning. Flash handles classification well, but Pro handles multi-step code authoring.
The instruction tells the agent to generate HTML fixes for the top 5 violations, citing WCAG success criteria.
output_key="fix_gen_report" stores the agent's output in session state under that key for later access.

Why Pro for fix-gen and Flash for triage?

Triage is classification and prioritization. That is a low-reasoning task, so Flash at $0.30/1M input tokens handles it efficiently.

Fix-gen requires understanding HTML semantics, WCAG criteria, and writing syntactically correct code. Pro at $1.25/1M input tokens justifies the cost for that deeper reasoning. A typical fix-gen call on 5 violations costs about $0.05.

Wire fix-gen into the Workflow edges

The Workflow edge sequence shorthand lets you chain nodes in order. Adding fix_gen_agent to the end of the tuple makes it the third stage.

In the same file, find this line inside root_agent:

edges=[("START", scan_url, triage_agent)],

Replace it with this line, adding fix_gen_agent to the sequence:

edges=[("START", scan_url, triage_agent, fix_gen_agent)],

What does edge sequence shorthand do?

The tuple ("START", scan_url, triage_agent, fix_gen_agent) is shorthand for three separate edges: START to scan_url, scan_url to triage_agent, and triage_agent to fix_gen_agent. Each node's output flows into the next node's input.

Save pipeline/agent.py.

✔️ Awesome, I've got everything!

Great. Make sure you have saved the file before continuing.

ⓧ I'd like to double check the full code

Your pipeline/agent.py should have these sections in order:

Imports: Agent, Workflow from google.adk, and scan_url from .tools.
triage_agent definition (unchanged from the previous step).
fix_gen_agent definition with gemini-2.5-pro model and output_key="fix_gen_report".
root_agent Workflow with edges including all three nodes: scan_url, triage_agent, fix_gen_agent.

Scroll up to confirm each chunk matches what you pasted.

Run the pipeline and examine the output

The 3-stage pipeline is wired up. Time to test it end-to-end.

Start the ADK web interface by running this command from the accessagent-scanner/ directory:

adk web .

In the ADK web interface, select pipeline from the agent dropdown.
Type a URL with known accessibility issues into the chat (e.g. a site you tested in Step 2).

Before you read the output, predict: do you think the fix-gen agent will produce accurate, usable HTML fixes with correct rule IDs and WCAG citations?

Read the fix-gen agent's output carefully. Look at the rule IDs, the HTML snippets, and the WCAG citations it produces.

You will notice the output is inconsistent. Some rule IDs do not match what the scanner actually found. Some HTML snippets are invented rather than pulled from the real scan data. WCAG citations may be missing or wrong.

This is the expected result. The fix-gen agent is receiving the triage output as unstructured free text, not as typed data it can parse programmatically.

Why is fix-gen producing unreliable output?

Without an output_schema on the triage agent, its output flows downstream as a types.Content object. That is free-form text, not structured data.

The fix-gen agent has to guess where rule IDs end and descriptions begin. It has to infer which HTML snippets belong to which violations. That guessing leads to hallucinated fixes and mismatched citations.

Not seeing any output from fix-gen?

Make sure you saved pipeline/agent.py after adding the fix_gen_agent and updating the edges.
Check that fix_gen_agent appears in the edges tuple after triage_agent. A missing comma between agents will cause a syntax error.
If you see an API error, confirm your GCP project has Vertex AI API enabled and your ADC credentials are valid.

Help me debug my fix-gen agent not producing output.

You have confirmed the quality gap. The pipeline runs end-to-end, but free text flowing between agents produces unreliable downstream results. In the next step, you will add Pydantic schemas that force structured data between stages and eliminate the guessing entirely.

Add Pydantic Schemas for Structured Data Flow

In the last step, you saw fix-gen produce unreliable output. It received free-form text from the triage agent and had to guess at the structure. Wrong rule IDs, invented HTML, missing WCAG citations.

The fix is output_schema. When you attach a Pydantic model as an agent's output schema, ADK forces the LLM to emit a validated JSON dict matching that model. The downstream agent receives typed, structured data it can consume deterministically.

In this step, get ready to:

Define Pydantic models (TriageReport, FixGenReport) as typed data contracts between agents.
Add output_schema to both agents so ADK enforces structured JSON output.
Run the pipeline and confirm fix-gen produces consistent, accurate code fixes.

Define the triage data contract

A data contract is a schema that both sides agree on. The triage agent promises to produce data matching this shape, and the fix-gen agent can trust that shape without parsing free text.

Open pipeline/schemas.py in VS Code.
Add the triage data models by pasting this code:

from pydantic import BaseModel, Field


class TriagedViolation(BaseModel):
    rule_id: str = Field(description="The axe-core rule ID")
    impact: str = Field(description="critical, serious, moderate, or minor")
    priority: str = Field(description="P0 for critical, P1 for serious, P2 for moderate, P3 for minor")
    description: str = Field(description="What the violation means")
    help_url: str = Field(description="Link to the axe-core rule documentation")
    wcag_tags: list[str] = Field(description="WCAG success criteria tags")
    affected_elements: int = Field(description="Number of elements affected")
    fix_suggestion: str = Field(description="Brief actionable fix suggestion")


class TriageReport(BaseModel):
    url: str = Field(description="The scanned URL")
    summary: str = Field(description="1-2 sentence accessibility health summary")
    total_violations: int = Field(description="Total number of violations found")
    violations: list[TriagedViolation] = Field(description="Prioritized list of violations")

What does this code do?

Each Field description tells the LLM exactly what value to place in that slot. ADK passes these descriptions to the model as part of its output instructions.

TriagedViolation defines the shape of a single violation: its rule ID, impact level, priority, WCAG tags, and a fix suggestion.
TriageReport wraps a list of violations with a URL, summary, and total count. This is what fix-gen will receive as structured input.

Define the fix-gen data contract

Below the TriageReport class in schemas.py, add the fix-gen models:

class CodeFix(BaseModel):
    rule_id: str = Field(description="The axe-core rule ID this fix addresses")
    wcag_criterion: str = Field(description="The specific WCAG success criterion e.g. 1.1.1")
    original_html: str = Field(description="The problematic HTML snippet")
    fixed_html: str = Field(description="The corrected HTML snippet")
    explanation: str = Field(description="Why this fix resolves the violation")


class FixGenReport(BaseModel):
    url: str = Field(description="The URL these fixes apply to")
    fixes: list[CodeFix] = Field(description="List of generated code fixes")
    summary: str = Field(description="Brief summary of all fixes generated")

What does this code do?

CodeFix defines one fix: the rule it addresses, the WCAG criterion, the before/after HTML, and an explanation.
FixGenReport wraps a list of fixes with the target URL and a summary. This is what the pipeline will store in session state.

Save schemas.py.
Verify the module loads by running this command in your terminal:

python -c "from pipeline.schemas import TriageReport, FixGenReport; print('Schemas OK')"

You should see Schemas OK printed.

Seeing an ImportError?

Check that you're running the command from the accessagent-scanner/ directory (the parent of pipeline/).
Confirm your virtual environment is active. You should see (.venv) in your terminal prompt.
Check for typos in the class names and make sure the from pydantic import BaseModel, Field line is at the top of the file.

Help me debug my schemas.py import error.

✔️ Awesome, I've got everything!

Great. Confirm you've saved schemas.py and move on.

ⓧ I'd like to double check the full code

Your schemas.py should have these sections in order:

The from pydantic import BaseModel, Field import at the top.
TriagedViolation class with 8 fields (rule_id through fix_suggestion).
TriageReport class with 4 fields (url, summary, total_violations, violations).
CodeFix class with 5 fields (rule_id, wcag_criterion, original_html, fixed_html, explanation).
FixGenReport class with 3 fields (url, fixes, summary).

Scroll up and confirm each chunk is present in order.

Wire schemas into agent.py

The schemas exist, but your agents don't know about them yet. You need to import the models and attach them as each agent's output_schema.

Open pipeline/agent.py.
Find this import line:

from .tools import scan_url

Add the schema import directly above it, so the imports section looks like this:

from google.adk import Agent, Workflow

from .schemas import FixGenReport, TriageReport
from .tools import scan_url

What does this import do?

This pulls in the two top-level report models. You'll pass them to output_schema on each agent. ADK uses the model's fields and descriptions to instruct the LLM on the exact JSON shape to produce.

In the triage_agent definition, find this line:

    output_key="triage_report",

Add output_schema=TriageReport, on a new line directly above it, so the end of the triage agent looks like:

    output_schema=TriageReport,
    output_key="triage_report",
)

In the fix_gen_agent definition, find the first line of the instruction:

You receive a triage report of WCAG violations as input.

Replace that line with:

You receive a structured triage report of WCAG violations as JSON input.

Why update the instruction wording?

With output_schema on the triage agent, fix-gen now receives a JSON dict instead of free text. Updating the instruction to say "structured" and "JSON input" aligns the prompt with reality. This helps the model use field names like rule_id directly.

Still in the fix_gen_agent definition, find this line:

    output_key="fix_gen_report",

Add output_schema=FixGenReport, on a new line directly above it, so the end of the fix-gen agent looks like:

    output_schema=FixGenReport,
    output_key="fix_gen_report",
)

Save agent.py.

Seeing a red underline on the import?

Make sure the class names match exactly: FixGenReport and TriageReport (capital letters matter).
Confirm you saved schemas.py before editing agent.py. VS Code can't resolve imports from unsaved files.

Help me fix my import in agent.py.

Run the pipeline and compare output quality

Before you run this, think back to the fix-gen output from the last step. It had wrong rule IDs, invented HTML, and missing WCAG citations. Do you think adding schemas will fix all three problems?

Start the ADK web interface by running:

adk web .

In the top-left dropdown, select pipeline.
Type a URL of a site with known accessibility issues into the chat (for example, a news site or an older government page).
Wait for the pipeline to complete all three stages.
Click the Events tab on the right side panel.

Look at the triage_report in session state. It is now a structured dict with typed fields, not a wall of free text.

What should I see?

In the Events tab, the triage_report state entry shows a JSON object with url, summary, total_violations, and a violations array. Each violation has rule_id, priority, wcag_tags, and fix_suggestion.

The fix-gen output now references the exact rule_id values from the triage report, shows real HTML from the scan data, and cites specific WCAG success criteria. Compare this to the inconsistent output from Step 3.

Why is the output so much better?

Without output_schema, the triage agent output a types.Content object (unstructured text) to the next node. Fix-gen had to parse that text and guess at field boundaries.

With output_schema set, ADK calls model_dump() on the validated Pydantic model and passes the resulting dict to the next node. Fix-gen receives a Python dict with named keys it can reference directly.

Press Ctrl+C in your terminal to stop the ADK web server.

Fix-gen still producing inconsistent output?

Confirm both agents have output_schema set. If only triage has it, triage output is structured but fix-gen still outputs free text.
Check the Events tab for validation errors. If the LLM produces output that doesn't match the schema, ADK raises an error instead of passing bad data downstream.
Restart the server after saving both files. The ADK web server does not always hot-reload schema changes.

Help me debug my output_schema setup.

Your pipeline now has typed data contracts enforcing structure at every handoff. The triage agent produces validated JSON, and fix-gen consumes it deterministically. Next up, you'll extend this pipeline with conditional routing to skip expensive processing when a site is clean.

Secret mission

Add Conditional Routing

Your pipeline sends every scan to Gemini 2.5 Pro for fix generation, even when the site has zero violations. That is wasted money. Add a routing node that short-circuits the pipeline when there is nothing to fix.

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

This project uses Vertex AI API calls that are charged per token, but there are no ongoing costs when the pipeline is idle. Each triage call costs approximately $0.003 (Flash) and each fix-gen call costs approximately $0.05-0.10 (Pro).

Resources you used:

pipeline/ directory (4 Python files and .env created in this project).
GCP Project with Vertex AI API enabled (shared with Project 2, needed for Project 4).

Keep everything running

No action needed. Choose this if you plan to continue to Project 4 (Human-in-the-loop review).

Your pipeline/ directory stays in place, ready for the next project.
Your GCP project and Vertex AI API remain enabled. No charges accrue until you run the pipeline again.
Your triage_agent/ directory from Project 2 remains intact alongside the new pipeline/ directory.

Pause - I'll come back to this later

Shut down any running processes to free up your terminal. No ongoing charges apply when the pipeline is idle.

If adk web is still running, press Ctrl+C in the terminal to stop it.
Your pipeline/ directory and GCP project remain intact for when you return.

Delete - I don't want to use this again

Remove the project files created in this project. Your GCP project and triage_agent/ from Project 2 remain intact.

If adk web is still running, press Ctrl+C to stop it.
Remove the pipeline/ directory by running the appropriate command for your OS.

macOS

Make sure you are in the accessagent-scanner/ directory, then run:

rm -rf pipeline/

Windows

Make sure you are in the accessagent-scanner/ directory, then run:

rmdir /s /q pipeline

Confirm the directory is removed by running:

ls

You should see triage_agent/ still listed, but no pipeline/ directory.

What about my GCP project?

Your GCP project with Vertex AI API is shared across the AccessAgent series. If you are continuing to Project 4, keep it enabled. If you are done with the entire series, you can delete the project from the Google Cloud Console under IAM & Admin > Settings > Shut down.

That's a Wrap!

Nice work! You've just built a multi-agent pipeline that scans any website for accessibility violations, triages them by severity, and generates concrete HTML code fixes.

You've learned how to:

Built a 3-stage ADK Workflow orchestrating Scanner, Triage, and Fix-Gen agents through graph-based edges with sequence shorthand.
Defined Pydantic output_schema models (TriageReport, FixGenReport) that enforce typed data contracts between agents, eliminating free-text parsing failures.
Matched models to task complexity, using Gemini 2.5 Flash for cost-efficient triage classification and Gemini 2.5 Pro for high-quality code fix generation.
Secret Mission: Added conditional routing with Event(route=...) and dict-based edge maps to skip expensive fix generation on clean sites.

Ready to quiz yourself?