ruff for python linting (#1255)

Co-authored-by: Taichiro Suzuki <taichirs@amazon.co.jp>

Author: tbrand

.github/workflows/python.yml

@@ -0,0 +1,56 @@
+# This workflow will install Python dependencies and run linting checks
+# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
+
+name: Python CI
+
+permissions:
+  contents: read
+
+on:
+  push:
+    branches: ['main']
+    paths: ['packages/cdk/lambda-python/**']
+  pull_request:
+    branches: ['main']
+    paths: ['packages/cdk/lambda-python/**']
+
+jobs:
+  lint:
+    name: 'Python Lint Check'
+    runs-on: ubuntu-latest
+
+    strategy:
+      matrix:
+        python-version: ['3.12']
+
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v4
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v3
+        with:
+          version: 'latest'
+
+      - name: Install dependencies
+        working-directory: packages/cdk/lambda-python/generic-agent-core-runtime
+        run: uv sync --group=lint
+
+      - name: Run ruff check
+        working-directory: packages/cdk/lambda-python/generic-agent-core-runtime
+        run: uv run ruff check --fix .
+
+      - name: Run ruff format
+        working-directory: packages/cdk/lambda-python/generic-agent-core-runtime
+        run: uv run ruff format .
+
+      - name: Check for changes
+        run: |
+          if [ -n "$(git status --porcelain)" ]; then
+            echo "Linting changes detected. Please run 'uv run ruff check --fix .' and 'uv run ruff format .' locally and commit the changes."
+            git diff
+            exit 1
+          fi

packages/cdk/lambda-python/generic-agent-core-runtime/.dockerignore

@@ -86,4 +86,7 @@ venv/
 **/*.swp
 
 # VS Code
-.vscode/
+.vscode/
+
+# Ruff
+.ruff_cache

packages/cdk/lambda-python/generic-agent-core-runtime/.gitignore

@@ -51,6 +51,7 @@ logs/
 tmp/
 temp/
 .tmp/
+.ruff_cache/
 
 # Docker
 .dockerignore

packages/cdk/lambda-python/generic-agent-core-runtime/app.py

@@ -3,11 +3,12 @@
 import json
 import logging
 import traceback
+
 from fastapi import FastAPI, Request
 from fastapi.responses import StreamingResponse
+
 from src.agent import AgentManager
-from src.utils import create_ws_directory, clean_ws_directory, create_error_response
-from src.types import AgentCoreRequest
+from src.utils import clean_ws_directory, create_error_response, create_ws_directory
 
 # Configure root logger
 logging.basicConfig(
@@ -71,12 +72,7 @@ async def invocations(request: Request):
         # Return streaming response
         async def generate():
             try:
-                async for chunk in agent_manager.process_request_streaming(
-                    messages=messages,
-                    system_prompt=system_prompt,
-                    prompt=prompt,
-                    model_info=model_info
-                ):
+                async for chunk in agent_manager.process_request_streaming(messages=messages, system_prompt=system_prompt, prompt=prompt, model_info=model_info):
                     yield chunk
             finally:
                 clean_ws_directory()
@@ -92,4 +88,5 @@ async def generate():
 
 if __name__ == "__main__":
     import uvicorn
+
     uvicorn.run(app, host="0.0.0.0", port=8080, log_level="warning", access_log=False)

packages/cdk/lambda-python/generic-agent-core-runtime/pyproject.toml

@@ -12,3 +12,28 @@ dependencies = [
   "pydantic",
   "aws-opentelemetry-distro>=0.10.1",
 ]
+
+[dependency-groups]
+lint = [
+  "ruff>=0.8.0",
+]
+
+[tool.ruff]
+target-version = "py313"
+line-length = 500
+
+[tool.ruff.lint]
+select = [
+    "E",  # pycodestyle errors
+    "W",  # pycodestyle warnings
+    "F",  # pyflakes
+    "I",  # isort
+    "B",  # flake8-bugbear
+    "C4", # flake8-comprehensions
+    "UP", # pyupgrade
+]
+ignore = []
+
+[tool.ruff.format]
+quote-style = "double"
+indent-style = "space"

packages/cdk/lambda-python/generic-agent-core-runtime/src/__init__.py

@@ -1 +1 @@
-# Empty init file for src package
+# Empty init file for src package

packages/cdk/lambda-python/generic-agent-core-runtime/src/agent.py

@@ -1,20 +1,18 @@
 """Agent management for the agent core runtime."""
 
-import boto3
 import json
 import logging
-from strands.models import BedrockModel
+from collections.abc import AsyncGenerator
+from typing import Any
+
+import boto3
 from strands import Agent as StrandsAgent
-from typing import List, Dict, Union, Any, Optional, AsyncGenerator
-from .config import get_system_prompt, extract_model_info
+from strands.models import BedrockModel
+
+from .config import extract_model_info, get_system_prompt
 from .tools import ToolManager
-from .utils import (
-    create_empty_response, 
-    create_error_response,
-    process_messages,
-    process_prompt
-)
-from .types import ModelInfo, Message
+from .types import Message, ModelInfo
+from .utils import process_messages, process_prompt
 
 logger = logging.getLogger(__name__)
 
@@ -31,22 +29,22 @@ def set_session_info(self, session_id: str, trace_id: str):
 
     async def process_request_streaming(
         self,
-        messages: Union[List[Message], List[Dict[str, Any]]],
-        system_prompt: Optional[str],
-        prompt: Union[str, List[Dict[str, Any]]],
+        messages: list[Message] | list[dict[str, Any]],
+        system_prompt: str | None,
+        prompt: str | list[dict[str, Any]],
         model_info: ModelInfo,
-    ) -> AsyncGenerator[str, None]:
+    ) -> AsyncGenerator[str]:
         """Process a request and yield streaming responses as raw events"""
         try:
             # Get model info
             model_id, region = extract_model_info(model_info)
-            
+
             # Combine system prompts
             combined_system_prompt = get_system_prompt(system_prompt)
-            
+
             # Get all tools
             tools = self.tool_manager.get_all_tools()
-            
+
             # Create boto3 session and Bedrock model
             session = boto3.Session(region_name=region)
             bedrock_model = BedrockModel(
@@ -55,11 +53,11 @@ async def process_request_streaming(
                 cache_prompt="default",
                 cache_tools="default",
             )
-            
+
             # Process messages and prompt using utility functions
             processed_messages = process_messages(messages)
             processed_prompt = process_prompt(prompt)
-            
+
             # Create Strands agent and stream response
             agent = StrandsAgent(
                 system_prompt=combined_system_prompt,

packages/cdk/lambda-python/generic-agent-core-runtime/src/config.py

@@ -1,8 +1,8 @@
 """Configuration and environment setup for the agent core runtime."""
 
-import os
 import logging
-from typing import Dict, Any
+import os
+from typing import Any
 
 # Configure root logger
 logging.basicConfig(
@@ -21,7 +21,7 @@
 """
 
 
-def get_aws_credentials() -> Dict[str, str]:
+def get_aws_credentials() -> dict[str, str]:
     """Get AWS credentials from environment or IAM role"""
     credentials = {}
 
@@ -37,7 +37,7 @@ def get_aws_credentials() -> Dict[str, str]:
     return credentials
 
 
-def get_uv_environment() -> Dict[str, str]:
+def get_uv_environment() -> dict[str, str]:
     """Get UV environment with AWS credentials"""
     aws_creds = get_aws_credentials()
     return {
@@ -62,14 +62,12 @@ def get_system_prompt(user_system_prompt: str = None) -> str:
 def extract_model_info(model_info: Any) -> tuple[str, str]:
     """Extract model ID and region from model info"""
     aws_creds = get_aws_credentials()
-    
+
     if isinstance(model_info, str):
         model_id = model_info
         region = aws_creds.get("AWS_REGION", "us-east-1")
     else:
-        model_id = model_info.get(
-            "modelId", "us.anthropic.claude-3-5-sonnet-20241022-v2:0"
-        )
+        model_id = model_info.get("modelId", "us.anthropic.claude-3-5-sonnet-20241022-v2:0")
         region = model_info.get("region", aws_creds.get("AWS_REGION", "us-east-1"))
 
-    return model_id, region
+    return model_id, region