feat: add ping support for a2a app

(cherry picked from commit dec0614b8faa9f3381070defb172af30f9176c89)

Author: innsd

agentkit/apps/a2a_app/a2a_app.py

@@ -14,9 +14,10 @@
 
 import logging
 import os
-from typing import Callable, override
-
 import uvicorn
+import inspect
+
+from typing import Callable, override
 from a2a.server.agent_execution import AgentExecutor
 from a2a.server.agent_execution.context import RequestContext
 from a2a.server.apps import A2AStarletteApplication
@@ -25,8 +26,9 @@
 from a2a.server.tasks import InMemoryTaskStore
 from a2a.server.tasks.task_store import TaskStore
 from a2a.types import AgentCard
+from starlette.requests import Request
 from starlette.applications import Starlette
-from starlette.responses import JSONResponse
+from starlette.responses import JSONResponse, Response
 from starlette.routing import Route
 
 from agentkit.apps.a2a_app.telemetry import telemetry
@@ -75,6 +77,7 @@ def __init__(self) -> None:
 
         self._agent_executor: AgentExecutor | None = None
         self._task_store: TaskStore | None = None
+        self._ping_func: Callable | None = None
 
     def agent_executor(self, **kwargs) -> Callable:
         """Wrap an AgentExecutor class, init it, then bind it to the app instance."""
@@ -136,6 +139,30 @@ def is_agentkit_runtime() -> bool:
         )
         app.routes.append(route)
 
+    def ping(self, func: Callable) -> Callable:
+        """Register a zero-argument health check function and expose it via GET /ping.
+
+        The function must accept no arguments and should return either a string or a dict.
+        The response shape mirrors SimpleApp: {"status": <str|dict>}.
+        """
+        # Ensure zero-argument function similar to SimpleApp
+        if len(list(inspect.signature(func).parameters.keys())) != 0:
+            raise AssertionError(
+                f"Health check function `{func.__name__}` should not receive any arguments."
+            )
+
+        self._ping_func = func
+        return func
+
+    def _format_ping_status(self, result: str | dict) -> dict:
+        # Align behavior with SimpleApp: always wrap into {"status": result}
+        if isinstance(result, (str, dict)):
+            return {"status": result}
+        logger.error(
+            f"Health check function {getattr(self._ping_func, '__name__', 'unknown')} must return `dict` or `str` type."
+        )
+        return {"status": "error", "message": "Invalid response type."}
+
     @override
     def run(self, agent_card: AgentCard, host: str, port: int = 8000):
         if not self._agent_executor:
@@ -155,6 +182,30 @@ def run(self, agent_card: AgentCard, host: str, port: int = 8000):
             ),
         ).build()
 
+        # Register /ping route consistent with SimpleApp behavior
+        async def _ping_handler(request: Request):
+            if not self._ping_func:
+                logger.error("Ping handler function is not set")
+                return Response(status_code=404)
+
+            try:
+                result = (
+                    await self._ping_func()
+                    if inspect.iscoroutinefunction(self._ping_func)
+                    else self._ping_func()
+                )
+                payload = self._format_ping_status(result)
+                return JSONResponse(content=payload, media_type="application/json")
+            except Exception as e:
+                logger.exception("Ping handler function failed: %s", e)
+                return JSONResponse(
+                    content={"status": "error", "message": str(e)},
+                    media_type="application/json",
+                    status_code=500,
+                )
+
+        a2a_app.add_route("/ping", _ping_handler, methods=["GET"])
+
         self.add_env_detect_route(a2a_app)
 
         uvicorn.run(a2a_app, host=host, port=port)

agentkit/toolkit/resources/samples/a2a.py

@@ -63,12 +63,15 @@
 class MyAgentExecutor(A2aAgentExecutor):
     pass
 
+@a2a_app.ping
+def ping() -> str:
+    return "pong!"
 
 if __name__ == "__main__":
     from a2a.types import AgentCard, AgentProvider, AgentSkill, AgentCapabilities
 
     agent_card = AgentCard(
-        capabilities=AgentCapabilities(streaming=True),  # 启用流式
+        capabilities=AgentCapabilities(streaming=True),
         description=agent.description,
         name=agent.name,
         defaultInputModes=["text"],

agentkit/toolkit/runners/ve_agentkit.py

@@ -248,7 +248,28 @@ def status(self, config: VeAgentkitRunnerConfig) -> StatusResult:
                         },
                         timeout=10,
                     )
-                    ping_status = ping_response.status_code == 200
+                    if ping_response.status_code == 200:
+                        ping_status = True
+                    elif ping_response.status_code in (404, 405):
+                        # Fallback: try /health for SimpleApp compatibility
+                        try:
+                            health_response = requests.get(
+                                urljoin(public_endpoint, "health"),
+                                headers={
+                                    "Authorization": f"Bearer {runner_config.runtime_apikey}"
+                                },
+                                timeout=10,
+                            )
+                            if health_response.status_code == 200:
+                                ping_status = True
+                            else:
+                                ping_status = None  # Endpoint reachable but health route not available
+                        except Exception:
+                            # Endpoint reachable (ping returned 404/405), but health check failed
+                            ping_status = None
+                    else:
+                        # Non-200 status indicates server responded but not healthy
+                        ping_status = False
                 except Exception as e:
                     logger.error(f"Failed to check endpoint connectivity: {str(e)}")
                     ping_status = False
@@ -265,11 +286,15 @@ def status(self, config: VeAgentkitRunnerConfig) -> StatusResult:
                 status=status,
                 endpoint_url=public_endpoint,
                 service_id=runner_config.runtime_id,
-                health="healthy"
-                if ping_status
-                else "unhealthy"
-                if ping_status is False
-                else None,
+                health=(
+                    "healthy"
+                    if ping_status is True
+                    else "unhealthy"
+                    if ping_status is False
+                    else "unknown"
+                    if ping_status is None
+                    else None
+                ),
                 metadata={
                     "runtime_id": runner_config.runtime_id,
                     "runtime_name": runtime.name