chore(signing-and-veryfing): python sample A2A 1.0.x update

samples/python/agents/signing_and_verifying/README.md

@@ -6,16 +6,24 @@ Read more about signing and verifying AgentCards here: [Agent Card Signing](http
 
 ## Getting started
 
-1. Start the server
+1. Setup the virtual environment and install dependencies:
 
    ```bash
-   uv run .
+   python3 -m venv .venv
+   source .venv/bin/activate
+   pip install -r requirements.txt
    ```
 
-2. Run the test client
+2. Start the server:
 
    ```bash
-   uv run test_client.py
+   python3 __main__.py
+   ```
+
+3. Run the test client:
+
+   ```bash
+   python3 test_client.py
    ```
 
 ## Build Container Image
@@ -45,11 +53,16 @@ Agent can also be built using a container file.
 
 ## Validate
 
-To validate in a separate terminal, run the A2A client:
+To validate in a separate terminal, run the A2A CLI host:
 
 ```bash
-cd samples/python/hosts/cli
-uv run . --agent http://localhost:9999
+cd ../../hosts/cli
+# Setup the CLI host's environment if you haven't already:
+python3 -m venv .venv
+source .venv/bin/activate
+pip install -r requirements.txt
+# Run the CLI host pointing to our agent:
+python3 __main__.py --agent http://localhost:9999
 ```
 
 

samples/python/agents/signing_and_verifying/__main__.py

@@ -1,11 +1,15 @@
 import json
 
 from pathlib import Path
+from typing import Any
 
 import uvicorn
 
-from a2a.server.apps import A2AStarletteApplication
 from a2a.server.request_handlers import DefaultRequestHandler
+from a2a.server.routes import (
+    create_agent_card_routes,
+    create_jsonrpc_routes,
+)
 from a2a.server.tasks import InMemoryTaskStore
 from a2a.types import (
     AgentCapabilities,
@@ -18,116 +22,135 @@
     SignedAgentExecutor,
 )
 from cryptography.hazmat.primitives import asymmetric, serialization
+from starlette.applications import Starlette
 from starlette.responses import FileResponse
 from starlette.routing import Route
 
 
-if __name__ == '__main__':
-    # Generate a private, public key pair
+def create_public_private_keys() -> tuple[str, str]:
+    """Generate EC private and public key pair as PEM-encoded strings."""
     private_key = asymmetric.ec.generate_private_key(asymmetric.ec.SECP256R1())
-    public_key = private_key.public_key()
-
-    # Save public key to a file
-    pem = public_key.public_bytes(
+    private_pem = private_key.private_bytes(
         encoding=serialization.Encoding.PEM,
-        format=serialization.PublicFormat.SubjectPublicKeyInfo,
+        format=serialization.PrivateFormat.PKCS8,
+        encryption_algorithm=serialization.NoEncryption(),
     ).decode('utf-8')
-    kid = 'my-key'
-    keys = {kid: pem}
-    with Path('public_keys.json').open('w') as f:
-        json.dump(keys, f, indent=2)
-
-    skill = AgentSkill(
-        id='reminder',
-        name='Verification Reminder',
-        description='Reminds the user to verify the Agent Card.',
-        tags=['verify me'],
-        examples=['Verify me!'],
+    public_pem = (
+        private_key.public_key()
+        .public_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PublicFormat.SubjectPublicKeyInfo,
+        )
+        .decode('utf-8')
     )
+    return private_pem, public_pem
 
-    extended_skill = AgentSkill(
-        id='reminder-please',
-        name='Verification Reminder Please!',
-        description='Politely reminds user to verify the Agent Card.',
-        tags=['verify me', 'pretty please', 'extended'],
-        examples=['Verify me, pretty please! :)', 'Please verify me.'],
-    )
 
-    public_agent_card = AgentCard(
-        name='Signed Agent',
-        description='An Agent that is signed',
-        icon_url='http://localhost:9999/',
-        version='1.0.0',
-        default_input_modes=['text'],
-        default_output_modes=['text'],
-        capabilities=AgentCapabilities(
-            streaming=True, extended_agent_card=True
-        ),
-        supported_interfaces=[
-            AgentInterface(
-                protocol_binding='JSONRPC',
-                url='http://localhost:9999',
-            )
-        ],
-        skills=[skill],
-    )
+# Generate a private, public key pair
+private_key, public_key = create_public_private_keys()
 
-    extended_agent_card = AgentCard(
-        name='Signed Agent - Extended Edition',
-        description='The full-featured signed agent for authenticated users.',
-        icon_url='http://localhost:9999/',
-        version='1.0.1',
-        default_input_modes=['text'],
-        default_output_modes=['text'],
-        capabilities=AgentCapabilities(
-            streaming=True, extended_agent_card=True
-        ),
-        supported_interfaces=[
-            AgentInterface(
-                protocol_binding='JSONRPC',
-                url='http://localhost:9999',
-            )
-        ],
-        skills=[
-            skill,
-            extended_skill,
-        ],
-    )
+# Save public key to a file
+kid = 'my-key'
+keys = {kid: public_key}
+with Path('public_keys.json').open('w') as f:
+    json.dump(keys, f, indent=2)
 
-    request_handler = DefaultRequestHandler(
-        agent_executor=SignedAgentExecutor(),
-        task_store=InMemoryTaskStore(),
-    )
+skill = AgentSkill(
+    id='reminder',
+    name='Verification Reminder',
+    description='Reminds the user to verify the Agent Card.',
+    tags=['verify me'],
+    examples=['Verify me!'],
+)
 
-    # Create singer function which will be used for AgentCard signing
-    signer = create_agent_card_signer(
-        signing_key=private_key,
-        protected_header={
-            'kid': kid,
-            'alg': 'ES256',
-            'jku': 'http://localhost:9999/public_keys.json',
-        },
-    )
+extended_skill = AgentSkill(
+    id='reminder-please',
+    name='Verification Reminder Please!',
+    description='Politely reminds user to verify the Agent Card.',
+    tags=['verify me', 'pretty please', 'extended'],
+    examples=['Verify me, pretty please! :)', 'Please verify me.'],
+)
 
-    server = A2AStarletteApplication(
-        agent_card=public_agent_card,
-        http_handler=request_handler,
-        card_modifier=signer,  # The signer function is used to sign the public Agent Card
-        extended_agent_card=extended_agent_card,
-        extended_card_modifier=lambda card, _: signer(
-            card
-        ),  # The signer function is also used to sign the extended Agent Card
-    )
+public_agent_card = AgentCard(
+    name='Signed Agent',
+    description='An Agent that is signed',
+    icon_url='http://localhost:9999/',
+    version='1.0.0',
+    default_input_modes=['text'],
+    default_output_modes=['text'],
+    capabilities=AgentCapabilities(streaming=True, extended_agent_card=True),
+    supported_interfaces=[
+        AgentInterface(
+            protocol_binding='JSONRPC',
+            url='http://localhost:9999',
+        )
+    ],
+    skills=[skill],
+)
 
-    app = server.build()
-    # Expose the public key for verification purposes
-    # Contents of public_keys.json will be fetched on the client side during AgentCard signatures verification
-    app.routes.append(
-        Route(
-            '/public_keys.json',
-            endpoint=FileResponse('public_keys.json'),
-            methods=['GET'],
+extended_agent_card = AgentCard(
+    name='Signed Agent - Extended Edition',
+    description='The full-featured signed agent for authenticated users.',
+    icon_url='http://localhost:9999/',
+    version='1.0.1',
+    default_input_modes=['text'],
+    default_output_modes=['text'],
+    capabilities=AgentCapabilities(streaming=True, extended_agent_card=True),
+    supported_interfaces=[
+        AgentInterface(
+            protocol_binding='JSONRPC',
+            url='http://localhost:9999',
         )
+    ],
+    skills=[
+        skill,
+        extended_skill,
+    ],
+)
+
+# Create singer function which will be used for AgentCard signing
+signer = create_agent_card_signer(
+    signing_key=private_key,
+    protected_header={
+        'kid': kid,
+        'alg': 'ES256',
+        'jku': 'http://localhost:9999/public_keys.json',
+    },
+)
+
+
+async def async_signer(card: AgentCard) -> AgentCard:
+    """Sign the public agent card."""
+    return signer(card)
+
+
+async def async_extended_signer(card: AgentCard, _: Any) -> AgentCard:
+    """Sign the extended agent card."""
+    return signer(card)
+
+
+request_handler = DefaultRequestHandler(
+    agent_executor=SignedAgentExecutor(),
+    task_store=InMemoryTaskStore(),
+    agent_card=public_agent_card,
+    extended_agent_card=extended_agent_card,
+    extended_card_modifier=async_extended_signer,
+)
+
+routes = []
+routes.extend(create_agent_card_routes(public_agent_card, card_modifier=async_signer))
+routes.extend(create_jsonrpc_routes(request_handler, '/'))
+
+app = Starlette(routes=routes)
+# Expose the public key for verification purposes
+# Contents of public_keys.json will be fetched on the client side during AgentCard signatures verification
+app.routes.append(
+    Route(
+        '/public_keys.json',
+        endpoint=FileResponse('public_keys.json'),
+        methods=['GET'],
     )
+)
 
+if __name__ == '__main__':
     uvicorn.run(app, host='127.0.0.1', port=9999)

samples/python/agents/signing_and_verifying/agent_executor.py

@@ -1,6 +1,7 @@
+from a2a.helpers import new_text_message
 from a2a.server.agent_execution import AgentExecutor, RequestContext
 from a2a.server.events import EventQueue
-from a2a.utils import new_agent_text_message
+from a2a.types import Role
 
 
 class SignedAgentExecutor(AgentExecutor):
@@ -12,10 +13,8 @@ async def execute(
         event_queue: EventQueue,
     ) -> None:
         """Execute the agent."""
-        await event_queue.enqueue_event(new_agent_text_message('Verify me!'))
+        await event_queue.enqueue_event(new_text_message('Verify me!', role=Role.ROLE_AGENT))
 
-    async def cancel(
-        self, context: RequestContext, event_queue: EventQueue
-    ) -> None:
-        """Cancel method is not supported."""
-        print('Cancel not supported.')
+    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
+        """Raise exception as cancel is not supported."""
+        raise NotImplementedError('Cancel is not supported.')