agent-framework-purview adds Microsoft Purview (Microsoft Graph dataSecurityAndGovernance) policy evaluation to the Microsoft Agent Framework. It lets you enforce data security / governance policies on both the prompt (user input + conversation history) and the model response before they proceed further in your workflow.

Status: Preview

• Middleware-based policy enforcement (agent-level and chat-client level)
• Blocks or allows content at both ingress (prompt) and egress (response)
• Works with any Agent / agent orchestration using the standard Agent Framework middleware pipeline
• Supports both synchronous TokenCredential and AsyncTokenCredential from azure-identity
• Configuration via PurviewSettings / PurviewAppLocation
• Built-in caching with configurable TTL and size limits for protection scopes in PurviewSettings
• Background processing for content activities and offline policy evaluation

Add Purview when you need to:

• Prevent sensitive data leaks: Inline blocking of sensitive content based on Data Loss Prevention (DLP) policies.
• Enable governance: Log AI interactions in Purview for Audit, Communication Compliance, Insider Risk Management, eDiscovery, and Data Lifecycle Management.
• Prevent sensitive or disallowed content from being sent to an LLM
• Prevent model output containing disallowed data from leaving the system
• Apply centrally managed policies without rewriting agent logic

• Microsoft Azure subscription with Microsoft Purview configured.
• Microsoft 365 subscription with an E5 license and pay-as-you-go billing setup.
  • For testing, you can use a Microsoft 365 Developer Program tenant. For more information, see Join the Microsoft 365 Developer Program.

PurviewClient uses the azure-identity library for token acquisition. You can use any TokenCredential or AsyncTokenCredential implementation.

• Entra registration: Register your agent and add the required Microsoft Graph permissions (dataSecurityAndGovernance) to the Service Principal. For more information, see Register an application in Microsoft Entra ID and dataSecurityAndGovernance resource type. You'll need the Microsoft Entra app ID in the next step.

• Graph Permissions:

• ProtectionScopes.Compute.All : userProtectionScopeContainer

• Content.Process.All : processContent

• ContentActivity.Write : contentActivity

• Purview policies: Configure Purview policies using the Microsoft Entra app ID to enable agent communications data to flow into Purview. For more information, see Configure Microsoft Purview.

PurviewSettings.get_scopes() derives the Graph scope list (currently https://graph.microsoft.com/.default style).

import asyncio
from agent_framework import Agent, Message, Role
from agent_framework.openai import OpenAIChatCompletionClient
from agent_framework.microsoft import PurviewPolicyMiddleware, PurviewSettings
from azure.identity import InteractiveBrowserCredential

async def main():
	client = OpenAIChatCompletionClient()  # uses environment for endpoint + deployment

	purview_middleware = PurviewPolicyMiddleware(
		credential=InteractiveBrowserCredential(),
		settings=PurviewSettings(app_name="My Sample App")
	)

	agent = Agent(
		client=client,
		instructions="You are a helpful assistant.",
		middleware=[purview_middleware]
	)

	response = await agent.run(Message("user", ["Summarize zero trust in one sentence."]))
	print(response)

asyncio.run(main())

If a policy violation is detected on the prompt, the middleware terminates the run and substitutes a system message: "Prompt blocked by policy". If on the response, the result becomes "Response blocked by policy".

PurviewSettings(
    app_name="My App",                         # Required: Display / logical name
    app_version=None,                          # Optional: Version string of the application
    tenant_id=None,                            # Optional: Tenant id (guid), used mainly for auth context
    purview_app_location=None,                 # Optional: PurviewAppLocation for scoping
    graph_base_uri="https://graph.microsoft.com/v1.0/",
    blocked_prompt_message="Prompt blocked by policy",      # Custom message for blocked prompts
    blocked_response_message="Response blocked by policy",  # Custom message for blocked responses
    ignore_exceptions=False,                   # If True, non-payment exceptions are logged but not thrown
    ignore_payment_required=False,             # If True, 402 payment required errors are logged but not thrown
    cache_ttl_seconds=14400,                   # Cache TTL in seconds (default 4 hours)
    max_cache_size_bytes=200 * 1024 * 1024     # Max cache size in bytes (default 200MB)
)

The Purview integration includes built-in caching for protection scopes responses to improve performance and reduce API calls:

• Default TTL: 4 hours (14400 seconds)
• Default Cache Size: 200MB
• Cache Provider: InMemoryCacheProvider is used by default, but you can provide a custom implementation via the CacheProvider protocol
• Cache Invalidation: Cache is automatically invalidated when protection scope state is modified
• Exception Caching: 402 Payment Required errors are cached to avoid repeated failed API calls

You can customize caching behavior in PurviewSettings:

from agent_framework.microsoft import PurviewSettings

settings = PurviewSettings(
    app_name="My App",
    cache_ttl_seconds=14400,           # 4 hours
    max_cache_size_bytes=200 * 1024 * 1024  # 200MB
)

Or provide your own cache provider:

from typing import Any
from agent_framework.microsoft import PurviewPolicyMiddleware, PurviewSettings, CacheProvider
from azure.identity import DefaultAzureCredential

class MyCustomCache(CacheProvider):
    async def get(self, key: str) -> Any | None:
        # Your implementation
        pass

    async def set(self, key: str, value: Any, ttl_seconds: int | None = None) -> None:
        # Your implementation
        pass

    async def remove(self, key: str) -> None:
        # Your implementation
        pass

credential = DefaultAzureCredential()
settings = PurviewSettings(app_name="MyApp")

middleware = PurviewPolicyMiddleware(
    credential=credential,
    settings=settings,
    cache_provider=MyCustomCache()
)

To scope evaluation by location (application, URL, or domain):

from agent_framework.microsoft import (
	PurviewAppLocation,
	PurviewLocationType,
	PurviewSettings,
)

settings = PurviewSettings(
	app_name="Contoso Support",
	purview_app_location=PurviewAppLocation(
		location_type=PurviewLocationType.APPLICATION,
		location_value="<app-client-id>"
	)
)

By default, when Purview blocks a prompt or response, the middleware returns a generic system message. You can customize these messages by providing your own text in the PurviewSettings:

from agent_framework.microsoft import PurviewSettings

settings = PurviewSettings(
	app_name="My App",
	blocked_prompt_message="Your request contains content that violates our policies. Please rephrase and try again.",
	blocked_response_message="The response was blocked due to policy restrictions. Please contact support if you need assistance."
)

The Purview integration provides fine-grained control over exception handling to support graceful degradation scenarios:

from agent_framework.microsoft import PurviewSettings

# Ignore all non-payment exceptions (continue execution even if policy check fails)
settings = PurviewSettings(
    app_name="My App",
    ignore_exceptions=True  # Log errors but don't throw
)

# Ignore only 402 Payment Required errors (useful for tenants without proper licensing)
settings = PurviewSettings(
    app_name="My App",
    ignore_payment_required=True  # Continue even without Purview Consumptive Billing Setup
)

# Both can be combined
settings = PurviewSettings(
    app_name="My App",
    ignore_exceptions=True,
    ignore_payment_required=True
)

Use the agent middleware when you already have / want the full agent pipeline:

from agent_framework import Agent
from agent_framework.openai import OpenAIChatCompletionClient
from agent_framework.microsoft import PurviewPolicyMiddleware, PurviewSettings
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
client = OpenAIChatCompletionClient()

agent = Agent(
	client=client,
	instructions="You are helpful.",
	middleware=[PurviewPolicyMiddleware(credential, PurviewSettings(app_name="My App"))]
)

Use the chat middleware when you attach directly to a chat client (e.g. minimal agent shell or custom orchestration):

import os
from agent_framework import Agent
from agent_framework.openai import OpenAIChatCompletionClient
from agent_framework.microsoft import PurviewChatPolicyMiddleware, PurviewSettings
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()

client = OpenAIChatCompletionClient(
	model=os.environ["AZURE_OPENAI_MODEL"],
	azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
	credential=credential,
	middleware=[
		PurviewChatPolicyMiddleware(credential, PurviewSettings(app_name="My App (Chat)"))
	],
)

agent = Agent(client=client, instructions="You are helpful.")

The policy logic is identical; the difference is only the hook point in the pipeline.

1. Before agent execution (prompt phase): all context.messages are evaluated.
   • If no valid user_id is found, processing is skipped (no policy evaluation)
   • Protection scopes are retrieved (with caching)
   • Applicable scopes are checked to determine execution mode
   • In inline mode: content is evaluated immediately
   • In offline mode: evaluation is queued in background
2. If blocked: context.result is replaced with a system message and context.terminate = True.
3. After successful agent execution (response phase): the produced messages are evaluated using the same user_id from the prompt phase.
4. If blocked: result messages are replaced with a blocking notice.

The user identifier is discovered from Message.additional_properties['user_id'] during the prompt phase and reused for the response phase, ensuring both evaluations map consistently to the same user. If no user_id is present, policy evaluation is skipped entirely.

You can customize the blocking messages using the blocked_prompt_message and blocked_response_message fields in PurviewSettings. For more advanced scenarios, you can wrap the middleware or post-process context.result in later middleware.

All exceptions inherit from PurviewServiceError. You can catch specific exceptions or use the base class:

from agent_framework.microsoft import (
    PurviewPaymentRequiredError,
    PurviewAuthenticationError,
    PurviewRateLimitError,
    PurviewRequestError,
    PurviewServiceError
)

try:
    # Your code here
    pass
except PurviewPaymentRequiredError as ex:
    # Handle licensing issues specifically
    print(f"Purview licensing required: {ex}")
except (PurviewAuthenticationError, PurviewRateLimitError, PurviewRequestError, PurviewServiceError) as ex:
    # Handle other errors
    print(f"Purview enforcement skipped: {ex}")

• User Identification: Provide a user_id per request (e.g. in Message(..., additional_properties={"user_id": "<guid>"})) for per-user policy scoping. If no user_id is provided, policy evaluation is skipped entirely.
• Blocking Messages: Can be customized via blocked_prompt_message and blocked_response_message in PurviewSettings. By default, they are "Prompt blocked by policy" and "Response blocked by policy" respectively.
• Streaming Responses: Post-response policy evaluation presently applies only to non-streaming chat responses.
• Error Handling: Use ignore_exceptions and ignore_payment_required settings for graceful degradation. When enabled, errors are logged but don't fail the request.
• Caching: Protection scopes responses and 402 errors are cached by default with a 4-hour TTL. Cache is automatically invalidated when protection scope state changes.
• Background Processing: Content Activities and offline Process Content requests are handled asynchronously using background tasks to avoid blocking the main execution flow.