.NET: Implement Purview middleware in dotnet

dotnet/agent-framework-dotnet.slnx

@@ -75,6 +75,9 @@
     <Project Path="samples/GettingStarted/AgentWithOpenAI/Agent_OpenAI_Step01_Running/Agent_OpenAI_Step01_Running.csproj" />
     <Project Path="samples/GettingStarted/AgentWithOpenAI/Agent_OpenAI_Step02_Reasoning/Agent_OpenAI_Step02_Reasoning.csproj" />
   </Folder>
+  <Folder Name="/Samples/Purview/AgentWithPurview/">
+    <Project Path="samples/Purview/AgentWithPurview/AgentWithPurview.csproj" Id="36a59e99-ac83-4a35-b5ae-8b60f009dc1e" />
+  </Folder>
   <Folder Name="/Samples/GettingStarted/AgentWithRAG/">
     <File Path="samples/GettingStarted/AgentWithRAG/README.md" />
     <Project Path="samples/GettingStarted/AgentWithRAG/AgentWithRAG_Step01_BasicTextRAG/AgentWithRAG_Step01_BasicTextRAG.csproj" />
@@ -291,6 +294,7 @@
     <Project Path="src/Microsoft.Agents.AI.Hosting/Microsoft.Agents.AI.Hosting.csproj" />
     <Project Path="src/Microsoft.Agents.AI.Mem0/Microsoft.Agents.AI.Mem0.csproj" />
     <Project Path="src/Microsoft.Agents.AI.OpenAI/Microsoft.Agents.AI.OpenAI.csproj" />
+    <Project Path="src/Microsoft.Agents.AI.Purview/Microsoft.Agents.AI.Purview.csproj" Id="e2fa3f42-2062-44cc-9329-0d7899ce8578" />
     <Project Path="src/Microsoft.Agents.AI.Workflows.Declarative/Microsoft.Agents.AI.Workflows.Declarative.csproj" />
     <Project Path="src/Microsoft.Agents.AI.Workflows/Microsoft.Agents.AI.Workflows.csproj" />
     <Project Path="src/Microsoft.Agents.AI/Microsoft.Agents.AI.csproj" />
@@ -318,6 +322,7 @@
     <Project Path="tests/Microsoft.Agents.AI.Hosting.UnitTests/Microsoft.Agents.AI.Hosting.UnitTests.csproj" />
     <Project Path="tests/Microsoft.Agents.AI.Mem0.UnitTests/Microsoft.Agents.AI.Mem0.UnitTests.csproj" />
     <Project Path="tests/Microsoft.Agents.AI.OpenAI.UnitTests/Microsoft.Agents.AI.OpenAI.UnitTests.csproj" />
+    <Project Path="tests/Microsoft.Agents.AI.Purview.UnitTests/Microsoft.Agents.AI.Purview.UnitTests.csproj" />
     <Project Path="tests/Microsoft.Agents.AI.UnitTests/Microsoft.Agents.AI.UnitTests.csproj" />
     <Project Path="tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests.csproj" />
     <Project Path="tests/Microsoft.Agents.AI.Workflows.UnitTests/Microsoft.Agents.AI.Workflows.UnitTests.csproj" />

dotnet/samples/GettingStarted/Workflows/_Foundational/07_MixedWorkflowAgentsAndExecutors/Program.cs

@@ -132,7 +132,7 @@ private static async Task ExecuteWorkflowAsync(Workflow workflow, string input)
         const bool ShowAgentThinking = false;
 
         // Execute in streaming mode to see real-time progress
-        await using StreamingRun run = await InProcessExecution.StreamAsync<string>(workflow, input);
+        await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, input);
 
         // Watch the workflow events
         await foreach (WorkflowEvent evt in run.WatchStreamAsync())

dotnet/samples/GettingStarted/Workflows/_Foundational/08_WriterCriticWorkflow/Program.cs

@@ -89,7 +89,7 @@ private static async Task Main()
     private static async Task ExecuteWorkflowAsync(Workflow workflow, string input)
     {
         // Execute in streaming mode to see real-time progress
-        await using StreamingRun run = await InProcessExecution.StreamAsync<string>(workflow, input);
+        await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, input);
 
         // Watch the workflow events
         await foreach (WorkflowEvent evt in run.WatchStreamAsync())

dotnet/samples/Purview/AgentWithPurview/AgentWithPurview.csproj

@@ -0,0 +1,21 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net9.0</TargetFramework>
+
+    <Nullable>enable</Nullable>
+    <ImplicitUsings>enable</ImplicitUsings>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Azure.AI.OpenAI" />
+    <PackageReference Include="Azure.Identity" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\..\src\Microsoft.Agents.AI.OpenAI\Microsoft.Agents.AI.OpenAI.csproj" />
+    <ProjectReference Include="..\..\..\src\Microsoft.Agents.AI.Purview\Microsoft.Agents.AI.Purview.csproj" />
+  </ItemGroup>
+
+</Project>

dotnet/samples/Purview/AgentWithPurview/Program.cs

@@ -0,0 +1,43 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+// This sample shows how to create and use a simple AI agent with Azure OpenAI Responses as the backend.
+
+using Azure.AI.OpenAI;
+using Azure.Core;
+using Azure.Identity;
+using Microsoft.Agents.AI;
+using Microsoft.Agents.AI.Purview;
+using Microsoft.Extensions.AI;
+
+var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
+var deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT_NAME") ?? "gpt-4o-mini";
+var purviewClientAppId = Environment.GetEnvironmentVariable("PURVIEW_CLIENT_APP_ID") ?? throw new InvalidOperationException("PURVIEW_CLIENT_APP_ID is not set.");
+
+// This will get a user token for an entra app configured to call the Purview API.
+// Any TokenCredential with permissions to call the Purview API can be used here.
+TokenCredential browserCredential = new InteractiveBrowserCredential(
+    new InteractiveBrowserCredentialOptions
+    {
+        ClientId = purviewClientAppId
+    });
+
+IChatClient client = new AzureOpenAIClient(
+    new Uri(endpoint),
+    new AzureCliCredential())
+    .GetOpenAIResponseClient(deploymentName)
+    .AsIChatClient()
+    .AsBuilder()
+    .WithPurview(browserCredential, new PurviewSettings("Agent Framework Test App"))
+    .Build();
+
+using (client)
+{
+    Console.WriteLine("Enter a prompt to send to the client:");
+    string? promptText = Console.ReadLine();
+
+    if (!string.IsNullOrEmpty(promptText))
+    {
+        // Invoke the agent and output the text result.
+        Console.WriteLine(await client.GetResponseAsync(promptText));
+    }
+}

dotnet/src/Microsoft.Agents.AI.Purview/BackgroundJobRunner.cs

@@ -0,0 +1,72 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using Microsoft.Agents.AI.Purview.Models.Jobs;
+using Microsoft.Extensions.Logging;
+
+namespace Microsoft.Agents.AI.Purview;
+
+/// <summary>
+/// Service that runs jobs in background threads.
+/// </summary>
+internal sealed class BackgroundJobRunner
+{
+    private readonly IChannelHandler _channelHandler;
+    private readonly IPurviewClient _purviewClient;
+    private readonly ILogger _logger;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="BackgroundJobRunner"/> class.
+    /// </summary>
+    /// <param name="channelHandler"></param>
+    /// <param name="purviewClient"></param>
+    /// <param name="logger"></param>
+    /// <param name="purviewSettings"></param>
+    public BackgroundJobRunner(IChannelHandler channelHandler, IPurviewClient purviewClient, ILogger logger, PurviewSettings purviewSettings)
+    {
+        this._channelHandler = channelHandler;
+        this._purviewClient = purviewClient;
+        this._logger = logger;
+
+        for (int i = 0; i < purviewSettings.MaxConcurrentJobConsumers; i++)
+        {
+            this._channelHandler.AddRunner(async (Channel<BackgroundJobBase> channel) =>
+            {
+                await foreach (BackgroundJobBase job in channel.Reader.ReadAllAsync().ConfigureAwait(false))
+                {
+                    try
+                    {
+                        await this.RunJobAsync(job).ConfigureAwait(false);
+                    }
+                    catch (Exception e) when (
+                        !(e is OperationCanceledException) &&
+                        !(e is SystemException))
+                    {
+                        this._logger.LogError(e, "Error running background job {BackgroundJobError}.", e.Message);
+                    }
+                }
+            });
+        }
+    }
+
+    /// <summary>
+    /// Runs a job.
+    /// </summary>
+    /// <param name="job"></param>
+    /// <returns></returns>
+    private async Task RunJobAsync(BackgroundJobBase job)
+    {
+        switch (job)
+        {
+            case ProcessContentJob processContentJob:
+                _ = await this._purviewClient.ProcessContentAsync(processContentJob.Request, CancellationToken.None).ConfigureAwait(false);
+                break;
+            case ContentActivityJob contentActivityJob:
+                _ = await this._purviewClient.SendContentActivitiesAsync(contentActivityJob.Request, CancellationToken.None).ConfigureAwait(false);
+                break;
+        }
+    }
+}

dotnet/src/Microsoft.Agents.AI.Purview/CacheProvider.cs

@@ -0,0 +1,89 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Text.Json;
+using System.Text.Json.Serialization.Metadata;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Agents.AI.Purview.Serialization;
+using Microsoft.Extensions.Caching.Distributed;
+
+namespace Microsoft.Agents.AI.Purview;
+
+/// <summary>
+/// Manages caching of values.
+/// </summary>
+internal sealed class CacheProvider : ICacheProvider
+{
+    private readonly IDistributedCache _cache;
+    private readonly PurviewSettings _purviewSettings;
+
+    /// <summary>
+    /// Create a new instance of the <see cref="CacheProvider"/> class.
+    /// </summary>
+    /// <param name="cache"></param>
+    /// <param name="purviewSettings"></param>
+    public CacheProvider(IDistributedCache cache, PurviewSettings purviewSettings)
+    {
+        this._cache = cache;
+        this._purviewSettings = purviewSettings;
+    }
+
+    /// <summary>
+    /// Get a value from the cache.
+    /// </summary>
+    /// <typeparam name="TKey">The type of the key in the cache. Used for serialization.</typeparam>
+    /// <typeparam name="TValue">The type of the value in the cache. Used for serialization.</typeparam>
+    /// <param name="key">The key to look up in the cache.</param>
+    /// <param name="cancellationToken">A cancellation token for the async operation.</param>
+    /// <returns>The value in the cache. Null or default if no value is present.</returns>
+    public async Task<TValue?> GetAsync<TKey, TValue>(TKey key, CancellationToken cancellationToken)
+    {
+        JsonTypeInfo<TKey> keyTypeInfo = (JsonTypeInfo<TKey>)PurviewSerializationUtils.SerializationSettings.GetTypeInfo(typeof(TKey));
+        string serializedKey = JsonSerializer.Serialize(key, keyTypeInfo);
+        byte[]? data = await this._cache.GetAsync(serializedKey, cancellationToken).ConfigureAwait(false);
+        if (data == null)
+        {
+            return default;
+        }
+
+        JsonTypeInfo<TValue> valueTypeInfo = (JsonTypeInfo<TValue>)PurviewSerializationUtils.SerializationSettings.GetTypeInfo(typeof(TValue));
+
+        return JsonSerializer.Deserialize(data, valueTypeInfo);
+    }
+
+    /// <summary>
+    /// Set a value in the cache.
+    /// </summary>
+    /// <typeparam name="TKey">The type of the key in the cache. Used for serialization.</typeparam>
+    /// <typeparam name="TValue">The type of the value in the cache. Used for serialization.</typeparam>
+    /// <param name="key">The key to identify the cache entry.</param>
+    /// <param name="value">The value to cache.</param>
+    /// <param name="cancellationToken">A cancellation token for the async operation.</param>
+    /// <returns>A task for the async operation.</returns>
+    public Task SetAsync<TKey, TValue>(TKey key, TValue value, CancellationToken cancellationToken)
+    {
+        JsonTypeInfo<TKey> keyTypeInfo = (JsonTypeInfo<TKey>)PurviewSerializationUtils.SerializationSettings.GetTypeInfo(typeof(TKey));
+        string serializedKey = JsonSerializer.Serialize(key, keyTypeInfo);
+        JsonTypeInfo<TValue> valueTypeInfo = (JsonTypeInfo<TValue>)PurviewSerializationUtils.SerializationSettings.GetTypeInfo(typeof(TValue));
+        byte[] serializedValue = JsonSerializer.SerializeToUtf8Bytes(value, valueTypeInfo);
+
+        DistributedCacheEntryOptions cacheOptions = new() { AbsoluteExpirationRelativeToNow = this._purviewSettings.CacheTTL };
+
+        return this._cache.SetAsync(serializedKey, serializedValue, cacheOptions, cancellationToken);
+    }
+
+    /// <summary>
+    /// Removes a value from the cache.
+    /// </summary>
+    /// <typeparam name="TKey">The type of the key.</typeparam>
+    /// <param name="key">The key to identify the cache entry.</param>
+    /// <param name="cancellationToken">The cancellation token for the async operation.</param>
+    /// <returns>A task for the async operation.</returns>
+    public Task RemoveAsync<TKey>(TKey key, CancellationToken cancellationToken)
+    {
+        JsonTypeInfo<TKey> keyTypeInfo = (JsonTypeInfo<TKey>)PurviewSerializationUtils.SerializationSettings.GetTypeInfo(typeof(TKey));
+        string serializedKey = JsonSerializer.Serialize(key, keyTypeInfo);
+
+        return this._cache.RemoveAsync(serializedKey, cancellationToken);
+    }
+}

dotnet/src/Microsoft.Agents.AI.Purview/ChannelHandler.cs

@@ -0,0 +1,101 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using System.Collections.Generic;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using Microsoft.Agents.AI.Purview.Exceptions;
+using Microsoft.Agents.AI.Purview.Models.Jobs;
+using Microsoft.Extensions.Logging;
+
+namespace Microsoft.Agents.AI.Purview;
+
+/// <summary>
+/// Handler class for background job management.
+/// </summary>
+internal class ChannelHandler : IChannelHandler
+{
+    private readonly Channel<BackgroundJobBase> _jobChannel;
+    private readonly List<Task> _channelListeners;
+    private readonly ILogger _logger;
+    private readonly PurviewSettings _purviewSettings;
+
+    /// <summary>
+    /// Creates a new instance of JobHandler.
+    /// </summary>
+    /// <param name="purviewSettings"></param>
+    /// <param name="logger"></param>
+    /// <param name="jobChannel"></param>
+    public ChannelHandler(PurviewSettings purviewSettings, ILogger logger, Channel<BackgroundJobBase> jobChannel)
+    {
+        this._purviewSettings = purviewSettings;
+        this._logger = logger;
+        this._jobChannel = jobChannel;
+
+        this._channelListeners = new List<Task>(this._purviewSettings.MaxConcurrentJobConsumers);
+    }
+
+    /// <inheritdoc/>
+    public void QueueJob(BackgroundJobBase job)
+    {
+        try
+        {
+            if (job == null)
+            {
+                throw new PurviewJobException("Cannot queue null job.");
+            }
+
+            if (this._channelListeners.Count == 0)
+            {
+                this._logger.LogWarning("No listeners are available to process the job.");
+                throw new PurviewJobException("No listeners are available to process the job.");
+            }
+
+            bool canQueue = this._jobChannel.Writer.TryWrite(job);
+
+            if (!canQueue)
+            {
+                int jobCount = this._jobChannel.Reader.Count;
+                this._logger.LogError("Could not queue a job for background processing.");
+
+                if (this._jobChannel.Reader.Completion.IsCompleted)
+                {
+                    throw new PurviewJobException("Job channel is closed or completed. Cannot queue job.");
+                }
+                else if (jobCount >= this._purviewSettings.PendingBackgroundJobLimit)
+                {
+                    throw new PurviewJobLimitExceededException($"Job queue is full. Current pending jobs: {jobCount}. Maximum number of queued jobs: {this._purviewSettings.PendingBackgroundJobLimit}");
+                }
+                else
+                {
+                    throw new PurviewJobException("Could not queue job for background processing.");
+                }
+            }
+        }
+        catch (Exception e)
+        {
+            if (this._purviewSettings.IgnoreExceptions)
+            {
+                this._logger.LogError(e, "Error queuing job: {ExceptionMessage}", e.Message);
+            }
+            else
+            {
+                throw;
+            }
+        }
+    }
+
+    /// <inheritdoc/>
+    public void AddRunner(Func<Channel<BackgroundJobBase>, Task> runnerTask)
+    {
+        this._channelListeners.Add(Task.Run(async () => await runnerTask(this._jobChannel).ConfigureAwait(false)));
+    }
+
+    /// <inheritdoc/>
+    public async Task StopAndWaitForCompletionAsync()
+    {
+        this._jobChannel.Writer.Complete();
+        await this._jobChannel.Reader.Completion.ConfigureAwait(false);
+        await Task.WhenAll(this._channelListeners).ConfigureAwait(false);
+    }
+}