"""Validate a deliberately small A2A 1.0 teaching contract offline. This module is not an implementation of the complete A2A specification and is not a substitute for an official SDK, Inspector, TCK, or cross-vendor test. """ from __future__ import annotations import argparse import base64 import binascii import json from pathlib import Path from typing import Any from urllib.parse import urlparse CORE_BINDINGS = {"JSONRPC", "GRPC", "HTTP+JSON"} CONTENT_FIELDS = {"text", "raw", "url", "data"} TASK_STATES = { "TASK_STATE_UNSPECIFIED", "TASK_STATE_SUBMITTED", "TASK_STATE_WORKING", "TASK_STATE_COMPLETED", "TASK_STATE_FAILED", "TASK_STATE_CANCELED", "TASK_STATE_INPUT_REQUIRED", "TASK_STATE_REJECTED", "TASK_STATE_AUTH_REQUIRED", } TERMINAL_STATES = { "TASK_STATE_COMPLETED", "TASK_STATE_FAILED", "TASK_STATE_CANCELED", "TASK_STATE_REJECTED", } def _is_non_empty_string(value: Any) -> bool: return isinstance(value, str) and bool(value.strip()) def _is_absolute_url(value: Any, *, https_only: bool = False) -> bool: if not _is_non_empty_string(value): return False parsed = urlparse(value) if https_only: return parsed.scheme == "https" and bool(parsed.netloc) return parsed.scheme in {"http", "https"} and bool(parsed.netloc) def _validate_string_list(value: Any, path: str, errors: list[str]) -> None: if not isinstance(value, list) or not value: errors.append(f"{path} must be a non-empty list") return if any(not _is_non_empty_string(item) for item in value): errors.append(f"{path} entries must be non-empty strings") def validate_agent_card(card: Any) -> list[str]: """Validate the documented A2A 1.0 subset plus local production policy.""" errors: list[str] = [] if not isinstance(card, dict): return ["agentCard must be an object"] for field in ("name", "description", "version"): if not _is_non_empty_string(card.get(field)): errors.append(f"agentCard.{field} must be a non-empty string") interfaces = card.get("supportedInterfaces") if not isinstance(interfaces, list) or not interfaces: errors.append("agentCard.supportedInterfaces must be a non-empty list") else: for index, interface in enumerate(interfaces): path = f"agentCard.supportedInterfaces[{index}]" if not isinstance(interface, dict): errors.append(f"{path} must be an object") continue if not _is_absolute_url(interface.get("url"), https_only=True): errors.append(f"{path}.url must be an absolute HTTPS URL for this production baseline") binding = interface.get("protocolBinding") is_custom_binding = _is_absolute_url(binding) if binding not in CORE_BINDINGS and not is_custom_binding: errors.append( f"{path}.protocolBinding must be a core binding or an absolute HTTP(S) URI" ) if interface.get("protocolVersion") != "1.0": errors.append(f"{path}.protocolVersion must equal the course baseline '1.0'") if not isinstance(card.get("capabilities"), dict): errors.append("agentCard.capabilities must be an object") _validate_string_list(card.get("defaultInputModes"), "agentCard.defaultInputModes", errors) _validate_string_list(card.get("defaultOutputModes"), "agentCard.defaultOutputModes", errors) skills = card.get("skills") if not isinstance(skills, list) or not skills: errors.append("agentCard.skills must be a non-empty list") else: seen_ids: set[str] = set() for index, skill in enumerate(skills): path = f"agentCard.skills[{index}]" if not isinstance(skill, dict): errors.append(f"{path} must be an object") continue for field in ("id", "name", "description"): if not _is_non_empty_string(skill.get(field)): errors.append(f"{path}.{field} must be a non-empty string") skill_id = skill.get("id") if _is_non_empty_string(skill_id): if skill_id in seen_ids: errors.append(f"{path}.id must be unique within the Agent Card") seen_ids.add(skill_id) _validate_string_list(skill.get("tags"), f"{path}.tags", errors) return errors def validate_part(part: Any, path: str) -> list[str]: """Validate the v1.0 member-based Part discrimination used in the course.""" if not isinstance(part, dict): return [f"{path} must be an object"] present = CONTENT_FIELDS.intersection(part) errors: list[str] = [] if len(present) != 1: errors.append(f"{path} must contain exactly one of text/raw/url/data") if "kind" in part: errors.append(f"{path}.kind is a v0.3 discriminator and is not valid in the v1.0 baseline") if "url" in part and not _is_absolute_url(part.get("url")): errors.append(f"{path}.url must be an absolute HTTP(S) URL") if "text" in part and not isinstance(part.get("text"), str): errors.append(f"{path}.text must be a string") if "raw" in part and not _is_non_empty_string(part.get("raw")): errors.append(f"{path}.raw must be a non-empty base64 string in JSON fixtures") elif "raw" in part: try: base64.b64decode(part["raw"], validate=True) except (binascii.Error, ValueError): errors.append(f"{path}.raw must be valid base64 in JSON fixtures") return errors def _validate_artifacts(artifacts: Any, path: str) -> list[str]: if not isinstance(artifacts, list) or not artifacts: return [f"{path} must be a non-empty list"] errors: list[str] = [] seen_ids: set[str] = set() for index, artifact in enumerate(artifacts): artifact_path = f"{path}[{index}]" if not isinstance(artifact, dict): errors.append(f"{artifact_path} must be an object") continue artifact_id = artifact.get("artifactId") if not _is_non_empty_string(artifact_id): errors.append(f"{artifact_path}.artifactId must be a non-empty string") elif artifact_id in seen_ids: errors.append(f"{artifact_path}.artifactId must be unique within the snapshot") else: seen_ids.add(artifact_id) parts = artifact.get("parts") if not isinstance(parts, list) or not parts: errors.append(f"{artifact_path}.parts must be a non-empty list") continue for part_index, part in enumerate(parts): errors.extend(validate_part(part, f"{artifact_path}.parts[{part_index}]")) return errors def validate_task_snapshots(snapshots: Any) -> list[str]: """Validate a local sequence policy over snapshots of one A2A Task.""" if not isinstance(snapshots, list) or not snapshots: return ["taskSnapshots must be a non-empty list"] errors: list[str] = [] expected_task_id: str | None = None expected_context_id: str | None = None previous_state: str | None = None for index, task in enumerate(snapshots): path = f"taskSnapshots[{index}]" if not isinstance(task, dict): errors.append(f"{path} must be an object") continue task_id = task.get("id") context_id = task.get("contextId") if not _is_non_empty_string(task_id): errors.append(f"{path}.id must be a non-empty string") if not _is_non_empty_string(context_id): errors.append(f"{path}.contextId must be a non-empty string") if index == 0: expected_task_id = task_id if _is_non_empty_string(task_id) else None expected_context_id = context_id if _is_non_empty_string(context_id) else None else: if expected_task_id is not None and task_id != expected_task_id: errors.append(f"{path}.id changed within one task snapshot sequence") if expected_context_id is not None and context_id != expected_context_id: errors.append(f"{path}.contextId changed within one task snapshot sequence") status = task.get("status") if not isinstance(status, dict): errors.append(f"{path}.status must be an object") previous_state = None continue state = status.get("state") if state not in TASK_STATES: errors.append(f"{path}.status.state must use an A2A 1.0 TASK_STATE_* enum") previous_state = None else: if state == "TASK_STATE_UNSPECIFIED": errors.append(f"{path}.status.state cannot be UNSPECIFIED in this local execution baseline") if previous_state in TERMINAL_STATES and state != previous_state: errors.append(f"{path}.status.state cannot transition away from terminal state {previous_state}") artifacts = task.get("artifacts") if state == "TASK_STATE_COMPLETED": errors.extend(_validate_artifacts(artifacts, f"{path}.artifacts")) elif artifacts is not None: errors.extend(_validate_artifacts(artifacts, f"{path}.artifacts")) previous_state = state if state in TASK_STATES else None return errors def validate_case(case: Any) -> list[str]: if not isinstance(case, dict): return ["case must be an object"] errors: list[str] = [] if "agentCard" in case: errors.extend(validate_agent_card(case["agentCard"])) if "taskSnapshots" in case: errors.extend(validate_task_snapshots(case["taskSnapshots"])) if "agentCard" not in case and "taskSnapshots" not in case: errors.append("case must contain agentCard and/or taskSnapshots") return errors def load_cases(path: Path) -> list[dict[str, Any]]: def reject_nonstandard_number(value: str) -> None: raise ValueError(f"fixture contains a non-standard JSON number: {value}") def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: result: dict[str, Any] = {} for key, value in pairs: if key in result: raise ValueError(f"fixture contains a duplicate JSON field: {key}") result[key] = value return result payload = json.loads( path.read_text(encoding="utf-8"), parse_constant=reject_nonstandard_number, object_pairs_hook=reject_duplicate_keys, ) cases = payload.get("cases") if isinstance(payload, dict) else None if not isinstance(cases, list) or not cases: raise ValueError("fixture must contain a non-empty 'cases' list") return cases def run_cases(cases: list[dict[str, Any]]) -> tuple[list[str], list[str]]: passed: list[str] = [] failed: list[str] = [] for index, case in enumerate(cases): name = case.get("name", f"case-{index}") if not _is_non_empty_string(name): name = f"case-{index}" errors = validate_case(case) expected_valid = case.get("expectedValid") expected_errors = case.get("expectedErrors", []) mismatch: list[str] = [] if not isinstance(expected_valid, bool): mismatch.append("expectedValid must be a boolean") elif expected_valid != (not errors): mismatch.append(f"expectedValid={expected_valid}, actualValid={not errors}") if not isinstance(expected_errors, list) or any( not _is_non_empty_string(item) for item in expected_errors ): mismatch.append("expectedErrors must be a list of non-empty strings") else: for fragment in expected_errors: if not any(fragment in error for error in errors): mismatch.append(f"missing expected error fragment: {fragment}") if mismatch: detail = "; ".join(mismatch) failed.append(f"{name}: {detail}; errors={errors}") else: passed.append(name) return passed, failed def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--cases", type=Path, required=True, help="Path to the JSON fixture") args = parser.parse_args() try: cases = load_cases(args.cases) passed, failed = run_cases(cases) except (OSError, ValueError, json.JSONDecodeError) as exc: print(f"ERROR: {exc}") return 2 for name in passed: print(f"PASS {name}") for detail in failed: print(f"FAIL {detail}") print(f"SUMMARY passed={len(passed)} failed={len(failed)}") return 1 if failed else 0 if __name__ == "__main__": raise SystemExit(main())