Deploy and orchestrate Vertex AI ADK agents using A2A protocol. Manages AgentCard discovery, task submission, Code Execution Sandbox, and Memory Bank. Use when asked to "deploy ADK agent" or "orchestrate agents".
Limited to specific tools
Additional assets for this skill
This skill is limited to using the following tools:
scripts/deploy-agent.shscripts/test-a2a-protocol.pyExpert in building and deploying production multi-agent systems using Google's Agent Development Kit (ADK). Handles agent orchestration (Sequential, Parallel, Loop), A2A protocol communication, Code Execution Sandbox for GCP operations, Memory Bank for stateful conversations, and deployment to Vertex AI Agent Engine.
adk deployUser Request → Analyze:
- Single agent vs multi-agent system?
- Tools needed (Code Exec, Memory Bank, custom tools)?
- Orchestration pattern (Sequential, Parallel, Loop)?
- Integration with LangChain/Genkit?
- Deployment target (local, Agent Engine, Cloud Run)?
Simple Agent (Python):
from google import adk
# Define agent with tools
agent = adk.Agent(
model="gemini-2.5-flash",
tools=[
adk.tools.CodeExecution(), # Secure sandbox
adk.tools.MemoryBank(), # Persistent memory
],
system_instruction="""
You are a GCP deployment specialist.
Help users deploy resources securely using gcloud commands.
"""
)
# Run agent
response = agent.run("Deploy a GKE cluster named prod in us-central1")
print(response)
Multi-Agent Orchestrator (Python):
from google import adk
# Define specialized sub-agents
validator_agent = adk.Agent(
model="gemini-2.5-flash",
system_instruction="Validate GCP configurations"
)
deployer_agent = adk.Agent(
model="gemini-2.5-flash",
tools=[adk.tools.CodeExecution()],
system_instruction="Deploy validated GCP resources"
)
monitor_agent = adk.Agent(
model="gemini-2.5-flash",
system_instruction="Monitor deployment status"
)
# Orchestrate with Sequential pattern
orchestrator = adk.SequentialAgent(
agents=[validator_agent, deployer_agent, monitor_agent],
system_instruction="Coordinate validation → deployment → monitoring"
)
result = orchestrator.run("Deploy a production GKE cluster")
The Code Execution Sandbox provides:
# Agent with Code Execution
agent = adk.Agent(
model="gemini-2.5-flash",
tools=[adk.tools.CodeExecution()],
system_instruction="""
Execute gcloud commands in the secure sandbox.
Remember previous operations in this session.
"""
)
# Turn 1: Create cluster
agent.run("Create GKE cluster named dev-cluster with 3 nodes")
# Sandbox executes: gcloud container clusters create dev-cluster --num-nodes=3
# Turn 2: Deploy to cluster (remembers cluster from Turn 1)
agent.run("Deploy my-app:latest to that cluster")
# Sandbox remembers dev-cluster, executes kubectl commands
Persistent conversation memory across sessions:
agent = adk.Agent(
model="gemini-2.5-flash",
tools=[adk.tools.MemoryBank()],
system_instruction="Remember user preferences and project context"
)
# Session 1 (Monday)
agent.run("I prefer deploying to us-central1 region", session_id="user-123")
# Session 2 (Wednesday) - same session_id
agent.run("Deploy a Cloud Run service", session_id="user-123")
# Agent remembers: uses us-central1 automatically
Deploy agent to Agent Engine with A2A endpoint:
# Install ADK
pip install google-adk
# Deploy with one command
adk deploy \
--agent-file agent.py \
--project-id my-project \
--region us-central1 \
--service-name gcp-deployer-agent
Agent Engine creates:
https://gcp-deployer-agent-{hash}.run.app/.well-known/agent-card metadata/v1/tasks:send for task submission/v1/tasks/{task_id} for pollingOnce deployed, Claude can invoke via A2A protocol:
# In Claude Code plugin / external script
import requests
def invoke_adk_agent(message, session_id=None):
"""
Call deployed ADK agent via A2A protocol.
"""
response = requests.post(
"https://gcp-deployer-agent-xyz.run.app/v1/tasks:send",
json={
"message": message,
"session_id": session_id or "claude-session-123",
"config": {
"enable_code_execution": True,
"enable_memory_bank": True,
}
},
headers={"Authorization": f"Bearer {get_token()}"}
)
return response.json()
# Use from Claude
result = invoke_adk_agent("Deploy GKE cluster named prod-api")
User: "Create an ADK agent that deploys GCP resources"
Implementation:
from google import adk
deployment_agent = adk.Agent(
model="gemini-2.5-flash",
tools=[
adk.tools.CodeExecution(),
adk.tools.MemoryBank(),
],
system_instruction="""
You are a GCP deployment specialist.
CAPABILITIES:
- Deploy GKE clusters
- Deploy Cloud Run services
- Deploy Vertex AI Pipelines
- Manage IAM permissions
- Monitor deployments
SECURITY:
- Validate all configurations before deployment
- Use least-privilege IAM
- Log all operations
- Never expose credentials
"""
)
# Deploy to Agent Engine
# $ adk deploy --agent-file deployment_agent.py --service-name gcp-deployer
User: "Build a RAG system with ADK orchestrating a LangChain retriever"
Implementation:
from google import adk
from langchain.retrievers import VertexAISearchRetriever
# Sub-Agent 1: LangChain RAG
class RAGAgent(adk.Agent):
def __init__(self):
self.retriever = VertexAISearchRetriever(...)
super().__init__(model="gemini-2.5-flash")
def retrieve_docs(self, query):
return self.retriever.get_relevant_documents(query)
# Sub-Agent 2: ADK Answer Generator
answer_agent = adk.Agent(
model="gemini-2.5-pro", # More powerful for final answer
system_instruction="Generate comprehensive answers from retrieved docs"
)
# Orchestrator
orchestrator = adk.SequentialAgent(
agents=[RAGAgent(), answer_agent],
system_instruction="First retrieve docs, then generate answer"
)
User: "Deploy a GKE cluster and monitor progress"
Implementation:
# Submit async task
task_response = invoke_adk_agent(
"Deploy GKE cluster named prod-api with 5 nodes in us-central1"
)
task_id = task_response["task_id"]
print(f"✅ Task submitted: {task_id}")
# Poll for status
import time
while True:
status = requests.get(
f"https://gcp-deployer-agent-xyz.run.app/v1/tasks/{task_id}",
headers={"Authorization": f"Bearer {get_token()}"}
).json()
if status["status"] == "SUCCESS":
print(f"✅ Cluster deployed!")
break
elif status["status"] == "FAILURE":
print(f"❌ Deployment failed: {status['error']}")
break
else:
print(f"⏳ Status: {status['status']} ({status.get('progress', 0)*100}%)")
time.sleep(10)
# Use Genkit for flows, ADK for orchestration
genkit_flow_agent = create_genkit_flow()
orchestrator = adk.SequentialAgent(
agents=[validator, genkit_flow_agent, monitor]
)
# LangChain for RAG, ADK for multi-agent coordination
langchain_rag = create_langchain_retriever()
orchestrator = adk.ParallelAgent(
agents=[langchain_rag, fact_checker, answer_generator]
)
# Install ADK
pip install google-adk # Python
go get google.golang.org/adk # Go
# Deploy to Agent Engine
adk deploy \
--agent-file my_agent.py \
--project-id my-project \
--region us-central1 \
--service-name my-agent
# Deploy to Cloud Run (custom)
gcloud run deploy my-agent \
--source . \
--region us-central1
# Deploy locally for testing
adk run --agent-file my_agent.py