chore: Update A2A CLI for SDK 0.2.16 (#259)

* Remove dependency on common for cli

* remove other usage of common

* Update to snake_case

Author: holtskinner

samples/python/hosts/cli/README.md

@@ -1,6 +1,6 @@
-## CLI
+# A2A CLI
 
-The CLI is a small host application that demonstrates the capabilities of an A2AClient. It supports reading a server's AgentCard and text-based collaboration with a remote agent. All content received from the A2A server is printed to the console. 
+The CLI is a small host application that demonstrates the capabilities of an `A2AClient`. It supports reading a server's `AgentCard` and text-based collaboration with a remote agent. All content received from the A2A server is printed to the console.
 
 The client will use streaming if the server supports it.
 
@@ -13,19 +13,23 @@ The client will use streaming if the server supports it.
 ## Running the CLI
 
 1. Navigate to the CLI sample directory:
+
     ```bash
     cd samples/python/hosts/cli
     ```
+
 2. Run the example client
-    ```
+
+    ```sh
     uv run . --agent [url-of-your-a2a-server]
     ```
 
-   for example `--agent http://localhost:10000`. More command line options are documented in the source code. 
+   for example `--agent https://sample-a2a-agent-908687846511.us-central1.run.app`. More command line options are documented in the source code.
 
 ## Disclaimer
+
 Important: The sample code provided is for demonstration purposes and illustrates the mechanics of the Agent-to-Agent (A2A) protocol. When building production applications, it is critical to treat any agent operating outside of your direct control as a potentially untrusted entity.
 
 All data received from an external agent—including but not limited to its AgentCard, messages, artifacts, and task statuses—should be handled as untrusted input. For example, a malicious agent could provide an AgentCard containing crafted data in its fields (e.g., description, name, skills.description). If this data is used without sanitization to construct prompts for a Large Language Model (LLM), it could expose your application to prompt injection attacks.  Failure to properly validate and sanitize this data before use can introduce security vulnerabilities into your application.
 
-Developers are responsible for implementing appropriate security measures, such as input validation and secure handling of credentials to protect their systems and users.
+Developers are responsible for implementing appropriate security measures, such as input validation and secure handling of credentials to protect their systems and users.

samples/python/hosts/cli/__main__.py

@@ -2,6 +2,7 @@
 import base64
 import os
 import urllib
+
 from uuid import uuid4
 
 import asyncclick as click
@@ -26,12 +27,16 @@
     TaskStatusUpdateEvent,
     TextPart,
 )
-from common.utils.push_notification_auth import PushNotificationReceiverAuth
+from push_notification_auth import PushNotificationReceiverAuth
 
 
 @click.command()
 @click.option('--agent', default='http://localhost:10000')
-@click.option('--bearer-token', help='Bearer token for authentication.', envvar='A2A_CLI_BEARER_TOKEN')
+@click.option(
+    '--bearer-token',
+    help='Bearer token for authentication.',
+    envvar='A2A_CLI_BEARER_TOKEN',
+)
 @click.option('--session', default=0)
 @click.option('--history', default=False)
 @click.option('--use_push_notifications', default=False)
@@ -88,7 +93,7 @@ async def cli(
 
         while continue_loop:
             print('=========  starting a new task ======== ')
-            continue_loop, _, taskId = await completeTask(
+            continue_loop, _, task_id = await completeTask(
                 client,
                 streaming,
                 use_push_notifications,
@@ -101,7 +106,7 @@ async def cli(
             if history and continue_loop:
                 print('========= history ======== ')
                 task_response = await client.get_task(
-                    {'id': taskId, 'historyLength': 10}
+                    {'id': task_id, 'historyLength': 10}
                 )
                 print(
                     task_response.model_dump_json(
@@ -116,8 +121,8 @@ async def completeTask(
     use_push_notifications: bool,
     notification_receiver_host: str,
     notification_receiver_port: int,
-    taskId,
-    contextId,
+    task_id,
+    context_id,
 ):
     prompt = click.prompt(
         '\nWhat do you want to send to the agent? (:q or quit to exit)'
@@ -128,9 +133,9 @@ async def completeTask(
     message = Message(
         role='user',
         parts=[TextPart(text=prompt)],
-        messageId=str(uuid4()),
-        taskId=taskId,
-        contextId=contextId,
+        message_id=str(uuid4()),
+        task_id=task_id,
+        context_id=context_id,
     )
 
     file_path = click.prompt(
@@ -155,7 +160,7 @@ async def completeTask(
         id=str(uuid4()),
         message=message,
         configuration=MessageSendConfiguration(
-            acceptedOutputModes=['text'],
+            accepted_output_modes=['text'],
         ),
     )
 
@@ -179,32 +184,39 @@ async def completeTask(
         )
         async for result in response_stream:
             if isinstance(result.root, JSONRPCErrorResponse):
-                print(f'Error: {result.root.error}, contextId: {contextId}, taskId: {taskId}')
-                return False, contextId, taskId
+                print(
+                    f'Error: {result.root.error}, context_id: {context_id}, task_id: {task_id}'
+                )
+                return False, context_id, task_id
             event = result.root.result
-            contextId = event.contextId
+            context_id = event.context_id
             if isinstance(event, Task):
-                taskId = event.id
+                task_id = event.id
             elif isinstance(event, TaskStatusUpdateEvent) or isinstance(
                 event, TaskArtifactUpdateEvent
             ):
-                taskId = event.taskId
-                if isinstance(event, TaskStatusUpdateEvent) and event.status.state == 'completed':
+                task_id = event.task_id
+                if (
+                    isinstance(event, TaskStatusUpdateEvent)
+                    and event.status.state == 'completed'
+                ):
                     task_completed = True
             elif isinstance(event, Message):
                 message = event
             print(f'stream event => {event.model_dump_json(exclude_none=True)}')
         # Upon completion of the stream. Retrieve the full task if one was made.
-        if taskId and not task_completed:
+        if task_id and not task_completed:
             taskResultResponse = await client.get_task(
                 GetTaskRequest(
                     id=str(uuid4()),
-                    params=TaskQueryParams(id=taskId),
+                    params=TaskQueryParams(id=task_id),
                 )
             )
             if isinstance(taskResultResponse.root, JSONRPCErrorResponse):
-                print(f'Error: {taskResultResponse.root.error}, contextId: {contextId}, taskId: {taskId}')
-                return False, contextId, taskId
+                print(
+                    f'Error: {taskResultResponse.root.error}, context_id: {context_id}, task_id: {task_id}'
+                )
+                return False, context_id, task_id
             taskResult = taskResultResponse.root.result
     else:
         try:
@@ -218,18 +230,18 @@ async def completeTask(
             event = event.root.result
         except Exception as e:
             print('Failed to complete the call', e)
-        if not contextId:
-            contextId = event.contextId
+        if not context_id:
+            context_id = event.context_id
         if isinstance(event, Task):
-            if not taskId:
-                taskId = event.id
+            if not task_id:
+                task_id = event.id
             taskResult = event
         elif isinstance(event, Message):
             message = event
 
     if message:
         print(f'\n{message.model_dump_json(exclude_none=True)}')
-        return True, contextId, taskId
+        return True, context_id, task_id
     if taskResult:
         # Don't print the contents of a file.
         task_content = taskResult.model_dump_json(
@@ -248,19 +260,23 @@ async def completeTask(
         ## if the result is that more input is required, loop again.
         state = TaskState(taskResult.status.state)
         if state.name == TaskState.input_required.name:
-            return await completeTask(
-                client,
-                streaming,
-                use_push_notifications,
-                notification_receiver_host,
-                notification_receiver_port,
-                taskId,
-                contextId,
-            ), contextId, taskId
+            return (
+                await completeTask(
+                    client,
+                    streaming,
+                    use_push_notifications,
+                    notification_receiver_host,
+                    notification_receiver_port,
+                    task_id,
+                    context_id,
+                ),
+                context_id,
+                task_id,
+            )
         ## task is complete
-        return True, contextId, taskId
+        return True, context_id, task_id
     ## Failure case, shouldn't reach
-    return True, contextId, taskId
+    return True, context_id, task_id
 
 
 if __name__ == '__main__':

samples/python/hosts/cli/push_notification_auth.py

@@ -0,0 +1,146 @@
+import hashlib
+import json
+import logging
+import time
+import uuid
+
+from typing import Any
+
+import httpx
+import jwt
+
+from jwcrypto import jwk
+from jwt import PyJWK, PyJWKClient
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+
+
+logger = logging.getLogger(__name__)
+AUTH_HEADER_PREFIX = 'Bearer '
+
+
+class PushNotificationAuth:
+    def _calculate_request_body_sha256(self, data: dict[str, Any]):
+        """Calculates the SHA256 hash of a request body.
+
+        This logic needs to be same for both the agent who signs the payload and the client verifier.
+        """
+        body_str = json.dumps(
+            data,
+            ensure_ascii=False,
+            allow_nan=False,
+            indent=None,
+            separators=(',', ':'),
+        )
+        return hashlib.sha256(body_str.encode()).hexdigest()
+
+
+class PushNotificationSenderAuth(PushNotificationAuth):
+    def __init__(self):
+        self.public_keys = []
+        self.private_key_jwk: PyJWK = None
+
+    @staticmethod
+    async def verify_push_notification_url(url: str) -> bool:
+        async with httpx.AsyncClient(timeout=10) as client:
+            try:
+                validation_token = str(uuid.uuid4())
+                response = await client.get(
+                    url, params={'validationToken': validation_token}
+                )
+                response.raise_for_status()
+                is_verified = response.text == validation_token
+
+                logger.info(
+                    f'Verified push-notification URL: {url} => {is_verified}'
+                )
+                return is_verified
+            except Exception as e:
+                logger.warning(
+                    f'Error during sending push-notification for URL {url}: {e}'
+                )
+
+        return False
+
+    def generate_jwk(self):
+        key = jwk.JWK.generate(
+            kty='RSA', size=2048, kid=str(uuid.uuid4()), use='sig'
+        )
+        self.public_keys.append(key.export_public(as_dict=True))
+        self.private_key_jwk = PyJWK.from_json(key.export_private())
+
+    def handle_jwks_endpoint(self, _request: Request):
+        """Allow clients to fetch public keys."""
+        return JSONResponse({'keys': self.public_keys})
+
+    def _generate_jwt(self, data: dict[str, Any]):
+        """JWT is generated by signing both the request payload SHA digest and time of token generation.
+
+        Payload is signed with private key and it ensures the integrity of payload for client.
+        Including iat prevents from replay attack.
+        """
+        iat = int(time.time())
+
+        return jwt.encode(
+            {
+                'iat': iat,
+                'request_body_sha256': self._calculate_request_body_sha256(
+                    data
+                ),
+            },
+            key=self.private_key_jwk,
+            headers={'kid': self.private_key_jwk.key_id},
+            algorithm='RS256',
+        )
+
+    async def send_push_notification(self, url: str, data: dict[str, Any]):
+        jwt_token = self._generate_jwt(data)
+        headers = {'Authorization': f'Bearer {jwt_token}'}
+        async with httpx.AsyncClient(timeout=10) as client:
+            try:
+                response = await client.post(url, json=data, headers=headers)
+                response.raise_for_status()
+                logger.info(f'Push-notification sent for URL: {url}')
+            except Exception as e:
+                logger.warning(
+                    f'Error during sending push-notification for URL {url}: {e}'
+                )
+
+
+class PushNotificationReceiverAuth(PushNotificationAuth):
+    def __init__(self):
+        self.public_keys_jwks = []
+        self.jwks_client = None
+
+    async def load_jwks(self, jwks_url: str):
+        self.jwks_client = PyJWKClient(jwks_url)
+
+    async def verify_push_notification(self, request: Request) -> bool:
+        auth_header = request.headers.get('Authorization')
+        if not auth_header or not auth_header.startswith(AUTH_HEADER_PREFIX):
+            print('Invalid authorization header')
+            return False
+
+        token = auth_header[len(AUTH_HEADER_PREFIX) :]
+        signing_key = self.jwks_client.get_signing_key_from_jwt(token)
+
+        decode_token = jwt.decode(
+            token,
+            signing_key,
+            options={'require': ['iat', 'request_body_sha256']},
+            algorithms=['RS256'],
+        )
+
+        actual_body_sha256 = self._calculate_request_body_sha256(
+            await request.json()
+        )
+        if actual_body_sha256 != decode_token['request_body_sha256']:
+            # Payload signature does not match the digest in signed token.
+            raise ValueError('Invalid request body')
+
+        if time.time() - decode_token['iat'] > 60 * 5:
+            # Do not allow push-notifications older than 5 minutes.
+            # This is to prevent replay attack.
+            raise ValueError('Token is expired')
+
+        return True

samples/python/hosts/cli/push_notification_listener.py

@@ -2,7 +2,7 @@
 import threading
 import traceback
 
-from common.utils.push_notification_auth import PushNotificationReceiverAuth
+from push_notification_auth import PushNotificationReceiverAuth
 from starlette.applications import Starlette
 from starlette.requests import Request
 from starlette.responses import Response

samples/python/hosts/cli/pyproject.toml

@@ -5,7 +5,7 @@ description = "A CLI application that demonstrates the capabilities of an A2ACli
 readme = "README.md"
 requires-python = ">=3.12"
 dependencies = [
-    "a2a-samples",
+    "a2a-sdk",
     "asyncclick>=8.1.8",
     "sse-starlette>=2.2.1",
     "starlette>=0.46.1",
@@ -14,9 +14,6 @@ dependencies = [
 [tool.hatch.build.targets.wheel]
 packages = ["."]
 
-[tool.uv.sources]
-a2a-samples = { workspace = true }
-
 [build-system]
 requires = ["hatchling"]
 build-backend = "hatchling.build"