chore: Formatting/Lint fixes

Author: holtskinner

examples/google_adk/birthday_planner/__main__.py

@@ -5,6 +5,7 @@
 
 import click
 import uvicorn
+
 from adk_agent_executor import ADKAgentExecutor
 from dotenv import load_dotenv
 
@@ -18,6 +19,7 @@
     AgentSkill,
 )
 
+
 load_dotenv()
 
 logging.basicConfig()
@@ -39,11 +41,12 @@ def wrapper(*args, **kwargs):
 )
 def main(host: str, port: int, calendar_agent: str):
     # Verify an API key is set. Not required if using Vertex AI APIs, since those can use gcloud credentials.
-    if not os.getenv('GOOGLE_GENAI_USE_VERTEXAI') == 'TRUE':
-        if not os.getenv('GOOGLE_API_KEY'):
-            raise Exception(
-                'GOOGLE_API_KEY environment variable not set and GOOGLE_GENAI_USE_VERTEXAI is not TRUE.'
-            )
+    if os.getenv('GOOGLE_GENAI_USE_VERTEXAI') != 'TRUE' and not os.getenv(
+        'GOOGLE_API_KEY'
+    ):
+        raise ValueError(
+            'GOOGLE_API_KEY environment variable not set and GOOGLE_GENAI_USE_VERTEXAI is not TRUE.'
+        )
 
     skill = AgentSkill(
         id='plan_parties',

examples/google_adk/birthday_planner/adk_agent_executor.py

@@ -1,10 +1,12 @@
 import asyncio
 import logging
-from collections.abc import AsyncGenerator
-from typing import Any, AsyncIterable
+
+from collections.abc import AsyncGenerator, AsyncIterable
+from typing import Any
 from uuid import uuid4
 
 import httpx
+
 from google.adk import Runner
 from google.adk.agents import LlmAgent, RunConfig
 from google.adk.artifacts import InMemoryArtifactService
@@ -42,14 +44,15 @@
 from a2a.utils import get_text_parts
 from a2a.utils.errors import ServerError
 
+
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.DEBUG)
 
 AUTH_TASK_POLLING_DELAY_SECONDS = 0.2
 
 
 class A2ARunConfig(RunConfig):
-    """Custom override of ADK RunConfig to smuggle extra data through the event loop"""
+    """Custom override of ADK RunConfig to smuggle extra data through the event loop."""
 
     model_config = ConfigDict(
         arbitrary_types_allowed=True,
@@ -66,7 +69,7 @@ def __init__(self, calendar_agent_url):
             name='birthday_planner_agent',
             description='An agent that helps manage birthday parties.',
             after_tool_callback=self._handle_auth_required_task,
-            instruction=f"""
+            instruction="""
     You are an agent that helps plan birthday parties.
 
     Your job as a party planner is to act as a sounding board and idea generator for
@@ -111,7 +114,7 @@ async def _handle_auth_required_task(
         tool_context: ToolContext,
         tool_response: dict,
     ) -> dict | None:
-        """Handle requests that return auth-required"""
+        """Handle requests that return auth-required."""
         if tool.name != 'message_calendar_agent':
             return None
         if not tool_context.state.get('task_suspended'):
@@ -165,7 +168,7 @@ async def _process_request(
                 task_updater.add_artifact(response)
                 task_updater.complete()
                 break
-            elif calls := event.get_function_calls():
+            if calls := event.get_function_calls():
                 for call in calls:
                     # Provide an update on what we're doing.
                     if call.name == 'message_calendar_agent':
@@ -314,23 +317,21 @@ def convert_a2a_part_to_genai(part: Part) -> types.Part:
     part = part.root
     if isinstance(part, TextPart):
         return types.Part(text=part.text)
-    elif isinstance(part, FilePart):
+    if isinstance(part, FilePart):
         if isinstance(part.file, FileWithUri):
             return types.Part(
                 file_data=types.FileData(
                     file_uri=part.file.uri, mime_type=part.file.mime_type
                 )
             )
-        elif isinstance(part.file, FileWithBytes):
+        if isinstance(part.file, FileWithBytes):
             return types.Part(
                 inline_data=types.Blob(
                     data=part.file.bytes, mime_type=part.file.mime_type
                 )
             )
-        else:
-            raise ValueError(f'Unsupported file type: {type(part.file)}')
-    else:
-        raise ValueError(f'Unsupported part type: {type(part)}')
+        raise ValueError(f'Unsupported file type: {type(part.file)}')
+    raise ValueError(f'Unsupported part type: {type(part)}')
 
 
 def convert_genai_parts_to_a2a(parts: list[types.Part]) -> list[Part]:
@@ -346,14 +347,14 @@ def convert_genai_part_to_a2a(part: types.Part) -> Part:
     """Convert a single Google GenAI Part type into an A2A Part type."""
     if part.text:
         return TextPart(text=part.text)
-    elif part.file_data:
+    if part.file_data:
         return FilePart(
             file=FileWithUri(
                 uri=part.file_data.file_uri,
                 mime_type=part.file_data.mime_type,
             )
         )
-    elif part.inline_data:
+    if part.inline_data:
         return Part(
             root=FilePart(
                 file=FileWithBytes(
@@ -362,5 +363,4 @@ def convert_genai_part_to_a2a(part: types.Part) -> Part:
                 )
             )
         )
-    else:
-        raise ValueError(f'Unsupported part type: {part}')
+    raise ValueError(f'Unsupported part type: {part}')

src/a2a/server/request_handlers/default_request_handler.py

@@ -1,6 +1,6 @@
 import asyncio
-import contextlib
 import logging
+
 from collections.abc import AsyncGenerator
 from typing import cast
 
@@ -10,7 +10,6 @@
     EventConsumer,
     EventQueue,
     InMemoryQueueManager,
-    NoTaskQueue,
     QueueManager,
     TaskQueueExists,
 )
@@ -29,6 +28,7 @@
 )
 from a2a.utils.errors import ServerError
 
+
 logger = logging.getLogger(__name__)
 
 
@@ -206,11 +206,11 @@ async def on_message_send_stream(
         finally:
             await self._cleanup_producer(producer_task, task_id)
 
-    async def _register_producer(self, task_id, producer_task):
+    async def _register_producer(self, task_id, producer_task) -> None:
         async with self._running_agents_lock:
             self._running_agents[task_id] = producer_task
 
-    async def _cleanup_producer(self, producer_task, task_id):
+    async def _cleanup_producer(self, producer_task, task_id) -> None:
         await producer_task
         await self._queue_manager.close(task_id)
         async with self._running_agents_lock:

src/a2a/server/tasks/result_aggregator.py

@@ -1,12 +1,13 @@
 import asyncio
 import logging
+
 from collections.abc import AsyncGenerator, AsyncIterator
-from typing import Tuple
 
 from a2a.server.events import Event, EventConsumer
 from a2a.server.tasks.task_manager import TaskManager
 from a2a.types import Message, Task, TaskState, TaskStatusUpdateEvent
 
+
 logger = logging.getLogger(__name__)
 
 
@@ -54,7 +55,7 @@ async def consume_all(
 
     async def consume_and_break_on_interrupt(
         self, consumer: EventConsumer
-    ) -> Tuple[Task | Message | None, bool]:
+    ) -> tuple[Task | Message | None, bool]:
         """Process the event stream until completion or an interruptable state is encountered."""
         event_stream = consumer.consume_all()
         interrupted = False
@@ -64,7 +65,7 @@ async def consume_and_break_on_interrupt(
                 return event, False
             await self.task_manager.process(event)
             if (
-                isinstance(event, (Task, TaskStatusUpdateEvent))
+                isinstance(event, Task | TaskStatusUpdateEvent)
                 and event.status.state == TaskState.auth_required
             ):
                 # auth-required is a special state: the message should be
@@ -82,16 +83,8 @@ async def consume_and_break_on_interrupt(
                 break
         return await self.task_manager.get_task(), interrupted
 
-    async def _continue_consuming(self, event_stream: AsyncIterator[Event]):
+    async def _continue_consuming(
+        self, event_stream: AsyncIterator[Event]
+    ) -> None:
         async for event in event_stream:
             await self.task_manager.process(event)
-
-    # async def consume_and_emit_task(
-    #     self, consumer: EventConsumer
-    # ) -> AsyncGenerator[Event, None]:
-    #     """Processes the event stream and emits the current state of the task."""
-    #     async for event in consumer.consume_all():
-    #         if isinstance(event, Message):
-    #             self._current_task_or_message = event
-    #             break
-    #         yield await self.task_manager.process(event)

src/a2a/utils/helpers.py

@@ -92,7 +92,7 @@ def decorator(function):
         def wrapper(self, *args, **kwargs):
             if not expression(self):
                 if not error_message:
-                    message = str(expression)
+                    error_message = str(expression)
                 logger.error(f'Unsupported Operation: {error_message}')
                 raise ServerError(
                     UnsupportedOperationError(message=error_message)