diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 48606919..58ddf26c 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -15,7 +15,7 @@ jobs: - name: Set up Python uses: actions/setup-python@v5 with: - python-version: '3.11' + python-version: '3.12' - name: Install dependencies run: | @@ -39,7 +39,7 @@ jobs: - name: Set up Python uses: actions/setup-python@v5 with: - python-version: '3.11' + python-version: '3.12' - name: Install dependencies run: | diff --git a/Docs/konflux-integration-guide.md b/Docs/konflux-integration-guide.md index 6491504d..863997c9 100644 --- a/Docs/konflux-integration-guide.md +++ b/Docs/konflux-integration-guide.md @@ -286,6 +286,7 @@ Key steps: | `ab-eval-db-credentials` | Store results in PostgreSQL | | `minio-credentials` | Upload artifacts to MinIO/S3 | | `monitoring-slack-webhook` | Send degradation alerts to Slack | +| `a2a-agent-credentials` | Bearer token for JWT-protected A2A agents (`eval-engine=a2a` only) | ### Creating Workload Cluster Credentials @@ -310,6 +311,22 @@ stringData: See `config/konflux/secrets-template.yaml` for the full template. +### Creating A2A Agent Credentials + +Only needed when `eval-engine=a2a` and the target agent requires a Bearer +token (e.g. JWT-protected endpoints). If this secret doesn't exist, the +`agent-auth-token-secret` lookup is optional and no `Authorization` header +is sent — existing no-auth agents are unaffected. + +```bash +oc create secret generic a2a-agent-credentials \ + --from-literal=token="" \ + -n +``` + +If your secret has a different name, pass it via the `agent-auth-token-secret` +pipeline parameter (default: `a2a-agent-credentials`). + ## Tekton Bundles The core tasks are published as Tekton Bundles to Quay.io: diff --git a/Docs/manual_trigger_guide.md b/Docs/manual_trigger_guide.md index 5c95354c..b662df52 100644 --- a/Docs/manual_trigger_guide.md +++ b/Docs/manual_trigger_guide.md @@ -197,6 +197,8 @@ A2A monitoring runs can be triggered three ways (plus manual runs). Choose the r **Existing agent** (`agent-endpoint`): The pipeline connects to an already-running agent (typically `http://lightspeed-agent.ab-eval-flow.svc:8000`). Do not set `agent-image`/`agent-tag` unless you want a fresh deploy. +**JWT-protected agent** (optional): If the agent requires a Bearer token, create an `a2a-agent-credentials` Secret with a `token` key in the pipeline's namespace (see `Docs/konflux-integration-guide.md#creating-a2a-agent-credentials`). No params need to change — the pipeline picks it up automatically via the `agent-auth-token-secret` param (default: `a2a-agent-credentials`). If the secret doesn't exist, no `Authorization` header is sent, so agents that don't require auth are unaffected. + ### 1. Quay Push Webhook Fires when a new image is pushed to `quay.io/ecosystem-appeng/google-lightspeed-agent` (excluding `sha256:` digest tags and `on-pr-*` PR tags). diff --git a/README.md b/README.md index cbbf6baf..c9f23f47 100644 --- a/README.md +++ b/README.md @@ -122,7 +122,7 @@ The pipeline is LLM-agnostic. Three modes are supported: - Container registry (Quay.io) with push credentials - Harbor fork with OpenShift backend - LLM access (one of the three modes above) -- Python 3.11+ +- Python 3.12+ ## Documentation diff --git a/abevalflow/harbor_agents/a2a_adapter.py b/abevalflow/harbor_agents/a2a_adapter.py index c5be7974..682afdbc 100644 --- a/abevalflow/harbor_agents/a2a_adapter.py +++ b/abevalflow/harbor_agents/a2a_adapter.py @@ -7,7 +7,9 @@ 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= \\ + --ak verify_ssl=true Usage in Harbor config YAML: agents: @@ -15,12 +17,17 @@ kwargs: endpoint: "https://my-agent.example.com" timeout: 120 + auth_token: "" # optional; falls back to AGENT_AUTH_TOKEN env + blocking: true # optional; default True, see A2AAgent.__init__ + verify_ssl: true # optional; default False (preserves prior behavior), + # enable when the endpoint has a CA-trusted cert """ from __future__ import annotations import json import logging +import os import uuid from pathlib import Path from typing import Any @@ -72,6 +79,9 @@ def __init__( context_id: str | None = None, model_name: str | None = None, extra_env: dict[str, str] | None = None, + auth_token: str | None = None, + blocking: bool = True, + verify_ssl: bool = False, **kwargs, ): """Initialize the A2A agent adapter. @@ -82,14 +92,28 @@ 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). + blocking: Whether to request synchronous completion via `configuration.blocking` + in the `message/send` request (default: True). Some A2A servers only + populate `result.artifacts` for the caller when this is set, otherwise the + response may come back before the agent has finished and appear empty. + verify_ssl: Whether to verify TLS certificates when calling the A2A endpoint + (default: False, matching prior behavior). Most internal OpenShift/Kubernetes + Routes use self-signed or cluster-internal certs, so verification is skipped + by default. Set to True when the endpoint has a certificate trusted by the + caller's CA bundle. **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.blocking = blocking + self.verify_ssl = verify_ssl 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: @@ -126,11 +150,12 @@ async def run( "jsonrpc": "2.0", "method": "message/send", "params": { + "configuration": {"blocking": self.blocking}, "message": { "messageId": message_id, "role": "user", - "parts": [{"text": instruction}], - } + "parts": [{"kind": "text", "text": instruction}], + }, }, "id": request_id, } @@ -164,13 +189,16 @@ 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"}, - ssl=False, + headers=headers, + ssl=self.verify_ssl, ) as response: response.raise_for_status() return await response.json() diff --git a/config/konflux/secrets-template.yaml b/config/konflux/secrets-template.yaml index 8089f72b..1bf3cc1c 100644 --- a/config/konflux/secrets-template.yaml +++ b/config/konflux/secrets-template.yaml @@ -13,6 +13,7 @@ # minio-credentials : Upload artifacts to MinIO/S3 # monitoring-slack-webhook: Send degradation alerts to Slack # promptfoo-cloud-credentials: Share red-team results to Promptfoo Cloud (optional) +# a2a-agent-credentials : Bearer token for JWT-protected A2A agents (eval-engine=a2a only) --- # workload-cluster-credentials # ONLY required when EVAL_MODE=remote (cross-cluster evaluation). @@ -55,6 +56,19 @@ type: Opaque stringData: token: "" --- +# a2a-agent-credentials (OPTIONAL) +# Bearer token for JWT-protected A2A agents. Only used when eval-engine=a2a +# and the target agent requires authentication. If this secret doesn't +# exist, no Authorization header is sent (unchanged default behavior). +apiVersion: v1 +kind: Secret +metadata: + name: a2a-agent-credentials + namespace: +type: Opaque +stringData: + token: "" +--- # promptfoo-cloud-credentials (OPTIONAL) # API key for Promptfoo Cloud to share red-team results. # If not configured, red-team runs locally without cloud sharing. diff --git a/docs/index.html b/docs/index.html index ddc19002..0568d92b 100644 --- a/docs/index.html +++ b/docs/index.html @@ -588,7 +588,7 @@

Agentic Eval Flow

Make AI artifact evaluation
automated, measurable, and observable

Tekton-orchestrated pipeline on OpenShift that evaluates skills, agents, and MCP servers through A/B testing, statistical analysis, and multi-gate certification.

- Python 3.11+ + Python 3.12+ Apache-2.0 Tekton / OpenShift PostgreSQL diff --git a/examples/a2a-agent-eval/README.md b/examples/a2a-agent-eval/README.md index 094b71a4..d1426eff 100644 --- a/examples/a2a-agent-eval/README.md +++ b/examples/a2a-agent-eval/README.md @@ -26,6 +26,29 @@ harbor run \ --n-attempts 3 ``` +#### Optional agent kwargs + +The A2A adapter accepts a few optional `--ak` flags beyond `endpoint`/`timeout`: + +| Kwarg | Default | Description | +|-------|---------|-------------| +| `auth_token` | none | Bearer token sent as `Authorization: Bearer ` for JWT-protected agents. Also read from the `AGENT_AUTH_TOKEN` env var if not set. | +| `blocking` | `true` | Requests synchronous completion (`configuration.blocking`) so `result.artifacts` is populated. Set to `false` only if your agent requires async/polling semantics. | +| `verify_ssl` | `false` | TLS certificate verification for the agent endpoint. Defaults to `false` because most internal OpenShift/Kubernetes Routes use self-signed or cluster-internal certs. Set to `true` if your endpoint has a CA-trusted certificate. | + +```bash +# Example: JWT-protected agent with a CA-trusted certificate +harbor run \ + -p examples/a2a-agent-eval/tasks/lightspeed-qa \ + --agent-import-path abevalflow.harbor_agents.a2a_adapter:A2AAgent \ + --ak endpoint=$A2A_ENDPOINT \ + --ak timeout=120 \ + --ak auth_token=$A2A_AUTH_TOKEN \ + --ak verify_ssl=true \ + -e podman \ + --n-attempts 3 +``` + ### Via Agentic Eval Flow Pipeline (Tekton) ```bash @@ -35,6 +58,13 @@ tkn pipeline start abevalflow-ci-pipeline \ --param submission-name=lightspeed-qa-eval ``` +If the agent requires a Bearer token, create an `a2a-agent-credentials` Secret +(with a `token` key) in the pipeline's namespace beforehand — see +[`Docs/konflux-integration-guide.md`](../../Docs/konflux-integration-guide.md#creating-a2a-agent-credentials). +No extra params are needed unless your secret has a different name (use +`--param agent-auth-token-secret=` in that case). If the secret doesn't +exist, the pipeline runs exactly as before with no `Authorization` header. + ## Task Structure ``` @@ -121,6 +151,7 @@ def grade() -> dict: | `LLM_JUDGE_MODEL` | Model for LLM-as-judge | `openai/claude-sonnet` | | `LLM_BASE_URL` | LiteLLM proxy URL | `http://litellm.ab-eval-flow.svc.cluster.local:4000` | | `A2A_ENDPOINT` | Agent endpoint URL | (required) | +| `AGENT_AUTH_TOKEN` | Bearer token for JWT-protected agents (fallback if `--ak auth_token` isn't set) | none | ## Troubleshooting diff --git a/pipeline/tasks/konflux/evaluate.yaml b/pipeline/tasks/konflux/evaluate.yaml index 8dca1465..dbe9d61c 100644 --- a/pipeline/tasks/konflux/evaluate.yaml +++ b/pipeline/tasks/konflux/evaluate.yaml @@ -76,6 +76,13 @@ spec: type: string default: "120" description: A2A agent request timeout in seconds + - name: agent-auth-token-secret + type: string + default: "a2a-agent-credentials" + description: >- + Name of a Secret (in this task's namespace) with a "token" key holding + a Bearer token for JWT-protected A2A agents. Optional: if the secret + doesn't exist, no Authorization header is sent (unchanged default behavior). - name: mcp-url type: string default: "" @@ -155,6 +162,12 @@ spec: name: llm-credentials key: api-key optional: true + - name: AGENT_AUTH_TOKEN + valueFrom: + secretKeyRef: + name: $(params.agent-auth-token-secret) + key: token + optional: true script: | #!/usr/bin/env bash set -euo pipefail @@ -165,6 +178,8 @@ spec: SUBMISSION_NAME="$(params.submission-name)" SUBMISSION_DIR="$(params.submission-dir)" AGENT_ENDPOINT="$(params.agent-endpoint)" + # Optional; unset when the secret/key doesn't exist (optional: true above). + AGENT_AUTH_TOKEN="${AGENT_AUTH_TOKEN:-}" MCP_URL="$(params.mcp-url)" COMMIT_SHA="$(params.commit-sha)" PIPELINE_RUN_ID="$(params.pipeline-run-id)" @@ -305,14 +320,17 @@ spec: echo "Task: \$(basename \$TASK_DIR) | Attempts: \$N_ATTEMPTS" - python3 - "\$RESULTS_DIR" "\$TASK_DIR" "\$N_ATTEMPTS" "$AGENT_ENDPOINT" "$(params.agent-timeout)" "$(params.llm-api-base)" "openai/$(params.llm-model)" <<'GENCFG' + python3 - "\$RESULTS_DIR" "\$TASK_DIR" "\$N_ATTEMPTS" "$AGENT_ENDPOINT" "$(params.agent-timeout)" "$(params.llm-api-base)" "openai/$(params.llm-model)" "$AGENT_AUTH_TOKEN" <<'GENCFG' import sys, yaml - results_dir, task_dir, n_attempts, endpoint, timeout, llm_api_base, llm_model = sys.argv[1:8] + results_dir, task_dir, n_attempts, endpoint, timeout, llm_api_base, llm_model, auth_token = sys.argv[1:9] + agent_kwargs = {"endpoint": endpoint, "timeout": int(timeout)} + if auth_token: + agent_kwargs["auth_token"] = auth_token config = { "job_name": "a2a-eval", "jobs_dir": results_dir, "n_attempts": int(n_attempts), - "agents": [{"import_path": "abevalflow.harbor_agents.a2a_adapter:A2AAgent", "kwargs": {"endpoint": endpoint, "timeout": int(timeout)}}], + "agents": [{"import_path": "abevalflow.harbor_agents.a2a_adapter:A2AAgent", "kwargs": agent_kwargs}], "tasks": [{"path": task_dir}], "environment": {"type": "local"}, "verifier": {"env": {"LLM_JUDGE_MODEL": llm_model, "LLM_API_BASE": llm_api_base, "OPENAI_API_KEY": "sk-dummy"}} @@ -532,14 +550,17 @@ spec: LLM_API_BASE="$(params.llm-api-base)" LLM_MODEL="openai/$(params.llm-model)" - python3 - "$CONFIG_FILE" "$TASK_DIR" "$RESULTS_DIR" "$N_ATTEMPTS" "$AGENT_ENDPOINT" "$(params.agent-timeout)" "$LLM_API_BASE" "$LLM_MODEL" <<'GENCFG' + python3 - "$CONFIG_FILE" "$TASK_DIR" "$RESULTS_DIR" "$N_ATTEMPTS" "$AGENT_ENDPOINT" "$(params.agent-timeout)" "$LLM_API_BASE" "$LLM_MODEL" "$AGENT_AUTH_TOKEN" <<'GENCFG' import sys, yaml - config_file, task_dir, results_dir, n_attempts, endpoint, timeout, llm_api_base, llm_model = sys.argv[1:9] + config_file, task_dir, results_dir, n_attempts, endpoint, timeout, llm_api_base, llm_model, auth_token = sys.argv[1:10] + agent_kwargs = {"endpoint": endpoint, "timeout": int(timeout)} + if auth_token: + agent_kwargs["auth_token"] = auth_token config = { "job_name": "a2a-eval", "jobs_dir": results_dir, "n_attempts": int(n_attempts), - "agents": [{"import_path": "abevalflow.harbor_agents.a2a_adapter:A2AAgent", "kwargs": {"endpoint": endpoint, "timeout": int(timeout)}}], + "agents": [{"import_path": "abevalflow.harbor_agents.a2a_adapter:A2AAgent", "kwargs": agent_kwargs}], "tasks": [{"path": task_dir}], "environment": {"type": "local"}, "verifier": {"env": {"LLM_JUDGE_MODEL": llm_model, "LLM_API_BASE": llm_api_base, "OPENAI_API_KEY": "sk-dummy"}} diff --git a/pipeline/tasks/phases/evaluate.yaml b/pipeline/tasks/phases/evaluate.yaml index 15cf76a5..6f8f5ebf 100644 --- a/pipeline/tasks/phases/evaluate.yaml +++ b/pipeline/tasks/phases/evaluate.yaml @@ -110,6 +110,13 @@ spec: type: string default: "120" description: A2A agent request timeout in seconds + - name: agent-auth-token-secret + type: string + default: "a2a-agent-credentials" + description: >- + Name of a Secret (in this task's namespace) with a "token" key holding + a Bearer token for JWT-protected A2A agents. Optional: if the secret + doesn't exist, no Authorization header is sent (unchanged default behavior). - name: agent-image type: string default: "quay.io/ecosystem-appeng/google-lightspeed-agent" @@ -1102,6 +1109,12 @@ spec: name: llm-credentials key: api-key optional: true + - name: AGENT_AUTH_TOKEN + valueFrom: + secretKeyRef: + name: $(params.agent-auth-token-secret) + key: token + optional: true script: | #!/usr/bin/env bash set -euo pipefail @@ -1112,6 +1125,9 @@ spec: exit 0 fi + # Optional; unset when the secret/key doesn't exist (optional: true above). + AGENT_AUTH_TOKEN="${AGENT_AUTH_TOKEN:-}" + # Read endpoint from deploy step or param ENDPOINT_FILE="$(workspaces.source.path)/_eval_tmp/agent-endpoint" if [ -f "$ENDPOINT_FILE" ]; then @@ -1178,11 +1194,15 @@ spec: LLM_API_BASE="$(params.llm-api-base)" LLM_MODEL="openai/$(params.llm-model)" - python3 - "$CONFIG_FILE" "$TASK_DIR" "$RESULTS_DIR" "$N_ATTEMPTS" "$AGENT_ENDPOINT" "$(params.agent-timeout)" "$LLM_API_BASE" "$LLM_MODEL" <<'GENCFG' + python3 - "$CONFIG_FILE" "$TASK_DIR" "$RESULTS_DIR" "$N_ATTEMPTS" "$AGENT_ENDPOINT" "$(params.agent-timeout)" "$LLM_API_BASE" "$LLM_MODEL" "$AGENT_AUTH_TOKEN" <<'GENCFG' import sys import yaml - config_file, task_dir, results_dir, n_attempts, endpoint, timeout, llm_api_base, llm_model = sys.argv[1:9] + config_file, task_dir, results_dir, n_attempts, endpoint, timeout, llm_api_base, llm_model, auth_token = sys.argv[1:10] + + agent_kwargs = {"endpoint": endpoint, "timeout": int(timeout)} + if auth_token: + agent_kwargs["auth_token"] = auth_token config = { "job_name": "a2a-eval", @@ -1190,10 +1210,7 @@ spec: "n_attempts": int(n_attempts), "agents": [{ "import_path": "abevalflow.harbor_agents.a2a_adapter:A2AAgent", - "kwargs": { - "endpoint": endpoint, - "timeout": int(timeout) - } + "kwargs": agent_kwargs }], "tasks": [{"path": task_dir}], "environment": { diff --git a/pyproject.toml b/pyproject.toml index 2d6c29fe..ef75a79b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ version = "0.1.0" description = "Automated Tekton pipeline on OpenShift for A/B evaluation of AI submissions" readme = "README.md" license = "Apache-2.0" -requires-python = ">=3.11" +requires-python = ">=3.12" dependencies = [ "pydantic>=2.0", "jinja2>=3.1", @@ -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] @@ -40,7 +43,7 @@ testpaths = ["tests"] pythonpath = ["."] [tool.ruff] -target-version = "py311" +target-version = "py312" line-length = 120 [tool.ruff.lint] diff --git a/submissions/a2a-agent-eval/README.md b/submissions/a2a-agent-eval/README.md index 094b71a4..d1426eff 100644 --- a/submissions/a2a-agent-eval/README.md +++ b/submissions/a2a-agent-eval/README.md @@ -26,6 +26,29 @@ harbor run \ --n-attempts 3 ``` +#### Optional agent kwargs + +The A2A adapter accepts a few optional `--ak` flags beyond `endpoint`/`timeout`: + +| Kwarg | Default | Description | +|-------|---------|-------------| +| `auth_token` | none | Bearer token sent as `Authorization: Bearer ` for JWT-protected agents. Also read from the `AGENT_AUTH_TOKEN` env var if not set. | +| `blocking` | `true` | Requests synchronous completion (`configuration.blocking`) so `result.artifacts` is populated. Set to `false` only if your agent requires async/polling semantics. | +| `verify_ssl` | `false` | TLS certificate verification for the agent endpoint. Defaults to `false` because most internal OpenShift/Kubernetes Routes use self-signed or cluster-internal certs. Set to `true` if your endpoint has a CA-trusted certificate. | + +```bash +# Example: JWT-protected agent with a CA-trusted certificate +harbor run \ + -p examples/a2a-agent-eval/tasks/lightspeed-qa \ + --agent-import-path abevalflow.harbor_agents.a2a_adapter:A2AAgent \ + --ak endpoint=$A2A_ENDPOINT \ + --ak timeout=120 \ + --ak auth_token=$A2A_AUTH_TOKEN \ + --ak verify_ssl=true \ + -e podman \ + --n-attempts 3 +``` + ### Via Agentic Eval Flow Pipeline (Tekton) ```bash @@ -35,6 +58,13 @@ tkn pipeline start abevalflow-ci-pipeline \ --param submission-name=lightspeed-qa-eval ``` +If the agent requires a Bearer token, create an `a2a-agent-credentials` Secret +(with a `token` key) in the pipeline's namespace beforehand — see +[`Docs/konflux-integration-guide.md`](../../Docs/konflux-integration-guide.md#creating-a2a-agent-credentials). +No extra params are needed unless your secret has a different name (use +`--param agent-auth-token-secret=` in that case). If the secret doesn't +exist, the pipeline runs exactly as before with no `Authorization` header. + ## Task Structure ``` @@ -121,6 +151,7 @@ def grade() -> dict: | `LLM_JUDGE_MODEL` | Model for LLM-as-judge | `openai/claude-sonnet` | | `LLM_BASE_URL` | LiteLLM proxy URL | `http://litellm.ab-eval-flow.svc.cluster.local:4000` | | `A2A_ENDPOINT` | Agent endpoint URL | (required) | +| `AGENT_AUTH_TOKEN` | Bearer token for JWT-protected agents (fallback if `--ak auth_token` isn't set) | none | ## Troubleshooting diff --git a/tests/test_a2a_adapter_auth.py b/tests/test_a2a_adapter_auth.py new file mode 100644 index 00000000..5554fc95 --- /dev/null +++ b/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"} diff --git a/tests/test_a2a_adapter_config.py b/tests/test_a2a_adapter_config.py new file mode 100644 index 00000000..a4b68b9f --- /dev/null +++ b/tests/test_a2a_adapter_config.py @@ -0,0 +1,120 @@ +"""Tests for configurable blocking/TLS behavior on A2AAgent. + +Covers three related fixes to abevalflow.harbor_agents.a2a_adapter.A2AAgent: + +1. Message parts sent in `message/send` now include an explicit "kind": "text" + discriminator, matching the A2A spec's Part schema. +2. `configuration.blocking` is now sent on every `message/send` request and is + configurable via the `blocking` kwarg (default True). Without it, some A2A + servers return before the agent has finished, yielding an empty response. +3. TLS verification is configurable via the `verify_ssl` kwarg (default False, + preserving the prior hardcoded `ssl=False` behavior) instead of being + permanently hardcoded with no way to opt into verification. +""" + +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: + path = tmp_path / "logs" + path.mkdir(parents=True, exist_ok=True) + return path + + +def _patched_session(mock_response: AsyncMock) -> tuple[AsyncMock, patch]: + """Build a mocked aiohttp.ClientSession context and the patch to install it.""" + 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 + + patcher = patch( + "abevalflow.harbor_agents.a2a_adapter.aiohttp.ClientSession", + return_value=mock_session_ctx, + ) + return mock_session, patcher + + +def _mock_response(result: dict | None = None) -> AsyncMock: + mock_response = AsyncMock() + mock_response.raise_for_status = MagicMock() + mock_response.json = AsyncMock(return_value={"result": result or {}}) + return mock_response + + +class TestA2ABlockingConfiguration: + def test_blocking_defaults_to_true(self, logs_dir: Path) -> None: + agent = A2AAgent(logs_dir, "https://agent.example.com") + assert agent.blocking is True + + def test_blocking_can_be_disabled(self, logs_dir: Path) -> None: + agent = A2AAgent(logs_dir, "https://agent.example.com", blocking=False) + assert agent.blocking is False + + @pytest.mark.asyncio + async def test_run_sends_blocking_configuration_and_kind(self, logs_dir: Path) -> None: + agent = A2AAgent(logs_dir, "https://agent.example.com") + mock_session, patcher = _patched_session(_mock_response()) + + with patcher: + await agent.run("do the thing", MagicMock(), MagicMock()) + + _, kwargs = mock_session.post.call_args + payload = kwargs["json"] + assert payload["params"]["configuration"] == {"blocking": True} + assert payload["params"]["message"]["parts"] == [{"kind": "text", "text": "do the thing"}] + + @pytest.mark.asyncio + async def test_run_respects_blocking_false(self, logs_dir: Path) -> None: + agent = A2AAgent(logs_dir, "https://agent.example.com", blocking=False) + mock_session, patcher = _patched_session(_mock_response()) + + with patcher: + await agent.run("do the thing", MagicMock(), MagicMock()) + + _, kwargs = mock_session.post.call_args + assert kwargs["json"]["params"]["configuration"] == {"blocking": False} + + +class TestA2AVerifySslConfiguration: + def test_verify_ssl_defaults_to_false(self, logs_dir: Path) -> None: + agent = A2AAgent(logs_dir, "https://agent.example.com") + assert agent.verify_ssl is False + + def test_verify_ssl_can_be_enabled(self, logs_dir: Path) -> None: + agent = A2AAgent(logs_dir, "https://agent.example.com", verify_ssl=True) + assert agent.verify_ssl is True + + @pytest.mark.asyncio + async def test_send_request_passes_verify_ssl_false_by_default(self, logs_dir: Path) -> None: + agent = A2AAgent(logs_dir, "https://agent.example.com") + mock_session, patcher = _patched_session(_mock_response()) + + with patcher: + await agent._send_request({"jsonrpc": "2.0", "id": "1"}) + + _, kwargs = mock_session.post.call_args + assert kwargs["ssl"] is False + + @pytest.mark.asyncio + async def test_send_request_passes_verify_ssl_true_when_enabled(self, logs_dir: Path) -> None: + agent = A2AAgent(logs_dir, "https://agent.example.com", verify_ssl=True) + mock_session, patcher = _patched_session(_mock_response()) + + with patcher: + await agent._send_request({"jsonrpc": "2.0", "id": "1"}) + + _, kwargs = mock_session.post.call_args + assert kwargs["ssl"] is True