Python: [Purview] Add Caching and background processing in Python Purview Middleware (#1844)

* [PythonPurview] Add Caching and background processing

* [PythonPurview] Updates based on comments

python/packages/purview/agent_framework_purview/__init__.py

@@ -1,7 +1,9 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+from ._cache import CacheProvider
 from ._exceptions import (
     PurviewAuthenticationError,
+    PurviewPaymentRequiredError,
     PurviewRateLimitError,
     PurviewRequestError,
     PurviewServiceError,
@@ -10,10 +12,12 @@ from ._middleware import PurviewChatPolicyMiddleware, PurviewPolicyMiddleware
 from ._settings import PurviewAppLocation, PurviewLocationType, PurviewSettings
 
 __all__ = [
+    "CacheProvider",
     "PurviewAppLocation",
     "PurviewAuthenticationError",
     "PurviewChatPolicyMiddleware",
     "PurviewLocationType",
+    "PurviewPaymentRequiredError",
     "PurviewPolicyMiddleware",
     "PurviewRateLimitError",
     "PurviewRequestError",

python/packages/purview/agent_framework_purview/_cache.py

@@ -0,0 +1,191 @@
+# Copyright (c) Microsoft. All rights reserved.
+"""Cache provider for Purview data."""
+
+from __future__ import annotations
+
+import hashlib
+import heapq
+import json
+import sys
+import time
+from typing import Any, Protocol
+
+from ._models import ProtectionScopesRequest
+
+
+class CacheProvider(Protocol):
+    """Protocol for cache providers used by Purview integration."""
+
+    async def get(self, key: str) -> Any | None:
+        """Get a value from the cache.
+
+        Args:
+            key: The cache key.
+
+        Returns:
+            The cached value or None if not found or expired.
+        """
+        ...
+
+    async def set(self, key: str, value: Any, ttl_seconds: int | None = None) -> None:
+        """Set a value in the cache.
+
+        Args:
+            key: The cache key.
+            value: The value to cache.
+            ttl_seconds: Time to live in seconds. If None, uses provider default.
+        """
+        ...
+
+    async def remove(self, key: str) -> None:
+        """Remove a value from the cache.
+
+        Args:
+            key: The cache key.
+        """
+        ...
+
+
+class InMemoryCacheProvider:
+    """Simple in-memory cache implementation for Purview data.
+
+    This implementation uses a dictionary with expiration tracking and size limits.
+    """
+
+    def __init__(self, default_ttl_seconds: int = 1800, max_size_bytes: int = 200 * 1024 * 1024):
+        """Initialize the in-memory cache.
+
+        Args:
+            default_ttl_seconds: Default time to live in seconds (default 1800 = 30 minutes).
+            max_size_bytes: Maximum cache size in bytes (default 200MB).
+        """
+        self._cache: dict[str, tuple[Any, float, int]] = {}  # key -> (value, expiry, size)
+        self._expiry_heap: list[tuple[float, str]] = []  # min-heap of (expiry_time, key)
+        self._default_ttl = default_ttl_seconds
+        self._max_size_bytes = max_size_bytes
+        self._current_size_bytes = 0
+
+    def _estimate_size(self, value: Any) -> int:
+        """Estimate the size of a cached value in bytes.
+
+        Args:
+            value: The value to estimate size for.
+
+        Returns:
+            Estimated size in bytes.
+        """
+        try:
+            if hasattr(value, "model_dump_json"):
+                return len(value.model_dump_json().encode("utf-8"))
+
+            return len(json.dumps(value, default=str).encode("utf-8"))
+        except Exception:
+            # Fallback to sys.getsizeof if JSON serialization fails
+            try:
+                return sys.getsizeof(value)
+            except Exception:
+                # Conservative fallback estimate
+                return 1024
+
+    def _evict_if_needed(self, required_size: int) -> None:
+        """Evict oldest entries if needed to make room for new entry.
+
+        Uses a min-heap to efficiently find and evict entries with earliest expiry times.
+        Also cleans up stale heap entries for keys that no longer exist in cache.
+
+        Args:
+            required_size: Size in bytes needed for new entry.
+        """
+        if self._current_size_bytes + required_size <= self._max_size_bytes:
+            return
+
+        while self._expiry_heap and self._current_size_bytes + required_size > self._max_size_bytes:
+            expiry_time, key = heapq.heappop(self._expiry_heap)
+
+            if key in self._cache:
+                _, cached_expiry, size = self._cache[key]
+                if cached_expiry == expiry_time:
+                    del self._cache[key]
+                    self._current_size_bytes -= size
+                # else: stale heap entry, already updated/removed, skip it
+
+    async def get(self, key: str) -> Any | None:
+        """Get a value from the cache.
+
+        Args:
+            key: The cache key.
+
+        Returns:
+            The cached value or None if not found or expired.
+        """
+        if key not in self._cache:
+            return None
+
+        value, expiry, size = self._cache[key]
+        if time.time() > expiry:
+            del self._cache[key]
+            self._current_size_bytes -= size
+            return None
+
+        return value
+
+    async def set(self, key: str, value: Any, ttl_seconds: int | None = None) -> None:
+        """Set a value in the cache.
+
+        Args:
+            key: The cache key.
+            value: The value to cache.
+            ttl_seconds: Time to live in seconds. If None, uses default TTL.
+        """
+        ttl = ttl_seconds if ttl_seconds is not None else self._default_ttl
+        expiry = time.time() + ttl
+        size = self._estimate_size(value)
+
+        # Remove old entry if exists
+        if key in self._cache:
+            old_size = self._cache[key][2]
+            self._current_size_bytes -= old_size
+
+        # Evict if needed
+        self._evict_if_needed(size)
+
+        self._cache[key] = (value, expiry, size)
+        self._current_size_bytes += size
+
+        heapq.heappush(self._expiry_heap, (expiry, key))
+
+    async def remove(self, key: str) -> None:
+        """Remove a value from the cache.
+
+        Args:
+            key: The cache key.
+        """
+        entry = self._cache.pop(key, None)
+        if entry is not None:
+            self._current_size_bytes -= entry[2]
+        self._cache.pop(key, None)
+
+
+def create_protection_scopes_cache_key(request: ProtectionScopesRequest) -> str:
+    """Create a cache key for a ProtectionScopesRequest.
+
+    The key is based on the serialized request content (excluding correlation_id).
+
+    Args:
+        request: The protection scopes request.
+
+    Returns:
+        A string cache key.
+    """
+    data = request.to_dict(exclude_none=True)
+
+    for field in ["correlation_id"]:
+        data.pop(field, None)
+
+    json_str = json.dumps(data, sort_keys=True)
+    return f"purview:protection_scopes:{hashlib.sha256(json_str.encode()).hexdigest()}"
+
+
+__all__ = [
+    "CacheProvider",
+]

python/packages/purview/agent_framework_purview/_client.py

@@ -5,15 +5,19 @@ import base64
 import inspect
 import json
 from typing import Any, cast
+from uuid import uuid4
 
 import httpx
 from agent_framework import AGENT_FRAMEWORK_USER_AGENT
+from agent_framework._logging import get_logger
 from agent_framework.observability import get_tracer
 from azure.core.credentials import TokenCredential
 from azure.core.credentials_async import AsyncTokenCredential
+from opentelemetry import trace
 
 from ._exceptions import (
     PurviewAuthenticationError,
+    PurviewPaymentRequiredError,
     PurviewRateLimitError,
     PurviewRequestError,
     PurviewServiceError,
@@ -28,6 +32,8 @@ from ._models import (
 )
 from ._settings import PurviewSettings
 
+logger = get_logger("agent_framework.purview")
+
 
 class PurviewClient:
     """Async client for calling Graph Purview endpoints.
@@ -85,13 +91,39 @@ class PurviewClient:
         with get_tracer().start_as_current_span("purview.process_content"):
             token = await self._get_token(tenant_id=request.tenant_id)
             url = f"{self._graph_uri}/users/{request.user_id}/dataSecurityAndGovernance/processContent"
-            return cast(ProcessContentResponse, await self._post(url, request, ProcessContentResponse, token))
+            headers = {}
+            # Add If-None-Match header if scope_identifier is present
+            if hasattr(request, "scope_identifier") and request.scope_identifier:
+                headers["If-None-Match"] = request.scope_identifier
+            # Add Prefer: evaluateInline header if process_inline is True
+            if hasattr(request, "process_inline") and request.process_inline:
+                headers["Prefer"] = "evaluateInline"
+
+            response = await self._post(
+                url, request, ProcessContentResponse, token, headers=headers, return_response=True
+            )
+
+            if isinstance(response, tuple) and len(response) == 2:
+                response_obj, _ = response
+                return cast(ProcessContentResponse, response_obj)
+
+            return cast(ProcessContentResponse, response)
 
     async def get_protection_scopes(self, request: ProtectionScopesRequest) -> ProtectionScopesResponse:
         with get_tracer().start_as_current_span("purview.get_protection_scopes"):
             token = await self._get_token()
             url = f"{self._graph_uri}/users/{request.user_id}/dataSecurityAndGovernance/protectionScopes/compute"
-            return cast(ProtectionScopesResponse, await self._post(url, request, ProtectionScopesResponse, token))
+            response = await self._post(url, request, ProtectionScopesResponse, token, return_response=True)
+
+            # Extract etag from response headers
+            if isinstance(response, tuple) and len(response) == 2:
+                response_obj, headers = response
+                if "etag" in headers:
+                    etag_value = headers["etag"].strip('"')
+                    response_obj.scope_identifier = etag_value
+                return cast(ProtectionScopesResponse, response_obj)
+
+            return cast(ProtectionScopesResponse, response)
 
     async def send_content_activities(self, request: ContentActivitiesRequest) -> ContentActivitiesResponse:
         with get_tracer().start_as_current_span("purview.send_content_activities"):
@@ -99,16 +131,44 @@ class PurviewClient:
             url = f"{self._graph_uri}/users/{request.user_id}/dataSecurityAndGovernance/activities/contentActivities"
             return cast(ContentActivitiesResponse, await self._post(url, request, ContentActivitiesResponse, token))
 
-    async def _post(self, url: str, model: Any, response_type: type[Any], token: str) -> Any:
+    async def _post(
+        self,
+        url: str,
+        model: Any,
+        response_type: type[Any],
+        token: str,
+        headers: dict[str, str] | None = None,
+        return_response: bool = False,
+    ) -> Any:
+        if hasattr(model, "correlation_id") and not model.correlation_id:
+            model.correlation_id = str(uuid4())
+
+        correlation_id = getattr(model, "correlation_id", None)
+        if correlation_id:
+            span = trace.get_current_span()
+            if span and span.is_recording():
+                span.set_attribute("correlation_id", correlation_id)
+            logger.info(f"Purview request to {url} with correlation_id: {correlation_id}")
+
         payload = model.model_dump(by_alias=True, exclude_none=True, mode="json")
-        headers = {
+        request_headers = {
             "Authorization": f"Bearer {token}",
             "User-Agent": AGENT_FRAMEWORK_USER_AGENT,
             "Content-Type": "application/json",
         }
-        resp = await self._client.post(url, json=payload, headers=headers)
+        if correlation_id:
+            request_headers["client-request-id"] = correlation_id
+
+        if headers:
+            request_headers.update(headers)
+        resp = await self._client.post(url, json=payload, headers=request_headers)
+
         if resp.status_code in (401, 403):
             raise PurviewAuthenticationError(f"Auth failure {resp.status_code}: {resp.text}")
+        if resp.status_code == 402:
+            if self._settings.ignore_payment_required:
+                return response_type()  # type: ignore[call-arg, no-any-return]
+            raise PurviewPaymentRequiredError(f"Payment required {resp.status_code}: {resp.text}")
         if resp.status_code == 429:
             raise PurviewRateLimitError(f"Rate limited {resp.status_code}: {resp.text}")
         if resp.status_code not in (200, 201, 202):
@@ -117,10 +177,21 @@ class PurviewClient:
             data = resp.json()
         except ValueError:
             data = {}
+
         try:
             # Prefer pydantic-style model_validate if present, else fall back to constructor.
             if hasattr(response_type, "model_validate"):
-                return response_type.model_validate(data)  # type: ignore[no-any-return]
-            return response_type(**data)  # type: ignore[call-arg, no-any-return]
-        except Exception as ex:  # pragma: no cover
+                response_obj = response_type.model_validate(data)  # type: ignore[no-any-return]
+            else:
+                response_obj = response_type(**data)  # type: ignore[call-arg, no-any-return]
+
+            # Extract correlation_id from response headers if response object supports it
+            if "client-request-id" in resp.headers and hasattr(response_obj, "correlation_id"):
+                response_obj.correlation_id = resp.headers["client-request-id"]
+                logger.info(f"Purview response from {url} with correlation_id: {response_obj.correlation_id}")
+
+            if return_response:
+                return (response_obj, resp.headers)
+            return response_obj
+        except Exception as ex:
             raise PurviewServiceError(f"Failed to deserialize Purview response: {ex}") from ex

python/packages/purview/agent_framework_purview/_exceptions.py

@@ -7,6 +7,7 @@ from agent_framework.exceptions import ServiceResponseException
 
 __all__ = [
     "PurviewAuthenticationError",
+    "PurviewPaymentRequiredError",
     "PurviewRateLimitError",
     "PurviewRequestError",
     "PurviewServiceError",
@@ -21,6 +22,10 @@ class PurviewAuthenticationError(PurviewServiceError):
     """Authentication / authorization failure (401/403)."""
 
 
+class PurviewPaymentRequiredError(PurviewServiceError):
+    """Payment required (402)."""
+
+
 class PurviewRateLimitError(PurviewServiceError):
     """Rate limiting or throttling (429)."""
 

python/packages/purview/agent_framework_purview/_middleware.py

@@ -8,7 +8,9 @@ from agent_framework._logging import get_logger
 from azure.core.credentials import TokenCredential
 from azure.core.credentials_async import AsyncTokenCredential
 
+from ._cache import CacheProvider
 from ._client import PurviewClient
+from ._exceptions import PurviewPaymentRequiredError
 from ._models import Activity
 from ._processor import ScopedContentProcessor
 from ._settings import PurviewSettings
@@ -38,9 +40,10 @@ class PurviewPolicyMiddleware(AgentMiddleware):
         self,
         credential: TokenCredential | AsyncTokenCredential,
         settings: PurviewSettings,
+        cache_provider: CacheProvider | None = None,
     ) -> None:
         self._client = PurviewClient(credential, settings)
-        self._processor = ScopedContentProcessor(self._client, settings)
+        self._processor = ScopedContentProcessor(self._client, settings, cache_provider)
         self._settings = settings
 
     async def process(
@@ -62,9 +65,14 @@ class PurviewPolicyMiddleware(AgentMiddleware):
                 )
                 context.terminate = True
                 return
+        except PurviewPaymentRequiredError as ex:
+            logger.error(f"Purview payment required error in policy pre-check: {ex}")
+            if not self._settings.ignore_payment_required:
+                raise
         except Exception as ex:
-            # Log and continue if there's an error in the pre-check
             logger.error(f"Error in Purview policy pre-check: {ex}")
+            if not self._settings.ignore_exceptions:
+                raise
 
         await next(context)
 
@@ -86,9 +94,14 @@ class PurviewPolicyMiddleware(AgentMiddleware):
             else:
                 # Streaming responses are not supported for post-checks
                 logger.debug("Streaming responses are not supported for Purview policy post-checks")
+        except PurviewPaymentRequiredError as ex:
+            logger.error(f"Purview payment required error in policy post-check: {ex}")
+            if not self._settings.ignore_payment_required:
+                raise
         except Exception as ex:
-            # Log and continue if there's an error in the post-check
             logger.error(f"Error in Purview policy post-check: {ex}")
+            if not self._settings.ignore_exceptions:
+                raise
 
 
 class PurviewChatPolicyMiddleware(ChatMiddleware):
@@ -118,9 +131,10 @@ class PurviewChatPolicyMiddleware(ChatMiddleware):
         self,
         credential: TokenCredential | AsyncTokenCredential,
         settings: PurviewSettings,
+        cache_provider: CacheProvider | None = None,
     ) -> None:
         self._client = PurviewClient(credential, settings)
-        self._processor = ScopedContentProcessor(self._client, settings)
+        self._processor = ScopedContentProcessor(self._client, settings, cache_provider)
         self._settings = settings
 
     async def process(
@@ -134,15 +148,20 @@ class PurviewChatPolicyMiddleware(ChatMiddleware):
                 context.messages, Activity.UPLOAD_TEXT
             )
             if should_block_prompt:
-                from agent_framework import ChatMessage
+                from agent_framework import ChatMessage, ChatResponse
 
-                context.result = [  # type: ignore[assignment]
-                    ChatMessage(role="system", text=self._settings.blocked_prompt_message)
-                ]
+                blocked_message = ChatMessage(role="system", text=self._settings.blocked_prompt_message)
+                context.result = ChatResponse(messages=[blocked_message])
                 context.terminate = True
                 return
+        except PurviewPaymentRequiredError as ex:
+            logger.error(f"Purview payment required error in policy pre-check: {ex}")
+            if not self._settings.ignore_payment_required:
+                raise
         except Exception as ex:
             logger.error(f"Error in Purview policy pre-check: {ex}")
+            if not self._settings.ignore_exceptions:
+                raise
 
         await next(context)
 
@@ -157,12 +176,17 @@ class PurviewChatPolicyMiddleware(ChatMiddleware):
                         messages, Activity.UPLOAD_TEXT, user_id=resolved_user_id
                     )
                     if should_block_response:
-                        from agent_framework import ChatMessage
+                        from agent_framework import ChatMessage, ChatResponse
 
-                        context.result = [  # type: ignore[assignment]
-                            ChatMessage(role="system", text=self._settings.blocked_response_message)
-                        ]
+                        blocked_message = ChatMessage(role="system", text=self._settings.blocked_response_message)
+                        context.result = ChatResponse(messages=[blocked_message])
             else:
                 logger.debug("Streaming responses are not supported for Purview policy post-checks")
+        except PurviewPaymentRequiredError as ex:
+            logger.error(f"Purview payment required error in policy post-check: {ex}")
+            if not self._settings.ignore_payment_required:
+                raise
         except Exception as ex:
             logger.error(f"Error in Purview policy post-check: {ex}")
+            if not self._settings.ignore_exceptions:
+                raise

python/packages/purview/agent_framework_purview/_models.py

@@ -642,7 +642,9 @@ class ContentToProcess(_AliasSerializable):
 
 class ProcessContentRequest(_AliasSerializable):
     _ALIASES: ClassVar[dict[str, str]] = {"content_to_process": "contentToProcess"}
-    DEFAULT_EXCLUDE: ClassVar[set[str]] = {"user_id", "tenant_id", "correlation_id", "process_inline"}
+    DEFAULT_EXCLUDE: ClassVar[set[str]] = {
+        "correlation_id",
+    }
 
     def __init__(
         self,
@@ -651,6 +653,7 @@ class ProcessContentRequest(_AliasSerializable):
         tenant_id: str,
         correlation_id: str | None = None,
         process_inline: bool | None = None,
+        scope_identifier: str | None = None,
         **kwargs: Any,
     ) -> None:
         # Extract aliased values from kwargs
@@ -668,10 +671,11 @@ class ProcessContentRequest(_AliasSerializable):
         self.tenant_id = tenant_id
         self.correlation_id = correlation_id
         self.process_inline = process_inline
+        self.scope_identifier = scope_identifier
 
 
 class ProtectionScopesRequest(_AliasSerializable):
-    DEFAULT_EXCLUDE: ClassVar[set[str]] = {"user_id", "tenant_id", "correlation_id", "scope_identifier"}
+    DEFAULT_EXCLUDE: ClassVar[set[str]] = {"correlation_id"}
     _ALIASES: ClassVar[dict[str, str]] = {
         "pivot_on": "pivotOn",
         "device_metadata": "deviceMetadata",
@@ -743,7 +747,7 @@ class ContentActivitiesRequest(_AliasSerializable):
         "scope_identifier": "scopeIdentifier",
         "content_to_process": "contentMetadata",
     }
-    DEFAULT_EXCLUDE: ClassVar[set[str]] = {"tenant_id", "correlation_id"}
+    DEFAULT_EXCLUDE: ClassVar[set[str]] = {"correlation_id"}
 
     def __init__(
         self,
@@ -800,12 +804,15 @@ class ProcessContentResponse(_AliasSerializable):
         "protection_scope_state": "protectionScopeState",
         "policy_actions": "policyActions",
         "processing_errors": "processingErrors",
+        "correlation_id": "correlationId",
     }
+    DEFAULT_EXCLUDE: ClassVar[set[str]] = {"correlation_id"}
 
     id: str | None
     protection_scope_state: ProtectionScopeState | None
     policy_actions: list[DlpActionInfo] | None
     processing_errors: list[ProcessingError] | None
+    correlation_id: str | None
 
     def __init__(
         self,
@@ -813,6 +820,7 @@ class ProcessContentResponse(_AliasSerializable):
         protection_scope_state: ProtectionScopeState | None = None,
         policy_actions: list[DlpActionInfo | MutableMapping[str, Any]] | None = None,
         processing_errors: list[ProcessingError | MutableMapping[str, Any]] | None = None,
+        correlation_id: str | None = None,
         **kwargs: Any,
     ) -> None:
         # Extract aliased values from kwargs
@@ -822,6 +830,8 @@ class ProcessContentResponse(_AliasSerializable):
             policy_actions = kwargs["policyActions"]
         if "processingErrors" in kwargs:
             processing_errors = kwargs["processingErrors"]
+        if "correlationId" in kwargs:
+            correlation_id = kwargs["correlationId"]
 
         # Convert to objects
         converted_policy_actions: list[DlpActionInfo] | None = None
@@ -838,12 +848,12 @@ class ProcessContentResponse(_AliasSerializable):
                 [pe if isinstance(pe, ProcessingError) else ProcessingError(**pe) for pe in processing_errors],
             )
 
-        # Call parent without explicit params with aliases
         super().__init__(**kwargs)
         self.id = id
         self.protection_scope_state = protection_scope_state
         self.policy_actions = converted_policy_actions
         self.processing_errors = converted_processing_errors
+        self.correlation_id = correlation_id
 
 
 class PolicyScope(_AliasSerializable):
@@ -909,15 +919,22 @@ class PolicyScope(_AliasSerializable):
 
 
 class ProtectionScopesResponse(_AliasSerializable):
-    _ALIASES: ClassVar[dict[str, str]] = {"scope_identifier": "scopeIdentifier", "scopes": "value"}
+    _ALIASES: ClassVar[dict[str, str]] = {
+        "scope_identifier": "scopeIdentifier",
+        "scopes": "value",
+        "correlation_id": "correlationId",
+    }
+    DEFAULT_EXCLUDE: ClassVar[set[str]] = {"correlation_id"}
 
     scope_identifier: str | None
     scopes: list[PolicyScope] | None
+    correlation_id: str | None
 
     def __init__(
         self,
         scope_identifier: str | None = None,
         scopes: list[PolicyScope | MutableMapping[str, Any]] | None = None,
+        correlation_id: str | None = None,
         **kwargs: Any,
     ) -> None:
         # Extract aliased values from kwargs before they're normalized by parent
@@ -925,6 +942,8 @@ class ProtectionScopesResponse(_AliasSerializable):
             scope_identifier = kwargs["scopeIdentifier"]
         if "value" in kwargs:
             scopes = kwargs["value"]
+        if "correlationId" in kwargs:
+            correlation_id = kwargs["correlationId"]
 
         converted_scopes: list[PolicyScope] | None = None
         if scopes is not None:
@@ -936,22 +955,32 @@ class ProtectionScopesResponse(_AliasSerializable):
         super().__init__(**kwargs)
         self.scope_identifier = scope_identifier
         self.scopes = converted_scopes
+        self.correlation_id = correlation_id
 
 
 class ContentActivitiesResponse(_AliasSerializable):
-    DEFAULT_EXCLUDE: ClassVar[set[str]] = {"status_code"}
+    DEFAULT_EXCLUDE: ClassVar[set[str]] = {"correlation_id"}
+    _ALIASES: ClassVar[dict[str, str]] = {"correlation_id": "correlationId"}
+
+    status_code: int | None
+    error: ErrorDetails | None
+    correlation_id: str | None
 
     def __init__(
         self,
         status_code: int | None = None,
         error: ErrorDetails | MutableMapping[str, Any] | None = None,
+        correlation_id: str | None = None,
         **kwargs: Any,
     ) -> None:
+        if "correlationId" in kwargs:
+            correlation_id = kwargs["correlationId"]
         if isinstance(error, MutableMapping):
             error = ErrorDetails(**error)
-        super().__init__(status_code=status_code, error=error, **kwargs)
+        super().__init__(status_code=status_code, error=error, correlation_id=correlation_id, **kwargs)
         self.status_code = status_code
         self.error = error  # type: ignore[assignment]
+        self.correlation_id = correlation_id
 
 
 __all__ = [

python/packages/purview/agent_framework_purview/_processor.py

@@ -1,13 +1,17 @@
 # Copyright (c) Microsoft. All rights reserved.
 from __future__ import annotations
 
+import asyncio
 import uuid
 from collections.abc import Iterable, MutableMapping
 from typing import Any
 
 from agent_framework import ChatMessage
+from agent_framework._logging import get_logger
 
+from ._cache import CacheProvider, InMemoryCacheProvider, create_protection_scopes_cache_key
 from ._client import PurviewClient
+from ._exceptions import PurviewPaymentRequiredError
 from ._models import (
     Activity,
     ActivityMetadata,
@@ -16,22 +20,25 @@ from ._models import (
     DeviceMetadata,
     DlpAction,
     DlpActionInfo,
+    ExecutionMode,
     IntegratedAppMetadata,
     OperatingSystemSpecifications,
     PolicyLocation,
     ProcessContentRequest,
     ProcessContentResponse,
     ProcessConversationMetadata,
-    ProcessingError,
     ProtectedAppMetadata,
     ProtectionScopesRequest,
     ProtectionScopesResponse,
+    ProtectionScopeState,
     PurviewTextContent,
     RestrictionAction,
     translate_activity,
 )
 from ._settings import PurviewSettings
 
+logger = get_logger("agent_framework.purview")
+
 
 def _is_valid_guid(value: str | None) -> bool:
     """Check if a string is a valid GUID/UUID format using uuid module."""
@@ -47,9 +54,13 @@ def _is_valid_guid(value: str | None) -> bool:
 class ScopedContentProcessor:
     """Combine protection scopes, process content, and content activities logic."""
 
-    def __init__(self, client: PurviewClient, settings: PurviewSettings):
+    def __init__(self, client: PurviewClient, settings: PurviewSettings, cache_provider: CacheProvider | None = None):
         self._client = client
         self._settings = settings
+        self._cache: CacheProvider = cache_provider or InMemoryCacheProvider(
+            default_ttl_seconds=settings.cache_ttl_seconds, max_size_bytes=settings.max_cache_size_bytes
+        )
+        self._background_tasks: set[asyncio.Task[Any]] = set()
 
     async def process_messages(
         self, messages: Iterable[ChatMessage], activity: Activity, user_id: str | None = None
@@ -173,7 +184,7 @@ class ScopedContentProcessor:
                 user_id=resolved_user_id,  # Use the resolved user_id for all messages
                 tenant_id=tenant_id,
                 correlation_id=meta.correlation_id,
-                process_inline=True if self._settings.process_inline else None,
+                process_inline=None,  # Will be set based on execution mode
             )
             results.append(req)
         return results, resolved_user_id
@@ -191,23 +202,86 @@ class ScopedContentProcessor:
             integrated_app_metadata=pc_request.content_to_process.integrated_app_metadata,
             correlation_id=pc_request.correlation_id,
         )
-        ps_resp = await self._client.get_protection_scopes(ps_req)
-        should_process, dlp_actions = self._check_applicable_scopes(pc_request, ps_resp)
+
+        # Check for tenant-level 402 exception cache first
+        tenant_payment_cache_key = f"purview:payment_required:{pc_request.tenant_id}"
+        cached_payment_exception = await self._cache.get(tenant_payment_cache_key)
+        if isinstance(cached_payment_exception, PurviewPaymentRequiredError):
+            raise cached_payment_exception
+
+        cache_key = create_protection_scopes_cache_key(ps_req)
+        cached_ps_resp = await self._cache.get(cache_key)
+
+        if cached_ps_resp is not None:
+            if isinstance(cached_ps_resp, ProtectionScopesResponse):
+                ps_resp = cached_ps_resp
+        else:
+            try:
+                ps_resp = await self._client.get_protection_scopes(ps_req)
+                await self._cache.set(cache_key, ps_resp, ttl_seconds=self._settings.cache_ttl_seconds)
+            except PurviewPaymentRequiredError as ex:
+                # Cache the exception at tenant level so all subsequent requests for this tenant fail fast
+                await self._cache.set(tenant_payment_cache_key, ex, ttl_seconds=self._settings.cache_ttl_seconds)
+                raise
+
+        if ps_resp.scope_identifier:
+            pc_request.scope_identifier = ps_resp.scope_identifier
+
+        should_process, dlp_actions, execution_mode = self._check_applicable_scopes(pc_request, ps_resp)
 
         if should_process:
+            # Set process_inline based on execution mode
+            pc_request.process_inline = execution_mode == ExecutionMode.EVALUATE_INLINE
+
+            # If execution mode is offline, queue the PC request in background
+            if execution_mode != ExecutionMode.EVALUATE_INLINE:
+                task = asyncio.create_task(self._process_content_background(pc_request, cache_key))
+                self._background_tasks.add(task)
+                task.add_done_callback(self._background_tasks.discard)
+                return ProcessContentResponse(id="204", correlation_id=pc_request.correlation_id)
+
             pc_resp = await self._client.process_content(pc_request)
+
+            if pc_request.scope_identifier and pc_resp.protection_scope_state == ProtectionScopeState.MODIFIED:
+                await self._cache.remove(cache_key)
+
             pc_resp.policy_actions = self._combine_policy_actions(pc_resp.policy_actions, dlp_actions)
             return pc_resp
+
+        # No applicable scopes - send content activities in background
         ca_req = ContentActivitiesRequest(
             user_id=pc_request.user_id,
             tenant_id=pc_request.tenant_id,
             content_to_process=pc_request.content_to_process,
             correlation_id=pc_request.correlation_id,
         )
-        ca_resp = await self._client.send_content_activities(ca_req)
-        if ca_resp.error:
-            return ProcessContentResponse(processing_errors=[ProcessingError(message=str(ca_resp.error))])
-        return ProcessContentResponse()
+
+        task = asyncio.create_task(self._send_content_activities_background(ca_req))
+        self._background_tasks.add(task)
+        task.add_done_callback(self._background_tasks.discard)
+        # Respond with HttpStatusCode 204(No Content)
+        return ProcessContentResponse(id="204", correlation_id=pc_request.correlation_id)
+
+    async def _process_content_background(self, pc_request: ProcessContentRequest, cache_key: str) -> None:
+        """Process content in background for offline execution mode."""
+        try:
+            pc_resp = await self._client.process_content(pc_request)
+
+            # If protection scope state is modified, make another PC request and invalidate cache
+            if pc_request.scope_identifier and pc_resp.protection_scope_state == ProtectionScopeState.MODIFIED:
+                await self._cache.remove(cache_key)
+                await self._client.process_content(pc_request)
+        except Exception as ex:
+            # Log errors but don't propagate since this is fire-and-forget
+            logger.warning(f"Background process content request failed: {ex}")
+
+    async def _send_content_activities_background(self, ca_req: ContentActivitiesRequest) -> None:
+        """Send content activities in background without blocking."""
+        try:
+            await self._client.send_content_activities(ca_req)
+        except Exception as ex:
+            # Log errors but don't propagate since this is fire-and-forget
+            logger.warning(f"Background content activities request failed: {ex}")
 
     @staticmethod
     def _combine_policy_actions(
@@ -225,11 +299,22 @@ class ScopedContentProcessor:
     @staticmethod
     def _check_applicable_scopes(
         pc_request: ProcessContentRequest, ps_response: ProtectionScopesResponse
-    ) -> tuple[bool, list[DlpActionInfo]]:
+    ) -> tuple[bool, list[DlpActionInfo], ExecutionMode]:
+        """Check if any scopes are applicable to the request.
+
+        Args:
+            pc_request: The process content request
+            ps_response: The protection scopes response
+
+        Returns:
+            A tuple of (should_process, dlp_actions, execution_mode)
+        """
         req_activity = translate_activity(pc_request.content_to_process.activity_metadata.activity)
         location = pc_request.content_to_process.protected_app_metadata.application_location
         should_process: bool = False
         dlp_actions: list[DlpActionInfo] = []
+        execution_mode: ExecutionMode = ExecutionMode.EVALUATE_OFFLINE  # Default to offline
+
         for scope in ps_response.scopes or []:
             # Check if all activities in req_activity are present in scope.activities using bitwise flags.
             activity_match = bool(scope.activities and (scope.activities & req_activity) == req_activity)
@@ -246,6 +331,11 @@ class ScopedContentProcessor:
                         break
             if activity_match and location_match:
                 should_process = True
+
+                # If any scope has EvaluateInline, upgrade to inline mode
+                if scope.execution_mode == ExecutionMode.EVALUATE_INLINE:
+                    execution_mode = ExecutionMode.EVALUATE_INLINE
+
                 if scope.policy_actions:
                     dlp_actions.extend(scope.policy_actions)
-        return should_process, dlp_actions
+        return should_process, dlp_actions, execution_mode

python/packages/purview/agent_framework_purview/_settings.py

@@ -41,18 +41,23 @@ class PurviewSettings(AFBaseSettings):
 
     Attributes:
         app_name: Public app name.
+        app_version: Optional version string of the application.
         tenant_id: Optional tenant id (guid) of the user making the request.
         purview_app_location: Optional app location for policy evaluation.
         graph_base_uri: Base URI for Microsoft Graph.
         blocked_prompt_message: Custom message to return when a prompt is blocked by policy.
         blocked_response_message: Custom message to return when a response is blocked by policy.
+        ignore_exceptions: If True, all Purview exceptions will be logged but not thrown in middleware.
+        ignore_payment_required: If True, 402 payment required errors will be logged but not thrown.
+        cache_ttl_seconds: Time to live for cache entries in seconds (default 14400 = 4 hours).
+        max_cache_size_bytes: Maximum cache size in bytes (default 200MB).
     """
 
     app_name: str = Field(...)
+    app_version: str | None = Field(default=None)
     tenant_id: str | None = Field(default=None)
     purview_app_location: PurviewAppLocation | None = Field(default=None)
     graph_base_uri: str = Field(default="https://graph.microsoft.com/v1.0/")
-    process_inline: bool = Field(default=False, description="Process content inline if supported.")
     blocked_prompt_message: str = Field(
         default="Prompt blocked by policy",
         description="Message to return when a prompt is blocked by policy.",
@@ -61,6 +66,22 @@ class PurviewSettings(AFBaseSettings):
         default="Response blocked by policy",
         description="Message to return when a response is blocked by policy.",
     )
+    ignore_exceptions: bool = Field(
+        default=False,
+        description="If True, all Purview exceptions will be logged but not thrown in middleware.",
+    )
+    ignore_payment_required: bool = Field(
+        default=False,
+        description="If True, 402 payment required errors will be logged but not thrown.",
+    )
+    cache_ttl_seconds: int = Field(
+        default=14400,
+        description="Time to live for cache entries in seconds (default 14400 = 4 hours).",
+    )
+    max_cache_size_bytes: int = Field(
+        default=200 * 1024 * 1024,
+        description="Maximum cache size in bytes (default 200MB).",
+    )
 
     model_config = SettingsConfigDict(populate_by_name=True, validate_assignment=True)