feat: add ImageLoader for Azure blob image processing and refactor poster validation logic

Author: bmoussaud

infra/main.bicep

@@ -651,3 +651,4 @@ output APIM_SERVICE_NAME string = apiManagement.name
 output TMDB_ENDPOINT string = 'https://${apiManagement.outputs.apiManagementProxyHostName}'
 output STORAGE_ACCOUNT_BLOB_URL string = storageAccountAzureRambi.outputs.primaryBlobEndpoint
 output AZURE_AI_PROJECT_ENDPOINT string = aiFoundryProject.outputs.projectEndpoint
+output AZURE_CLIENT_ID string = azrStorageContributor.properties.clientId

src/movie_poster_agent_svc/ai_tools.py

@@ -0,0 +1,121 @@
+from io import BytesIO
+from agent_framework import AgentRunResponse, ChatAgent, ChatMessage, ai_function
+from agent_framework.openai import OpenAIResponsesClient
+
+import asyncio
+import os
+import httpx
+import logging
+import base64
+from urllib.parse import urlparse
+from random import randrange
+from typing import TYPE_CHECKING, Annotated, Any
+from azure.storage.blob import BlobServiceClient
+from PIL import Image
+from azure.identity import ManagedIdentityCredential, DefaultAzureCredential
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+
+@ai_function
+def get_image_content(url: Annotated[str, "url to image stored in Azure Blob Storage"]) -> str:
+    """Get the current weather for a given location."""
+    # Simulate weather data
+    client_id = os.getenv("AZURE_CLIENT_ID", None)
+    return ImageLoader(client_id).encode_image_from_url(url)
+
+
+class ImageLoader:
+    def __init__(self, client_id: str = None):
+        # Use managed identity for blob storage access
+        if client_id:
+            logger.info(f"Using managed identity {client_id} for blob storage access")
+            self._credential = ManagedIdentityCredential(client_id=client_id)
+        else:
+            logger.info("Using default Azure credential for blob storage access")
+            self._credential = DefaultAzureCredential()
+        logger.info(f"Credential initialized: {self._credential}")
+        
+
+    def _blob_service_client(self, storage_account_url: str) -> str:
+        return BlobServiceClient(
+            account_url=storage_account_url,
+            credential=self._credential
+        )
+
+    def encode_image_from_url(self, image_url: str) -> str:
+        logger.info(f"Encoding image from URL: {image_url}")
+        """Encode image from URL to base64, using managed identity for Azure blob storage URLs."""
+        try:
+            # Check if this is an Azure blob storage URL
+            if self._is_azure_blob_url(image_url):
+                logger.info("Detected Azure blob storage URL, using authenticated access")
+                return self._encode_image_from_blob_url(image_url)
+            else:
+                # Regular HTTP URL - use direct access
+                logger.info("Using direct HTTP access for non-blob URL")
+                with httpx.Client() as client:
+                    response = client.get(image_url)
+                    response.raise_for_status()
+                    
+                    logger.info(f"Image fetched successfully: {len(response.content)} bytes")
+                    image = Image.open(BytesIO(response.content))
+                    
+                    # Convert to base64
+                    logger.info("Encoding image to base64")
+                    return base64.b64encode(response.content).decode('utf-8')
+        except Exception as e:
+            logger.error(f"Error encoding image from URL {image_url}: {str(e)}")
+            raise RuntimeError(f"Failed to process image from URL: {str(e)}") from e
+    
+    def _is_azure_blob_url(self, url: str) -> bool:
+        """Check if URL is an Azure blob storage URL."""
+        try:
+            parsed = urlparse(url)
+            logger.info(f"Parsed URL netloc: {parsed.netloc}")
+            return 'blob.core.windows.net' in parsed.netloc
+        except Exception:
+            return False
+    
+    def _encode_image_from_blob_url(self, blob_url: str) -> str:
+        """Encode image from Azure blob storage URL using managed identity."""
+        try:
+            # Parse the blob URL to extract container and blob name
+            parsed = urlparse(blob_url)
+            path_parts = parsed.path.lstrip('/').split('/')
+            logger.info(f"Parsed blob URL path parts: {path_parts}")
+
+            if len(path_parts) < 2:
+                raise RuntimeError("Invalid blob URL format: {blob_url}")
+            
+            container_name = path_parts[0]
+            blob_name = '/'.join(path_parts[1:])
+            
+            logger.info(f"Accessing blob: container={container_name}, blob={blob_name}")
+            
+            # Get blob client and download content
+            blob_client = self._blob_service_client(parsed.netloc).get_blob_client(
+                container=container_name,
+                blob=blob_name
+            )
+            
+            # Download blob content
+            with blob_client:
+                blob_data = blob_client.download_blob()
+                content = blob_data.readall()
+            
+            logger.info(f"Blob downloaded successfully: {len(content)} bytes")
+            
+            # Validate it's an image
+            image = Image.open(BytesIO(content))
+            
+            # Convert to base64
+            logger.info("Encoding blob image to base64")
+            return base64.b64encode(content).decode('utf-8')
+            
+        except Exception as e:
+            logger.error(f"Error accessing blob {blob_url}: {str(e)}")
+            raise RuntimeError(
+                f"Failed to access blob with managed identity: {str(e)}" 
+            ) from e
+    

src/movie_poster_agent_svc/main.py

@@ -24,13 +24,13 @@
 from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
 from agent_framework import ChatMessage, TextContent, UriContent, DataContent, Role
 from urllib.parse import urlparse
-
+from ai_tools import ImageLoader
 load_dotenv()
 
 # Configure logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
-logger.setLevel(logging.INFO    )
+logger.setLevel(logging.INFO)
 
 # Configure telemetry
 if os.getenv("APPLICATIONINSIGHTS_CONNECTION_STRING"):
@@ -73,35 +73,17 @@ class PosterValidationAgent:
 
     def __init__(self):
         """Initialize the validation agent."""
+        
         self.project_endpoint = os.getenv("AZURE_AI_PROJECT_ENDPOINT")
         self.model_deployment = os.getenv("AZURE_AI_MODEL_DEPLOYMENT", "gpt-4o")
-        
         if not self.project_endpoint:
             raise ValueError("AZURE_AI_PROJECT_ENDPOINT environment variable is required")
-        else:
-            logger.info(f"Initializing agent with endpoint: {self.project_endpoint}")
-            logger.info(f"Using model deployment: {self.model_deployment}")
 
-        # Initialize blob storage for image processing with managed identity
-        self.blob_service_client = None
-        self.storage_account_url = os.getenv("STORAGE_ACCOUNT_BLOB_URL")
-        
-        if self.storage_account_url:
-            # Use managed identity for blob storage access
-            client_id = os.getenv("AZURE_CLIENT_ID")
-            if client_id:
-                logger.info(f"Using managed identity {client_id} for blob storage access")
-                credential = ManagedIdentityCredential(client_id=client_id)
-            else:
-                logger.info("Using default Azure credential for blob storage access")
-                credential = DefaultAzureCredential()
-            
-            self.blob_service_client = BlobServiceClient(
-                account_url=self.storage_account_url,
-                credential=credential
-            )
-            logger.info(f"Blob service client initialized with URL: {self.storage_account_url}")
-    
+        logger.info(f"Initializing agent with endpoint: {self.project_endpoint}")
+        logger.info(f"Using model deployment: {self.model_deployment}")
+
+        self._image_loader = ImageLoader(os.getenv("AZURE_CLIENT_ID"))
+       
     async def create_agent(self) -> ChatAgent:
         """Create and configure the chat agent."""
         agent_instructions = """
@@ -229,7 +211,7 @@ async def _encode_image_from_blob_url(self, blob_url: str) -> str:
     async def validate_poster(self, request: PosterValidationRequest) -> PosterValidationResponse:
         """Validate a movie poster using the AI agent."""
         try:
-            image_base64 = await self.encode_image_from_url(request.poster_url)
+            image_base64 = self._image_loader.encode_image_from_url(request.poster_url) if request.poster_url else None
 
             async with await self.create_agent() as agent:
                 # Build the validation prompt

src/movie_poster_agent_svc/test_blob_access.py

@@ -7,7 +7,7 @@
 import logging
 from dotenv import load_dotenv
 from main import PosterValidationAgent, PosterValidationRequest
-
+from ai_tools import ImageLoader
 # Load environment variables
 load_dotenv()
 
@@ -19,11 +19,11 @@ async def test_validate_poster():
     """Test the updated validate_poster method."""
     try:
         # Initialize the agent
-        agent = PosterValidationAgent()
+        agent = ImageLoader()
 
         # Create a test request
         test_request = PosterValidationRequest(
-            poster_url="https://nazrambihxklazmdpap4s.blob.core.windows.net/movieposters/test_image.png",
+            poster_url="https://nazrambihxklazmdpap4s.blob.core.windows.net/movieposters/525_3170_Romance_87490.png",
             poster_description="A dramatic movie poster featuring action scenes",
             movie_title="Test Movie",
             movie_genre="Action",