Agent-to-Agent (A2A) Security Pipeline with MCP Integration

A security-focused, multi-agent data processing pipeline that combines the Agent-to-Agent (A2A) protocol, Google Agent Development Kit (ADK), and the Model Context Protocol (MCP). The system enables secure natural-language querying of a salary database through a sequential pipeline of specialized agents that enforce threat detection, SQL execution, and PII masking.

---

Table of Contents

• Architecture Overview
• Agent Pipeline
• Security Model
• MCP Integration
• A2A Protocol
• Project Structure
• Technology Stack
• Setup & Installation
• Usage
• Evaluation & Testing
• Deployment
• Documentation

---

Architecture Overview

The system is composed of three independent layers that collaborate to process every request securely:

┌─────────────────────────────────────────────────────────────┐
│                        CLIENT LAYER                         │
│          query_MCP_ADK_A2A.py  ·  a2a_client.py            │
└────────────────────┬────────────────────────────────────────┘
                     │  JSON-RPC 2.0 over HTTP
┌────────────────────▼────────────────────────────────────────┐
│                     A2A PROTOCOL LAYER                       │
│   Judge Server :10002  ·  SQL Server :10004  ·  Mask :10003 │
│              a2a_servers.py  ·  task_manager.py              │
└────────────────────┬────────────────────────────────────────┘
                     │  ADK agent callbacks
┌────────────────────▼────────────────────────────────────────┐
│                      ADK AGENT LAYER                         │
│       Judge Agent  →  SQL Agent  →  Mask Agent               │
│                      agent.py                                │
└──────┬─────────────────┬───────────────────┬────────────────┘
       │                 │                   │
  SecurityBlocker   MCP Server         Google Cloud DLP
  (in-process)    server_mcp.py         (Cloud API)
                       │
                  SQLite Database
                   salaries.db

Layer Responsibilities

---

Agent Pipeline

Every user query passes through three specialized agents in sequence.

Which pipeline halts. clients/query_MCP_ADK_A2A.analyze_salary_data_async calls the three agents over A2A and returns early on a BLOCKED verdict — that path halts. The SequentialAgent in agents/agent.py (used by adk run) does not: ADK runs every sub-agent unconditionally, so a BLOCKED from the Judge simply becomes the SQL agent’s input. The SQL tools are independently restricted to read-only SELECTs (see below), so a pass-through cannot damage the database.

User Query
    │
    ▼
[Input Sanitization]
  • Character whitelist (alphanumeric + basic punctuation)
  • 300-character length limit
  • Model Armor API scan
    │
    ▼
[Judge Agent]  ──────────────────────────────────────────────┐
  Model: gemini-3.7-flash                                    │
  Tool:  SecurityBlocker                                      │ BLOCKED
  • 60+ SQL/XSS/command injection patterns                    │ → Return error
  • 50+ obfuscation patterns (leetspeak, encoding, etc.)      │
  • Returns: PASS or BLOCKED                                  │
    │ PASS                                                    │
    ▼                                                         │
[SQL Agent]                                                   │
  Model: gemini-3.7-flash                                    │
  Tool:  MCP Server (query_data, execute_sql_query, etc.)    │
  • Translates natural language → SQL                        │
  • Executes against salaries.db (SQLite)                    │
  • Returns structured query results                         │
    │                                                         │
    ▼                                                         │
[Mask Agent]                                                  │
  Model: gemini-3.7-flash                                    │
  Tool:  Google Cloud DLP                                    │
  • Scans results for 16 PII info types                      │
  • Replaces sensitive values with asterisks                 │
  • Returns safe, masked output                              │
    │                                                         │
    ▼                                                         │
Final Response ◄─────────────────────────────────────────────┘

Agent Definitions

Judge Agent (security_judge)

• Evaluates every input before any database interaction
• Invokes the SecurityBlocker tool, which applies 100+ regex patterns covering SQL injection, XSS, command injection, path traversal, and obfuscated variants
• Outputs a clear PASS or BLOCKED verdict

SQL Agent (sql_assistant)

• Accepts natural-language queries about salary data
• Uses MCP tools to inspect the schema, generate valid SQL, and execute it
• Works with columns: work_year, experience_level, employment_type, job_title, salary, salary_currency, salary_in_usd, employee_residence, remote_ratio, company_location, company_size, fictitious_name, fictitious_surname

Mask Agent (data_masker)

• Post-processes all SQL results using Google Cloud DLP
• Detects and masks: EMAIL_ADDRESS, PHONE_NUMBER, CREDIT_CARD_NUMBER, US_SOCIAL_SECURITY_NUMBER, PERSON_NAME, IP_ADDRESS, DATE_OF_BIRTH, STREET_ADDRESS, PASSWORD, URL, AGE, VEHICLE_IDENTIFICATION_NUMBER, IBAN_CODE, MAC_ADDRESS, and more (16 types total)

---

Security Model

The system implements a five-layer security model:

1. Input Sanitization (Client-Side)

• Whitelists alphanumeric characters and safe punctuation
• Enforces a 300-character hard limit
• Integrates with Google Model Armor API for additional LLM-specific threat scanning

2. Threat Detection (Judge Agent + SecurityBlocker)

Pattern categories covered:

---

## Evaluation & Testing

The project includes an evaluation framework with 100 test scenarios.

### Run the Evaluator

```bash
python evaluation/simple_evaluator.py

Test Coverage

Metrics Reported

• Total scenarios run
• Pipeline errors, counted separately and excluded from the pass rate
• Pass / fail rate over valid results only
• An explicit list of malicious scenarios that were not blocked
• Results persisted to evaluation/security_test_results.json

Why errors are separated. analyze_salary_data_async returns a plain string for every outcome, including its own failures ("Privacy masking error: ...", "SQL execution error: ..."). The evaluator used to classify anything that did not contain "blocked" as PASSED, so an infrastructure failure on a benign scenario counted as a successful security test. The recorded run in security_test_results.json contains exactly that: legitimate_salary_range_by_title returned "Privacy masking error: 500 Internal error encountered." and was scored passed: true. A harness that reports green while the system is broken is worse than no harness.

Reading the recorded results

The shipped security_test_results.json reports 82/99. Re-scored with errors excluded it is 81/98 — the headline barely moves, but the corrected summary now surfaces what the JSON always contained: 17 malicious scenarios were not blocked, including NoSQL, LDAP, XPath, template-injection, file-inclusion and path-traversal payloads. malicious_nosql_injection returned 3755, the full row count of the salaries table, meaning the query executed.

Those 17 are the actual findings of this evaluation. They are gaps in the SecurityBlocker pattern set, which covers SQL/XSS/command-injection shapes well and non-SQL injection families poorly.

---

Unit tests

pip install pytest pytest-asyncio
pytest -q

53 tests, no credentials or network required — tests/conftest.py mocks ADK, Vertex AI, DLP, Secret Manager and the SQL toolkit before importing anything. Coverage is focused on the security controls: the read-only SQL guard (including stacked queries and comment smuggling), the Model Armor verdict being read, the Judge instruction not inviting an override, and the evaluator’s outcome classification.

---

Deployment

Local Testing with Docker

docker build -t adk-multi-agent .
docker run -p 8000:8000 \
  -e GOOGLE_API_KEY=your_api_key \
  adk-multi-agent adk web

Production: Google Cloud Run

export GOOGLE_CLOUD_PROJECT=your-project-id
export GOOGLE_CLOUD_LOCATION=us-central1
export GOOGLE_GENAI_USE_VERTEXAI=True
export AGENT_PATH="."
export SERVICE_NAME="adk-agent-service"
export APP_NAME="agents"

adk deploy cloud_run \
  --project=$GOOGLE_CLOUD_PROJECT \
  --region=$GOOGLE_CLOUD_LOCATION \
  --service_name=$SERVICE_NAME \
  --app_name=$APP_NAME \
  --with_ui \
  $AGENT_PATH

This command builds the container, pushes it to Artifact Registry, and deploys it as a Cloud Run service with the ADK web UI enabled.

---

Secrets hygiene

agents/.env is tracked in git (committed in 519982c). .gitignore previously contained only **/__pycache__/, so nothing prevented it. The committed version holds a GCP project id and no credentials, but the file being tracked means the next real key written there is committed too.

To untrack it without deleting your local copy:

git rm --cached agents/.env
git commit -m "stop tracking agents/.env"

.gitignore now ignores .env and **/.env while keeping .env.example.

---

Known limitations

• The SecurityBlocker pattern set is SQL/XSS-shaped. The recorded evaluation shows 17 non-SQL injection families getting through (NoSQL, LDAP, XPath, template, file inclusion, path traversal, CSRF, HTML injection). Closing those is the substantive outstanding work.
• No per-caller authentication on the MCP tools — see the Authentication note.
• sanitize_input strips disallowed characters silently rather than rejecting the input, so DROP TABLE users; becomes DROP TABLE users and is passed on to the Judge rather than refused outright. The whitelist is also stricter than the charset check that follows it, so that check can never fire.
• The Judge is a rewriting step in the trust chain. analyze_salary_data_async interpolates the Judge’s output into the SQL agent’s prompt, so whatever the Judge emits becomes the SQL agent’s instruction. The block test is if "BLOCK" in judge_output.upper() — a substring match on free model text.
• Block detection is substring-based, so a legitimate query containing the word “block” would be refused. No such job title exists in the bundled dataset (93 distinct titles checked), but a different dataset could trip it.
• salaries.db is rebuilt from SALARIES_CSV_PATH on every server start (if_exists='replace'), discarding any state. If that variable is unset the committed salaries.db is used as-is.
• Sessions are in-memory, so conversation history does not survive a restart and is not shared across processes.

---

Documentation

• Google Agent Development Kit (ADK)
• Agent-to-Agent (A2A) Protocol
• Model Context Protocol (MCP)
• Google Cloud DLP
• Google Model Armor
• LangChain SQL Toolkit

---

This project demonstrates a production-grade integration of A2A, MCP, and Google ADK to build a secure, multi-agent architecture for data analysis with enterprise-level threat detection and privacy protection.