adk_a2a_developer

Name: adk_a2a_developer
Author: VeerMuchandi

15stars1forks

How to install adk_a2a_developer?

veermuchandi/rad-skills/adk_a2a_developer

$npx -y skills add VeerMuchandi/rad-skills --skill adk_a2a_developer

Installs into the current project.

Summary

Expert guide for making ADK agents A2A compatible and testing them on Agent Engine.

Files · 1

View on GitHub

SKILL.md

1	# ADK A2A Developer Skill
2
3	This skill equips you with the knowledge to take a standard ADK agent, make it A2A (Agent-to-Agent) compatible, and deploy it to Vertex AI Agent Engine.
4
5	## A2A Protocol
6
7	A2A (Agent-to-Agent) is a protocol for communication between agents or between a client and an agent using structured messages (usually JSON-RPC). It handles message routing, sessions, and tasks. This guide focuses on deploying agents that exchange text messages and structured data via A2A.
8
9	## Step 1: Create a Custom Agent Executor
10
11	To make an ADK agent A2A compatible on Vertex AI Agent Engine, you need to override the default execution loop with a custom `AgentExecutor`.
12
13	Create a file named `agent_executor.py` in your agent directory.
14
15	Here is a template for an A2A executor:
16
17	```python
18	"""Agent executor for ADK agents adapted for A2A compatibility."""
19
20	import logging
21	from a2a import types
22	from a2a import utils
23	from a2a.server import agent_execution
24	from a2a.server import events
25	from a2a.server import tasks
26	from a2a.utils import errors as a2a_errors
27
28	import agent # Import your ADK agent
29	from google.adk import runners
30	from google.adk.artifacts import in_memory_artifact_service
31	from google.adk.memory import in_memory_memory_service
32	from google.adk.sessions import in_memory_session_service
33	from google.genai import types as genai_types
34
35	logger = logging.getLogger(__name__)
36
37	class AdkAgentToA2AExecutor(agent_execution.AgentExecutor):
38	"""An agent executor for ADK agents to make them A2A compatible."""
39
40	_runner: runners.Runner
41
42	def __init__(self):
43	self._agent = agent.root_agent
44	self._runner = runners.Runner(
45	app_name=self._agent.name,
46	agent=self._agent,
47	session_service=in_memory_session_service.InMemorySessionService(),
48	artifact_service=in_memory_artifact_service.InMemoryArtifactService(),
49	memory_service=in_memory_memory_service.InMemoryMemoryService(),
50	)
51	self._user_id = "remote_agent"
52
53	async def execute(
54	self,
55	context: agent_execution.RequestContext,
56	event_queue: events.EventQueue,
57	) -> None:
58	query = context.get_user_input()
59	task = context.current_task
60	logger.info("[DEBUG] Query: %s", query)
61
62	if not task:
63	if not context.message:
64	return
65
66	task = utils.new_task(context.message)
67	await event_queue.enqueue_event(task)
68
69	updater = tasks.TaskUpdater(event_queue, task.id, task.context_id)
70	session_id = task.context_id
71
72	session = await self._runner.session_service.get_session(
73	app_name=self._agent.name,
74	user_id=self._user_id,
75	session_id=session_id,
76	)
77	if session is None:
78	session = await self._runner.session_service.create_session(
79	app_name=self._agent.name,
80	user_id=self._user_id,
81	state={},
82	session_id=session_id,
83	)
84
85	await updater.start_work()
86
87	content = genai_types.Content(
88	role="user", parts=[{"text": query}]
89	)
90
91	final_response_content = ""
92
93	try:
94	async for event in self._runner.run_async(
95	user_id=self._user_id, session_id=session.id, new_message=content
96	):
97	if event.is_final_response():
98	if (
99	event.content
100	and event.content.parts
101	and event.content.parts[0].text
102	):
103	final_response_content += event.content.parts[0].text
104
105	except Exception as e:
106	await updater.failed(
107	message=utils.new_agent_text_message(
108	f"Task failed with error: {str(e)}"
109	)
110	)
111	return
112
113	if not final_response_content:
114	await updater.failed(
115	message=utils.new_agent_text_message("No response generated.")
116	)
117	return
118
119	# Yield the text response as a standard A2A TextPart
120	parts = [types.Part(root=types.TextPart(text=final_response_content))]
121	await updater.add_artifact(parts, name="response")
122	await updater.complete()
123
124	async def cancel(
125	self,
126	context: agent_execution.RequestContext,
127	event_queue: events.EventQueue,
128	) -> None:
129	raise a2a_errors.ServerError(error=types.UnsupportedOperationError())
130	```
131
132	## Step 2: Deploy Using Python SDK
133
134	Use the `vertexai.Client` to deploy your agent using the `A2aAgent` template.
135
136	Create a file named `deploy_ae.py` in your agent directory:
137
138	```python
139	import os
140	import vertexai
141	from vertexai.preview.reasoning_engines import A2aAgent
142	from vertexai.preview.reasoning_engines.templates.a2a import create_agent_card
143	from a2a.types import AgentSkill
144	from dotenv import load_dotenv
145	from google.genai import types
146
147	import agent_executor
148
149	def deploy():
150	load_dotenv()
151
152	project_id = os.environ.get("PROJECT_ID")
153	location = os.environ.get("LOCATION") or "us-central1"
154	storage = os.environ.get("STORAGE_BUCKET")
155
156	vertexai.init(
157	pr

People who liked this also viewed

6 picks

Type

hyperframes-cliUse the HyperFrames CLI development loop: init, add, catalog, capture, lint, check, snapshot, compare, grade-compare, preview, play, present, beats, keyframes, single or batch render, publish, cloud, SkillsJul 2026262k37k
request-refactor-planSkills for Real Engineers. Straight from my .agents directory.SkillsJul 2026165k181k
analyze-projectRigor Analyze / Rigor Audit read-only skill for deep learning research repositories. Use when the user wants to read and understand a repository, inspect model structure and training or inference entrSkillsMay 2026161k507
explore-codeRigor Improve implementation leaf skill for auditable candidate implementation in deep learning research repositories. Use when the researcher explicitly authorizes exploratory work on an isolated braSkillsMay 2026161k507
ubiquitous-languageSkills for Real Engineers. Straight from my .agents directory.SkillsJul 2026160k181k
scaffold-exercisesSkills for Real Engineers. Straight from my .agents directory.SkillsJul 2026149k181k

adk_a2a_developer

15stars1forks

View on GitHub

How to install adk_a2a_developer?

veermuchandi/rad-skills/adk_a2a_developer

$npx -y skills add VeerMuchandi/rad-skills --skill adk_a2a_developer

Installs into the current project.

Summary

Expert guide for making ADK agents A2A compatible and testing them on Agent Engine.

Files · 1

View on GitHub

SKILL.md

1	# ADK A2A Developer Skill
2
3	This skill equips you with the knowledge to take a standard ADK agent, make it A2A (Agent-to-Agent) compatible, and deploy it to Vertex AI Agent Engine.
4
5	## A2A Protocol
6
7	A2A (Agent-to-Agent) is a protocol for communication between agents or between a client and an agent using structured messages (usually JSON-RPC). It handles message routing, sessions, and tasks. This guide focuses on deploying agents that exchange text messages and structured data via A2A.
8
9	## Step 1: Create a Custom Agent Executor
10
11	To make an ADK agent A2A compatible on Vertex AI Agent Engine, you need to override the default execution loop with a custom `AgentExecutor`.
12
13	Create a file named `agent_executor.py` in your agent directory.
14
15	Here is a template for an A2A executor:
16
17	```python
18	"""Agent executor for ADK agents adapted for A2A compatibility."""
19
20	import logging
21	from a2a import types
22	from a2a import utils
23	from a2a.server import agent_execution
24	from a2a.server import events
25	from a2a.server import tasks
26	from a2a.utils import errors as a2a_errors
27
28	import agent # Import your ADK agent
29	from google.adk import runners
30	from google.adk.artifacts import in_memory_artifact_service
31	from google.adk.memory import in_memory_memory_service
32	from google.adk.sessions import in_memory_session_service
33	from google.genai import types as genai_types
34
35	logger = logging.getLogger(__name__)
36
37	class AdkAgentToA2AExecutor(agent_execution.AgentExecutor):
38	"""An agent executor for ADK agents to make them A2A compatible."""
39
40	_runner: runners.Runner
41
42	def __init__(self):
43	self._agent = agent.root_agent
44	self._runner = runners.Runner(
45	app_name=self._agent.name,
46	agent=self._agent,
47	session_service=in_memory_session_service.InMemorySessionService(),
48	artifact_service=in_memory_artifact_service.InMemoryArtifactService(),
49	memory_service=in_memory_memory_service.InMemoryMemoryService(),
50	)
51	self._user_id = "remote_agent"
52
53	async def execute(
54	self,
55	context: agent_execution.RequestContext,
56	event_queue: events.EventQueue,
57	) -> None:
58	query = context.get_user_input()
59	task = context.current_task
60	logger.info("[DEBUG] Query: %s", query)
61
62	if not task:
63	if not context.message:
64	return
65
66	task = utils.new_task(context.message)
67	await event_queue.enqueue_event(task)
68
69	updater = tasks.TaskUpdater(event_queue, task.id, task.context_id)
70	session_id = task.context_id
71
72	session = await self._runner.session_service.get_session(
73	app_name=self._agent.name,
74	user_id=self._user_id,
75	session_id=session_id,
76	)
77	if session is None:
78	session = await self._runner.session_service.create_session(
79	app_name=self._agent.name,
80	user_id=self._user_id,
81	state={},
82	session_id=session_id,
83	)
84
85	await updater.start_work()
86
87	content = genai_types.Content(
88	role="user", parts=[{"text": query}]
89	)
90
91	final_response_content = ""
92
93	try:
94	async for event in self._runner.run_async(
95	user_id=self._user_id, session_id=session.id, new_message=content
96	):
97	if event.is_final_response():
98	if (
99	event.content
100	and event.content.parts
101	and event.content.parts[0].text
102	):
103	final_response_content += event.content.parts[0].text
104
105	except Exception as e:
106	await updater.failed(
107	message=utils.new_agent_text_message(
108	f"Task failed with error: {str(e)}"
109	)
110	)
111	return
112
113	if not final_response_content:
114	await updater.failed(
115	message=utils.new_agent_text_message("No response generated.")
116	)
117	return
118
119	# Yield the text response as a standard A2A TextPart
120	parts = [types.Part(root=types.TextPart(text=final_response_content))]
121	await updater.add_artifact(parts, name="response")
122	await updater.complete()
123
124	async def cancel(
125	self,
126	context: agent_execution.RequestContext,
127	event_queue: events.EventQueue,
128	) -> None:
129	raise a2a_errors.ServerError(error=types.UnsupportedOperationError())
130	```
131
132	## Step 2: Deploy Using Python SDK
133
134	Use the `vertexai.Client` to deploy your agent using the `A2aAgent` template.
135
136	Create a file named `deploy_ae.py` in your agent directory:
137
138	```python
139	import os
140	import vertexai
141	from vertexai.preview.reasoning_engines import A2aAgent
142	from vertexai.preview.reasoning_engines.templates.a2a import create_agent_card
143	from a2a.types import AgentSkill
144	from dotenv import load_dotenv
145	from google.genai import types
146
147	import agent_executor
148
149	def deploy():
150	load_dotenv()
151
152	project_id = os.environ.get("PROJECT_ID")
153	location = os.environ.get("LOCATION") or "us-central1"
154	storage = os.environ.get("STORAGE_BUCKET")
155
156	vertexai.init(
157	pr

People who liked this also viewed

6 picks

Type