Merge pull request #64 from Serverless-Devs/feat-default-health-api

feat(server): add health check endpoint with tests

Author: OhYee

agentrun/server/agui_protocol.py

@@ -206,11 +206,6 @@ async def run_agent(request: Request):
                     headers=sse_headers,
                 )
 
-        @router.get("/health")
-        async def health_check():
-            """健康检查端点"""
-            return {"status": "ok", "protocol": "ag-ui", "version": "1.0"}
-
         return router
 
     async def parse_request(

agentrun/server/server.py

@@ -132,6 +132,9 @@ def __init__(
 
         self.agent_invoker = AgentInvoker(invoke_agent)
 
+        # 注册 health check 路由
+        self._register_health_check()
+
         # 配置 CORS
         self._setup_cors(config.cors_origins if config else None)
 
@@ -145,6 +148,13 @@ def __init__(
         # 挂载所有协议的 Router
         self._mount_protocols(protocols)
 
+    def _register_health_check(self):
+        """注册 /health 健康检查路由 / Register /health health check route"""
+
+        @self.app.get("/health")
+        async def health_check():
+            return {"status": "ok"}
+
     def _wrap_with_memory(
         self,
         invoke_agent: InvokeAgentHandler,

tests/unittests/integration/langchain/test_agent_invoke_methods.py

@@ -100,7 +100,7 @@ def _start_server(app: FastAPI) -> tuple:
     base_url = f"http://127.0.0.1:{port}"
     for i in range(50):
         try:
-            httpx.get(f"{base_url}/ag-ui/agent/health", timeout=0.2)
+            httpx.get(f"{base_url}/health", timeout=0.2)
             break
         except Exception:
             if i == 49:

tests/unittests/integration/test_langchain_agui_integration.py

@@ -671,7 +671,7 @@ async def invoke_agent(request: AgentRequest):
         base_url = f"http://127.0.0.1:{port}"
         for i in range(50):
             try:
-                httpx.get(f"{base_url}/ag-ui/agent/health", timeout=0.2)
+                httpx.get(f"{base_url}/health", timeout=0.2)
                 break
             except Exception:
                 if i == 49:

tests/unittests/server/test_agui_protocol.py

@@ -43,22 +43,6 @@ def get_client(self, invoke_agent):
         server = AgentRunServer(invoke_agent=invoke_agent)
         return TestClient(server.as_fastapi_app())
 
-    @pytest.mark.asyncio
-    async def test_health_check(self):
-        """测试健康检查端点"""
-
-        def invoke_agent(request: AgentRequest):
-            return "Hello"
-
-        client = self.get_client(invoke_agent)
-        response = client.get("/ag-ui/agent/health")
-
-        assert response.status_code == 200
-        data = response.json()
-        assert data["status"] == "ok"
-        assert data["protocol"] == "ag-ui"
-        assert data["version"] == "1.0"
-
     @pytest.mark.asyncio
     async def test_value_error_handling(self):
         """测试 ValueError 处理"""

tests/unittests/server/test_server.py

@@ -124,6 +124,43 @@ def get_client(self, invoke_agent):
 
         return TestClient(app)
 
+    async def test_health_check(self):
+        """测试 /health 健康检查路由"""
+
+        client = self.get_client(self.get_invoke_agent_non_streaming())
+
+        response = client.get("/health")
+
+        assert response.status_code == 200
+        assert response.json() == {"status": "ok"}
+
+    async def test_health_check_post_not_allowed(self):
+        """测试 POST /health 不被允许"""
+
+        client = self.get_client(self.get_invoke_agent_non_streaming())
+
+        response = client.post("/health")
+
+        # FastAPI 对不匹配的方法返回 405
+        assert response.status_code == 405
+
+    async def test_health_check_with_custom_protocols(self):
+        """测试自定义协议列表时 /health 仍可用"""
+        from agentrun.server.openai_protocol import OpenAIProtocolHandler
+
+        server = AgentRunServer(
+            invoke_agent=self.get_invoke_agent_non_streaming(),
+            protocols=[OpenAIProtocolHandler()],
+        )
+        from fastapi.testclient import TestClient
+
+        client = TestClient(server.as_fastapi_app())
+
+        response = client.get("/health")
+
+        assert response.status_code == 200
+        assert response.json() == {"status": "ok"}
+
     async def test_server_non_streaming_protocols(self):
         """测试非流式的 OpenAI 和 AGUI 服务器响应功能"""