This project implements a security-focused data processing pipeline using the Agent-to-Agent (A2A) protocol combined with Model Context Protocol (MCP) integration. The system provides secure database queries through a multi-agent architecture.
The system utilizes a layered architecture:
- A2A Protocol Layer: Provides standardized communication between clients and agent services
- ADK Framework Layer: Manages agent behavior and tool integration
- MCP Server Layer: Provides specialized SQL and data processing tools
- A2A Servers: Handle client requests and agent communication
- ADK Agents: Process natural language requests using specialized tools
- MCP Server: Provides SQL query and database interaction tools
- Task Managers: Coordinate task execution across agents
The system implements a security pipeline with three specialized agents:
- Judge Agent: Evaluates input for security threats (SQL injection, XSS, etc.)
- SQL Agent: Performs database queries and analysis using MCP tools
- Mask Agent: Applies privacy protection to sensitive data in results
Client Request → A2A Server → Judge Agent → SQL Agent → Mask Agent → Client Response
- Security Threat Detection: Identifies and blocks malicious inputs via tool and Model Armor
- SQL Query Analysis: Processes database queries using natural language
- PII Data Protection: Masks personally identifiable information in results using DLP
- A2A Protocol Compliance: Implements standardized agent communication
- MCP Integration: Leverages Model Context Protocol tools for enhanced capabilities
- Python 3.8+
- aiohttp
- FastAPI
- Google ADK
- Google Generative AI packages
- uvicorn
- Clone the repository
- Install dependencies:
pip install -r requirements.txt - Configure API keys in environment variables (even better, use secret manager)
- Have fun
Run the adk web script to run the chat interface:
adk webThis will start:
- Judge Server (port 10002)
- Mask Server (port 10003)
- SQL Server (port 10004)
- MCP Server
Run the query_MCP_ADK_A2A.py script to query the multi-agent system:
python ./clients/query_MCP_ADK_A2A.pyThis will use the a2a_client.py module to make requests to the pipeline:
a2a_client.py: Client for A2A communicationa2a_servers.py: Server implementations for A2A protocolquery_MCP_ADK_A2A.py: Main pipeline implementationrun_servers.py: Server startup and coordinationserver_mcp.py: MCP server implementationtask_manager.py: Task coordination for agent communicationmcp_agent.py: Integration between ADK and MCP
The system integrates with Model Context Protocol (MCP) for enhanced SQL capabilities:
# Connect to MCP server
tools, exit_stack = await MCPToolset.from_server(
connection_params=StdioServerParameters(
command='python',
args=["server_mcp.py"],
)
)
# Create ADK agent with MCP tools
agent = LlmAgent(
model='gemini-2.5-pro-preview-03-25',
name='sql_assistant',
instruction="...",
tools=tools,
)- Pattern-based security threat detection
- PII identification and masking (emails, names, addresses, etc.)
- Input sanitation with whitelist approach
- Model Armor API integration for additional protection
Testing:
docker build -t adk-multi-agent .
docker run -p 8000:8000 -e GOOGLE_API_KEY=your_api_key adk-multi-agent adk web
Production:
export GOOGLE_CLOUD_PROJECT=next-project25
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
Agent Development Kit Documentation
- Fork the repository
- Create a feature branch
- Submit a pull request with comprehensive description
This project demonstrates integration between A2A protocol and MCP server capabilities, creating a secure and flexible agent architecture for data processing.