fix: make changes based on review comments.

- Move `canonicalize_agent_card` from `signing` to `helpers'
- Add `PyJWT` liberary to dev in pyproject.toml
- Remove `last_error` from `signature_verifier`
- Update tests

Author: sokoliva

pyproject.toml

@@ -35,8 +35,8 @@ grpc = ["grpcio>=1.60", "grpcio-tools>=1.60", "grpcio_reflection>=1.7.0"]
 telemetry = ["opentelemetry-api>=1.33.0", "opentelemetry-sdk>=1.33.0"]
 postgresql = ["sqlalchemy[asyncio,postgresql-asyncpg]>=2.0.0"]
 mysql = ["sqlalchemy[asyncio,aiomysql]>=2.0.0"]
-sqlite = ["sqlalchemy[asyncio,aiosqlite]>=2.0.0"]
 signing = ["PyJWT>=2.0.0"]
+sqlite = ["sqlalchemy[asyncio,aiosqlite]>=2.0.0"]
 
 sql = ["a2a-sdk[postgresql,mysql,sqlite]"]
 
@@ -88,6 +88,7 @@ style = "pep440"
 dev = [
   "datamodel-code-generator>=0.30.0",
   "mypy>=1.15.0",
+  "PyJWT>=2.0.0"
   "pytest>=8.3.5",
   "pytest-asyncio>=0.26.0",
   "pytest-cov>=6.1.1",

src/a2a/utils/helpers.py

@@ -2,13 +2,15 @@
 
 import functools
 import inspect
+import json
 import logging
 
 from collections.abc import Callable
 from typing import Any
 from uuid import uuid4
 
 from a2a.types import (
+    AgentCard,
     Artifact,
     MessageSendParams,
     Part,
@@ -340,3 +342,29 @@ def are_modalities_compatible(
         return True
 
     return any(x in server_output_modes for x in client_output_modes)
+
+
+def _clean_empty(d: Any) -> Any:
+    """Recursively remove empty strings, lists and dicts from a dictionary."""
+    if isinstance(d, dict):
+        cleaned_dict: dict[Any, Any] = {
+            k: _clean_empty(v) for k, v in d.items()
+        }
+        return {k: v for k, v in cleaned_dict.items() if v}
+    if isinstance(d, list):
+        cleaned_list: list[Any] = [_clean_empty(v) for v in d]
+        return [v for v in cleaned_list if v]
+    return d if d not in ['', [], {}] else None
+
+
+def canonicalize_agent_card(agent_card: AgentCard) -> str:
+    """Canonicalizes the Agent Card JSON according to RFC 8785 (JCS)."""
+    card_dict = agent_card.model_dump(
+        exclude={'signatures'},
+        exclude_defaults=True,
+        exclude_none=True,
+        by_alias=True,
+    )
+    # Recursively remove empty values
+    cleaned_dict = _clean_empty(card_dict)
+    return json.dumps(cleaned_dict, separators=(',', ':'), sort_keys=True)

src/a2a/utils/signing.py

@@ -3,6 +3,8 @@
 from collections.abc import Callable
 from typing import Any, TypedDict
 
+from a2a.utils.helpers import canonicalize_agent_card
+
 
 try:
     import jwt
@@ -48,30 +50,6 @@ class ProtectedHeader(TypedDict):
     """
 
 
-def clean_empty(d: Any) -> Any:
-    """Recursively remove empty strings, lists and dicts from a dictionary."""
-    if isinstance(d, dict):
-        cleaned_dict: dict[Any, Any] = {k: clean_empty(v) for k, v in d.items()}
-        return {k: v for k, v in cleaned_dict.items() if v}
-    if isinstance(d, list):
-        cleaned_list: list[Any] = [clean_empty(v) for v in d]
-        return [v for v in cleaned_list if v]
-    return d if d not in ['', [], {}] else None
-
-
-def canonicalize_agent_card(agent_card: AgentCard) -> str:
-    """Canonicalizes the Agent Card JSON according to RFC 8785 (JCS)."""
-    card_dict = agent_card.model_dump(
-        exclude={'signatures'},
-        exclude_defaults=True,
-        exclude_none=True,
-        by_alias=True,
-    )
-    # Recursively remove empty values
-    cleaned_dict = clean_empty(card_dict)
-    return json.dumps(cleaned_dict, separators=(',', ':'), sort_keys=True)
-
-
 def create_agent_card_signer(
     signing_key: PyJWK | str | bytes,
     protected_header: ProtectedHeader,
@@ -141,7 +119,6 @@ def signature_verifier(
         if not agent_card.signatures:
             raise NoSignatureError('AgentCard has no signatures to verify.')
 
-        last_error = None
         for agent_card_signature in agent_card.signatures:
             try:
                 # get verification key
@@ -166,13 +143,10 @@ def signature_verifier(
                 )
                 # Found a valid signature, exit the loop and function
                 break
-            except PyJWTError as e:
-                last_error = e
+            except PyJWTError:
                 continue
         else:
             # This block runs only if the loop completes without a break
-            raise InvalidSignaturesError(
-                'No valid signature found'
-            ) from last_error
+            raise InvalidSignaturesError('No valid signature found')
 
     return signature_verifier

tests/utils/test_helpers.py

@@ -7,6 +7,10 @@
 
 from a2a.types import (
     Artifact,
+    AgentCard,
+    AgentCardSignature,
+    AgentCapabilities,
+    AgentSkill,
     Message,
     MessageSendParams,
     Part,
@@ -23,6 +27,7 @@
     build_text_artifact,
     create_task_obj,
     validate,
+    canonicalize_agent_card,
 )
 
 
@@ -45,6 +50,34 @@
     'type': 'task',
 }
 
+SAMPLE_AGENT_CARD: dict[str, Any] = {
+    'name': 'Test Agent',
+    'description': 'A test agent',
+    'url': 'http://localhost',
+    'version': '1.0.0',
+    'capabilities': AgentCapabilities(
+        streaming=None,
+        push_notifications=True,
+    ),
+    'default_input_modes': ['text/plain'],
+    'default_output_modes': ['text/plain'],
+    'documentation_url': None,
+    'icon_url': '',
+    'skills': [
+        AgentSkill(
+            id='skill1',
+            name='Test Skill',
+            description='A test skill',
+            tags=['test'],
+        )
+    ],
+    'signatures': [
+        AgentCardSignature(
+            protected='protected_header', signature='test_signature'
+        )
+    ],
+}
+
 
 # Test create_task_obj
 def test_create_task_obj():
@@ -328,3 +361,22 @@ def test_are_modalities_compatible_both_empty():
         )
         is True
     )
+
+
+def test_canonicalize_agent_card():
+    """Test canonicalize_agent_card with defaults, optionals, and exceptions.
+
+    - extensions is omitted as it's not set and optional.
+    - protocolVersion is included because it's always added by canonicalize_agent_card.
+    - signatures should be omitted.
+    """
+    agent_card = AgentCard(**SAMPLE_AGENT_CARD)
+    expected_jcs = (
+        '{"capabilities":{"pushNotifications":true},'
+        '"defaultInputModes":["text/plain"],"defaultOutputModes":["text/plain"],'
+        '"description":"A test agent","name":"Test Agent",'
+        '"skills":[{"description":"A test skill","id":"skill1","name":"Test Skill","tags":["test"]}],'
+        '"url":"http://localhost","version":"1.0.0"}'
+    )
+    result = canonicalize_agent_card(agent_card)
+    assert result == expected_jcs

tests/utils/test_signing.py

@@ -9,12 +9,7 @@
     AgentSkill,
     AgentCardSignature,
 )
-from a2a.utils.signing import (
-    canonicalize_agent_card,
-    create_agent_card_signer,
-    create_signature_verifier,
-    InvalidSignaturesError,
-)
+from a2a.utils import signing
 from typing import Any
 from jwt.utils import base64url_encode
 
@@ -63,7 +58,7 @@ def test_signer_and_verifier_symmetric(sample_agent_card: AgentCard):
     key = 'key12345'  # Using a simple symmetric key for HS256
     wrong_key = 'wrongkey'
 
-    agent_card_signer = create_agent_card_signer(
+    agent_card_signer = signing.create_agent_card_signer(
         signing_key=key,
         protected_header={
             'alg': 'HS384',
@@ -81,19 +76,19 @@ def test_signer_and_verifier_symmetric(sample_agent_card: AgentCard):
     assert signature.signature is not None
 
     # Verify the signature
-    verifier = create_signature_verifier(
+    verifier = signing.create_signature_verifier(
         create_key_provider(key), ['HS256', 'HS384', 'ES256', 'RS256']
     )
     try:
         verifier(signed_card)
-    except InvalidSignaturesError:
+    except signing.InvalidSignaturesError:
         pytest.fail('Signature verification failed with correct key')
 
     # Verify with wrong key
-    verifier_wrong_key = create_signature_verifier(
+    verifier_wrong_key = signing.create_signature_verifier(
         create_key_provider(wrong_key), ['HS256', 'HS384', 'ES256', 'RS256']
     )
-    with pytest.raises(InvalidSignaturesError):
+    with pytest.raises(signing.InvalidSignaturesError):
         verifier_wrong_key(signed_card)
 
 
@@ -111,7 +106,7 @@ def test_signer_and_verifier_symmetric_multiple_signatures(
     key = 'key12345'  # Using a simple symmetric key for HS256
     wrong_key = 'wrongkey'
 
-    agent_card_signer = create_agent_card_signer(
+    agent_card_signer = signing.create_agent_card_signer(
         signing_key=key,
         protected_header={
             'alg': 'HS384',
@@ -129,19 +124,19 @@ def test_signer_and_verifier_symmetric_multiple_signatures(
     assert signature.signature is not None
 
     # Verify the signature
-    verifier = create_signature_verifier(
+    verifier = signing.create_signature_verifier(
         create_key_provider(key), ['HS256', 'HS384', 'ES256', 'RS256']
     )
     try:
         verifier(signed_card)
-    except InvalidSignaturesError:
+    except signing.InvalidSignaturesError:
         pytest.fail('Signature verification failed with correct key')
 
     # Verify with wrong key
-    verifier_wrong_key = create_signature_verifier(
+    verifier_wrong_key = signing.create_signature_verifier(
         create_key_provider(wrong_key), ['HS256', 'HS384', 'ES256', 'RS256']
     )
-    with pytest.raises(InvalidSignaturesError):
+    with pytest.raises(signing.InvalidSignaturesError):
         verifier_wrong_key(signed_card)
 
 
@@ -156,7 +151,7 @@ def test_signer_and_verifier_asymmetric(sample_agent_card: AgentCard):
     )
     public_key_error = private_key_error.public_key()
 
-    agent_card_signer = create_agent_card_signer(
+    agent_card_signer = signing.create_agent_card_signer(
         signing_key=private_key,
         protected_header={
             'alg': 'ES256',
@@ -173,43 +168,18 @@ def test_signer_and_verifier_asymmetric(sample_agent_card: AgentCard):
     assert signature.protected is not None
     assert signature.signature is not None
 
-    verifier = create_signature_verifier(
+    verifier = signing.create_signature_verifier(
         create_key_provider(public_key), ['HS256', 'HS384', 'ES256', 'RS256']
     )
     try:
         verifier(signed_card)
-    except InvalidSignaturesError:
+    except signing.InvalidSignaturesError:
         pytest.fail('Signature verification failed with correct key')
 
     # Verify with wrong key
-    verifier_wrong_key = create_signature_verifier(
+    verifier_wrong_key = signing.create_signature_verifier(
         create_key_provider(public_key_error),
         ['HS256', 'HS384', 'ES256', 'RS256'],
     )
-    with pytest.raises(InvalidSignaturesError):
+    with pytest.raises(signing.InvalidSignaturesError):
         verifier_wrong_key(signed_card)
-
-
-def test_canonicalize_agent_card(
-    sample_agent_card: AgentCard,
-):
-    """Test canonicalize_agent_card with defaults, optionals, and exceptions.
-
-    - extensions is omitted as it's not set and optional.
-    - protocolVersion is included because it's always added by canonicalize_agent_card.
-    - signatures should be omitted.
-    """
-    sample_agent_card.signatures = [
-        AgentCardSignature(
-            protected='protected_header', signature='test_signature'
-        )
-    ]
-    expected_jcs = (
-        '{"capabilities":{"pushNotifications":true},'
-        '"defaultInputModes":["text/plain"],"defaultOutputModes":["text/plain"],'
-        '"description":"A test agent","name":"Test Agent",'
-        '"skills":[{"description":"A test skill","id":"skill1","name":"Test Skill","tags":["test"]}],'
-        '"url":"http://localhost","version":"1.0.0"}'
-    )
-    result = canonicalize_agent_card(sample_agent_card)
-    assert result == expected_jcs