Add lint fixes

Author: holtskinner

scripts/grpc_gen_post_processor.py

@@ -47,7 +47,7 @@ def process_generated_code(src_folder: str = 'src/a2a/grpc') -> None:
             else:
                 print('No changes needed')
 
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001
             print(f'Error processing file {file}: {e}')
             sys.exit(1)
 

src/a2a/extensions/__init__.py

@@ -317,8 +317,7 @@ async def _handle_requests(self, request: Request) -> Response:  # noqa: PLR0911
                 )
             raise e
         except Exception as e:
-            logger.exception(f'Unhandled exception: {e}')
-            traceback.print_exc()
+            logger.exception('Unhandled exception')
             return self._generate_error_response(
                 request_id, A2AError(root=InternalError(message=str(e)))
             )

src/a2a/server/apps/jsonrpc/jsonrpc_app.py

@@ -140,13 +140,11 @@ async def consume_all(self) -> AsyncGenerator[Event]:
                 # python 3.12 and get a queue empty error on an open queue
                 if self.queue.is_closed():
                     break
-            except ValidationError as e:
-                logger.exception(f'Invalid event format received: {e}')
+            except ValidationError:
+                logger.exception('Invalid event format received')
                 continue
             except Exception as e:
-                logger.exception(
-                    f'Stopping event consumption due to exception: {e}'
-                )
+                logger.exception('Stopping event consumption due to exception')
                 self._exception = e
                 continue
 

src/a2a/server/events/event_consumer.py

@@ -147,8 +147,9 @@ async def close(self) -> None:
         # Otherwise, join the queue
         else:
             tasks = [asyncio.create_task(self.queue.join())]
-            for child in self._children:
-                tasks.append(asyncio.create_task(child.close()))
+            tasks.extend(
+                asyncio.create_task(child.close()) for child in self._children
+            )
             await asyncio.wait(tasks, return_when=asyncio.ALL_COMPLETED)
 
     def is_closed(self) -> bool:

src/a2a/server/events/event_queue.py

@@ -302,8 +302,8 @@ async def on_message_send(
                 task_id, result_aggregator
             )
 
-        except Exception as e:
-            logger.exception(f'Agent execution failed. Error: {e}')
+        except Exception:
+            logger.exception('Agent execution failed')
             raise
         finally:
             if interrupted_or_non_blocking:
@@ -478,16 +478,12 @@ async def on_list_task_push_notification_config(
             params.id
         )
 
-        task_push_notification_config = []
-        if push_notification_config_list:
-            for config in push_notification_config_list:
-                task_push_notification_config.append(
-                    TaskPushNotificationConfig(
-                        task_id=params.id, push_notification_config=config
-                    )
-                )
-
-        return task_push_notification_config
+        return [
+            TaskPushNotificationConfig(
+                task_id=params.id, push_notification_config=config
+            )
+            for config in push_notification_config_list
+        ]
 
     async def on_delete_task_push_notification_config(
         self,

src/a2a/server/request_handlers/default_request_handler.py

@@ -64,9 +64,9 @@ async def _dispatch_notification(
             logger.info(
                 f'Push-notification sent for task_id={task.id} to URL: {url}'
             )
-            return True
-        except Exception as e:
+        except Exception:
             logger.exception(
-                f'Error sending push-notification for task_id={task.id} to URL: {url}. Error: {e}'
+                f'Error sending push-notification for task_id={task.id} to URL: {url}.'
             )
             return False
+        return True

src/a2a/server/tasks/base_push_notification_sender.py

@@ -252,12 +252,11 @@ async def get_info(self, task_id: str) -> list[PushNotificationConfig]:
             for model in models:
                 try:
                     configs.append(self._from_orm(model))
-                except ValueError as e:
+                except ValueError:
                     logger.exception(
-                        'Could not deserialize push notification config for task %s, config %s: %s',
+                        'Could not deserialize push notification config for task %s, config %s',
                         model.task_id,
                         model.config_id,
-                        e,
                     )
             return configs
 

src/a2a/server/tasks/database_push_notification_config_store.py

@@ -80,7 +80,7 @@ async def wrapper(*args: Any, **kwargs: Any) -> Response:
             return JSONResponse(
                 content={'message': error.message}, status_code=http_code
             )
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001
             logger.log(logging.ERROR, f'Unknown error occurred {e}')
             return JSONResponse(
                 content={'message': 'unknown exception'}, status_code=500

src/a2a/utils/error_handlers.py

@@ -338,16 +338,12 @@ def security(
     ) -> list[a2a_pb2.Security] | None:
         if not security:
             return None
-        rval: list[a2a_pb2.Security] = []
-        for s in security:
-            rval.append(
-                a2a_pb2.Security(
-                    schemes={
-                        k: a2a_pb2.StringList(list=v) for (k, v) in s.items()
-                    }
-                )
+        return [
+            a2a_pb2.Security(
+                schemes={k: a2a_pb2.StringList(list=v) for (k, v) in s.items()}
             )
-        return rval
+            for s in security
+        ]
 
     @classmethod
     def security_schemes(
@@ -774,10 +770,9 @@ def security(
     ) -> list[dict[str, list[str]]] | None:
         if not security:
             return None
-        rval: list[dict[str, list[str]]] = []
-        for s in security:
-            rval.append({k: list(v.list) for (k, v) in s.schemes.items()})
-        return rval
+        return [
+            {k: list(v.list) for (k, v) in s.schemes.items()} for s in security
+        ]
 
     @classmethod
     def provider(