feat: add health check endpoint for agent authentication status

This commit is contained in:
Abhimanyu Saharan
2026-02-16 00:42:15 +05:30
parent cd68446c42
commit 1d63bd0148
4 changed files with 133 additions and 0 deletions

View File

@@ -45,6 +45,7 @@ from app.schemas.gateway_coordination import (
GatewayMainAskUserRequest,
GatewayMainAskUserResponse,
)
from app.schemas.health import AgentHealthStatusResponse
from app.schemas.pagination import DefaultLimitOffsetPage
from app.schemas.tags import TagRef
from app.schemas.tasks import TaskCommentCreate, TaskCommentRead, TaskCreate, TaskRead, TaskUpdate
@@ -186,6 +187,73 @@ def _guard_task_access(agent_ctx: AgentAuthContext, task: Task) -> None:
OpenClawAuthorizationPolicy.require_board_write_access(allowed=allowed)
@router.get(
"/healthz",
response_model=AgentHealthStatusResponse,
tags=AGENT_ALL_ROLE_TAGS,
summary="Agent Auth Health Check",
description=(
"Token-authenticated liveness probe for agent API clients.\n\n"
"Use this endpoint when the caller needs to verify both service availability "
"and agent-token validity in one request."
),
openapi_extra={
"x-llm-intent": "agent_auth_health",
"x-when-to-use": [
"Verify agent token validity before entering an automation loop",
"Confirm agent API availability with caller identity context",
],
"x-when-not-to-use": [
"General infrastructure liveness checks that do not require auth context",
"Task, board, or messaging workflow actions",
],
"x-required-actor": "any_agent",
"x-prerequisites": [
"Authenticated agent token via X-Agent-Token header",
],
"x-side-effects": [
"May refresh agent last-seen presence metadata via auth middleware",
],
"x-negative-guidance": [
"Do not parse this response as an array.",
"Do not use this endpoint for task routing decisions.",
],
"x-routing-policy": [
"Use this as the first probe for agent-scoped automation health.",
"Use /healthz only for unauthenticated service-level liveness checks.",
],
"x-routing-policy-examples": [
{
"input": {
"intent": "agent startup probe with token verification",
"required_privilege": "any_agent",
},
"decision": "agent_auth_health",
},
{
"input": {
"intent": "platform-level probe with no agent token",
"required_privilege": "none",
},
"decision": "service_healthz",
},
],
},
)
def agent_healthz(
agent_ctx: AgentAuthContext = AGENT_CTX_DEP,
) -> AgentHealthStatusResponse:
"""Return authenticated liveness metadata for the current agent token."""
return AgentHealthStatusResponse(
ok=True,
agent_id=agent_ctx.agent.id,
board_id=agent_ctx.agent.board_id,
gateway_id=agent_ctx.agent.gateway_id,
status=agent_ctx.agent.status,
is_board_lead=agent_ctx.agent.is_board_lead,
)
@router.get(
"/boards",
response_model=DefaultLimitOffsetPage[BoardRead],

View File

@@ -2,6 +2,8 @@
from __future__ import annotations
from uuid import UUID
from pydantic import Field
from sqlmodel import SQLModel
@@ -13,3 +15,29 @@ class HealthStatusResponse(SQLModel):
description="Indicates whether the probe check succeeded.",
examples=[True],
)
class AgentHealthStatusResponse(HealthStatusResponse):
"""Agent-authenticated liveness payload for agent route probes."""
agent_id: UUID = Field(
description="Authenticated agent id derived from `X-Agent-Token`.",
examples=["aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"],
)
board_id: UUID | None = Field(
default=None,
description="Board scope for the authenticated agent, when applicable.",
examples=["bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"],
)
gateway_id: UUID = Field(
description="Gateway owning the authenticated agent.",
examples=["cccccccc-cccc-cccc-cccc-cccccccccccc"],
)
status: str = Field(
description="Current persisted lifecycle status for the authenticated agent.",
examples=["online", "healthy", "updating"],
)
is_board_lead: bool = Field(
description="Whether the authenticated agent is the board lead.",
examples=[False],
)

View File

@@ -0,0 +1,34 @@
from __future__ import annotations
from uuid import UUID, uuid4
from app.api import agent as agent_api
from app.core.agent_auth import AgentAuthContext
from app.models.agents import Agent
def _agent_ctx(*, board_id: UUID | None, status: str, is_board_lead: bool) -> AgentAuthContext:
return AgentAuthContext(
actor_type="agent",
agent=Agent(
id=uuid4(),
board_id=board_id,
gateway_id=uuid4(),
name="Health Probe Agent",
status=status,
is_board_lead=is_board_lead,
),
)
def test_agent_healthz_returns_authenticated_agent_context() -> None:
agent_ctx = _agent_ctx(board_id=uuid4(), status="online", is_board_lead=True)
response = agent_api.agent_healthz(agent_ctx=agent_ctx)
assert response.ok is True
assert response.agent_id == agent_ctx.agent.id
assert response.board_id == agent_ctx.agent.board_id
assert response.gateway_id == agent_ctx.agent.gateway_id
assert response.status == "online"
assert response.is_board_lead is True

View File

@@ -39,6 +39,8 @@ def test_openapi_agent_role_tags_are_exposed() -> None:
path="/api/v1/agent/boards",
method="get",
)
health_tags = _op_tags(schema, path="/api/v1/agent/healthz", method="get")
assert {"agent-lead", "agent-worker", "agent-main"} <= health_tags
assert "agent-main" in _op_tags(
schema,
path="/api/v1/agent/boards/{board_id}",
@@ -106,6 +108,7 @@ def test_openapi_agent_tool_endpoints_include_llm_hints() -> None:
expected_paths = [
("/api/v1/agent/boards", "get"),
("/api/v1/agent/healthz", "get"),
("/api/v1/agent/boards/{board_id}", "get"),
("/api/v1/agent/agents", "get"),
("/api/v1/agent/heartbeat", "post"),