perf: optimize logging performance and modernize string formatting (#411)

- Replace f-strings with `%` formatting in logging calls for lazy
evaluation
- Convert `.format()` to f-strings in `server/models.py` `__repr__`
methods
- Add `test_venv/` to `.gitignore`
- Improves performance by avoiding string formatting when log levels are
disabled

---------

Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: Holt Skinner <13262395+holtskinner@users.noreply.github.com>
Co-authored-by: Holt Skinner <holtskinner@google.com>

Author: 3 people

.gitignore

@@ -6,6 +6,7 @@ __pycache__
 .pytest_cache
 .ruff_cache
 .venv
+test_venv/
 coverage.xml
 .nox
-spec.json
+spec.json

.ruff.toml

@@ -30,7 +30,6 @@ ignore = [
     "ANN003",
     "ANN401",
     "TRY003",
-    "G004",
     "TRY201",
     "FIX002",
 ]

src/a2a/client/auth/interceptor.py

@@ -62,7 +62,9 @@ async def intercept(
                         ):
                             headers['Authorization'] = f'Bearer {credential}'
                             logger.debug(
-                                f"Added Bearer token for scheme '{scheme_name}' (type: {scheme_def.type})."
+                                "Added Bearer token for scheme '%s' (type: %s).",
+                                scheme_name,
+                                scheme_def.type,
                             )
                             http_kwargs['headers'] = headers
                             return request_payload, http_kwargs
@@ -74,7 +76,9 @@ async def intercept(
                         ):
                             headers['Authorization'] = f'Bearer {credential}'
                             logger.debug(
-                                f"Added Bearer token for scheme '{scheme_name}' (type: {scheme_def.type})."
+                                "Added Bearer token for scheme '%s' (type: %s).",
+                                scheme_name,
+                                scheme_def.type,
                             )
                             http_kwargs['headers'] = headers
                             return request_payload, http_kwargs
@@ -83,7 +87,8 @@ async def intercept(
                         case APIKeySecurityScheme(in_=In.header):
                             headers[scheme_def.name] = credential
                             logger.debug(
-                                f"Added API Key Header for scheme '{scheme_name}'."
+                                "Added API Key Header for scheme '%s'.",
+                                scheme_name,
                             )
                             http_kwargs['headers'] = headers
                             return request_payload, http_kwargs

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

@@ -233,9 +233,13 @@ def _generate_error_response(
         )
         logger.log(
             log_level,
-            f'Request Error (ID: {request_id}): '
-            f"Code={error_resp.error.code}, Message='{error_resp.error.message}'"
-            f'{", Data=" + str(error_resp.error.data) if error_resp.error.data else ""}',
+            "Request Error (ID: %s): Code=%s, Message='%s'%s",
+            request_id,
+            error_resp.error.code,
+            error_resp.error.message,
+            ', Data=' + str(error_resp.error.data)
+            if error_resp.error.data
+            else '',
         )
         return JSONResponse(
             error_resp.model_dump(mode='json', exclude_none=True),
@@ -422,7 +426,7 @@ async def _process_non_streaming_request(
                 )
             case _:
                 logger.error(
-                    f'Unhandled validated request type: {type(request_obj)}'
+                    'Unhandled validated request type: %s', type(request_obj)
                 )
                 error = UnsupportedOperationError(
                     message=f'Request type {type(request_obj).__name__} is unknown.'
@@ -497,8 +501,10 @@ async def _handle_get_agent_card(self, request: Request) -> JSONResponse:
         """
         if request.url.path == PREV_AGENT_CARD_WELL_KNOWN_PATH:
             logger.warning(
-                f"Deprecated agent card endpoint '{PREV_AGENT_CARD_WELL_KNOWN_PATH}' accessed. "
-                f"Please use '{AGENT_CARD_WELL_KNOWN_PATH}' instead. This endpoint will be removed in a future version."
+                "Deprecated agent card endpoint '%s' accessed. "
+                "Please use '%s' instead. This endpoint will be removed in a future version.",
+                PREV_AGENT_CARD_WELL_KNOWN_PATH,
+                AGENT_CARD_WELL_KNOWN_PATH,
             )
 
         card_to_serve = self.agent_card

src/a2a/server/events/event_consumer.py

@@ -62,7 +62,7 @@ async def consume_one(self) -> Event:
                 InternalError(message='Agent did not return any response')
             ) from e
 
-        logger.debug(f'Dequeued event of type: {type(event)} in consume_one.')
+        logger.debug('Dequeued event of type: %s in consume_one.', type(event))
 
         self.queue.task_done()
 
@@ -95,7 +95,7 @@ async def consume_all(self) -> AsyncGenerator[Event]:
                     self.queue.dequeue_event(), timeout=self._timeout
                 )
                 logger.debug(
-                    f'Dequeued event of type: {type(event)} in consume_all.'
+                    'Dequeued event of type: %s in consume_all.', type(event)
                 )
                 self.queue.task_done()
                 logger.debug(

src/a2a/server/events/event_queue.py

@@ -54,7 +54,7 @@ async def enqueue_event(self, event: Event) -> None:
                 logger.warning('Queue is closed. Event will not be enqueued.')
                 return
 
-        logger.debug(f'Enqueuing event of type: {type(event)}')
+        logger.debug('Enqueuing event of type: %s', type(event))
 
         # Make sure to use put instead of put_nowait to avoid blocking the event loop.
         await self.queue.put(event)
@@ -103,13 +103,13 @@ async def dequeue_event(self, no_wait: bool = False) -> Event:
             logger.debug('Attempting to dequeue event (no_wait=True).')
             event = self.queue.get_nowait()
             logger.debug(
-                f'Dequeued event (no_wait=True) of type: {type(event)}'
+                'Dequeued event (no_wait=True) of type: %s', type(event)
             )
             return event
 
         logger.debug('Attempting to dequeue event (waiting).')
         event = await self.queue.get()
-        logger.debug(f'Dequeued event (waited) of type: {type(event)}')
+        logger.debug('Dequeued event (waited) of type: %s', type(event))
         return event
 
     def task_done(self) -> None:
@@ -193,7 +193,9 @@ async def clear_events(self, clear_child_queues: bool = True) -> None:
                 while True:
                     event = self.queue.get_nowait()
                     logger.debug(
-                        f'Discarding unprocessed event of type: {type(event)}, content: {event}'
+                        'Discarding unprocessed event of type: %s, content: %s',
+                        type(event),
+                        event,
                     )
                     self.queue.task_done()
                     cleared_count += 1
@@ -211,7 +213,8 @@ async def clear_events(self, clear_child_queues: bool = True) -> None:
 
             if cleared_count > 0:
                 logger.debug(
-                    f'Cleared {cleared_count} unprocessed events from EventQueue.'
+                    'Cleared %d unprocessed events from EventQueue.',
+                    cleared_count,
                 )
 
         # Clear all child queues (lock released before awaiting child tasks)

src/a2a/server/models.py

@@ -147,14 +147,9 @@ def task_metadata(cls) -> Mapped[dict[str, Any] | None]:
     @override
     def __repr__(self) -> str:
         """Return a string representation of the task."""
-        repr_template = (
-            '<{CLS}(id="{ID}", context_id="{CTX_ID}", status="{STATUS}")>'
-        )
-        return repr_template.format(
-            CLS=self.__class__.__name__,
-            ID=self.id,
-            CTX_ID=self.context_id,
-            STATUS=self.status,
+        return (
+            f'<{self.__class__.__name__}(id="{self.id}", '
+            f'context_id="{self.context_id}", status="{self.status}")>'
         )
 
 
@@ -188,12 +183,9 @@ class TaskModel(TaskMixin, base):  # type: ignore
         @override
         def __repr__(self) -> str:
             """Return a string representation of the task."""
-            repr_template = '<TaskModel[{TABLE}](id="{ID}", context_id="{CTX_ID}", status="{STATUS}")>'
-            return repr_template.format(
-                TABLE=table_name,
-                ID=self.id,
-                CTX_ID=self.context_id,
-                STATUS=self.status,
+            return (
+                f'<TaskModel[{table_name}](id="{self.id}", '
+                f'context_id="{self.context_id}", status="{self.status}")>'
             )
 
     # Set a dynamic name for better debugging
@@ -221,11 +213,9 @@ class PushNotificationConfigMixin:
     @override
     def __repr__(self) -> str:
         """Return a string representation of the push notification config."""
-        repr_template = '<{CLS}(task_id="{TID}", config_id="{CID}")>'
-        return repr_template.format(
-            CLS=self.__class__.__name__,
-            TID=self.task_id,
-            CID=self.config_id,
+        return (
+            f'<{self.__class__.__name__}(task_id="{self.task_id}", '
+            f'config_id="{self.config_id}")>'
         )
 
 
@@ -241,11 +231,9 @@ class PushNotificationConfigModel(PushNotificationConfigMixin, base):  # type: i
         @override
         def __repr__(self) -> str:
             """Return a string representation of the push notification config."""
-            repr_template = '<PushNotificationConfigModel[{TABLE}](task_id="{TID}", config_id="{CID}")>'
-            return repr_template.format(
-                TABLE=table_name,
-                TID=self.task_id,
-                CID=self.config_id,
+            return (
+                f'<PushNotificationConfigModel[{table_name}]('
+                f'task_id="{self.task_id}", config_id="{self.config_id}")>'
             )
 
     PushNotificationConfigModel.__name__ = (

src/a2a/server/request_handlers/default_request_handler.py

@@ -244,7 +244,9 @@ def _validate_task_id_match(self, task_id: str, event_task_id: str) -> None:
         """Validates that agent-generated task ID matches the expected task ID."""
         if task_id != event_task_id:
             logger.error(
-                f'Agent generated task_id={event_task_id} does not match the RequestContext task_id={task_id}.'
+                'Agent generated task_id=%s does not match the RequestContext task_id=%s.',
+                event_task_id,
+                task_id,
             )
             raise ServerError(
                 InternalError(message='Task ID mismatch in agent response')

src/a2a/server/tasks/base_push_notification_sender.py

@@ -44,7 +44,7 @@ async def send_notification(self, task: Task) -> None:
 
         if not all(results):
             logger.warning(
-                f'Some push notifications failed to send for task_id={task.id}'
+                'Some push notifications failed to send for task_id=%s', task.id
             )
 
     async def _dispatch_notification(
@@ -62,11 +62,13 @@ async def _dispatch_notification(
             )
             response.raise_for_status()
             logger.info(
-                f'Push-notification sent for task_id={task.id} to URL: {url}'
+                'Push-notification sent for task_id=%s to URL: %s', task.id, url
             )
         except Exception:
             logger.exception(
-                f'Error sending push-notification for task_id={task.id} to URL: {url}.'
+                'Error sending push-notification for task_id=%s to URL: %s.',
+                task.id,
+                url,
             )
             return False
         return True

src/a2a/server/tasks/database_push_notification_config_store.py

@@ -78,7 +78,8 @@ def __init__(
                 The key must be a URL-safe base64-encoded 32-byte key.
         """
         logger.debug(
-            f'Initializing DatabasePushNotificationConfigStore with existing engine, table: {table_name}'
+            'Initializing DatabasePushNotificationConfigStore with existing engine, table: %s',
+            table_name,
         )
         self.engine = engine
         self.async_session_maker = async_sessionmaker(
@@ -235,7 +236,9 @@ async def set_info(
         async with self.async_session_maker.begin() as session:
             await session.merge(db_config)
             logger.debug(
-                f'Push notification config for task {task_id} with config id {config_to_save.id} saved/updated.'
+                'Push notification config for task %s with config id %s saved/updated.',
+                task_id,
+                config_to_save.id,
             )
 
     async def get_info(self, task_id: str) -> list[PushNotificationConfig]:
@@ -280,9 +283,13 @@ async def delete_info(
 
             if result.rowcount > 0:
                 logger.info(
-                    f'Deleted {result.rowcount} push notification config(s) for task {task_id}.'
+                    'Deleted %s push notification config(s) for task %s.',
+                    result.rowcount,
+                    task_id,
                 )
             else:
                 logger.warning(
-                    f'Attempted to delete push notification config for task {task_id} with config_id: {config_id} that does not exist.'
+                    'Attempted to delete push notification config for task %s with config_id: %s that does not exist.',
+                    task_id,
+                    config_id,
                 )