feat: support a2a agent template

- Add new 'a2a' template with A2A protocol support and sample implementation
- Add messageId field to A2A payload construction using UUID
- Move header processing before payload construction in invoke command
- Add support for parsing agent role from status.message.parts in streaming responses
- Remove redundant "Stream marked final" console output
- Auto-detect invoke path based on agent_type: A2A agents use '/', others use '/invoke'
-

Author: innsd

agentkit/apps/simple_app/simple_app_handlers.py

@@ -133,7 +133,10 @@ def _convert_to_sse(self, obj) -> bytes:
         Returns:
             bytes: SSE-formatted data ready for streaming
         """
-        json_string = safe_serialize_to_json_string(obj)
+        if isinstance(obj, str):
+            json_string = obj
+        else:
+            json_string = safe_serialize_to_json_string(obj)
         sse_data = f"data: {json_string}\n\n"
         return sse_data.encode("utf-8")
 

agentkit/toolkit/cli/cli_invoke.py

@@ -20,6 +20,7 @@
 import typer
 from rich.console import Console
 import random
+import uuid
 from agentkit.toolkit.config import get_config
 import logging
 
@@ -64,6 +65,7 @@ def build_a2a_payload(message: Optional[str], payload: Optional[str], headers: d
         "params": {
             "message": {
                 "role": "user",
+                "messageId": str(uuid.uuid4()),
                 "parts": [
                     {"kind": "text", "text": text}
                 ]
@@ -117,15 +119,6 @@ def invoke_command(
     config = get_config(config_path=config_file)
     common_config = config.get_common_config()
 
-    final_payload = build_standard_payload(message, payload)
-    agent_type = getattr(common_config, "agent_type", "") or getattr(common_config, "template_type", "")
-    is_a2a = isinstance(agent_type, str) and "a2a" in agent_type.lower()
-
-    # If it's an A2A Agent, reconstruct payload using A2A constructor
-    if is_a2a:
-        console.print("[cyan]Detected A2A agent type - constructing A2A JSON-RPC envelope[/cyan]")
-        final_payload = build_a2a_payload(message, payload, final_headers)
-
     # Process headers
     final_headers = {"user_id": "agentkit_user", "session_id": "agentkit_sample_session"}
     if headers:
@@ -137,6 +130,16 @@ def invoke_command(
             raise typer.Exit(1)
     else:
         console.print(f"[blue]Using default headers: {final_headers}[/blue]")
+
+    final_payload = build_standard_payload(message, payload)
+    agent_type = getattr(common_config, "agent_type", "") or getattr(common_config, "template_type", "")
+    is_a2a = isinstance(agent_type, str) and "a2a" in agent_type.lower()
+
+    # If it's an A2A Agent, reconstruct payload using A2A constructor
+    if is_a2a:
+        console.print("[cyan]Detected A2A agent type - constructing A2A JSON-RPC envelope[/cyan]")
+        final_payload = build_a2a_payload(message, payload, final_headers)
+
 
     # Set execution context - CLI uses ConsoleReporter (with colored output and progress)
     from agentkit.toolkit.context import ExecutionContext
@@ -194,6 +197,10 @@ def invoke_command(
                     parts = event["message"].get("parts", [])
                 elif isinstance(event.get("content"), dict):
                     parts = event["content"].get("parts", [])
+                elif isinstance(event.get("status"), dict):
+                    role = event["status"].get("message", {}).get("role")
+                    if role == "agent":
+                        parts = event["status"].get("message", {}).get("parts", [])
                 if not event.get("partial", True):
                     logger.info("Partial event: %s", event)  # Log partial events
                     continue
@@ -216,7 +223,6 @@ def invoke_command(
 
                 # Handle status updates (e.g., final flag or completed status)
                 if event.get("final") is True:
-                    console.print("\n[cyan]Stream marked final[/cyan]")
                     break
 
                 status = event.get("status")

agentkit/toolkit/executors/init_executor.py

@@ -56,6 +56,15 @@
         "type": "Stream App",
         "extra_requirements": ["# google-adk"],
     },
+    "a2a": {
+        "file": "a2a.py",
+        "name": "A2A Agent App",
+        "language": "Python",
+        "language_version": "3.12",
+        "description": "支持A2A协议的Agent应用",
+        "type": "A2A App",
+        "extra_requirements": ["# google-adk"],
+    },
     # "eino_a2a": {
     #     "filepath": "eino_a2a",
     #     "name": "Eino A2A Agent App",

agentkit/toolkit/resources/samples/a2a.py

@@ -0,0 +1,86 @@
+# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+
+from veadk import Agent, Runner
+from veadk.a2a.agent_card import get_agent_card
+from veadk.prompts.agent_default_prompt import DEFAULT_DESCRIPTION, DEFAULT_INSTRUCTION
+from google.adk.a2a.executor.a2a_agent_executor import A2aAgentExecutor
+from agentkit.apps import AgentkitA2aApp
+
+logger = logging.getLogger(__name__)
+
+a2a_app = AgentkitA2aApp()
+
+agent_name = "{{ agent_name | default('Agent') }}"
+{% if description %}description = "{{ description }}" {% else %}description = DEFAULT_DESCRIPTION {% endif %}
+{% if system_prompt %}system_prompt = "{{ system_prompt }}" {% else %}system_prompt = DEFAULT_INSTRUCTION {% endif %}
+{% if model_name %}model_name = "{{ model_name }}"{% endif %}
+
+tools = []
+{% if tools %}
+{% if 'web_search' in tools %}
+from veadk.tools.builtin_tools.web_search import web_search
+tools.append(web_search)
+{% endif %}
+{% if 'run_code' in tools %}
+from veadk.tools.builtin_tools.run_code import run_code
+tools.append(run_code)
+{% endif %}
+{% if 'get_weather' in tools %}
+# from veadk.tools.builtin_tools.get_weather import get_weather
+# tools.append(get_weather)
+{% endif %}
+{% else %}
+# from veadk.tools.builtin_tools.web_search import web_search
+# tools.append(web_search)
+{% endif %}
+
+agent = Agent(
+    name=agent_name,
+    description=description,
+    instruction=system_prompt,
+{%- if model_name %}
+    model_name=model_name,
+{%- endif %}
+    tools=tools,
+)
+runner = Runner(agent=agent)
+
+@a2a_app.agent_executor(runner=runner)
+class MyAgentExecutor(A2aAgentExecutor):
+    pass
+
+
+if __name__ == "__main__":
+    from a2a.types import AgentCard, AgentProvider, AgentSkill, AgentCapabilities
+    
+    agent_card = AgentCard(
+        capabilities=AgentCapabilities(streaming=True),  # 启用流式
+        description=agent.description,
+        name=agent.name,
+        defaultInputModes=["text"],
+        defaultOutputModes=["text"],
+        provider=AgentProvider(organization="veadk", url=""),
+        skills=[AgentSkill(id="0", name="chat", description="Chat", tags=["chat"])],
+        url="http://0.0.0.0:8000",
+        version="1.0.0",
+    )
+    
+    a2a_app.run(
+        agent_card=agent_card,
+        host="0.0.0.0",
+        port=8000,
+    )

agentkit/toolkit/resources/samples/basic.py

@@ -12,15 +12,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-'''
-**simple agent demo app**
-
-Before running, the user must set the following environment variables; otherwise, runtime exceptions will inevitably occur:
-- MODEL_AGENT_NAME  # model id in Volcano Engine Ark platform
-- MODEL_AGENT_API_KEY  # model api key in Volcano Engine Ark platform
-
-MODEL_AGENT_NAME and MODEL_AGENT_API_KEY are used to access the model service of the Volcano Engine Ark platform.
-'''
 import logging
 
 from veadk import Agent, Runner

agentkit/toolkit/resources/samples/basic_stream.py

@@ -99,12 +99,12 @@ async def run(payload: dict, headers: dict):
             # Format as SSE data
             sse_event = event.model_dump_json(exclude_none=True, by_alias=True)
             logger.debug("Generated event in agent run streaming: %s", sse_event)
-            yield f"data: {sse_event}\n\n"
+            yield sse_event
     except Exception as e:
         logger.exception("Error in event_generator: %s", e)
         # You might want to yield an error event here
         error_data = json.dumps({"error": str(e)})
-        yield f'data: {error_data}\n\n'
+        yield error_data
 
 
 @app.ping

agentkit/toolkit/resources/samples/simple_a2a_veadk.py

@@ -530,9 +530,14 @@ def invoke(self, config: LocalDockerRunnerConfig, payload: Dict[str, Any], heade
                 )
 
             # Build invocation endpoint
+            # Auto-detect invoke path based on agent_type: A2A agents use '/', others use '/invoke'
             port = docker_config.invoke_port or 8000
             endpoint = f"http://127.0.0.1:{port}/"
-            invoke_endpoint = urljoin(endpoint, "invoke")
+            
+            agent_type = getattr(common_config, 'agent_type', '') or ''
+            is_a2a = 'a2a' in agent_type.lower()
+            invoke_path = "/" if is_a2a else "/invoke"
+            invoke_endpoint = urljoin(endpoint, invoke_path.lstrip('/')) if invoke_path != "/" else endpoint
 
             # Prepare default request headers
             if headers is None:

agentkit/toolkit/runners/local_docker.py

@@ -308,7 +308,12 @@ def invoke(self, config: VeAgentkitRunnerConfig, payload: Dict[str, Any], header
                     )
 
             # Construct invoke endpoint URL
-            invoke_endpoint = urljoin(endpoint, "invoke")
+            # Auto-detect invoke path based on agent_type: A2A agents use '/', others use '/invoke'
+            common_config = runner_config.common_config
+            agent_type = getattr(common_config, 'agent_type', '') or '' if common_config else ''
+            is_a2a = 'a2a' in agent_type.lower()
+            invoke_path = "/" if is_a2a else "/invoke"
+            invoke_endpoint = urljoin(endpoint, invoke_path.lstrip('/')) if invoke_path != "/" else endpoint
 
             # Prepare request headers
             if headers is None: