feat(a2a): add optional Bearer auth support for JWT-protected agents

abevalflow/harbor_agents/a2a_adapter.py

@@ -7,20 +7,23 @@
     harbor run -p tasks/my-eval \\
         --agent-import-path abevalflow.harbor_agents.a2a_adapter:A2AAgent \\
         --ak endpoint=https://my-agent.example.com \\
-        --ak timeout=120
+        --ak timeout=120 \\
+        --ak auth_token=<bearer-jwt>
 
 Usage in Harbor config YAML:
     agents:
       - import_path: "abevalflow.harbor_agents.a2a_adapter:A2AAgent"
         kwargs:
           endpoint: "https://my-agent.example.com"
           timeout: 120
+          auth_token: "<bearer-jwt>"  # optional; falls back to AGENT_AUTH_TOKEN env
 """
 
 from __future__ import annotations
 
 import json
 import logging
+import os
 import uuid
 from pathlib import Path
 from typing import Any
@@ -72,6 +75,7 @@ def __init__(
         context_id: str | None = None,
         model_name: str | None = None,
         extra_env: dict[str, str] | None = None,
+        auth_token: str | None = None,
         **kwargs,
     ):
         """Initialize the A2A agent adapter.
@@ -82,14 +86,20 @@ def __init__(
             timeout: Request timeout in seconds (default: 120).
             context_id: Optional context ID for conversation continuity.
             model_name: Optional model name for logging/tracking.
-            extra_env: Extra environment variables (unused but accepted for compatibility).
+            auth_token: Optional bearer token for Authorization header (also reads AGENT_AUTH_TOKEN env).
             **kwargs: Additional arguments passed to BaseAgent.
         """
         super().__init__(logs_dir=logs_dir, model_name=model_name, **kwargs)
         self.endpoint = endpoint.rstrip("/")
         self.timeout = timeout
         self.context_id = context_id
         self._extra_env = extra_env or {}
+        self._auth_token = (
+            auth_token
+            or self._extra_env.get("AGENT_AUTH_TOKEN")
+            or os.environ.get("AGENT_AUTH_TOKEN")
+            or ""
+        )
 
     @staticmethod
     def name() -> str:
@@ -164,12 +174,15 @@ async def _send_request(self, payload: dict[str, Any]) -> dict[str, Any]:
             The JSON response from the A2A agent.
         """
         timeout = aiohttp.ClientTimeout(total=self.timeout)
+        headers = {"Content-Type": "application/json"}
+        if self._auth_token:
+            headers["Authorization"] = f"Bearer {self._auth_token}"
 
         async with aiohttp.ClientSession(timeout=timeout) as session:
             async with session.post(
                 self.endpoint,
                 json=payload,
-                headers={"Content-Type": "application/json"},
+                headers=headers,
                 ssl=False,
             ) as response:
                 response.raise_for_status()

pyproject.toml

@@ -19,6 +19,7 @@ dependencies = [
     "psycopg[binary]>=3.1",
     "minio>=7.0",
     "openai>=1.0",
+    "aiohttp>=3.9",
 ]
 
 [project.optional-dependencies]
@@ -33,6 +34,8 @@ dev = [
     "pytest-cov>=4.1",
     "pytest-asyncio>=0.23",
     "ruff>=0.4",
+    "aiohttp>=3.9",
+    "harbor>=0.13.1",
 ]
 
 [tool.pytest.ini_options]

tests/test_a2a_adapter_auth.py

@@ -0,0 +1,100 @@
+"""Tests for optional Bearer auth on abevalflow.harbor_agents.a2a_adapter.A2AAgent."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from abevalflow.harbor_agents.a2a_adapter import A2AAgent
+
+
+@pytest.fixture
+def logs_dir(tmp_path: Path) -> Path:
+    return tmp_path / "logs"
+
+
+class TestA2AAuthTokenResolution:
+    def test_auth_token_kwarg(self, logs_dir: Path) -> None:
+        agent = A2AAgent(logs_dir, "https://agent.example.com", auth_token="jwt-kwarg")
+        assert agent._auth_token == "jwt-kwarg"
+
+    def test_extra_env_agent_auth_token(self, logs_dir: Path) -> None:
+        agent = A2AAgent(
+            logs_dir,
+            "https://agent.example.com",
+            extra_env={"AGENT_AUTH_TOKEN": "jwt-extra"},
+        )
+        assert agent._auth_token == "jwt-extra"
+
+    def test_env_agent_auth_token(self, logs_dir: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("AGENT_AUTH_TOKEN", "jwt-env")
+        agent = A2AAgent(logs_dir, "https://agent.example.com")
+        assert agent._auth_token == "jwt-env"
+
+    def test_kwarg_overrides_env(self, logs_dir: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("AGENT_AUTH_TOKEN", "jwt-env")
+        agent = A2AAgent(logs_dir, "https://agent.example.com", auth_token="jwt-kwarg")
+        assert agent._auth_token == "jwt-kwarg"
+
+    def test_no_token(self, logs_dir: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.delenv("AGENT_AUTH_TOKEN", raising=False)
+        agent = A2AAgent(logs_dir, "https://agent.example.com")
+        assert agent._auth_token == ""
+
+
+class TestA2ASendRequestHeaders:
+    @pytest.mark.asyncio
+    async def test_includes_bearer_header_when_token_set(self, logs_dir: Path) -> None:
+        agent = A2AAgent(logs_dir, "https://agent.example.com", auth_token="secret-jwt")
+
+        mock_response = AsyncMock()
+        mock_response.raise_for_status = MagicMock()
+        mock_response.json = AsyncMock(return_value={"result": {}})
+
+        mock_post_ctx = AsyncMock()
+        mock_post_ctx.__aenter__.return_value = mock_response
+
+        mock_session = AsyncMock()
+        mock_session.post = MagicMock(return_value=mock_post_ctx)
+
+        mock_session_ctx = AsyncMock()
+        mock_session_ctx.__aenter__.return_value = mock_session
+
+        with patch(
+            "abevalflow.harbor_agents.a2a_adapter.aiohttp.ClientSession",
+            return_value=mock_session_ctx,
+        ):
+            await agent._send_request({"jsonrpc": "2.0", "id": "1"})
+
+        _, kwargs = mock_session.post.call_args
+        assert kwargs["headers"]["Authorization"] == "Bearer secret-jwt"
+        assert kwargs["headers"]["Content-Type"] == "application/json"
+
+    @pytest.mark.asyncio
+    async def test_omits_authorization_without_token(self, logs_dir: Path) -> None:
+        agent = A2AAgent(logs_dir, "https://agent.example.com")
+
+        mock_response = AsyncMock()
+        mock_response.raise_for_status = MagicMock()
+        mock_response.json = AsyncMock(return_value={"result": {}})
+
+        mock_post_ctx = AsyncMock()
+        mock_post_ctx.__aenter__.return_value = mock_response
+
+        mock_session = AsyncMock()
+        mock_session.post = MagicMock(return_value=mock_post_ctx)
+
+        mock_session_ctx = AsyncMock()
+        mock_session_ctx.__aenter__.return_value = mock_session
+
+        with patch(
+            "abevalflow.harbor_agents.a2a_adapter.aiohttp.ClientSession",
+            return_value=mock_session_ctx,
+        ):
+            await agent._send_request({"jsonrpc": "2.0", "id": "1"})
+
+        _, kwargs = mock_session.post.call_args
+        assert "Authorization" not in kwargs["headers"]
+        assert kwargs["headers"] == {"Content-Type": "application/json"}