Check the python format for adk samples

.github/workflows/python_samples_build.yml

@@ -45,6 +45,10 @@ jobs:
           python -m pip install --upgrade pip
           pip install uv
 
+      - name: Check Formatting
+        working-directory: samples/agent/adk
+        run: uv run pyink --check .
+
       - name: Build contact_lookup
         working-directory: samples/agent/adk/contact_lookup
         run: uv build .

samples/agent/adk/component_gallery/__main__.py

@@ -1,5 +1,5 @@
-
 """Main entry point for the Component Gallery agent."""
+
 import logging
 import os
 import sys
@@ -29,72 +29,74 @@
 
 logger = logging.getLogger(__name__)
 
+
 @click.command()
 @click.option("--host", default="localhost")
 @click.option("--port", default=10005)
 def main(host, port):
-    try:
-        capabilities = AgentCapabilities(
-            streaming=True,
-            extensions=[get_a2ui_agent_extension()],
-        )
-
-        # Skill definition
-        skill = AgentSkill(
-            id="component_gallery",
-            name="Component Gallery",
-            description="Demonstrates A2UI components.",
-            tags=["gallery", "demo"],
-            examples=["Show me the gallery"],
-        )
-
-        base_url = f"http://{host}:{port}"
-
-        agent_card = AgentCard(
-            name="Component Gallery Agent",
-            description="A2UI Component Gallery",
-            url=base_url,
-            version="0.0.1",
-            default_input_modes=["text"],
-            default_output_modes=["text"],
-            capabilities=capabilities,
-            skills=[skill],
-        )
-
-        agent_executor = ComponentGalleryExecutor(base_url=base_url)
-
-        request_handler = DefaultRequestHandler(
-            agent_executor=agent_executor,
-            task_store=InMemoryTaskStore(),
-        )
-
-        server = A2AStarletteApplication(
-            agent_card=agent_card, http_handler=request_handler
-        )
-
-        app = server.build()
-
-        app.add_middleware(
-            CORSMiddleware,
-            allow_origins=["http://localhost:5173"],
-            allow_credentials=True,
-            allow_methods=["*"],
-            allow_headers=["*"],
-        )
-
-        # Mount assets directory
-        assets_dir = os.path.join(os.path.dirname(__file__), "assets")
-        if os.path.exists(assets_dir):
-            app.mount("/assets", StaticFiles(directory=assets_dir), name="assets")
-        else:
-            logger.warning(f"Assets directory not found at {assets_dir}")
-
-        print(f"Starting Component Gallery Agent on port {port}...")
-        uvicorn.run(app, host=host, port=port)
-
-    except Exception as e:
-        logger.error(f"An error occurred during server startup: {e}")
-        exit(1)
+  try:
+    capabilities = AgentCapabilities(
+        streaming=True,
+        extensions=[get_a2ui_agent_extension()],
+    )
+
+    # Skill definition
+    skill = AgentSkill(
+        id="component_gallery",
+        name="Component Gallery",
+        description="Demonstrates A2UI components.",
+        tags=["gallery", "demo"],
+        examples=["Show me the gallery"],
+    )
+
+    base_url = f"http://{host}:{port}"
+
+    agent_card = AgentCard(
+        name="Component Gallery Agent",
+        description="A2UI Component Gallery",
+        url=base_url,
+        version="0.0.1",
+        default_input_modes=["text"],
+        default_output_modes=["text"],
+        capabilities=capabilities,
+        skills=[skill],
+    )
+
+    agent_executor = ComponentGalleryExecutor(base_url=base_url)
+
+    request_handler = DefaultRequestHandler(
+        agent_executor=agent_executor,
+        task_store=InMemoryTaskStore(),
+    )
+
+    server = A2AStarletteApplication(
+        agent_card=agent_card, http_handler=request_handler
+    )
+
+    app = server.build()
+
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["http://localhost:5173"],
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    # Mount assets directory
+    assets_dir = os.path.join(os.path.dirname(__file__), "assets")
+    if os.path.exists(assets_dir):
+      app.mount("/assets", StaticFiles(directory=assets_dir), name="assets")
+    else:
+      logger.warning(f"Assets directory not found at {assets_dir}")
+
+    print(f"Starting Component Gallery Agent on port {port}...")
+    uvicorn.run(app, host=host, port=port)
+
+  except Exception as e:
+    logger.error(f"An error occurred during server startup: {e}")
+    exit(1)
+
 
 if __name__ == "__main__":
-    main()
+  main()

samples/agent/adk/component_gallery/a2ui_schema.py

@@ -15,7 +15,7 @@
 
 # a2ui_schema.py
 
-A2UI_SCHEMA = r'''
+A2UI_SCHEMA = r"""
 {
   "title": "A2UI Message Schema",
   "description": "Describes a JSON payload for an A2UI (Agent to UI) message, which is used to dynamically construct and update user interfaces. A message MUST contain exactly ONE of the action properties: 'beginRendering', 'surfaceUpdate', 'dataModelUpdate', or 'deleteSurface'.",
@@ -789,4 +789,4 @@
     }
   }
 }
-'''
+"""

samples/agent/adk/component_gallery/agent.py

@@ -1,5 +1,5 @@
-
 """Agent logic for the Component Gallery."""
+
 import logging
 import json
 from collections.abc import AsyncIterable
@@ -12,69 +12,67 @@
 
 logger = logging.getLogger(__name__)
 
+
 class ComponentGalleryAgent:
-    """An agent that displays a component gallery."""
-
-    def __init__(self, base_url: str):
-        self.base_url = base_url
-
-    async def stream(self, query: str, session_id: str) -> AsyncIterable[dict[str, Any]]:
-        """Streams the gallery or responses to actions."""
-
-        logger.info(f"Stream called with query: {query}")
-
-        # Initial Load or Reset
-        if "WHO_ARE_YOU" in query or "START" in query: # Simple trigger for initial load
-             gallery_json = get_gallery_json()
-             yield {
-                "is_task_complete": True,
-                "payload": {
-                    "text": "Here is the component gallery.",
-                    "json_string": gallery_json
-                }
-             }
-             return
-
-        # Handle Actions
-        if query.startswith("ACTION:"):
-             action_name = query
-             # Create a response update for the second surface
-
-             # Simulate network/processing delay
-             await asyncio.sleep(0.5)
-
-             timestamp = datetime.datetime.now().strftime("%H:%M:%S")
-
-             response_update = [
-                 {
-                     "surfaceUpdate": {
-                         "surfaceId": "response-surface",
-                         "components": [
-                             {
-                                 "id": "response-text",
-                                 "component": {
-                                     "Text": { "text": { "literalString": f"Agent Processed Action: {action_name} at {timestamp}" } }
-                                 }
-                             }
-                         ]
-                     }
-                 }
-             ]
-
-
-             yield {
-                "is_task_complete": True,
-                "payload": {
-                    "text": "Action processed.",
-                    "json_data": response_update
-                }
-             }
-             return
-
-        # Fallback for text
-        yield {
-             "is_task_complete": True,
-             "payload": {
-                 "text": "I am the Component Gallery Agent."
-             }
-        }
+  """An agent that displays a component gallery."""
+
+  def __init__(self, base_url: str):
+    self.base_url = base_url
+
+  async def stream(self, query: str, session_id: str) -> AsyncIterable[dict[str, Any]]:
+    """Streams the gallery or responses to actions."""
+
+    logger.info(f"Stream called with query: {query}")
+
+    # Initial Load or Reset
+    if "WHO_ARE_YOU" in query or "START" in query:  # Simple trigger for initial load
+      gallery_json = get_gallery_json()
+      yield {
+          "is_task_complete": True,
+          "payload": {
+              "text": "Here is the component gallery.",
+              "json_string": gallery_json,
+          },
+      }
+      return
+
+    # Handle Actions
+    if query.startswith("ACTION:"):
+      action_name = query
+      # Create a response update for the second surface
+
+      # Simulate network/processing delay
+      await asyncio.sleep(0.5)
+
+      timestamp = datetime.datetime.now().strftime("%H:%M:%S")
+
+      response_update = [{
+          "surfaceUpdate": {
+              "surfaceId": "response-surface",
+              "components": [{
+                  "id": "response-text",
+                  "component": {
+                      "Text": {
+                          "text": {
+                              "literalString": (
+                                  f"Agent Processed Action: {action_name} at"
+                                  f" {timestamp}"
+                              )
+                          }
+                      }
+                  },
+              }],
+          }
+      }]
+
+      yield {
+          "is_task_complete": True,
+          "payload": {"text": "Action processed.", "json_data": response_update},
+      }
+      return
+
+    # Fallback for text
+    yield {
+        "is_task_complete": True,
+        "payload": {"text": "I am the Component Gallery Agent."},
+    }