Compare commits

..
Author SHA1 Message Date
Tao Chen ed79671309 Comments 2026-03-31 17:00:56 -07:00
Tao Chen 408efac0e4 Fix formatting 2026-03-31 16:28:48 -07:00
Tao Chen 8862be263f Fix migration samples 2 2026-03-31 16:25:01 -07:00
Tao Chen 5e930f97ca Fix migration samples 2026-03-31 16:18:27 -07:00
164 changed files with 983 additions and 7203 deletions
+12 -12
View File
@@ -94,23 +94,23 @@ Create a simple Azure Responses Agent that writes a haiku about the Microsoft Ag
# Use `az login` to authenticate with Azure CLI
import os
import asyncio
from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from azure.identity import AzureCliCredential
async def main():
# Initialize a chat agent with Microsoft Foundry
# Initialize a chat agent with Azure OpenAI Responses
# the endpoint, deployment name, and api version can be set via environment variables
# or they can be passed in directly to the FoundryChatClient constructor
agent = Agent(
client=FoundryChatClient(
credential=AzureCliCredential(),
# project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
# model=os.environ["FOUNDRY_MODEL_DEPLOYMENT_NAME"],
),
name="HaikuBot",
instructions="You are an upbeat assistant that writes beautifully.",
# or they can be passed in directly to the AzureOpenAIResponsesClient constructor
agent = AzureOpenAIResponsesClient(
# endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
# deployment_name=os.environ["AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME"],
# api_version=os.environ["AZURE_OPENAI_API_VERSION"],
# api_key=os.environ["AZURE_OPENAI_API_KEY"], # Optional if using AzureCliCredential
credential=AzureCliCredential(), # Optional, if using api_key
).as_agent(
name="HaikuBot",
instructions="You are an upbeat assistant that writes beautifully.",
)
print(await agent.run("Write a haiku about Microsoft Agent Framework."))
-213
View File
@@ -1,213 +0,0 @@
---
name: verify-samples-tool
description: How to use the verify-samples tool to run, verify, and manage sample definitions in the Agent Framework repository. Use this when adding, updating, or running sample verification.
---
# verify-samples Tool
The `verify-samples` project (`dotnet/eng/verify-samples/`) is an automated tool that runs sample projects and verifies their output using deterministic checks and AI-powered verification.
## Running verify-samples
```bash
cd dotnet
# Run all samples across all categories
dotnet run --project eng/verify-samples -- --log results.log --csv results.csv
# Run a specific category
dotnet run --project eng/verify-samples -- --category 02-agents --log results.log
# Run specific samples by name
dotnet run --project eng/verify-samples -- Agent_Step02_StructuredOutput Agent_Step09_AsFunctionTool
# Control parallelism (default 8)
dotnet run --project eng/verify-samples -- --parallel 8 --log results.log
# Combine options
dotnet run --project eng/verify-samples -- --category 03-workflows --parallel 4 --log results.log --csv results.csv
```
### Required Environment Variables
The tool itself needs:
- `AZURE_OPENAI_ENDPOINT` — for the AI verification agent
- `AZURE_OPENAI_DEPLOYMENT_NAME` (optional, defaults to `gpt-5-mini`)
Individual samples require their own env vars (e.g., `AZURE_AI_PROJECT_ENDPOINT`). The tool automatically checks and skips samples with missing env vars.
### Output Files
- `--log results.log` — detailed per-sample log with stdout/stderr, AI reasoning, and a summary
- `--csv results.csv` — tabular summary with Sample, ProjectPath, Status, FailedChecks, and Failures columns
## Sample Categories
Definitions are in the `dotnet/eng/verify-samples/` directory:
| Category | Config File | Registered Key |
|----------|-------------|----------------|
| 01-get-started | `GetStartedSamples.cs` | `01-get-started` |
| 02-agents | `AgentsSamples.cs` | `02-agents` |
| 03-workflows | `WorkflowSamples.cs` | `03-workflows` |
Categories are registered in `VerifyOptions.cs` in the `s_sampleSets` dictionary.
## SampleDefinition Properties
Each sample is defined as a `SampleDefinition` in the appropriate config file. Key properties:
```csharp
new SampleDefinition
{
// Required: Display name for the sample
Name = "Agent_Step02_StructuredOutput",
// Required: Relative path from dotnet/ to the sample project directory
ProjectPath = "samples/02-agents/Agents/Agent_Step02_StructuredOutput",
// Environment variables the sample requires (throws if missing)
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
// Environment variables with defaults that would prompt on console if unset
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
// Skip this sample with a reason (for structural issues only)
SkipReason = null, // or "Requires external service X."
// Deterministic checks: substrings that must appear in stdout
MustContain = ["=== Section Header ==="],
// Substrings that must NOT appear in stdout
MustNotContain = [],
// If true, only MustContain checks are used (no AI verification)
IsDeterministic = false,
// AI verification: natural-language descriptions of expected output
// Each entry describes one aspect to verify independently
ExpectedOutputDescription =
[
"The output should show structured person information with Name, Age, and Occupation fields.",
"The output should not contain error messages or stack traces.",
],
// Stdin inputs to feed to the sample (for interactive samples)
Inputs = ["Y", "Y", "Y"],
// Delay between stdin inputs in ms (default 2000, increase for LLM calls between inputs)
InputDelayMs = 3000,
}
```
## How to Add a New Sample Definition
1. **Check the sample's Program.cs** to understand:
- What environment variables it reads (look for `GetEnvironmentVariable`)
- Whether it needs stdin input (look for `Console.ReadLine`, `Application.GetInput`)
- Whether it has an external loop (look for `EXIT` patterns in YAML workflows)
- What output it produces (section headers, markers, expected behavior)
- Whether it exits on its own or runs as a server
2. **Choose the right verification strategy:**
- **Deterministic** (`IsDeterministic = true`): Use `MustContain` for samples with fixed output strings. No AI verification.
- **AI-verified** (default): Use `ExpectedOutputDescription` with semantic descriptions. Write expectations that are flexible enough for non-deterministic LLM output.
- **Both**: Use `MustContain` for fixed markers AND `ExpectedOutputDescription` for LLM-generated content.
3. **Set `SkipReason` only for structural issues:**
- Web servers that don't exit
- Multi-process client/server architectures
- Samples requiring external infrastructure (MCP servers you can't reach, Docker, etc.)
- Do NOT skip for missing env vars — the tool checks those dynamically.
4. **For interactive samples, provide `Inputs`:**
- Samples using `Application.GetInput(args)` need one initial input
- Samples with `Console.ReadLine()` approval loops need `"Y"` inputs
- YAML workflows with `externalLoop` need `"EXIT"` as the last input
- Set `InputDelayMs` to 3000-8000ms for samples with LLM calls between inputs
5. **Add the definition** to the appropriate config file (e.g., `AgentsSamples.cs`) in the `All` list.
6. **Register new categories** (if needed) in `VerifyOptions.cs` `s_sampleSets` dictionary.
### Writing Good ExpectedOutputDescription
- Write descriptions that are **semantically flexible** — LLM output varies between runs
- Each array entry should describe **one independent aspect** to verify
- Always include `"The output should not contain error messages or stack traces."` as the last entry
- Avoid exact wording expectations — use "should mention", "should contain information about", "should show"
- Bad: `"The output should say 'The weather in Amsterdam is cloudy with a high of 15°C'"`
- Good: `"The output should contain weather information about Amsterdam mentioning cloudy weather with a high of 15°C."`
### Example: Simple LLM Sample
```csharp
new SampleDefinition
{
Name = "Agent_With_AzureOpenAIChatCompletion",
ProjectPath = "samples/02-agents/AgentProviders/Agent_With_AzureOpenAIChatCompletion",
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
ExpectedOutputDescription =
[
"The output should contain a joke about a pirate.",
"The output should not contain error messages or stack traces.",
],
},
```
### Example: Deterministic Sample
```csharp
new SampleDefinition
{
Name = "Workflow_Declarative_GenerateCode",
ProjectPath = "samples/03-workflows/Declarative/GenerateCode",
IsDeterministic = true,
MustContain = ["WORKFLOW: Parsing", "WORKFLOW: Defined"],
ExpectedOutputDescription = ["The output should show a YAML workflow being parsed and C# code being generated from it."],
},
```
### Example: Interactive Sample with Approval Loop
```csharp
new SampleDefinition
{
Name = "FoundryAgent_Hosted_MCP",
ProjectPath = "samples/02-agents/ModelContextProtocol/FoundryAgent_Hosted_MCP",
RequiredEnvironmentVariables = ["AZURE_AI_PROJECT_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
Inputs = ["Y", "Y", "Y", "Y", "Y"],
InputDelayMs = 5000,
ExpectedOutputDescription = ["The output should show an agent using the Microsoft Learn MCP tool with approval prompts."],
},
```
### Example: Declarative Workflow with External Loop
```csharp
new SampleDefinition
{
Name = "Workflow_Declarative_FunctionTools",
ProjectPath = "samples/03-workflows/Declarative/FunctionTools",
RequiredEnvironmentVariables = ["AZURE_AI_PROJECT_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
Inputs = ["What are today's specials?", "EXIT"],
InputDelayMs = 8000,
ExpectedOutputDescription = ["The output should show a workflow calling function tools to answer a question about restaurant specials."],
},
```
### Example: Skipped Sample
```csharp
new SampleDefinition
{
Name = "Agent_MCP_Server",
ProjectPath = "samples/02-agents/ModelContextProtocol/Agent_MCP_Server",
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
SkipReason = "Runs as an MCP stdio server that does not exit on its own.",
},
```
-2
View File
@@ -7,7 +7,6 @@
<Folder Name="/Samples/">
<File Path="samples/AGENTS.md" />
<File Path="samples/README.md" />
<Project Path="eng/verify-samples/verify-samples.csproj" />
</Folder>
<Folder Name="/Samples/01-get-started/">
<Project Path="samples/01-get-started/01_hello_agent/01_hello_agent.csproj" />
@@ -172,7 +171,6 @@
<Project Path="samples/02-agents/AgentWithRAG/AgentWithRAG_Step02_CustomVectorStoreRAG/AgentWithRAG_Step02_CustomVectorStoreRAG.csproj" />
<Project Path="samples/02-agents/AgentWithRAG/AgentWithRAG_Step03_CustomRAGDataSource/AgentWithRAG_Step03_CustomRAGDataSource.csproj" />
<Project Path="samples/02-agents/AgentWithRAG/AgentWithRAG_Step04_FoundryServiceRAG/AgentWithRAG_Step04_FoundryServiceRAG.csproj" />
<Project Path="samples/02-agents/AgentWithRAG/AgentWithRAG_Step05_Neo4jGraphRAG/AgentWithRAG_Step05_Neo4jGraphRAG.csproj" />
</Folder>
<Folder Name="/Samples/02-agents/ModelContextProtocol/">
<File Path="samples/02-agents/ModelContextProtocol/README.md" />
File diff suppressed because it is too large Load Diff
@@ -1,95 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
namespace VerifySamples;
/// <summary>
/// Thread-safe console output with sample-name prefixes and colored status.
/// </summary>
internal sealed class ConsoleReporter
{
private readonly object _lock = new();
/// <summary>
/// Writes a complete prefixed line atomically to the console.
/// </summary>
public void WriteLineWithPrefix(string sampleName, string message, ConsoleColor? color = null)
{
lock (this._lock)
{
Console.ForegroundColor = ConsoleColor.Cyan;
Console.Write($"[{sampleName}] ");
if (color.HasValue)
{
Console.ForegroundColor = color.Value;
}
else
{
Console.ResetColor();
}
Console.WriteLine(message);
Console.ResetColor();
}
}
/// <summary>
/// Prints the final summary table and elapsed time to the console.
/// </summary>
public void PrintSummary(
IReadOnlyList<VerificationResult> orderedResults,
IReadOnlyList<(string Name, string Reason)> skipped,
TimeSpan elapsed)
{
var passCount = orderedResults.Count(r => r.Passed);
var failCount = orderedResults.Count(r => !r.Passed);
Console.WriteLine();
Console.WriteLine(new string('─', 60));
Console.ForegroundColor = ConsoleColor.White;
Console.WriteLine("SUMMARY");
Console.ResetColor();
foreach (var result in orderedResults)
{
Console.ForegroundColor = result.Passed ? ConsoleColor.Green : ConsoleColor.Red;
Console.Write(result.Passed ? " ✓ " : " ✗ ");
Console.ResetColor();
Console.WriteLine($"{result.SampleName}: {result.Summary}");
}
foreach (var (name, reason) in skipped)
{
Console.ForegroundColor = ConsoleColor.Yellow;
Console.Write(" ○ ");
Console.ResetColor();
Console.WriteLine($"{name}: Skipped — {reason}");
}
Console.WriteLine();
Console.Write("Results: ");
Console.ForegroundColor = ConsoleColor.Green;
Console.Write($"{passCount} passed");
Console.ResetColor();
if (failCount > 0)
{
Console.Write(", ");
Console.ForegroundColor = ConsoleColor.Red;
Console.Write($"{failCount} failed");
Console.ResetColor();
}
if (skipped.Count > 0)
{
Console.Write(", ");
Console.ForegroundColor = ConsoleColor.Yellow;
Console.Write($"{skipped.Count} skipped");
Console.ResetColor();
}
Console.WriteLine();
Console.ForegroundColor = ConsoleColor.DarkGray;
Console.WriteLine($"Elapsed: {elapsed.Hours:D2}:{elapsed.Minutes:D2}:{elapsed.Seconds:D2}");
Console.ResetColor();
}
}
@@ -1,56 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Text;
namespace VerifySamples;
/// <summary>
/// Writes a CSV summary of sample verification results.
/// </summary>
internal static class CsvResultWriter
{
/// <summary>
/// Writes the results to a CSV file at the specified path.
/// </summary>
public static async Task WriteAsync(
string path,
IReadOnlyList<VerificationResult> orderedResults,
IReadOnlyList<(string Name, string Reason)> skipped,
IReadOnlyList<SampleDefinition> samples)
{
var pathLookup = samples.ToDictionary(s => s.Name, s => s.ProjectPath);
var sb = new StringBuilder();
sb.AppendLine("Sample,ProjectPath,Status,FailedChecks,Failures");
foreach (var result in orderedResults)
{
var status = result.Passed ? "PASSED" : "FAILED";
var failedChecks = result.Failures.Count;
var failures = string.Join("; ", result.Failures);
pathLookup.TryGetValue(result.SampleName, out var projectPath);
sb.AppendLine($"{CsvEscape(result.SampleName)},{CsvEscape(projectPath ?? "")},{status},{failedChecks},{CsvEscape(failures)}");
}
foreach (var (name, reason) in skipped)
{
pathLookup.TryGetValue(name, out var projectPath);
sb.AppendLine($"{CsvEscape(name)},{CsvEscape(projectPath ?? "")},SKIPPED,0,{CsvEscape(reason)}");
}
await File.WriteAllTextAsync(path, sb.ToString());
}
/// <summary>
/// Escapes a value for CSV: wraps in quotes if it contains commas, quotes, or newlines.
/// </summary>
private static string CsvEscape(string value)
{
if (value.Contains('"') || value.Contains(',') || value.Contains('\n') || value.Contains('\r'))
{
return $"\"{value.Replace("\"", "\"\"")}\"";
}
return value;
}
}
@@ -1,105 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
namespace VerifySamples;
/// <summary>
/// Defines the expected behavior for each sample in 01-get-started.
/// </summary>
internal static class GetStartedSamples
{
public static IReadOnlyList<SampleDefinition> All { get; } =
[
new SampleDefinition
{
Name = "05_first_workflow",
ProjectPath = "samples/01-get-started/05_first_workflow",
RequiredEnvironmentVariables = [],
IsDeterministic = true,
MustContain =
[
"UppercaseExecutor: HELLO, WORLD!",
"ReverseTextExecutor: !DLROW ,OLLEH",
],
},
new SampleDefinition
{
Name = "01_hello_agent",
ProjectPath = "samples/01-get-started/01_hello_agent",
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
ExpectedOutputDescription =
[
"The output should contain a joke about a pirate.",
"There should be two separate joke responses — one from a non-streaming call and one from a streaming call.",
"The output should not contain error messages or stack traces.",
],
},
new SampleDefinition
{
Name = "02_add_tools",
ProjectPath = "samples/01-get-started/02_add_tools",
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
MustContain = [],
ExpectedOutputDescription =
[
"The output should contain information about the weather in Amsterdam.",
"The response should mention that it is cloudy with a high of 15°C (or equivalent), since this comes from a tool that returns a canned response.",
"There should be two responses — one from a non-streaming call and one from a streaming call.",
"The output should not contain error messages or stack traces.",
],
},
new SampleDefinition
{
Name = "03_multi_turn",
ProjectPath = "samples/01-get-started/03_multi_turn",
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
ExpectedOutputDescription =
[
"The output should contain a joke about a pirate.",
"After the initial joke, there should be a modified version that includes emojis and is told in the voice of a pirate's parrot.",
"The pattern repeats: first a non-streaming pirate joke + parrot version, then a streaming pirate joke + parrot version.",
"The output should not contain error messages or stack traces.",
],
},
new SampleDefinition
{
Name = "04_memory",
ProjectPath = "samples/01-get-started/04_memory",
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
MustContain =
[
">> Use session with blank memory",
">> Use deserialized session with previously created memories",
">> Read memories using memory component",
"MEMORY - User Name:",
"MEMORY - User Age:",
">> Use new session with previously created memories",
],
ExpectedOutputDescription =
[
"In the 'Use session with blank memory' section, the agent should respond to the user's messages. It may ask for the user's name or age if not yet known.",
"In the 'Use deserialized session with previously created memories' section, the agent should correctly recall that the user's name is Ruaidhrí and age is 20.",
"The 'MEMORY - User Name:' line should show 'Ruaidhrí' (or a close transliteration).",
"The 'MEMORY - User Age:' line should show '20'.",
"In the 'Use new session with previously created memories' section, the agent should know the user's name and age from the transferred memory.",
"The output should not contain error messages or stack traces.",
],
},
new SampleDefinition
{
Name = "06_host_your_agent",
ProjectPath = "samples/01-get-started/06_host_your_agent",
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
SkipReason = "Requires Azure Functions Core Tools runtime and starts a web server.",
},
];
}
-153
View File
@@ -1,153 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Text;
namespace VerifySamples;
/// <summary>
/// Incrementally writes a sequential (non-interleaved) log file, appending after each sample completes.
/// Thread-safe: multiple parallel tasks may call write methods concurrently.
/// </summary>
internal sealed class LogFileWriter : IDisposable
{
private readonly string _path;
private readonly SemaphoreSlim _writeLock = new(1, 1);
public LogFileWriter(string path)
{
this._path = path;
}
/// <inheritdoc />
public void Dispose()
{
this._writeLock.Dispose();
}
/// <summary>
/// Writes the log file header. Call once at the start of the run.
/// </summary>
public async Task WriteHeaderAsync()
{
var sb = new StringBuilder();
sb.AppendLine($"Sample Verification Log — {DateTime.UtcNow:yyyy-MM-dd HH:mm:ss} UTC");
sb.AppendLine(new string('═', 72));
sb.AppendLine();
await File.WriteAllTextAsync(this._path, sb.ToString());
}
/// <summary>
/// Appends a skipped-sample entry to the log file.
/// </summary>
public async Task WriteSkippedAsync(string name, string reason)
{
var sb = new StringBuilder();
sb.AppendLine($"── {name} ──");
sb.AppendLine($"Status: SKIPPED — {reason}");
sb.AppendLine();
await this.AppendAsync(sb.ToString());
}
/// <summary>
/// Appends a completed sample's full output section to the log file.
/// </summary>
public async Task WriteSampleResultAsync(VerificationResult result)
{
var sb = new StringBuilder();
sb.AppendLine(new string('─', 72));
sb.AppendLine($"── {result.SampleName} ──");
sb.AppendLine($"Status: {(result.Passed ? "PASSED" : "FAILED")}");
sb.AppendLine();
foreach (var line in result.LogLines)
{
sb.AppendLine(line);
}
sb.AppendLine();
if (!string.IsNullOrWhiteSpace(result.Stdout))
{
sb.AppendLine("--- stdout ---");
sb.AppendLine(result.Stdout.TrimEnd());
sb.AppendLine("--- end stdout ---");
sb.AppendLine();
}
if (!string.IsNullOrWhiteSpace(result.Stderr))
{
sb.AppendLine("--- stderr ---");
sb.AppendLine(result.Stderr.TrimEnd());
sb.AppendLine("--- end stderr ---");
sb.AppendLine();
}
if (result.Failures.Count > 0)
{
sb.AppendLine("Failures:");
foreach (var failure in result.Failures)
{
sb.AppendLine($" ✗ {failure}");
}
sb.AppendLine();
}
if (result.AIReasoning is not null)
{
sb.AppendLine("AI Reasoning:");
sb.AppendLine(result.AIReasoning);
sb.AppendLine();
}
await this.AppendAsync(sb.ToString());
}
/// <summary>
/// Appends the final summary section and elapsed time to the log file.
/// </summary>
public async Task WriteSummaryAsync(
IReadOnlyList<VerificationResult> orderedResults,
IReadOnlyList<(string Name, string Reason)> skipped,
TimeSpan elapsed)
{
var passCount = orderedResults.Count(r => r.Passed);
var failCount = orderedResults.Count(r => !r.Passed);
var sb = new StringBuilder();
sb.AppendLine(new string('═', 72));
sb.AppendLine("SUMMARY");
sb.AppendLine();
foreach (var result in orderedResults)
{
sb.AppendLine($" {(result.Passed ? "" : "")} {result.SampleName}: {result.Summary}");
}
foreach (var (name, reason) in skipped)
{
sb.AppendLine($" ○ {name}: Skipped — {reason}");
}
sb.AppendLine();
sb.AppendLine($"Results: {passCount} passed{(failCount > 0 ? $", {failCount} failed" : "")}{(skipped.Count > 0 ? $", {skipped.Count} skipped" : "")}");
sb.AppendLine($"Elapsed: {elapsed.Hours:D2}:{elapsed.Minutes:D2}:{elapsed.Seconds:D2}");
await this.AppendAsync(sb.ToString());
}
private async Task AppendAsync(string text)
{
await this._writeLock.WaitAsync();
try
{
await File.AppendAllTextAsync(this._path, text);
}
finally
{
this._writeLock.Release();
}
}
}
-98
View File
@@ -1,98 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
// This tool runs the 01-get-started, 02-agents, and 03-workflows samples and verifies their output.
// Deterministic samples are verified with exact string matching.
// Non-deterministic (LLM) samples are verified using an agent-framework agent.
//
// Usage:
// dotnet run # Run all samples
// dotnet run -- 01_hello_agent 05_first_workflow # Run specific samples by name
// dotnet run -- --category 01-get-started # Run the 01-get-started category
// dotnet run -- --category 02-agents # Run the 02-agents category
// dotnet run -- --category 03-workflows # Run the 03-workflows category
// dotnet run -- --parallel 16 # Run up to 16 samples concurrently
// dotnet run -- --log results.log # Write sequential log to file
// dotnet run -- --csv results.csv # Write CSV summary to file
//
// Required environment variables (for AI-powered samples):
// AZURE_OPENAI_ENDPOINT
// AZURE_OPENAI_DEPLOYMENT_NAME (optional, defaults to gpt-5-mini)
using System.Diagnostics;
using Azure.AI.OpenAI;
using Azure.Identity;
using VerifySamples;
var options = VerifyOptions.Parse(args);
if (options is null)
{
return 1;
}
var stopwatch = Stopwatch.StartNew();
// Resolve the dotnet/ root directory (verify-samples is at dotnet/eng/verify-samples/)
var dotnetRoot = Path.GetFullPath(Path.Combine(AppContext.BaseDirectory, "..", "..", "..", "..", ".."));
if (!File.Exists(Path.Combine(dotnetRoot, "agent-framework-dotnet.slnx")))
{
dotnetRoot = Path.GetFullPath(Path.Combine(Directory.GetCurrentDirectory(), "..", ".."));
}
// Set up the AI verifier
var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT");
var deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT_NAME") ?? "gpt-5-mini";
OpenAI.Chat.ChatClient? chatClient = null;
if (!string.IsNullOrEmpty(endpoint))
{
chatClient = new AzureOpenAIClient(new Uri(endpoint), new DefaultAzureCredential())
.GetChatClient(deploymentName);
}
// Set up optional log file writer
LogFileWriter? logWriter = null;
if (options.LogFilePath is not null)
{
logWriter = new LogFileWriter(options.LogFilePath);
await logWriter.WriteHeaderAsync();
}
try
{
// Run all samples
var reporter = new ConsoleReporter();
var verifier = new SampleVerifier(chatClient);
var orchestrator = new VerificationOrchestrator(verifier, reporter, dotnetRoot, TimeSpan.FromMinutes(3), logWriter);
var run = await orchestrator.RunAllAsync(options.Samples, options.MaxParallelism);
stopwatch.Stop();
// Print summary
var orderedResults = run.SampleOrder
.Where(run.Results.ContainsKey)
.Select(name => run.Results[name])
.ToList();
reporter.PrintSummary(orderedResults, run.Skipped, stopwatch.Elapsed);
// Write log file summary
if (logWriter is not null)
{
await logWriter.WriteSummaryAsync(orderedResults, run.Skipped, stopwatch.Elapsed);
Console.WriteLine($"Log written to: {options.LogFilePath}");
}
// Write CSV summary
if (options.CsvFilePath is not null)
{
await CsvResultWriter.WriteAsync(options.CsvFilePath, orderedResults, run.Skipped, options.Samples);
Console.WriteLine($"CSV written to: {options.CsvFilePath}");
}
return orderedResults.Any(r => !r.Passed) ? 1 : 0;
}
finally
{
logWriter?.Dispose();
}
@@ -1,79 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
namespace VerifySamples;
/// <summary>
/// Describes a sample to verify, including its expected output.
/// </summary>
internal sealed class SampleDefinition
{
/// <summary>
/// Display name for the sample (e.g., "01_hello_agent").
/// </summary>
public required string Name { get; init; }
/// <summary>
/// Relative path from the dotnet/ directory to the sample project directory.
/// </summary>
public required string ProjectPath { get; init; }
/// <summary>
/// Environment variables that the sample requires for a meaningful run.
/// The runner checks these before running and will skip the sample if any are unset,
/// recording a skip reason that indicates which required variables are missing.
/// </summary>
public string[] RequiredEnvironmentVariables { get; init; } = [];
/// <summary>
/// Environment variables that the sample can use but typically has fallbacks or defaults for.
/// If these are not set, the sample might prompt or behave interactively, which could cause
/// automated verification to hang. The runner checks these and skips the sample if they are unset
/// to avoid non-deterministic or blocking behavior in automated runs.
/// </summary>
public string[] OptionalEnvironmentVariables { get; init; } = [];
/// <summary>
/// If set, the sample is skipped with this reason.
/// Use only for structural reasons (e.g., web server, multi-process, needs external service).
/// Do NOT use for missing environment variables — those are checked dynamically.
/// </summary>
public string? SkipReason { get; init; }
/// <summary>
/// Substrings that must appear in stdout for the sample to pass.
/// Used for deterministic verification.
/// </summary>
public string[] MustContain { get; init; } = [];
/// <summary>
/// Substrings that must not appear in stdout for the sample to pass.
/// </summary>
public string[] MustNotContain { get; init; } = [];
/// <summary>
/// If true, <see cref="MustContain"/> entries cover the entire expected output —
/// no AI verification is needed.
/// </summary>
public bool IsDeterministic { get; init; }
/// <summary>
/// Natural-language description of what the sample output should look like.
/// Used by the AI verifier for non-deterministic samples.
/// Each entry describes one aspect of the expected output that should be verified.
/// </summary>
public string[] ExpectedOutputDescription { get; init; } = [];
/// <summary>
/// Sequence of stdin inputs to feed to the sample process.
/// Each entry is written as a line (followed by newline) to the process stdin.
/// A <c>null</c> entry inserts a delay without writing anything.
/// Inputs are sent with a short delay between each to allow the process to prompt.
/// </summary>
public string?[] Inputs { get; init; } = [];
/// <summary>
/// Delay in milliseconds between each input line. Default is 2000ms.
/// Increase for samples that need more time between prompts (e.g., LLM calls between inputs).
/// </summary>
public int InputDelayMs { get; init; } = 2000;
}
-132
View File
@@ -1,132 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics;
namespace VerifySamples;
/// <summary>
/// Result of running a sample process.
/// </summary>
internal sealed record SampleRunResult(
string Stdout,
string Stderr,
int ExitCode,
TimeSpan Elapsed);
/// <summary>
/// Runs a sample project via <c>dotnet run</c> and captures its output.
/// </summary>
internal static class SampleRunner
{
/// <summary>
/// Runs <c>dotnet run --framework net10.0</c> in the given project directory.
/// </summary>
public static Task<SampleRunResult> RunAsync(
string projectPath,
TimeSpan timeout,
CancellationToken cancellationToken = default)
=> RunAsync(projectPath, "run --framework net10.0", timeout, inputs: null, inputDelayMs: 0, cancellationToken: cancellationToken);
/// <summary>
/// Runs <c>dotnet run --framework net10.0</c> with stdin inputs.
/// </summary>
public static Task<SampleRunResult> RunAsync(
string projectPath,
TimeSpan timeout,
string?[]? inputs,
int inputDelayMs = 2000,
CancellationToken cancellationToken = default)
=> RunAsync(projectPath, "run --framework net10.0", timeout, inputs, inputDelayMs, cancellationToken);
/// <summary>
/// Runs an arbitrary <c>dotnet</c> command in the given working directory.
/// </summary>
public static async Task<SampleRunResult> RunAsync(
string workingDirectory,
string dotnetArgs,
TimeSpan timeout,
string?[]? inputs = null,
int inputDelayMs = 0,
CancellationToken cancellationToken = default)
{
var psi = new ProcessStartInfo
{
FileName = "dotnet",
Arguments = dotnetArgs,
WorkingDirectory = workingDirectory,
RedirectStandardOutput = true,
RedirectStandardError = true,
RedirectStandardInput = inputs is { Length: > 0 },
UseShellExecute = false,
CreateNoWindow = true,
};
var sw = Stopwatch.StartNew();
using var process = new Process { StartInfo = psi };
process.Start();
var stdoutTask = process.StandardOutput.ReadToEndAsync(cancellationToken);
var stderrTask = process.StandardError.ReadToEndAsync(cancellationToken);
// Feed stdin inputs with delays if configured
if (inputs is { Length: > 0 })
{
_ = Task.Run(async () =>
{
try
{
foreach (var input in inputs)
{
await Task.Delay(inputDelayMs, cancellationToken);
if (input is not null)
{
await process.StandardInput.WriteLineAsync(input.AsMemory(), cancellationToken);
await process.StandardInput.FlushAsync(cancellationToken);
}
}
process.StandardInput.Close();
}
catch (Exception ex) when (ex is IOException or ObjectDisposedException or OperationCanceledException)
{
// Process may have exited before all inputs were sent
}
}, cancellationToken);
}
using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
cts.CancelAfter(timeout);
try
{
await process.WaitForExitAsync(cts.Token);
}
catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
{
// Timeout — kill the process
try
{
process.Kill(entireProcessTree: true);
}
catch
{
// Best effort
}
sw.Stop();
return new SampleRunResult(
Stdout: await stdoutTask,
Stderr: $"TIMEOUT: Sample did not complete within {timeout.TotalSeconds}s.\n{await stderrTask}",
ExitCode: -1,
Elapsed: sw.Elapsed);
}
sw.Stop();
return new SampleRunResult(
Stdout: await stdoutTask,
Stderr: await stderrTask,
ExitCode: process.ExitCode,
Elapsed: sw.Elapsed);
}
}
-202
View File
@@ -1,202 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Text.Json.Serialization;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;
using OpenAI.Chat;
namespace VerifySamples;
/// <summary>
/// Verifies sample output using deterministic checks and an AI agent
/// for non-deterministic output validation.
/// </summary>
internal sealed class SampleVerifier
{
private readonly AIAgent? _verifierAgent;
/// <summary>
/// Creates a verifier. If <paramref name="chatClient"/> is provided,
/// AI-based verification is available for non-deterministic samples.
/// </summary>
public SampleVerifier(ChatClient? chatClient = null)
{
if (chatClient is not null)
{
this._verifierAgent = chatClient.AsAIAgent(
instructions: """
You are a test output verifier. You will be given:
1. The actual stdout output of a program
2. A list of expectations about what the output should contain or demonstrate
Your job is to determine whether the actual output satisfies each expectation.
Be reasonable the output comes from an LLM so exact wording won't match, but the
semantic intent should be clearly satisfied.
""",
name: "OutputVerifier");
}
}
/// <summary>
/// Verifies the output of a sample run against its definition.
/// </summary>
public async Task<VerificationResult> VerifyAsync(SampleDefinition sample, SampleRunResult run)
{
var failures = new List<string>();
// 1. Exit code check
if (run.ExitCode != 0)
{
failures.Add($"Exit code was {run.ExitCode}, expected 0. Stderr: {Truncate(run.Stderr, 500)}");
}
// 2. Must-contain checks
foreach (var expected in sample.MustContain)
{
if (!run.Stdout.Contains(expected, StringComparison.Ordinal))
{
failures.Add($"Output missing expected substring: \"{expected}\"");
}
}
// 3. Must-not-contain checks
foreach (var unexpected in sample.MustNotContain)
{
if (run.Stdout.Contains(unexpected, StringComparison.Ordinal))
{
failures.Add($"Output contains unexpected substring: \"{unexpected}\"");
}
}
// 4. AI verification for non-deterministic samples
string? aiReasoning = null;
if (!sample.IsDeterministic && sample.ExpectedOutputDescription.Length > 0)
{
if (this._verifierAgent is null)
{
failures.Add("AI verification required but no AI agent configured (missing AZURE_OPENAI_ENDPOINT).");
}
else
{
var aiResult = await this.VerifyWithAIAsync(run.Stdout, sample.ExpectedOutputDescription);
aiReasoning = aiResult.Reasoning;
foreach (var unmet in aiResult.UnmetExpectations)
{
failures.Add($"AI expectation not met: {unmet}");
}
}
}
bool passed = failures.Count == 0;
return new VerificationResult
{
SampleName = sample.Name,
Passed = passed,
Summary = passed ? "All checks passed" : $"{failures.Count} check(s) failed",
Failures = failures,
AIReasoning = aiReasoning,
};
}
private async Task<(string Reasoning, List<string> UnmetExpectations)> VerifyWithAIAsync(
string actualOutput,
string[] expectations)
{
var expectationList = string.Join("\n", expectations.Select((e, i) => $" {i + 1}. {e}"));
var prompt = $"""
Actual program output:
---
{Truncate(actualOutput, 4000)}
---
Expectations to verify:
{expectationList}
Does the output satisfy all expectations?
""";
try
{
var response = await this._verifierAgent!.RunAsync<AIVerificationResponse>(prompt);
var result = response.Result;
if (result is null)
{
return ($"AI verification returned null result. Raw: {response.Text}", ["AI verification returned null result."]);
}
var reasoning = result.Reasoning ?? "(no reasoning provided)";
// Collect unmet expectations as individual failures
var unmet = new List<string>();
if (result.ExpectationResults is { Count: > 0 })
{
foreach (var er in result.ExpectationResults.Where(er => !er.Met))
{
var detail = string.IsNullOrWhiteSpace(er.Detail) ? er.Expectation : $"{er.Expectation} — {er.Detail}";
unmet.Add(detail ?? "Unknown expectation");
}
// If the model flagged overall failure but all individual expectations were met,
// still treat as failure using the overall reasoning.
if (unmet.Count == 0 && !result.Pass)
{
unmet.Add(reasoning);
}
}
else if (!result.Pass)
{
// Fallback: no per-expectation detail but overall pass is false
unmet.Add(reasoning);
}
return (reasoning, unmet);
}
catch (Exception ex)
{
return ($"AI verification error: {ex.Message}", [$"AI verification error: {ex.Message}"]);
}
}
private static string Truncate(string text, int maxLength)
=> text.Length <= maxLength ? text : text[..maxLength] + "... (truncated)";
}
/// <summary>
/// Structured response from the AI verification agent.
/// </summary>
[System.Diagnostics.CodeAnalysis.SuppressMessage("Performance", "CA1812:Avoid uninstantiated internal classes", Justification = "Instantiated by JSON deserialization via RunAsync<T>.")]
internal sealed class AIVerificationResponse
{
/// <summary>Whether all expectations were met.</summary>
[JsonPropertyName("pass")]
public bool Pass { get; set; }
/// <summary>Brief explanation of the overall assessment.</summary>
[JsonPropertyName("reasoning")]
public string? Reasoning { get; set; }
/// <summary>Per-expectation results.</summary>
[JsonPropertyName("expectation_results")]
public List<ExpectationResult>? ExpectationResults { get; set; }
}
/// <summary>
/// Result for an individual expectation check.
/// </summary>
[System.Diagnostics.CodeAnalysis.SuppressMessage("Performance", "CA1812:Avoid uninstantiated internal classes", Justification = "Instantiated by JSON deserialization via RunAsync<T>.")]
internal sealed class ExpectationResult
{
/// <summary>The expectation text that was evaluated.</summary>
[JsonPropertyName("expectation")]
public string? Expectation { get; set; }
/// <summary>Whether this expectation was met.</summary>
[JsonPropertyName("met")]
public bool Met { get; set; }
/// <summary>Detail about how the expectation was or was not met.</summary>
[JsonPropertyName("detail")]
public string? Detail { get; set; }
}
@@ -1,197 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Collections.Concurrent;
namespace VerifySamples;
/// <summary>
/// Orchestrates sample verification: filters, runs in parallel, and collects results.
/// </summary>
internal sealed class VerificationOrchestrator
{
private readonly SampleVerifier _verifier;
private readonly ConsoleReporter _reporter;
private readonly LogFileWriter? _logWriter;
private readonly string _dotnetRoot;
private readonly TimeSpan _timeout;
public VerificationOrchestrator(
SampleVerifier verifier,
ConsoleReporter reporter,
string dotnetRoot,
TimeSpan timeout,
LogFileWriter? logWriter = null)
{
this._verifier = verifier;
this._reporter = reporter;
this._logWriter = logWriter;
this._dotnetRoot = dotnetRoot;
this._timeout = timeout;
}
/// <summary>
/// The result of running all samples through the orchestrator.
/// </summary>
internal sealed record RunAllResult(
ConcurrentDictionary<string, VerificationResult> Results,
List<(string Name, string Reason)> Skipped,
List<string> SampleOrder);
/// <summary>
/// Filters samples, runs the runnable ones in parallel, and returns all results.
/// </summary>
public async Task<RunAllResult> RunAllAsync(
IReadOnlyList<SampleDefinition> samples,
int maxParallelism)
{
var skipped = new List<(string Name, string Reason)>();
var runnableSamples = new List<SampleDefinition>();
var sampleOrder = new List<string>();
// Separate samples into skipped and runnable
foreach (var sample in samples)
{
sampleOrder.Add(sample.Name);
if (sample.SkipReason is not null)
{
skipped.Add((sample.Name, sample.SkipReason));
this._reporter.WriteLineWithPrefix(sample.Name, $"SKIPPED — {sample.SkipReason}", ConsoleColor.Yellow);
if (this._logWriter is not null)
{
await this._logWriter.WriteSkippedAsync(sample.Name, sample.SkipReason);
}
continue;
}
var missingRequired = sample.RequiredEnvironmentVariables
.Where(v => string.IsNullOrEmpty(Environment.GetEnvironmentVariable(v)))
.ToList();
var missingOptional = sample.OptionalEnvironmentVariables
.Where(v => string.IsNullOrEmpty(Environment.GetEnvironmentVariable(v)))
.ToList();
if (missingRequired.Count > 0 || missingOptional.Count > 0)
{
var reasons = new List<string>();
if (missingRequired.Count > 0)
{
reasons.Add($"Missing required: {string.Join(", ", missingRequired)}");
}
if (missingOptional.Count > 0)
{
reasons.Add($"Missing optional (would cause console prompt hang): {string.Join(", ", missingOptional)}");
}
var skipReason = string.Join("; ", reasons);
skipped.Add((sample.Name, skipReason));
this._reporter.WriteLineWithPrefix(sample.Name, $"SKIPPED — {skipReason}", ConsoleColor.Yellow);
if (this._logWriter is not null)
{
await this._logWriter.WriteSkippedAsync(sample.Name, skipReason);
}
continue;
}
runnableSamples.Add(sample);
}
// Run samples in parallel
var results = new ConcurrentDictionary<string, VerificationResult>();
var semaphore = new SemaphoreSlim(maxParallelism);
this._reporter.WriteLineWithPrefix(
"runner", $"Running {runnableSamples.Count} samples (max {maxParallelism} parallel)...");
try
{
var tasks = runnableSamples.Select(sample => this.RunSingleAsync(sample, results, semaphore)).ToArray();
await Task.WhenAll(tasks);
}
finally
{
semaphore.Dispose();
}
return new RunAllResult(results, skipped, sampleOrder);
}
private async Task RunSingleAsync(
SampleDefinition sample,
ConcurrentDictionary<string, VerificationResult> results,
SemaphoreSlim semaphore)
{
await semaphore.WaitAsync();
try
{
var log = new List<string>();
log.Add($"[{sample.Name}] Running...");
this._reporter.WriteLineWithPrefix(sample.Name, "Running...");
var projectPath = Path.Combine(this._dotnetRoot, sample.ProjectPath);
var run = sample.Inputs.Length > 0
? await SampleRunner.RunAsync(projectPath, this._timeout, sample.Inputs, sample.InputDelayMs)
: await SampleRunner.RunAsync(projectPath, this._timeout);
log.Add($"[{sample.Name}] Completed ({run.Elapsed.TotalSeconds:F1}s, exit={run.ExitCode})");
this._reporter.WriteLineWithPrefix(
sample.Name, $"Completed ({run.Elapsed.TotalSeconds:F1}s, exit={run.ExitCode}). Verifying...");
var result = await this._verifier.VerifyAsync(sample, run);
if (result.Passed)
{
log.Add($"[{sample.Name}] PASSED");
this._reporter.WriteLineWithPrefix(sample.Name, "PASSED", ConsoleColor.Green);
}
else
{
log.Add($"[{sample.Name}] FAILED");
this._reporter.WriteLineWithPrefix(sample.Name, "FAILED", ConsoleColor.Red);
foreach (var failure in result.Failures)
{
log.Add($"[{sample.Name}] ✗ {failure}");
this._reporter.WriteLineWithPrefix(sample.Name, $" ✗ {failure}", ConsoleColor.Red);
}
}
if (result.AIReasoning is not null)
{
log.Add($"[{sample.Name}] AI: {result.AIReasoning}");
this._reporter.WriteLineWithPrefix(
sample.Name, $" AI: {Truncate(result.AIReasoning, 300)}", ConsoleColor.DarkGray);
}
var verificationResult = new VerificationResult
{
SampleName = result.SampleName,
Passed = result.Passed,
Summary = result.Summary,
Failures = result.Failures,
AIReasoning = result.AIReasoning,
Stdout = run.Stdout,
Stderr = run.Stderr,
LogLines = log,
};
results[sample.Name] = verificationResult;
if (this._logWriter is not null)
{
await this._logWriter.WriteSampleResultAsync(verificationResult);
}
}
finally
{
semaphore.Release();
}
}
private static string Truncate(string text, int maxLength)
=> text.Length <= maxLength ? text : text[..maxLength] + "...";
}
@@ -1,31 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
namespace VerifySamples;
/// <summary>
/// The result of verifying a single sample.
/// </summary>
internal sealed class VerificationResult
{
public required string SampleName { get; init; }
public required bool Passed { get; init; }
public required string Summary { get; init; }
public List<string> Failures { get; init; } = [];
public string? AIReasoning { get; init; }
/// <summary>
/// The sample's stdout output, captured for log file output.
/// </summary>
public string? Stdout { get; init; }
/// <summary>
/// The sample's stderr output, captured for log file output.
/// </summary>
public string? Stderr { get; init; }
/// <summary>
/// Per-sample log lines, buffered during parallel execution
/// and written sequentially to the log file.
/// </summary>
public List<string> LogLines { get; init; } = [];
}
-124
View File
@@ -1,124 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
namespace VerifySamples;
/// <summary>
/// Parsed command-line options for the sample verification tool.
/// </summary>
internal sealed class VerifyOptions
{
/// <summary>
/// Maximum number of samples to run concurrently.
/// </summary>
public int MaxParallelism { get; init; } = 8;
/// <summary>
/// Path to write a CSV summary file, or <c>null</c> to skip.
/// </summary>
public string? CsvFilePath { get; init; }
/// <summary>
/// Path to write a sequential log file, or <c>null</c> to skip.
/// </summary>
public string? LogFilePath { get; init; }
/// <summary>
/// The filtered list of samples to process.
/// </summary>
public required IReadOnlyList<SampleDefinition> Samples { get; init; }
/// <summary>
/// All known sample set registries, keyed by category name.
/// </summary>
private static readonly Dictionary<string, IReadOnlyList<SampleDefinition>> s_sampleSets =
new(StringComparer.OrdinalIgnoreCase)
{
["01-get-started"] = GetStartedSamples.All,
["02-agents"] = AgentsSamples.All,
["03-workflows"] = WorkflowSamples.All,
};
/// <summary>
/// Parses command-line arguments and resolves the sample list.
/// Returns <c>null</c> and writes to stderr if the arguments are invalid.
/// </summary>
public static VerifyOptions? Parse(string[] args)
{
var argList = args.ToList();
var categoryFilter = ExtractArg(argList, "--category");
var logFilePath = ExtractArg(argList, "--log");
var csvFilePath = ExtractArg(argList, "--csv");
int maxParallelism = 8;
var parallelArg = ExtractArg(argList, "--parallel");
if (parallelArg is not null && int.TryParse(parallelArg, out var p) && p > 0)
{
maxParallelism = p;
}
HashSet<string>? nameFilter = null;
if (argList.Count > 0)
{
nameFilter = argList.ToHashSet(StringComparer.OrdinalIgnoreCase);
}
// Build the sample list
IReadOnlyList<SampleDefinition> samples;
if (categoryFilter is not null)
{
if (!s_sampleSets.TryGetValue(categoryFilter, out var categoryList))
{
Console.Error.WriteLine(
$"Unknown category '{categoryFilter}'. Available: {string.Join(", ", s_sampleSets.Keys)}");
return null;
}
samples = categoryList;
}
else
{
samples = s_sampleSets.Values.SelectMany(s => s).ToList();
}
if (nameFilter is not null)
{
samples = samples.Where(s => nameFilter.Contains(s.Name)).ToList();
}
if (samples.Count == 0)
{
var allNames = s_sampleSets.Values.SelectMany(s => s).Select(s => s.Name);
Console.Error.WriteLine($"No matching samples found. Available: {string.Join(", ", allNames)}");
return null;
}
return new VerifyOptions
{
MaxParallelism = maxParallelism,
LogFilePath = logFilePath,
CsvFilePath = csvFilePath,
Samples = samples,
};
}
private static string? ExtractArg(List<string> list, string flag)
{
var idx = list.IndexOf(flag);
if (idx < 0)
{
return null;
}
if (idx + 1 >= list.Count)
{
Console.Error.WriteLine($"Missing value for {flag}.");
list.RemoveAt(idx);
return null;
}
var value = list[idx + 1];
list.RemoveRange(idx, 2);
return value;
}
}
@@ -1,525 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
namespace VerifySamples;
/// <summary>
/// Defines the expected behavior for each sample in 03-workflows.
/// </summary>
internal static class WorkflowSamples
{
public static IReadOnlyList<SampleDefinition> All { get; } =
[
// ───────────────────────────────────────────────────────────────────
// _StartHere
// ───────────────────────────────────────────────────────────────────
new SampleDefinition
{
Name = "Workflow_StartHere_01_Streaming",
ProjectPath = "samples/03-workflows/_StartHere/01_Streaming",
RequiredEnvironmentVariables = [],
IsDeterministic = true,
MustContain =
[
"UppercaseExecutor: HELLO, WORLD!",
"ReverseTextExecutor: !DLROW ,OLLEH",
],
},
new SampleDefinition
{
Name = "Workflow_StartHere_02_AgentsInWorkflows",
ProjectPath = "samples/03-workflows/_StartHere/02_AgentsInWorkflows",
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
ExpectedOutputDescription =
[
"The output should show agent responses from a translation workflow.",
"The output should not contain error messages or stack traces.",
],
},
new SampleDefinition
{
Name = "Workflow_StartHere_03_AgentWorkflowPatterns",
ProjectPath = "samples/03-workflows/_StartHere/03_AgentWorkflowPatterns",
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
Inputs = ["sequential"],
InputDelayMs = 3000,
ExpectedOutputDescription =
[
"The output should show a sequential workflow pattern with multiple agents executing tasks in order.",
"The output should not contain error messages or stack traces.",
],
},
new SampleDefinition
{
Name = "Workflow_StartHere_04_MultiModelService",
ProjectPath = "samples/03-workflows/_StartHere/04_MultiModelService",
RequiredEnvironmentVariables = ["BEDROCK_ACCESS_KEY", "BEDROCK_SECRET_KEY", "ANTHROPIC_API_KEY", "OPENAI_API_KEY"],
SkipReason = "Requires multiple external provider API keys (Bedrock, Anthropic, OpenAI).",
},
new SampleDefinition
{
Name = "Workflow_StartHere_05_SubWorkflows",
ProjectPath = "samples/03-workflows/_StartHere/05_SubWorkflows",
RequiredEnvironmentVariables = [],
IsDeterministic = true,
MustContain =
[
"=== Sub-Workflow Demonstration ===",
"Final Output:",
"=== Main Workflow Completed ===",
"Sample Complete: Workflows can be composed hierarchically using sub-workflows",
],
},
new SampleDefinition
{
Name = "Workflow_StartHere_06_MixedWorkflowAgentsAndExecutors",
ProjectPath = "samples/03-workflows/_StartHere/06_MixedWorkflowAgentsAndExecutors",
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
Inputs = ["What is 2 plus 2?"],
InputDelayMs = 3000,
ExpectedOutputDescription =
[
"The output should show agents and executors working together to process a user question.",
"The output should not contain error messages or stack traces.",
],
},
new SampleDefinition
{
Name = "Workflow_StartHere_07_WriterCriticWorkflow",
ProjectPath = "samples/03-workflows/_StartHere/07_WriterCriticWorkflow",
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
MustContain = ["=== Writer-Critic Iteration Workflow ==="],
ExpectedOutputDescription =
[
"The output should show a writer-critic iteration workflow with writer and critic sections.",
"The critic should either approve or request revisions.",
"The output should not contain error messages or stack traces.",
],
},
// ───────────────────────────────────────────────────────────────────
// Agents
// ───────────────────────────────────────────────────────────────────
new SampleDefinition
{
Name = "Workflow_Agents_CustomAgentExecutors",
ProjectPath = "samples/03-workflows/Agents/CustomAgentExecutors",
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
ExpectedOutputDescription =
[
"The output should show custom workflow events including slogan generation and feedback.",
"The output should not contain error messages or stack traces.",
],
},
new SampleDefinition
{
Name = "Workflow_Agents_FoundryAgent",
ProjectPath = "samples/03-workflows/Agents/FoundryAgent",
RequiredEnvironmentVariables = ["AZURE_AI_PROJECT_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
SkipReason = "Requires Azure AI Foundry project endpoint.",
},
new SampleDefinition
{
Name = "Workflow_Agents_GroupChatToolApproval",
ProjectPath = "samples/03-workflows/Agents/GroupChatToolApproval",
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
MustContain = ["Starting group chat workflow for software deployment..."],
ExpectedOutputDescription =
[
"The output should show a group chat workflow with QA and DevOps agents for software deployment.",
"There should be approval requests for tool calls.",
"The workflow should show interaction between QA and DevOps agents toward deployment.",
"The output should not contain error messages or stack traces.",
],
},
new SampleDefinition
{
Name = "Workflow_Agents_WorkflowAsAnAgent",
ProjectPath = "samples/03-workflows/Agents/WorkflowAsAnAgent",
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
Inputs = ["hello", "exit"],
InputDelayMs = 5000,
ExpectedOutputDescription =
[
"The output should show a conversational workflow responding to the user's hello message.",
"The output should not contain error messages or stack traces.",
],
},
// ───────────────────────────────────────────────────────────────────
// Checkpoint
// ───────────────────────────────────────────────────────────────────
new SampleDefinition
{
Name = "Workflow_Checkpoint_CheckpointAndRehydrate",
ProjectPath = "samples/03-workflows/Checkpoint/CheckpointAndRehydrate",
RequiredEnvironmentVariables = [],
IsDeterministic = true,
MustContain =
[
"Workflow completed with result:",
"Number of checkpoints created:",
"Hydrating a new workflow instance from the 6th checkpoint.",
],
},
new SampleDefinition
{
Name = "Workflow_Checkpoint_CheckpointAndResume",
ProjectPath = "samples/03-workflows/Checkpoint/CheckpointAndResume",
RequiredEnvironmentVariables = [],
IsDeterministic = true,
MustContain =
[
"Workflow completed with result:",
"Number of checkpoints created:",
"Restoring from the 6th checkpoint.",
],
},
new SampleDefinition
{
Name = "Workflow_Checkpoint_CheckpointWithHumanInTheLoop",
ProjectPath = "samples/03-workflows/Checkpoint/CheckpointWithHumanInTheLoop",
RequiredEnvironmentVariables = [],
Inputs = ["50", "25", "40", "45", "42", "50", "25", "40", "45", "42"],
InputDelayMs = 1000,
MustContain = ["found in"],
ExpectedOutputDescription =
[
"The output should show a number guessing game with higher/lower hints that eventually reaches the correct number.",
"The output should demonstrate checkpoint save and restore behavior.",
],
},
// ───────────────────────────────────────────────────────────────────
// Concurrent
// ───────────────────────────────────────────────────────────────────
new SampleDefinition
{
Name = "Workflow_Concurrent_Concurrent",
ProjectPath = "samples/03-workflows/Concurrent/Concurrent",
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
ExpectedOutputDescription =
[
"The output should show results from concurrent agent processing.",
"The output should not contain error messages or stack traces.",
],
},
new SampleDefinition
{
Name = "Workflow_Concurrent_MapReduce",
ProjectPath = "samples/03-workflows/Concurrent/MapReduce",
RequiredEnvironmentVariables = [],
MustContain =
[
"=== RUNNING WORKFLOW ===",
],
},
// ───────────────────────────────────────────────────────────────────
// ConditionalEdges
// ───────────────────────────────────────────────────────────────────
new SampleDefinition
{
Name = "Workflow_ConditionalEdges_01_EdgeCondition",
ProjectPath = "samples/03-workflows/ConditionalEdges/01_EdgeCondition",
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
ExpectedOutputDescription =
[
"The output should show an email being classified as spam or not spam and processed accordingly.",
"The output should not contain error messages or stack traces.",
],
},
new SampleDefinition
{
Name = "Workflow_ConditionalEdges_02_SwitchCase",
ProjectPath = "samples/03-workflows/ConditionalEdges/02_SwitchCase",
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
ExpectedOutputDescription =
[
"The output should show an ambiguous email being classified as spam, not spam, or uncertain.",
"The output should not contain error messages or stack traces.",
],
},
new SampleDefinition
{
Name = "Workflow_ConditionalEdges_03_MultiSelection",
ProjectPath = "samples/03-workflows/ConditionalEdges/03_MultiSelection",
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
ExpectedOutputDescription =
[
"The output should show an email being classified and potentially routed to multiple handlers.",
"The output should not contain error messages or stack traces.",
],
},
// ───────────────────────────────────────────────────────────────────
// HumanInTheLoop
// ───────────────────────────────────────────────────────────────────
new SampleDefinition
{
Name = "Workflow_HumanInTheLoop_Basic",
ProjectPath = "samples/03-workflows/HumanInTheLoop/HumanInTheLoopBasic",
RequiredEnvironmentVariables = [],
Inputs = ["50", "25", "40", "45", "42"],
InputDelayMs = 1000,
MustContain = ["found in"],
ExpectedOutputDescription =
[
"The output should show a number guessing game with higher/lower hints that eventually reaches the correct number 42.",
],
},
// ───────────────────────────────────────────────────────────────────
// Loop
// ───────────────────────────────────────────────────────────────────
new SampleDefinition
{
Name = "Workflow_Loop",
ProjectPath = "samples/03-workflows/Loop",
RequiredEnvironmentVariables = [],
MustContain = ["Result:"],
},
// ───────────────────────────────────────────────────────────────────
// SharedStates
// ───────────────────────────────────────────────────────────────────
new SampleDefinition
{
Name = "Workflow_SharedStates",
ProjectPath = "samples/03-workflows/SharedStates",
RequiredEnvironmentVariables = [],
IsDeterministic = true,
MustContain =
[
"Total Paragraphs:",
"Total Words:",
],
},
// ───────────────────────────────────────────────────────────────────
// Visualization
// ───────────────────────────────────────────────────────────────────
new SampleDefinition
{
Name = "Workflow_Visualization",
ProjectPath = "samples/03-workflows/Visualization",
RequiredEnvironmentVariables = [],
IsDeterministic = true,
MustContain =
[
"Generating workflow visualization...",
"Mermaid string:",
"DiGraph string:",
],
},
// ───────────────────────────────────────────────────────────────────
// Observability
// ───────────────────────────────────────────────────────────────────
new SampleDefinition
{
Name = "Workflow_Observability_ApplicationInsights",
ProjectPath = "samples/03-workflows/Observability/ApplicationInsights",
RequiredEnvironmentVariables = ["APPLICATIONINSIGHTS_CONNECTION_STRING"],
SkipReason = "Requires Application Insights connection string.",
},
new SampleDefinition
{
Name = "Workflow_Observability_AspireDashboard",
ProjectPath = "samples/03-workflows/Observability/AspireDashboard",
RequiredEnvironmentVariables = [],
SkipReason = "Requires Aspire Dashboard / OTLP endpoint.",
},
new SampleDefinition
{
Name = "Workflow_Observability_WorkflowAsAnAgent",
ProjectPath = "samples/03-workflows/Observability/WorkflowAsAnAgent",
RequiredEnvironmentVariables = ["AZURE_OPENAI_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_OPENAI_DEPLOYMENT_NAME"],
SkipReason = "Interactive console with ReadLine loop; requires OTLP endpoint.",
},
// ───────────────────────────────────────────────────────────────────
// Declarative
// ───────────────────────────────────────────────────────────────────
new SampleDefinition
{
Name = "Workflow_Declarative_ConfirmInput",
ProjectPath = "samples/03-workflows/Declarative/ConfirmInput",
RequiredEnvironmentVariables = ["AZURE_AI_PROJECT_ENDPOINT"],
Inputs = ["hello", "hello"],
InputDelayMs = 8000,
ExpectedOutputDescription = ["The output should show a confirmation prompt and a user response."],
},
new SampleDefinition
{
Name = "Workflow_Declarative_CustomerSupport",
ProjectPath = "samples/03-workflows/Declarative/CustomerSupport",
RequiredEnvironmentVariables = ["AZURE_AI_PROJECT_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
Inputs = ["My laptop won't start"],
InputDelayMs = 3000,
ExpectedOutputDescription = ["The output should show a customer support workflow processing a laptop issue, with agent responses providing troubleshooting or support."],
},
new SampleDefinition
{
Name = "Workflow_Declarative_DeepResearch",
ProjectPath = "samples/03-workflows/Declarative/DeepResearch",
RequiredEnvironmentVariables = ["AZURE_AI_PROJECT_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
SkipReason = "Requires external weather API (wttr.in).",
},
new SampleDefinition
{
Name = "Workflow_Declarative_ExecuteCode",
ProjectPath = "samples/03-workflows/Declarative/ExecuteCode",
RequiredEnvironmentVariables = ["AZURE_AI_PROJECT_ENDPOINT"],
Inputs = ["What is 12 * 34?"],
InputDelayMs = 5000,
ExpectedOutputDescription = ["The output should show a declarative workflow executing generated code, processing a math question and producing a result."],
},
new SampleDefinition
{
Name = "Workflow_Declarative_ExecuteWorkflow",
ProjectPath = "samples/03-workflows/Declarative/ExecuteWorkflow",
RequiredEnvironmentVariables = ["AZURE_AI_PROJECT_ENDPOINT"],
SkipReason = "Requires a workflow file path as a CLI argument.",
},
new SampleDefinition
{
Name = "Workflow_Declarative_FunctionTools",
ProjectPath = "samples/03-workflows/Declarative/FunctionTools",
RequiredEnvironmentVariables = ["AZURE_AI_PROJECT_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
Inputs = ["What are today's specials?", "EXIT"],
InputDelayMs = 8000,
ExpectedOutputDescription = ["The output should show a workflow calling function tools (e.g. a menu plugin) to answer a question about restaurant specials."],
},
new SampleDefinition
{
Name = "Workflow_Declarative_GenerateCode",
ProjectPath = "samples/03-workflows/Declarative/GenerateCode",
IsDeterministic = true,
MustContain = ["WORKFLOW: Parsing", "WORKFLOW: Defined"],
ExpectedOutputDescription = ["The output should show a YAML workflow being parsed and C# code being generated from it."],
},
new SampleDefinition
{
Name = "Workflow_Declarative_HostedWorkflow",
ProjectPath = "samples/03-workflows/Declarative/HostedWorkflow",
RequiredEnvironmentVariables = ["AZURE_AI_PROJECT_ENDPOINT"],
SkipReason = "Hosts a persistent workflow server that does not exit.",
},
new SampleDefinition
{
Name = "Workflow_Declarative_InputArguments",
ProjectPath = "samples/03-workflows/Declarative/InputArguments",
RequiredEnvironmentVariables = ["AZURE_AI_PROJECT_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
Inputs = ["I'd like to visit Seattle", "EXIT"],
InputDelayMs = 8000,
ExpectedOutputDescription = ["The output should show a workflow capturing location input and providing travel-related information about Seattle."],
},
new SampleDefinition
{
Name = "Workflow_Declarative_InvokeFunctionTool",
ProjectPath = "samples/03-workflows/Declarative/InvokeFunctionTool",
RequiredEnvironmentVariables = ["AZURE_AI_PROJECT_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
Inputs = ["What's the soup of the day?", "EXIT"],
InputDelayMs = 8000,
ExpectedOutputDescription = ["The output should show a workflow invoking a function tool (e.g. a menu plugin) to answer a question about the soup of the day."],
},
new SampleDefinition
{
Name = "Workflow_Declarative_InvokeMcpTool",
ProjectPath = "samples/03-workflows/Declarative/InvokeMcpTool",
RequiredEnvironmentVariables = ["AZURE_AI_PROJECT_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
Inputs = ["Search for .NET tutorials on Microsoft Learn"],
InputDelayMs = 3000,
ExpectedOutputDescription = ["The output should show a workflow using MCP tools to search Microsoft Learn documentation and provide a summary of results."],
},
new SampleDefinition
{
Name = "Workflow_Declarative_Marketing",
ProjectPath = "samples/03-workflows/Declarative/Marketing",
RequiredEnvironmentVariables = ["AZURE_AI_PROJECT_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
Inputs = ["A smart water bottle that tracks hydration"],
InputDelayMs = 3000,
ExpectedOutputDescription = ["The output should show a marketing workflow generating content about a smart water bottle product."],
},
new SampleDefinition
{
Name = "Workflow_Declarative_StudentTeacher",
ProjectPath = "samples/03-workflows/Declarative/StudentTeacher",
RequiredEnvironmentVariables = ["AZURE_AI_PROJECT_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
Inputs = ["What is 18 + 27?"],
InputDelayMs = 3000,
ExpectedOutputDescription = ["The output should show a student-teacher workflow where a student asks a math question and a teacher provides the answer."],
},
new SampleDefinition
{
Name = "Workflow_Declarative_ToolApproval",
ProjectPath = "samples/03-workflows/Declarative/ToolApproval",
RequiredEnvironmentVariables = ["AZURE_AI_PROJECT_ENDPOINT"],
OptionalEnvironmentVariables = ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
Inputs = ["Search for .NET tutorials", "EXIT"],
InputDelayMs = 8000,
ExpectedOutputDescription = ["The output should show a workflow using an MCP tool with approval to search Microsoft Learn, followed by an exit from the input loop."],
},
];
}
@@ -1,24 +0,0 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFrameworks>net10.0</TargetFrameworks>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<IsPackable>false</IsPackable>
<IsAotCompatible>false</IsAotCompatible>
<!-- This is a top-level console app; ConfigureAwait is unnecessary -->
<NoWarn>$(NoWarn);CA2007</NoWarn>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Azure.AI.OpenAI" />
<PackageReference Include="Azure.Identity" />
<PackageReference Include="Microsoft.Extensions.AI.OpenAI" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\src\Microsoft.Agents.AI.OpenAI\Microsoft.Agents.AI.OpenAI.csproj" />
</ItemGroup>
</Project>
@@ -5,8 +5,8 @@ This sample demonstrates how to create an AIAgent using Anthropic Claude models
The sample supports three deployment scenarios:
1. **Anthropic Public API** - Direct connection to Anthropic's public API
2. **Microsoft Foundry with API Key** - Anthropic models deployed through Microsoft Foundry using API key authentication
3. **Microsoft Foundry with Azure CLI** - Anthropic models deployed through Microsoft Foundry using Azure CLI credentials
2. **Azure Foundry with API Key** - Anthropic models deployed through Azure Foundry using API key authentication
3. **Azure Foundry with Azure CLI** - Anthropic models deployed through Azure Foundry using Azure CLI credentials
## Prerequisites
@@ -25,29 +25,29 @@ $env:ANTHROPIC_API_KEY="your-anthropic-api-key" # Replace with your Anthropic A
$env:ANTHROPIC_CHAT_MODEL_NAME="claude-haiku-4-5" # Optional, defaults to claude-haiku-4-5
```
### For Microsoft Foundry with API Key
### For Azure Foundry with API Key
- Microsoft Foundry service endpoint and deployment configured
- Azure Foundry service endpoint and deployment configured
- Anthropic API key
Set the following environment variables:
```powershell
$env:ANTHROPIC_RESOURCE="your-foundry-resource-name" # Replace with your Microsoft Foundry resource name (subdomain before .services.ai.azure.com)
$env:ANTHROPIC_RESOURCE="your-foundry-resource-name" # Replace with your Azure Foundry resource name (subdomain before .services.ai.azure.com)
$env:ANTHROPIC_API_KEY="your-anthropic-api-key" # Replace with your Anthropic API key
$env:ANTHROPIC_CHAT_MODEL_NAME="claude-haiku-4-5" # Optional, defaults to claude-haiku-4-5
```
### For Microsoft Foundry with Azure CLI
### For Azure Foundry with Azure CLI
- Microsoft Foundry service endpoint and deployment configured
- Azure Foundry service endpoint and deployment configured
- Azure CLI installed and authenticated (for Azure credential authentication)
Set the following environment variables:
```powershell
$env:ANTHROPIC_RESOURCE="your-foundry-resource-name" # Replace with your Microsoft Foundry resource name (subdomain before .services.ai.azure.com)
$env:ANTHROPIC_RESOURCE="your-foundry-resource-name" # Replace with your Azure Foundry resource name (subdomain before .services.ai.azure.com)
$env:ANTHROPIC_CHAT_MODEL_NAME="claude-haiku-4-5" # Optional, defaults to claude-haiku-4-5
```
**Note**: When using Microsoft Foundry with Azure CLI, make sure you're logged in with `az login` and have access to the Microsoft Foundry resource. For more information, see the [Azure CLI documentation](https://learn.microsoft.com/cli/azure/authenticate-azure-cli-interactively).
**Note**: When using Azure Foundry with Azure CLI, make sure you're logged in with `az login` and have access to the Azure Foundry resource. For more information, see the [Azure CLI documentation](https://learn.microsoft.com/cli/azure/authenticate-azure-cli-interactively).
@@ -2,7 +2,7 @@
#pragma warning disable CS0618 // Type or member is obsolete - sample uses deprecated PersistentAgentsClientExtensions
// This sample shows how to create and use a simple AI agent with Microsoft Foundry Agents as the backend.
// This sample shows how to create and use a simple AI agent with Azure Foundry Agents as the backend.
using Azure.AI.Agents.Persistent;
using Azure.Identity;
@@ -13,14 +13,14 @@ Below is a comparison between the classic and new Foundry Agents approaches:
Before you begin, ensure you have the following prerequisites:
- .NET 10 SDK or later
- Microsoft Foundry service endpoint and deployment configured
- Azure Foundry service endpoint and deployment configured
- Azure CLI installed and authenticated (for Azure credential authentication)
**Note**: This demo uses Azure CLI credentials for authentication. Make sure you're logged in with `az login` and have access to the Microsoft Foundry resource. For more information, see the [Azure CLI documentation](https://learn.microsoft.com/cli/azure/authenticate-azure-cli-interactively).
**Note**: This demo uses Azure CLI credentials for authentication. Make sure you're logged in with `az login` and have access to the Azure Foundry resource. For more information, see the [Azure CLI documentation](https://learn.microsoft.com/cli/azure/authenticate-azure-cli-interactively).
Set the following environment variables:
```powershell
$env:AZURE_AI_PROJECT_ENDPOINT="https://your-foundry-service.services.ai.azure.com/api/projects/your-foundry-project" # Replace with your Microsoft Foundry resource endpoint
$env:AZURE_AI_PROJECT_ENDPOINT="https://your-foundry-service.services.ai.azure.com/api/projects/your-foundry-project" # Replace with your Azure Foundry resource endpoint
$env:AZURE_AI_MODEL_DEPLOYMENT_NAME="gpt-4o-mini" # Optional, defaults to gpt-4o-mini
```
@@ -1,6 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
// This sample shows how to create and use AI agents with Microsoft Foundry Agents as the backend.
// This sample shows how to create and use a AI agents with Azure Foundry Agents as the backend.
using Azure.AI.Projects;
using Azure.AI.Projects.Agents;
@@ -13,7 +13,7 @@ var deploymentName = Environment.GetEnvironmentVariable("AZURE_AI_MODEL_DEPLOYME
const string JokerName = "JokerAgent";
// Get a client to create/retrieve/delete server side agents with Microsoft Foundry Agents.
// Get a client to create/retrieve/delete server side agents with Azure Foundry Agents.
// WARNING: DefaultAzureCredential is convenient for development but requires careful consideration in production.
// In production, consider using a specific credential (e.g., ManagedIdentityCredential) to avoid
// latency issues, unintended credential probing, and potential security risks from fallback mechanisms.
@@ -13,14 +13,14 @@ Below is a comparison between the classic and new Foundry Agents approaches:
Before you begin, ensure you have the following prerequisites:
- .NET 10 SDK or later
- Microsoft Foundry service endpoint and deployment configured
- Azure Foundry service endpoint and deployment configured
- Azure CLI installed and authenticated (for Azure credential authentication)
**Note**: This demo uses Azure CLI credentials for authentication. Make sure you're logged in with `az login` and have access to the Microsoft Foundry resource. For more information, see the [Azure CLI documentation](https://learn.microsoft.com/cli/azure/authenticate-azure-cli-interactively).
**Note**: This demo uses Azure CLI credentials for authentication. Make sure you're logged in with `az login` and have access to the Azure Foundry resource. For more information, see the [Azure CLI documentation](https://learn.microsoft.com/cli/azure/authenticate-azure-cli-interactively).
Set the following environment variables:
```powershell
$env:AZURE_AI_PROJECT_ENDPOINT="https://your-foundry-service.services.ai.azure.com/api/projects/your-foundry-project" # Replace with your Microsoft Foundry resource endpoint
$env:AZURE_AI_PROJECT_ENDPOINT="https://your-foundry-service.services.ai.azure.com/api/projects/your-foundry-project" # Replace with your Azure Foundry resource endpoint
$env:AZURE_AI_MODEL_DEPLOYMENT_NAME="gpt-4o-mini" # Optional, defaults to gpt-4o-mini
```
@@ -1,7 +1,7 @@
// Copyright (c) Microsoft. All rights reserved.
// This sample shows how to use the OpenAI SDK to create and use a simple AI agent with any model hosted in Microsoft Foundry.
// You could use models from Microsoft, OpenAI, DeepSeek, Hugging Face, Meta, xAI or any other model you have deployed in your Microsoft Foundry resource.
// This sample shows how to use the OpenAI SDK to create and use a simple AI agent with any model hosted in Azure AI Foundry.
// You could use models from Microsoft, OpenAI, DeepSeek, Hugging Face, Meta, xAI or any other model you have deployed in your Azure AI Foundry resource.
// Note: Ensure that you pick a model that suits your needs. For example, if you want to use function calling, ensure that the model you pick supports function calling.
using System.ClientModel;
@@ -15,7 +15,7 @@ var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") ?? th
var apiKey = Environment.GetEnvironmentVariable("AZURE_OPENAI_API_KEY");
var model = Environment.GetEnvironmentVariable("AZURE_AI_MODEL_DEPLOYMENT_NAME") ?? "Phi-4-mini-instruct";
// Since we are using the OpenAI Client SDK, we need to override the default endpoint to point to Microsoft Foundry.
// Since we are using the OpenAI Client SDK, we need to override the default endpoint to point to Azure Foundry.
var clientOptions = new OpenAIClientOptions() { Endpoint = new Uri(endpoint) };
// Create the OpenAI client with either an API key or Azure CLI credential.
@@ -1,8 +1,8 @@
## Overview
This sample shows how to use the OpenAI SDK to create and use a simple AI agent with any model hosted in Microsoft Foundry.
This sample shows how to use the OpenAI SDK to create and use a simple AI agent with any model hosted in Azure AI Foundry.
You could use models from Microsoft, OpenAI, DeepSeek, Hugging Face, Meta, xAI or any other model you have deployed in Microsoft Foundry.
You could use models from Microsoft, OpenAI, DeepSeek, Hugging Face, Meta, xAI or any other model you have deployed in Azure AI Foundry.
**Note**: Ensure that you pick a model that suits your needs. For example, if you want to use function calling, ensure that the model you pick supports function calling.
@@ -11,19 +11,19 @@ You could use models from Microsoft, OpenAI, DeepSeek, Hugging Face, Meta, xAI o
Before you begin, ensure you have the following prerequisites:
- .NET 10 SDK or later
- Microsoft Foundry resource
- A model deployment in your Microsoft Foundry resource. This example defaults to using the `Phi-4-mini-instruct` model,
- Azure AI Foundry resource
- A model deployment in your Azure AI Foundry resource. This example defaults to using the `Phi-4-mini-instruct` model,
so if you want to use a different model, ensure that you set your `AZURE_AI_MODEL_DEPLOYMENT_NAME` environment
variable to the name of your deployed model.
- An API key or role based authentication to access the Microsoft Foundry resource
- An API key or role based authentication to access the Azure AI Foundry resource
See [here](https://learn.microsoft.com/en-us/azure/ai-foundry/quickstarts/get-started-code?tabs=csharp) for more info on setting up these prerequisites
Set the following environment variables:
```powershell
# Replace with your Microsoft Foundry resource endpoint
# Ensure that you have the "/openai/v1/" path in the URL, since this is required when using the OpenAI SDK to access Microsoft Foundry models.
# Replace with your Azure AI Foundry resource endpoint
# Ensure that you have the "/openai/v1/" path in the URL, since this is required when using the OpenAI SDK to access Azure Foundry models.
$env:AZURE_OPENAI_ENDPOINT="https://ai-foundry-<myresourcename>.services.ai.azure.com/openai/v1/"
# Optional, defaults to using Azure CLI for authentication if not provided
@@ -18,7 +18,7 @@ See the README.md for each sample for the prerequisites for that sample.
|[Creating an AIAgent with Anthropic](./Agent_With_Anthropic/)|This sample demonstrates how to create an AIAgent using Anthropic Claude models as the underlying inference service|
|[Creating an AIAgent with Foundry Agents using Azure.AI.Agents.Persistent](./Agent_With_AzureAIAgentsPersistent/)|This sample demonstrates how to create a Foundry Persistent agent and expose it as an AIAgent using the Azure.AI.Agents.Persistent SDK|
|[Creating an AIAgent with Foundry Agents using Azure.AI.Project](./Agent_With_AzureAIProject/)|This sample demonstrates how to create an Foundry Project agent and expose it as an AIAgent using the Azure.AI.Project SDK|
|[Creating an AIAgent with Foundry Model](./Agent_With_AzureFoundryModel/)|This sample demonstrates how to use any model deployed to Microsoft Foundry to create an AIAgent|
|[Creating an AIAgent with AzureFoundry Model](./Agent_With_AzureFoundryModel/)|This sample demonstrates how to use any model deployed to Azure Foundry to create an AIAgent|
|[Creating an AIAgent with Azure OpenAI ChatCompletion](./Agent_With_AzureOpenAIChatCompletion/)|This sample demonstrates how to create an AIAgent using Azure OpenAI ChatCompletion as the underlying inference service|
|[Creating an AIAgent with Azure OpenAI Responses](./Agent_With_AzureOpenAIResponses/)|This sample demonstrates how to create an AIAgent using Azure OpenAI Responses as the underlying inference service|
|[Creating an AIAgent with a custom implementation](./Agent_With_CustomImplementation/)|This sample demonstrates how to create an AIAgent with a custom implementation|
@@ -18,9 +18,9 @@ Before you begin, ensure you have the following prerequisites:
**Note**: These samples use Anthropic Claude models. For more information, see [Anthropic documentation](https://docs.anthropic.com/).
## Using Anthropic with Microsoft Foundry
## Using Anthropic with Azure Foundry
To use Anthropic with Microsoft Foundry, you can check the sample [AgentProviders/Agent_With_Anthropic](../AgentProviders/Agent_With_Anthropic/README.md) for more details.
To use Anthropic with Azure Foundry, you can check the sample [AgentProviders/Agent_With_Anthropic](../AgentProviders/Agent_With_Anthropic/README.md) for more details.
## Samples
@@ -1,10 +1,10 @@
// Copyright (c) Microsoft. All rights reserved.
// This sample shows how to use the FoundryMemoryProvider to persist and recall memories for an agent.
// The sample stores conversation messages in a Microsoft Foundry memory store and retrieves relevant
// The sample stores conversation messages in an Azure AI Foundry memory store and retrieves relevant
// memories for subsequent invocations, even across new sessions.
//
// Note: Memory extraction in Microsoft Foundry is asynchronous and takes time. This sample demonstrates
// Note: Memory extraction in Azure AI Foundry is asynchronous and takes time. This sample demonstrates
// a simple polling approach to wait for memory updates to complete before querying.
using System.Text.Json;
@@ -62,7 +62,7 @@ await memoryProvider.EnsureStoredMemoriesDeletedAsync(session);
Console.WriteLine(await agent.RunAsync("Hi there! My name is Taylor and I'm planning a hiking trip to Patagonia in November.", session));
Console.WriteLine(await agent.RunAsync("I'm travelling with my sister and we love finding scenic viewpoints.", session));
// Memory extraction in Microsoft Foundry is asynchronous and takes time to process.
// Memory extraction in Azure AI Foundry is asynchronous and takes time to process.
// WhenUpdatesCompletedAsync polls all pending updates and waits for them to complete.
Console.WriteLine("\nWaiting for Foundry Memory to process updates...");
await memoryProvider.WhenUpdatesCompletedAsync();
@@ -1,6 +1,6 @@
# Agent with Memory Using Microsoft Foundry
# Agent with Memory Using Azure AI Foundry
This sample demonstrates how to create and run an agent that uses Microsoft Foundry's managed memory service to extract and retrieve individual memories across sessions.
This sample demonstrates how to create and run an agent that uses Azure AI Foundry's managed memory service to extract and retrieve individual memories across sessions.
## Features Demonstrated
@@ -13,7 +13,7 @@ This sample demonstrates how to create and run an agent that uses Microsoft Foun
## Prerequisites
1. Azure subscription with Microsoft Foundry project
1. Azure subscription with Azure AI Foundry project
2. Azure OpenAI resource with a chat model deployment (e.g., gpt-4o-mini) and an embedding model deployment (e.g., text-embedding-ada-002)
3. .NET 10.0 SDK
4. Azure CLI logged in (`az login`)
@@ -21,7 +21,7 @@ This sample demonstrates how to create and run an agent that uses Microsoft Foun
## Environment Variables
```bash
# Microsoft Foundry project endpoint and memory store name
# Azure AI Foundry project endpoint and memory store name
export AZURE_AI_PROJECT_ENDPOINT="https://your-account.services.ai.azure.com/api/projects/your-project"
export AZURE_AI_MEMORY_STORE_ID="my_memory_store"
@@ -48,10 +48,10 @@ The agent will:
## Key Differences from Mem0
| Aspect | Mem0 | Microsoft Foundry Memory |
| Aspect | Mem0 | Azure AI Foundry Memory |
|--------|------|------------------------|
| Authentication | API Key | Azure Identity (DefaultAzureCredential) |
| Scope | ApplicationId, UserId, AgentId, ThreadId | Single `Scope` string |
| Memory Types | Single memory store | User Profile + Chat Summary |
| Hosting | Mem0 cloud or self-hosted | Microsoft Foundry managed service |
| Hosting | Mem0 cloud or self-hosted | Azure AI Foundry managed service |
| Store Creation | N/A (automatic) | Explicit via `EnsureMemoryStoreCreatedAsync` |
@@ -7,7 +7,7 @@ These samples show how to create an agent with the Agent Framework that uses Mem
|[Chat History memory](./AgentWithMemory_Step01_ChatHistoryMemory/)|This sample demonstrates how to enable an agent to remember messages from previous conversations.|
|[Memory with MemoryStore](./AgentWithMemory_Step02_MemoryUsingMem0/)|This sample demonstrates how to create and run an agent that uses the Mem0 service to extract and retrieve individual memories.|
|[Custom Memory Implementation](../../01-get-started/04_memory/)|This sample demonstrates how to create a custom memory component and attach it to an agent.|
|[Memory with Microsoft Foundry](./AgentWithMemory_Step04_MemoryUsingFoundry/)|This sample demonstrates how to create and run an agent that uses Microsoft Foundry's managed memory service to extract and retrieve individual memories.|
|[Memory with Azure AI Foundry](./AgentWithMemory_Step04_MemoryUsingFoundry/)|This sample demonstrates how to create and run an agent that uses Azure AI Foundry's managed memory service to extract and retrieve individual memories.|
|[Bounded Chat History with Overflow](./AgentWithMemory_Step05_BoundedChatHistory/)|This sample demonstrates how to create a bounded chat history provider that overflows older messages to a vector store and recalls them as memories.|
> **See also**: [Memory Search with Foundry Agents](../AgentsWithFoundry/Agent_Step22_MemorySearch/) - demonstrates using the built-in Memory Search tool with Microsoft Foundry agents.
> **See also**: [Memory Search with Foundry Agents](../AgentsWithFoundry/Agent_Step22_MemorySearch/) - demonstrates using the built-in Memory Search tool with Azure Foundry agents.
@@ -13,7 +13,7 @@ This sample uses Qdrant for the vector store, but this can easily be swapped out
- User has the `Cognitive Services OpenAI Contributor` role for the Azure OpenAI resource.
- An existing Qdrant instance. You can use a managed service or run a local instance using Docker, but the sample assumes the instance is running locally.
**Note**: These samples use Azure OpenAI models. For more information, see [how to deploy Azure OpenAI models with Microsoft Foundry](https://learn.microsoft.com/en-us/azure/ai-foundry/how-to/deploy-models-openai).
**Note**: These samples use Azure OpenAI models. For more information, see [how to deploy Azure OpenAI models with Azure AI Foundry](https://learn.microsoft.com/en-us/azure/ai-foundry/how-to/deploy-models-openai).
**Note**: These samples use Azure CLI credentials for authentication. Make sure you're logged in with `az login` and have access to the Azure OpenAI resource and have the `Cognitive Services OpenAI Contributor` role. For more information, see the [Azure CLI documentation](https://learn.microsoft.com/cli/azure/authenticate-azure-cli-interactively).
@@ -1,54 +0,0 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFrameworks>net10.0</TargetFrameworks>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<ManagePackageVersionsCentrally>false</ManagePackageVersionsCentrally>
</PropertyGroup>
<ItemGroup>
<PackageReference Remove="Microsoft.CodeAnalysis.NetAnalyzers" />
<PackageReference Remove="Microsoft.VisualStudio.Threading.Analyzers" />
<PackageReference Remove="xunit.analyzers" />
<PackageReference Remove="Moq.Analyzers" />
<PackageReference Remove="Roslynator.Analyzers" />
<PackageReference Remove="Roslynator.CodeAnalysis.Analyzers" />
<PackageReference Remove="Roslynator.Formatting.Analyzers" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="Azure.AI.OpenAI" Version="2.9.0-beta.1" />
<PackageReference Include="Azure.Identity" Version="1.19.0" />
<PackageReference Include="Microsoft.Agents.AI.OpenAI" Version="1.0.0-rc4" />
<PackageReference Include="Microsoft.Extensions.AI.OpenAI" Version="10.4.0" />
<PackageReference Include="Neo4j.AgentFramework.GraphRAG" Version="0.1.0-preview.2" />
<PackageReference Include="Neo4j.Driver" Version="5.28.0" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.CodeAnalysis.NetAnalyzers" Version="10.0.100">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
<PackageReference Include="Microsoft.VisualStudio.Threading.Analyzers" Version="17.14.15">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
<PackageReference Include="Roslynator.Analyzers" Version="4.14.1">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
<PackageReference Include="Roslynator.CodeAnalysis.Analyzers" Version="4.14.1">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
<PackageReference Include="Roslynator.Formatting.Analyzers" Version="4.14.1">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
</ItemGroup>
</Project>
@@ -1,77 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using Azure.AI.OpenAI;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;
using Neo4j.AgentFramework.GraphRAG;
using Neo4j.Driver;
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 neo4jUri = Environment.GetEnvironmentVariable("NEO4J_URI") ?? throw new InvalidOperationException("NEO4J_URI is not set.");
var neo4jUsername = Environment.GetEnvironmentVariable("NEO4J_USERNAME") ?? "neo4j";
var neo4jPassword = Environment.GetEnvironmentVariable("NEO4J_PASSWORD") ?? throw new InvalidOperationException("NEO4J_PASSWORD is not set.");
var fulltextIndex = Environment.GetEnvironmentVariable("NEO4J_FULLTEXT_INDEX_NAME") ?? "search_chunks";
const string RetrievalQuery = """
MATCH (node)-[:FROM_DOCUMENT]->(doc:Document)<-[:FILED]-(company:Company)
OPTIONAL MATCH (company)-[:FACES_RISK]->(risk:RiskFactor)
WITH node, score, company, doc, collect(DISTINCT risk.name)[0..5] AS risks
OPTIONAL MATCH (company)-[:MENTIONS]->(product:Product)
WITH node, score, company, doc, risks, collect(DISTINCT product.name)[0..5] AS products
RETURN
node.text AS text,
score,
company.name AS company,
company.ticker AS ticker,
doc.title AS title,
risks,
products
ORDER BY score DESC
""";
await using var driver = GraphDatabase.Driver(new Uri(neo4jUri), AuthTokens.Basic(neo4jUsername, neo4jPassword));
await driver.VerifyConnectivityAsync();
await using var provider = new Neo4jContextProvider(
driver,
new Neo4jContextProviderOptions
{
IndexName = fulltextIndex,
IndexType = IndexType.Fulltext,
RetrievalQuery = RetrievalQuery,
TopK = 5,
ContextPrompt = "Use the retrieved Neo4j graph context to answer accurately and call out when context is missing."
});
// WARNING: DefaultAzureCredential is convenient for development but requires careful consideration in production.
// In production, consider using a specific credential (e.g., ManagedIdentityCredential) to avoid
// latency issues, unintended credential probing, and potential security risks from fallback mechanisms.
AIAgent agent = new AzureOpenAIClient(
new Uri(endpoint),
new DefaultAzureCredential())
.GetChatClient(deploymentName)
.AsIChatClient()
.AsAIAgent(new ChatClientAgentOptions
{
ChatOptions = new()
{
Instructions = "You are a helpful assistant that answers questions using Neo4j graph context."
},
AIContextProviders = [provider]
});
AgentSession session = await agent.CreateSessionAsync();
foreach (var question in new[]
{
"What products does Microsoft offer?",
"What risks does Apple face?",
"Tell me about NVIDIA's AI business and risk factors."
})
{
Console.WriteLine($">> {question}\n");
Console.WriteLine(await agent.RunAsync(question, session));
Console.WriteLine();
}
@@ -1,32 +0,0 @@
# Agent Framework Retrieval Augmented Generation (RAG) with Neo4j GraphRAG
This sample demonstrates how to create and run an agent that uses the [Neo4j GraphRAG context provider](https://github.com/neo4j-labs/neo4j-maf-provider) with Microsoft Agent Framework for .NET.
The sample uses a Neo4j fulltext index for retrieval and a Cypher `RetrievalQuery` to enrich results with related companies, products, and risk factors.
## Prerequisites
- .NET 10 SDK or later
- Azure OpenAI endpoint and chat deployment
- Azure CLI installed and authenticated
- A Neo4j database with chunked documents and a fulltext index such as `search_chunks`
## Environment variables
```powershell
$env:AZURE_OPENAI_ENDPOINT="https://your-resource.openai.azure.com/"
$env:AZURE_OPENAI_DEPLOYMENT_NAME="gpt-4o-mini"
$env:NEO4J_URI="neo4j+s://your-instance.databases.neo4j.io"
$env:NEO4J_USERNAME="neo4j"
$env:NEO4J_PASSWORD="your-password"
$env:NEO4J_FULLTEXT_INDEX_NAME="search_chunks"
```
## Build and run
```powershell
dotnet build
dotnet run --framework net10.0 --no-build
```
The sample issues a few questions against the graph-backed retrieval provider and prints the responses to the console.
@@ -8,4 +8,3 @@ These samples show how to create an agent with the Agent Framework that uses Ret
|[RAG with Vector Store and custom schema](./AgentWithRAG_Step02_CustomVectorStoreRAG/)|This sample demonstrates how to create and run an agent that uses Retrieval Augmented Generation (RAG) with a vector store. It also uses a custom schema for the documents stored in the vector store.|
|[RAG with custom RAG data source](./AgentWithRAG_Step03_CustomRAGDataSource/)|This sample demonstrates how to create and run an agent that uses Retrieval Augmented Generation (RAG) with a custom RAG data source.|
|[RAG with Foundry VectorStore service](./AgentWithRAG_Step04_FoundryServiceRAG/)|This sample demonstrates how to create and run an agent that uses Retrieval Augmented Generation (RAG) with the Foundry VectorStore service.|
|[RAG with Neo4j GraphRAG](./AgentWithRAG_Step05_Neo4jGraphRAG/)|This sample demonstrates how to create and run an agent that uses a Neo4j-backed GraphRAG context provider with graph-enriched retrieval.|
@@ -18,7 +18,7 @@ Before you begin, ensure you have the following prerequisites:
- Azure CLI installed and authenticated (for Azure credential authentication)
- User has the `Cognitive Services OpenAI Contributor` role for the Azure OpenAI resource
**Note**: This sample uses Azure OpenAI models. For more information, see [how to deploy Azure OpenAI models with Microsoft Foundry](https://learn.microsoft.com/en-us/azure/ai-foundry/how-to/deploy-models-openai).
**Note**: This sample uses Azure OpenAI models. For more information, see [how to deploy Azure OpenAI models with Azure AI Foundry](https://learn.microsoft.com/en-us/azure/ai-foundry/how-to/deploy-models-openai).
**Note**: This demo uses Azure CLI credentials for authentication. Make sure you're logged in with `az login` and have access to the Azure OpenAI resource and have the `Cognitive Services OpenAI Contributor` role. For more information, see the [Azure CLI documentation](https://learn.microsoft.com/cli/azure/authenticate-azure-cli-interactively).
@@ -20,8 +20,8 @@ To use the [MCP Inspector](https://modelcontextprotocol.io/docs/tools/inspector)
MCP Inspector is up and running at http://127.0.0.1:6274
```
1. Open a web browser and navigate to the URL displayed in the terminal. If not opened automatically, this will open the MCP Inspector interface.
1. In the MCP Inspector interface, add the following environment variables to allow your MCP server to access Microsoft Foundry Project to create and run the agent:
- AZURE_AI_PROJECT_ENDPOINT = https://your-resource.openai.azure.com/ # Replace with your Microsoft Foundry Project endpoint
1. In the MCP Inspector interface, add the following environment variables to allow your MCP server to access Azure AI Foundry Project to create and run the agent:
- AZURE_AI_PROJECT_ENDPOINT = https://your-resource.openai.azure.com/ # Replace with your Azure AI Foundry Project endpoint
- AZURE_AI_MODEL_DEPLOYMENT_NAME = gpt-4o-mini # Replace with your model deployment name
1. Find and click the `Connect` button in the MCP Inspector interface to connect to the MCP server.
1. As soon as the connection is established, open the `Tools` tab in the MCP Inspector interface and select the `Joker` tool from the list.
@@ -13,7 +13,7 @@ using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;
// Get Microsoft Foundry configuration from environment variables
// Get Azure AI Foundry configuration from environment variables
var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
var deploymentName = System.Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT_NAME") ?? "gpt-4o";
@@ -3,7 +3,7 @@
// This sample shows how to use a chat history reducer to keep the context within model size limits.
// Any implementation of Microsoft.Extensions.AI.IChatReducer can be used to customize how the chat history is reduced.
// NOTE: this feature is only supported where the chat history is stored locally, such as with OpenAI Chat Completion.
// Where the chat history is stored server side, such as with Microsoft Foundry Agents, the service must manage the chat history size.
// Where the chat history is stored server side, such as with Azure Foundry Agents, the service must manage the chat history size.
using Azure.AI.OpenAI;
using Azure.Identity;
@@ -2,7 +2,7 @@
#pragma warning disable CS0618 // Type or member is obsolete - sample uses deprecated PersistentAgentsClientExtensions
// This sample shows how to create a Microsoft Foundry Agent with the Deep Research Tool.
// This sample shows how to create an Azure AI Foundry Agent with the Deep Research Tool.
using Azure.AI.Agents.Persistent;
using Azure.Identity;
@@ -11,10 +11,10 @@ Key features:
Before running this sample, ensure you have:
1. A Microsoft Foundry project set up
1. An Azure AI Foundry project set up
2. A deep research model deployment (e.g., o3-deep-research)
3. A model deployment (e.g., gpt-4o)
4. A Bing Connection configured in your Microsoft Foundry project
4. A Bing Connection configured in your Azure AI Foundry project
5. Azure CLI installed and authenticated
**Important**: Please visit the following documentation for detailed setup instructions:
@@ -29,14 +29,14 @@ Pay special attention to the purple `Note` boxes in the Azure documentation.
/subscriptions/<sub-id>/resourceGroups/<rg>/providers/Microsoft.CognitiveServices/accounts/<account>/projects/<project>/connections/<connection-name>
```
You can find this in the Microsoft Foundry portal under **Management > Connected resources**, or retrieve it programmatically via the connections API (`.id` property).
You can find this in the Azure AI Foundry portal under **Management > Connected resources**, or retrieve it programmatically via the connections API (`.id` property).
## Environment Variables
Set the following environment variables:
```powershell
# Replace with your Microsoft Foundry project endpoint
# Replace with your Azure AI Foundry project endpoint
$env:AZURE_AI_PROJECT_ENDPOINT="https://your-project.services.ai.azure.com/"
# Replace with your Bing Grounding connection ID (full ARM resource URI)
+1 -1
View File
@@ -18,7 +18,7 @@ Before you begin, ensure you have the following prerequisites:
- Azure CLI installed and authenticated (for Azure credential authentication)
- User has the `Cognitive Services OpenAI Contributor` role for the Azure OpenAI resource.
**Note**: These samples use Azure OpenAI models. For more information, see [how to deploy Azure OpenAI models with Microsoft Foundry](https://learn.microsoft.com/en-us/azure/ai-foundry/how-to/deploy-models-openai).
**Note**: These samples use Azure OpenAI models. For more information, see [how to deploy Azure OpenAI models with Azure AI Foundry](https://learn.microsoft.com/en-us/azure/ai-foundry/how-to/deploy-models-openai).
**Note**: These samples use Azure CLI credentials for authentication. Make sure you're logged in with `az login` and have access to the Azure OpenAI resource and have the `Cognitive Services OpenAI Contributor` role. For more information, see the [Azure CLI documentation](https://learn.microsoft.com/cli/azure/authenticate-azure-cli-interactively).
@@ -1,7 +1,7 @@
// Copyright (c) Microsoft. All rights reserved.
// This sample shows how to create, use, and clean up a FoundryAgent backed by a server-side
// versioned agent in Microsoft Foundry. It demonstrates the full lifecycle:
// versioned agent in Azure AI Foundry. It demonstrates the full lifecycle:
// create agent version -> wrap as FoundryAgent -> run -> delete.
using Azure.AI.Projects;
@@ -1,6 +1,6 @@
# Getting started with Foundry Agents
These samples demonstrate how to use Microsoft Foundry with Agent Framework.
These samples demonstrate how to use Azure AI Foundry with Agent Framework.
## Quick start
@@ -1,7 +1,7 @@
// Copyright (c) Microsoft. All rights reserved.
// This sample shows how to create and use a simple AI agent with Microsoft Foundry Agents as the backend, that uses a Hosted MCP Tool.
// In this case the Microsoft Foundry Agents service will invoke any MCP tools as required. MCP tools are not invoked by the Agent Framework.
// This sample shows how to create and use a simple AI agent with Azure Foundry Agents as the backend, that uses a Hosted MCP Tool.
// In this case the Azure Foundry Agents service will invoke any MCP tools as required. MCP tools are not invoked by the Agent Framework.
// The sample first shows how to use MCP tools with auto approval, and then how to set up a tool that requires approval before it can be invoked and how to approve such a tool.
using Azure.AI.Projects;
@@ -3,14 +3,14 @@
Before you begin, ensure you have the following prerequisites:
- .NET 10 SDK or later
- Microsoft Foundry service endpoint and deployment configured
- Azure Foundry service endpoint and deployment configured
- Azure CLI installed and authenticated (for Azure credential authentication)
**Note**: This demo uses Azure CLI credentials for authentication. Make sure you're logged in with `az login` and have access to the Microsoft Foundry resource. For more information, see the [Azure CLI documentation](https://learn.microsoft.com/cli/azure/authenticate-azure-cli-interactively).
**Note**: This demo uses Azure CLI credentials for authentication. Make sure you're logged in with `az login` and have access to the Azure Foundry resource. For more information, see the [Azure CLI documentation](https://learn.microsoft.com/cli/azure/authenticate-azure-cli-interactively).
Set the following environment variables:
```powershell
$env:AZURE_AI_PROJECT_ENDPOINT="https://your-foundry-service.services.ai.azure.com/api/projects/your-foundry-project" # Replace with your Microsoft Foundry resource endpoint
$env:AZURE_AI_PROJECT_ENDPOINT="https://your-foundry-service.services.ai.azure.com/api/projects/your-foundry-project" # Replace with your Azure Foundry resource endpoint
$env:AZURE_AI_MODEL_DEPLOYMENT_NAME="gpt-4.1-mini" # Optional, defaults to gpt-4.1-mini
```
@@ -11,7 +11,7 @@ Before you begin, ensure you have the following prerequisites:
- Azure CLI installed and authenticated (for Azure credential authentication)
- User has the `Cognitive Services OpenAI Contributor` role for the Azure OpenAI resource.
**Note**: These samples use Azure OpenAI models. For more information, see [how to deploy Azure OpenAI models with Microsoft Foundry](https://learn.microsoft.com/en-us/azure/ai-foundry/how-to/deploy-models-openai).
**Note**: These samples use Azure OpenAI models. For more information, see [how to deploy Azure OpenAI models with Azure AI Foundry](https://learn.microsoft.com/en-us/azure/ai-foundry/how-to/deploy-models-openai).
**Note**: These samples use Azure CLI credentials for authentication. Make sure you're logged in with `az login` and have access to the Azure OpenAI resource and have the `Cognitive Services OpenAI Contributor` role. For more information, see the [Azure CLI documentation](https://learn.microsoft.com/cli/azure/authenticate-azure-cli-interactively).
@@ -11,12 +11,12 @@ using Microsoft.Extensions.AI;
namespace WorkflowFoundryAgentSample;
/// <summary>
/// This sample shows how to use Microsoft Foundry Agents within a workflow.
/// This sample shows how to use Azure Foundry Agents within a workflow.
/// </summary>
/// <remarks>
/// Pre-requisites:
/// - Foundational samples should be completed first.
/// - A Microsoft Foundry project endpoint and model ID.
/// - An Azure Foundry project endpoint and model id.
/// </remarks>
public static class Program
{
@@ -30,7 +30,7 @@ namespace Demo.Workflows.Declarative.InvokeMcpTool;
/// <item>Integrating with MCP-compatible services</item>
/// </list>
/// <para>
/// This sample uses the Microsoft Learn MCP server to search Azure documentation and the Microsoft Foundry MCP server to get AI model details.
/// This sample uses the Microsoft Learn MCP server to search Azure documentation and the Azure foundry MCP server to get AI model details.
/// When you run the sample, provide an AI model (e.g. gpt-4.1-mini) as input,
/// The workflow will use the MCP tools to find relevant information about the model from Microsoft Learn and foundry, then an agent will summarize the results.
/// </para>
@@ -6,7 +6,7 @@ to build a `Workflow` that may be executed using the same pattern as any code-ba
## Configuration
These samples must be configured to create and use agents your
[Microsoft Foundry Project](https://learn.microsoft.com/azure/ai-foundry).
[Azure Foundry Project](https://learn.microsoft.com/azure/ai-foundry).
### Settings
@@ -18,9 +18,9 @@ The configuraton required by the samples is:
|Setting Name| Description|
|:--|:--|
|AZURE_AI_PROJECT_ENDPOINT| The endpoint URL of your Microsoft Foundry Project.|
|AZURE_AI_PROJECT_ENDPOINT| The endpoint URL of your Azure Foundry Project.|
|AZURE_AI_MODEL_DEPLOYMENT_NAME| The name of the model deployment to use
|AZURE_AI_BING_CONNECTION_ID| The name of the Bing Grounding connection configured in your Microsoft Foundry Project.|
|AZURE_AI_BING_CONNECTION_ID| The name of the Bing Grounding connection configured in your Azure Foundry Project.|
To set your secrets with .NET Secret Manager:
@@ -42,13 +42,13 @@ To set your secrets with .NET Secret Manager:
dotnet user-secrets init
```
4. Define setting that identifies your Microsoft Foundry Project (endpoint):
4. Define setting that identifies your Azure Foundry Project (endpoint):
```
dotnet user-secrets set "AZURE_AI_PROJECT_ENDPOINT" "https://..."
```
5. Define setting that identifies your Microsoft Foundry Model Deployment (endpoint):
5. Define setting that identifies your Azure Foundry Model Deployment (endpoint):
```
dotnet user-secrets set "AZURE_AI_MODEL_DEPLOYMENT_NAME" "gpt-5"
@@ -70,7 +70,7 @@ $env:AZURE_AI_BING_CONNECTION_ID="mybinggrounding"
### Authorization
Use [_Azure CLI_](https://learn.microsoft.com/cli/azure/authenticate-azure-cli) to authorize access to your Microsoft Foundry Project:
Use [_Azure CLI_](https://learn.microsoft.com/cli/azure/authenticate-azure-cli) to authorize access to your Azure Foundry Project:
```
az login
+1 -1
View File
@@ -26,7 +26,7 @@ Once completed, please proceed to the other samples listed below.
| Sample | Concepts |
|--------|----------|
| [Foundry Agents in Workflows](./Agents/FoundryAgent) | Demonstrates using Microsoft Foundry agents in a workflow through `ChatClientAgent` |
| [Foundry Agents in Workflows](./Agents/FoundryAgent) | Demonstrates using Azure Foundry agents in a workflow through `ChatClientAgent` |
| [Custom Agent Executors](./Agents/CustomAgentExecutors) | Shows how to create a custom agent executor for more complex scenarios |
| [Workflow as an Agent](./Agents/WorkflowAsAnAgent) | Illustrates how to encapsulate a workflow as an agent |
| [Group Chat with Tool Approval](./Agents/GroupChatToolApproval) | Shows multi-agent group chat with tool approval requests and human-in-the-loop interaction |
@@ -51,7 +51,7 @@ dotnet run --urls "http://localhost:5002;https://localhost:5012" --agentType "lo
### Configuring for use with Azure AI Agents
You must create the agents in a Microsoft Foundry project and then provide the project endpoint and agent IDs. The instructions for each agent are as follows:
You must create the agents in an Azure AI Foundry project and then provide the project endpoint and agents ids. The instructions for each agent are as follows:
- Invoice Agent
```
@@ -1,7 +1,7 @@
// Copyright (c) Microsoft. All rights reserved.
// Seattle Hotel Agent - A simple agent with a tool to find hotels in Seattle.
// Uses Microsoft Agent Framework with Microsoft Foundry.
// Uses Microsoft Agent Framework with Azure AI Foundry.
// Ready for deployment to Foundry Hosted Agent service.
using System.ClientModel.Primitives;
@@ -4,7 +4,7 @@ This sample demonstrates how to build a hosted agent that uses local C# function
Key features:
- Defining local C# functions as agent tools using `AIFunctionFactory`
- Using `AIProjectClient` to discover the OpenAI connection from the Microsoft Foundry project
- Using `AIProjectClient` to discover the OpenAI connection from the Azure AI Foundry project
- Building a `ChatClientAgent` with custom instructions and tools
- Deploying to the Foundry Hosted Agent service
@@ -15,7 +15,7 @@ Key features:
Before running this sample, ensure you have:
1. .NET 10 SDK installed
2. A Microsoft Foundry Project with a chat model deployed (e.g., gpt-4o-mini)
2. An Azure AI Foundry Project with a chat model deployed (e.g., gpt-4o-mini)
3. Azure CLI installed and authenticated (`az login`)
## Environment Variables
@@ -23,7 +23,7 @@ Before running this sample, ensure you have:
Set the following environment variables:
```powershell
# Replace with your Microsoft Foundry project endpoint
# Replace with your Azure AI Foundry project endpoint
$env:AZURE_AI_PROJECT_ENDPOINT="https://your-project.services.ai.azure.com/api/projects/your-project-name"
# Optional, defaults to gpt-4o-mini
@@ -1,7 +1,7 @@
// Copyright (c) Microsoft. All rights reserved.
// This sample demonstrates a multi-agent workflow with Writer and Reviewer agents
// using Microsoft Foundry AIProjectClient and the Agent Framework WorkflowBuilder.
// using Azure AI Foundry AIProjectClient and the Agent Framework WorkflowBuilder.
#pragma warning disable CA2252 // AIProjectClient and Agents API require opting into preview features
@@ -42,7 +42,7 @@ which provisions a REST API endpoint compatible with the OpenAI Responses protoc
Before running this sample, ensure you have:
1. **Microsoft Foundry Project**
1. **Azure AI Foundry Project**
- Project created.
- Chat model deployed (e.g., `gpt-4o` or `gpt-4.1`)
- Note your project endpoint URL and model deployment name
@@ -4,7 +4,7 @@ name: FoundryMultiAgent
displayName: "Foundry Multi-Agent Workflow"
description: >
A multi-agent workflow featuring a Writer and Reviewer that collaborate
to create and refine content using Microsoft Foundry PersistentAgentsClient.
to create and refine content using Azure AI Foundry PersistentAgentsClient.
metadata:
authors:
- Microsoft Agent Framework Team
@@ -1,7 +1,7 @@
// Copyright (c) Microsoft. All rights reserved.
// Seattle Hotel Agent - A simple agent with a tool to find hotels in Seattle.
// Uses Microsoft Agent Framework with Microsoft Foundry.
// Uses Microsoft Agent Framework with Azure AI Foundry.
// Ready for deployment to Foundry Hosted Agent service.
#pragma warning disable CA2252 // AIProjectClient and Agents API require opting into preview features
@@ -39,7 +39,7 @@ which provisions a REST API endpoint compatible with the OpenAI Responses protoc
Before running this sample, ensure you have:
1. **Microsoft Foundry Project**
1. **Azure AI Foundry Project**
- Project created.
- Chat model deployed (e.g., `gpt-4o` or `gpt-4.1`)
- Note your project endpoint URL and model deployment name
@@ -57,7 +57,7 @@ Before running this sample, ensure you have:
Set the following environment variables (matching `agent.yaml`):
- `AZURE_AI_PROJECT_ENDPOINT` - Your Microsoft Foundry project endpoint URL (required)
- `AZURE_AI_PROJECT_ENDPOINT` - Your Azure AI Foundry project endpoint URL (required)
- `MODEL_DEPLOYMENT_NAME` - The deployment name for your chat model (defaults to `gpt-4o-mini`)
**PowerShell:**
@@ -20,7 +20,7 @@ Before running any sample, ensure you have:
1. **.NET 10 SDK** or later — [Download](https://dotnet.microsoft.com/download/dotnet/10.0)
2. **Azure CLI** installed — [Install guide](https://learn.microsoft.com/cli/azure/install-azure-cli)
3. **Azure OpenAI** or **Microsoft Foundry project** with a chat model deployed (e.g., `gpt-4o-mini`)
3. **Azure OpenAI** or **Azure AI Foundry project** with a chat model deployed (e.g., `gpt-4o-mini`)
### Authenticate with Azure CLI
@@ -39,14 +39,14 @@ Most samples require one or more of these environment variables:
|----------|---------|-------------|
| `AZURE_OPENAI_ENDPOINT` | Most samples | Your Azure OpenAI resource endpoint URL |
| `AZURE_OPENAI_DEPLOYMENT_NAME` | Most samples | Chat model deployment name (defaults to `gpt-4o-mini`) |
| `AZURE_AI_PROJECT_ENDPOINT` | AgentWithLocalTools, FoundryMultiAgent, FoundrySingleAgent | Microsoft Foundry project endpoint |
| `AZURE_AI_PROJECT_ENDPOINT` | AgentWithLocalTools, FoundryMultiAgent, FoundrySingleAgent | Azure AI Foundry project endpoint |
| `MODEL_DEPLOYMENT_NAME` | AgentWithLocalTools, FoundryMultiAgent, FoundrySingleAgent | Chat model deployment name (defaults to `gpt-4o-mini`) |
See each sample's README for the specific variables required.
## Microsoft Foundry Setup (for samples that use Foundry)
## Azure AI Foundry Setup (for samples that use Foundry)
Some samples (`AgentWithLocalTools`, `FoundrySingleAgent`, `FoundryMultiAgent`) connect to a Microsoft Foundry project. If you're using these samples, you'll need additional setup.
Some samples (`AgentWithLocalTools`, `FoundrySingleAgent`, `FoundryMultiAgent`) connect to an Azure AI Foundry project. If you're using these samples, you'll need additional setup.
### Azure AI Developer Role
@@ -61,7 +61,7 @@ az role assignment create `
> **Note**: You need **Owner** or **User Access Administrator** permissions on the resource to assign roles. If you don't have this, you may need to request JIT (Just-In-Time) elevated access via [Azure PIM](https://portal.azure.com/#view/Microsoft_Azure_PIMCommon/ActivationMenuBlade/~/aadmigratedresource).
For more details on permissions, see [Microsoft Foundry Permissions](https://aka.ms/FoundryPermissions).
For more details on permissions, see [Azure AI Foundry Permissions](https://aka.ms/FoundryPermissions).
## Running a Sample
+1 -1
View File
@@ -28,7 +28,7 @@ dotnet/samples/
│ ├── AGUI/ # AG-UI protocol samples
│ ├── DeclarativeAgents/ # Declarative agent definitions
│ ├── DevUI/ # DevUI samples
│ ├── AgentsWithFoundry/ # Microsoft Foundry samples (FoundryAgent + AsAIAgent extensions)
│ ├── AgentsWithFoundry/ # Azure AI Foundry samples (FoundryAgent + AsAIAgent extensions)
│ └── ModelContextProtocol/ # MCP server/client patterns
├── 03-workflows/ # Workflow patterns
│ ├── _StartHere/ # Introductory workflow samples
@@ -21,15 +21,6 @@ internal interface ICheckpointingHandle
/// <summary>
/// Restores the system state from the specified checkpoint asynchronously.
/// </summary>
/// <remarks>
/// This contract is used by live runtime restore paths. Implementations may re-emit pending
/// external request events as part of the restore once the active event stream is ready to
/// observe them.
///
/// Initial resume paths that create a new event stream should restore state first and defer
/// any replay until after the subscriber is attached, rather than calling this contract
/// directly before the stream is ready.
/// </remarks>
/// <param name="checkpointInfo">The checkpoint information that identifies the state to restore. Cannot be null.</param>
/// <param name="cancellationToken">A cancellation token that can be used to cancel the restore operation.</param>
/// <returns>A <see cref="ValueTask"/> that represents the asynchronous restore operation.</returns>
@@ -36,10 +36,9 @@ internal sealed class AsyncRunHandle : ICheckpointingHandle, IAsyncDisposable
this._eventStream.Start();
// If there are already unprocessed messages or unserviced requests (e.g., from a
// checkpoint restore that happened before this handle was created), signal the run
// loop to start processing them
if (stepRunner.HasUnprocessedMessages || stepRunner.HasUnservicedRequests)
// If there are already unprocessed messages (e.g., from a checkpoint restore that happened
// before this handle was created), signal the run loop to start processing them
if (stepRunner.HasUnprocessedMessages)
{
this.SignalInputToRunLoop();
}
@@ -193,17 +192,13 @@ internal sealed class AsyncRunHandle : ICheckpointingHandle, IAsyncDisposable
{
streamingEventStream.ClearBufferedEvents();
}
else if (this._eventStream is LockstepRunEventStream lockstepEventStream)
{
lockstepEventStream.ClearBufferedEvents();
}
// Restore the workflow state through the live runtime-restore path.
// This can re-emit pending requests into the already-active event stream.
// Restore the workflow state - this will republish unserviced requests as new events
await this._checkpointingHandle.RestoreCheckpointAsync(checkpointInfo, cancellationToken).ConfigureAwait(false);
// After restore, signal the run loop to process any restored messages. Initial resume
// paths handle this separately when they create the event stream after restoring state.
// After restore, signal the run loop to process any restored messages
// This is necessary because ClearBufferedEvents() doesn't signal, and the restored
// queued messages won't automatically wake up the run loop
this.SignalInputToRunLoop();
}
}
@@ -27,14 +27,6 @@ internal interface ISuperStepRunner
ConcurrentEventSink OutgoingEvents { get; }
/// <summary>
/// Re-emits <see cref="RequestInfoEvent"/>s for any pending external requests.
/// Called by event streams after subscribing to <see cref="OutgoingEvents"/> so that
/// requests restored from a checkpoint are observable even when the restore happened
/// before the subscription was active.
/// </summary>
ValueTask RepublishPendingEventsAsync(CancellationToken cancellationToken = default);
ValueTask<bool> RunSuperStepAsync(CancellationToken cancellationToken);
// This cannot be cancelled
@@ -15,7 +15,6 @@ internal sealed class LockstepRunEventStream : IRunEventStream
{
private readonly CancellationTokenSource _stopCancellation = new();
private readonly InputWaiter _inputWaiter = new();
private ConcurrentQueue<WorkflowEvent> _eventSink = new();
private int _isDisposed;
private readonly ISuperStepRunner _stepRunner;
@@ -36,8 +35,6 @@ internal sealed class LockstepRunEventStream : IRunEventStream
// doesn't leak into caller code via AsyncLocal.
Activity? previousActivity = Activity.Current;
this._stepRunner.OutgoingEvents.EventRaised += this.OnWorkflowEventAsync;
this._sessionActivity = this._stepRunner.TelemetryContext.StartWorkflowSessionActivity();
this._sessionActivity?.SetTag(Tags.WorkflowId, this._stepRunner.StartExecutorId)
.SetTag(Tags.SessionId, this._stepRunner.SessionId);
@@ -59,6 +56,10 @@ internal sealed class LockstepRunEventStream : IRunEventStream
using CancellationTokenSource linkedSource = CancellationTokenSource.CreateLinkedTokenSource(this._stopCancellation.Token, cancellationToken);
ConcurrentQueue<WorkflowEvent> eventSink = [];
this._stepRunner.OutgoingEvents.EventRaised += OnWorkflowEventAsync;
// Re-establish session as parent so the run activity nests correctly.
Activity.Current = this._sessionActivity;
@@ -72,31 +73,7 @@ internal sealed class LockstepRunEventStream : IRunEventStream
runActivity?.AddEvent(new ActivityEvent(EventNames.WorkflowStarted));
// Emit WorkflowStartedEvent to the event stream for consumers
this._eventSink.Enqueue(new WorkflowStartedEvent());
// Re-emit any pending external requests that were restored from a checkpoint
// before this subscription was active. For non-resume starts this is a no-op.
// This runs after WorkflowStartedEvent so consumers always see the started event first.
await this._stepRunner.RepublishPendingEventsAsync(linkedSource.Token).ConfigureAwait(false);
// When resuming from a checkpoint with only pending requests (no queued messages),
// the inner processing loop won't execute, so we must drain events now.
// For normal starts this is a no-op since the inner loop handles the drain.
if (!this._stepRunner.HasUnprocessedMessages)
{
var (drainedEvents, shouldHalt) = this.DrainAndFilterEvents();
foreach (WorkflowEvent raisedEvent in drainedEvents)
{
yield return raisedEvent;
}
if (shouldHalt)
{
yield break;
}
this.RunStatus = this._stepRunner.HasUnservicedRequests ? RunStatus.PendingRequests : RunStatus.Idle;
}
eventSink.Enqueue(new WorkflowStartedEvent());
do
{
@@ -130,19 +107,26 @@ internal sealed class LockstepRunEventStream : IRunEventStream
yield break; // Exit if cancellation is requested
}
var (drainedEvents, shouldHalt) = this.DrainAndFilterEvents();
foreach (WorkflowEvent raisedEvent in drainedEvents)
bool hadRequestHaltEvent = false;
foreach (WorkflowEvent raisedEvent in Interlocked.Exchange(ref eventSink, []))
{
if (linkedSource.Token.IsCancellationRequested)
{
yield break; // Exit if cancellation is requested
}
yield return raisedEvent;
// TODO: Do we actually want to interpret this as a termination request?
if (raisedEvent is RequestHaltEvent)
{
hadRequestHaltEvent = true;
}
else
{
yield return raisedEvent;
}
}
if (shouldHalt || linkedSource.Token.IsCancellationRequested)
if (hadRequestHaltEvent || linkedSource.Token.IsCancellationRequested)
{
// If we had a completion event, we are done.
yield break;
@@ -167,23 +151,25 @@ internal sealed class LockstepRunEventStream : IRunEventStream
finally
{
this.RunStatus = this._stepRunner.HasUnservicedRequests ? RunStatus.PendingRequests : RunStatus.Idle;
this._stepRunner.OutgoingEvents.EventRaised -= OnWorkflowEventAsync;
// Explicitly dispose the Activity so Activity.Stop fires deterministically,
// regardless of how the async iterator enumerator is disposed.
runActivity?.Dispose();
}
ValueTask OnWorkflowEventAsync(object? sender, WorkflowEvent e)
{
eventSink.Enqueue(e);
return default;
}
// If we are Idle or Ended, we should break out of the loop
// If we are PendingRequests and not blocking on pending requests, we should break out of the loop
// If cancellation is requested, we should break out of the loop
bool ShouldBreak() => this.RunStatus is RunStatus.Idle or RunStatus.Ended ||
(this.RunStatus == RunStatus.PendingRequests && !blockOnPendingRequest) ||
linkedSource.Token.IsCancellationRequested;
}
internal void ClearBufferedEvents()
{
Interlocked.Exchange(ref this._eventSink, new ConcurrentQueue<WorkflowEvent>());
(this.RunStatus == RunStatus.PendingRequests && !blockOnPendingRequest) ||
linkedSource.Token.IsCancellationRequested;
}
/// <summary>
@@ -206,7 +192,6 @@ internal sealed class LockstepRunEventStream : IRunEventStream
if (Interlocked.Exchange(ref this._isDisposed, 1) == 0)
{
this._stopCancellation.Cancel();
this._stepRunner.OutgoingEvents.EventRaised -= this.OnWorkflowEventAsync;
// Stop the session activity
if (this._sessionActivity is not null)
@@ -222,32 +207,4 @@ internal sealed class LockstepRunEventStream : IRunEventStream
return default;
}
private ValueTask OnWorkflowEventAsync(object? sender, WorkflowEvent e)
{
this._eventSink.Enqueue(e);
return default;
}
// Atomically drains the event sink and separates workflow events from halt signals.
// Used by both the early-drain (resume with pending requests only) and
// the inner superstep drain to keep halt-detection logic in one place.
private (List<WorkflowEvent> Events, bool ShouldHalt) DrainAndFilterEvents()
{
List<WorkflowEvent> events = [];
bool shouldHalt = false;
foreach (WorkflowEvent e in Interlocked.Exchange(ref this._eventSink, new ConcurrentQueue<WorkflowEvent>()))
{
if (e is RequestHaltEvent)
{
shouldHalt = true;
}
else
{
events.Add(e);
}
}
return (events, shouldHalt);
}
}
@@ -23,8 +23,7 @@ internal sealed class StreamingRunEventStream : IRunEventStream
private readonly CancellationTokenSource _runLoopCancellation;
private readonly bool _disableRunLoop;
private Task? _runLoopTask;
private volatile RunStatus _runStatus = RunStatus.NotStarted;
private RunStatus _runStatus = RunStatus.NotStarted;
private int _completionEpoch; // Tracks which completion signal belongs to which consumer iteration
public StreamingRunEventStream(ISuperStepRunner stepRunner, bool disableRunLoop = false)
@@ -61,10 +60,6 @@ internal sealed class StreamingRunEventStream : IRunEventStream
// Subscribe to events - they will flow directly to the channel as they're raised
this._stepRunner.OutgoingEvents.EventRaised += OnEventRaisedAsync;
// Re-emit any pending external requests that were restored from a checkpoint
// before this subscription was active. For non-resume starts this is a no-op.
await this._stepRunner.RepublishPendingEventsAsync(linkedSource.Token).ConfigureAwait(false);
// Start the session-level activity that spans the entire run loop lifetime.
// Individual run-stage activities are nested within this session activity.
Activity? sessionActivity = this._stepRunner.TelemetryContext.StartWorkflowSessionActivity();
@@ -128,7 +123,7 @@ internal sealed class StreamingRunEventStream : IRunEventStream
// Wait for next input from the consumer
// Works for both Idle (no work) and PendingRequests (waiting for responses)
await this._inputWaiter.WaitForInputAsync(linkedSource.Token).ConfigureAwait(false);
await this._inputWaiter.WaitForInputAsync(TimeSpan.FromSeconds(1), linkedSource.Token).ConfigureAwait(false);
// When signaled, resume running
this._runStatus = RunStatus.Running;
@@ -210,10 +205,7 @@ internal sealed class StreamingRunEventStream : IRunEventStream
[EnumeratorCancellation] CancellationToken cancellationToken = default)
{
// Get the current epoch - we'll only respond to completion signals from this epoch or later
int currentEpoch = Volatile.Read(ref this._completionEpoch);
bool expectingFreshWork = this._stepRunner.HasUnprocessedMessages || this._runStatus == RunStatus.Running;
int myEpoch = expectingFreshWork ? currentEpoch + 1 : currentEpoch;
int myEpoch = Volatile.Read(ref this._completionEpoch) + 1;
// Use custom async enumerable to avoid exceptions on cancellation.
NonThrowingChannelReaderAsyncEnumerable<WorkflowEvent> eventStream = new(this._eventChannel.Reader);
@@ -50,13 +50,10 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen
return runner.BeginStreamAsync(this.ExecutionMode, cancellationToken);
}
internal ValueTask<AsyncRunHandle> ResumeRunAsync(Workflow workflow, CheckpointInfo fromCheckpoint, IEnumerable<Type> knownValidInputTypes, CancellationToken cancellationToken = default)
=> this.ResumeRunAsync(workflow, fromCheckpoint, knownValidInputTypes, republishPendingEvents: true, cancellationToken);
internal ValueTask<AsyncRunHandle> ResumeRunAsync(Workflow workflow, CheckpointInfo fromCheckpoint, IEnumerable<Type> knownValidInputTypes, bool republishPendingEvents, CancellationToken cancellationToken = default)
internal ValueTask<AsyncRunHandle> ResumeRunAsync(Workflow workflow, CheckpointInfo fromCheckpoint, IEnumerable<Type> knownValidInputTypes, CancellationToken cancellationToken)
{
InProcessRunner runner = InProcessRunner.CreateTopLevelRunner(workflow, this.CheckpointManager, fromCheckpoint.SessionId, this.EnableConcurrentRuns, knownValidInputTypes);
return runner.ResumeStreamAsync(this.ExecutionMode, fromCheckpoint, republishPendingEvents, cancellationToken);
return runner.ResumeStreamAsync(this.ExecutionMode, fromCheckpoint, cancellationToken);
}
/// <inheritdoc/>
@@ -107,32 +104,6 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen
return new(runHandle);
}
/// <summary>
/// Resumes a streaming workflow run from a checkpoint with control over whether
/// pending request events are republished through the event stream.
/// </summary>
/// <param name="workflow">The workflow to resume.</param>
/// <param name="fromCheckpoint">The checkpoint to resume from.</param>
/// <param name="republishPendingEvents">
/// When <see langword="true"/>, any pending request events are republished through the event
/// stream after subscribing. When <see langword="false"/>, the caller is responsible for
/// handling pending requests (e.g., <see cref="WorkflowSession"/> already sends responses).
/// </param>
/// <param name="cancellationToken">Cancellation token.</param>
internal async ValueTask<StreamingRun> ResumeStreamingInternalAsync(
Workflow workflow,
CheckpointInfo fromCheckpoint,
bool republishPendingEvents,
CancellationToken cancellationToken = default)
{
this.VerifyCheckpointingConfigured();
AsyncRunHandle runHandle = await this.ResumeRunAsync(workflow, fromCheckpoint, [], republishPendingEvents, cancellationToken)
.ConfigureAwait(false);
return new(runHandle);
}
private async ValueTask<AsyncRunHandle> BeginRunHandlingChatProtocolAsync<TInput>(Workflow workflow,
TInput input,
string? sessionId = null,
@@ -71,28 +71,6 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle
/// <inheritdoc cref="ISuperStepRunner.StartExecutorId"/>
public string StartExecutorId { get; }
/// <summary>
/// Gating flag for deferred event republishing after checkpoint restore.
/// </summary>
/// <remarks>
/// <para>
/// Written with <see cref="Volatile.Write(ref int, int)"/> in <see cref="ResumeStreamAsync(ExecutionMode, CheckpointInfo, bool, CancellationToken)"/>
/// and consumed atomically with <see cref="Interlocked.Exchange(ref int, int)"/> in
/// <see cref="ISuperStepRunner.RepublishPendingEventsAsync"/>. The write does not need a full
/// memory barrier because it is sequenced before the <see cref="AsyncRunHandle"/> constructor
/// by the <see langword="await"/> in <see cref="ResumeStreamAsync(ExecutionMode, CheckpointInfo, bool, CancellationToken)"/>. The constructor is the
/// only code path that triggers consumption (via the event stream's subscribe and republish flow).
/// </para>
/// <para>
/// Note: <see cref="AsyncRunHandle"/> also reads <see cref="ISuperStepRunner.HasUnservicedRequests"/>
/// in its constructor to signal the run loop, but that property reads from
/// <see cref="InProcessRunnerContext"/>'s request dictionary (restored during
/// <see cref="RestoreCheckpointCoreAsync"/>), not from this flag. The two are independent:
/// <c>HasUnservicedRequests</c> triggers the run loop; <c>_needsRepublish</c> triggers event emission.
/// </para>
/// </remarks>
private int _needsRepublish;
/// <inheritdoc cref="ISuperStepRunner.TelemetryContext"/>
public WorkflowTelemetryContext TelemetryContext => this.Workflow.TelemetryContext;
@@ -167,10 +145,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle
return new(new AsyncRunHandle(this, this, mode));
}
public ValueTask<AsyncRunHandle> ResumeStreamAsync(ExecutionMode mode, CheckpointInfo fromCheckpoint, CancellationToken cancellationToken = default)
=> this.ResumeStreamAsync(mode, fromCheckpoint, republishPendingEvents: true, cancellationToken);
public async ValueTask<AsyncRunHandle> ResumeStreamAsync(ExecutionMode mode, CheckpointInfo fromCheckpoint, bool republishPendingEvents, CancellationToken cancellationToken = default)
public async ValueTask<AsyncRunHandle> ResumeStreamAsync(ExecutionMode mode, CheckpointInfo fromCheckpoint, CancellationToken cancellationToken = default)
{
this.RunContext.CheckEnded();
Throw.IfNull(fromCheckpoint);
@@ -179,18 +154,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle
throw new InvalidOperationException("This runner was not configured with a CheckpointManager, so it cannot restore checkpoints.");
}
// Restore checkpoint state without republishing pending request events.
// The event stream will republish them after subscribing so that events
// are never lost to an absent subscriber.
await this.RestoreCheckpointCoreAsync(fromCheckpoint, cancellationToken).ConfigureAwait(false);
if (republishPendingEvents)
{
// Signal the event stream to republish pending requests after subscribing.
// This is consumed atomically by RepublishPendingEventsAsync.
Volatile.Write(ref this._needsRepublish, 1);
}
await this.RestoreCheckpointAsync(fromCheckpoint, cancellationToken).ConfigureAwait(false);
return new AsyncRunHandle(this, this, mode);
}
@@ -199,16 +163,6 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle
bool ISuperStepRunner.TryGetResponsePortExecutorId(string portId, out string? executorId)
=> this.RunContext.TryGetResponsePortExecutorId(portId, out executorId);
ValueTask ISuperStepRunner.RepublishPendingEventsAsync(CancellationToken cancellationToken)
{
if (Interlocked.Exchange(ref this._needsRepublish, 0) != 0)
{
return this.RunContext.RepublishUnservicedRequestsAsync(cancellationToken);
}
return default;
}
public bool IsCheckpointingEnabled => this.RunContext.IsCheckpointingEnabled;
public IReadOnlyList<CheckpointInfo> Checkpoints => this._checkpoints;
@@ -356,31 +310,7 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle
this._checkpoints.Add(this._lastCheckpointInfo);
}
/// <summary>
/// Restores checkpoint state and re-emits any pending external request events.
/// </summary>
/// <remarks>
/// This is the <see cref="ICheckpointingHandle"/> implementation used for runtime restores
/// where the event stream subscription is already active. For initial resumes,
/// <see cref="ResumeStreamAsync(ExecutionMode, CheckpointInfo, CancellationToken)"/> calls
/// <see cref="RestoreCheckpointCoreAsync"/> directly and defers republishing to the event stream.
/// </remarks>
public async ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellationToken = default)
{
await this.RestoreCheckpointCoreAsync(checkpointInfo, cancellationToken).ConfigureAwait(false);
// Republish pending request events. This is safe for runtime restores where
// the event stream is already subscribed. For initial resumes the event stream
// handles republishing itself, so ResumeStreamAsync calls RestoreCheckpointCoreAsync directly.
await this.RunContext.RepublishUnservicedRequestsAsync(cancellationToken).ConfigureAwait(false);
}
/// <summary>
/// Restores checkpoint state (queued messages, executor state, edge state, etc.)
/// without republishing pending request events. The caller is responsible for
/// ensuring events are republished after an event subscriber is attached.
/// </summary>
private async ValueTask RestoreCheckpointCoreAsync(CheckpointInfo checkpointInfo, CancellationToken cancellationToken = default)
{
this.RunContext.CheckEnded();
Throw.IfNull(checkpointInfo);
@@ -405,9 +335,11 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle
await this.RunContext.ImportStateAsync(checkpoint).ConfigureAwait(false);
Task executorNotifyTask = this.RunContext.NotifyCheckpointLoadedAsync(cancellationToken);
ValueTask republishRequestsTask = this.RunContext.RepublishUnservicedRequestsAsync(cancellationToken);
await this.EdgeMap.ImportStateAsync(checkpoint).ConfigureAwait(false);
await Task.WhenAll(executorNotifyTask,
republishRequestsTask.AsTask(),
restoreCheckpointIndexTask.AsTask()).ConfigureAwait(false);
this._lastCheckpointInfo = checkpointInfo;
@@ -14,7 +14,6 @@ internal sealed class RequestPortOptions;
internal sealed class RequestInfoExecutor : Executor
{
private const string WrappedRequestsStateKey = nameof(WrappedRequestsStateKey);
private readonly Dictionary<string, ExternalRequest> _wrappedRequests = [];
private RequestPort Port { get; }
private IExternalRequestSink? RequestSink { get; set; }
@@ -125,46 +124,22 @@ internal sealed class RequestInfoExecutor : Executor
return null;
}
if (this._allowWrapped && this._wrappedRequests.TryGetValue(message.RequestId, out ExternalRequest? originalRequest))
{
await context.SendMessageAsync(originalRequest.RewrapResponse(message), cancellationToken: cancellationToken).ConfigureAwait(false);
}
else
{
await context.SendMessageAsync(message, cancellationToken: cancellationToken).ConfigureAwait(false);
}
if (!message.Data.IsType(this.Port.Response, out object? data))
{
throw this.Port.CreateExceptionForType(message);
}
if (this._allowWrapped && this._wrappedRequests.TryGetValue(message.RequestId, out ExternalRequest? originalRequest))
{
await context.SendMessageAsync(originalRequest.RewrapResponse(message), cancellationToken: cancellationToken).ConfigureAwait(false);
this._wrappedRequests.Remove(message.RequestId);
}
else
{
await context.SendMessageAsync(message, cancellationToken: cancellationToken).ConfigureAwait(false);
await context.SendMessageAsync(data, cancellationToken: cancellationToken).ConfigureAwait(false);
}
await context.SendMessageAsync(data, cancellationToken: cancellationToken).ConfigureAwait(false);
return message;
}
protected internal override async ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default)
{
await context.QueueStateUpdateAsync(WrappedRequestsStateKey,
new Dictionary<string, ExternalRequest>(this._wrappedRequests, StringComparer.Ordinal),
cancellationToken: cancellationToken).ConfigureAwait(false);
await base.OnCheckpointingAsync(context, cancellationToken).ConfigureAwait(false);
}
protected internal override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default)
{
await base.OnCheckpointRestoredAsync(context, cancellationToken).ConfigureAwait(false);
this._wrappedRequests.Clear();
Dictionary<string, ExternalRequest> wrappedRequests =
await context.ReadStateAsync<Dictionary<string, ExternalRequest>>(WrappedRequestsStateKey, cancellationToken: cancellationToken)
.ConfigureAwait(false) ?? [];
foreach (KeyValuePair<string, ExternalRequest> wrappedRequest in wrappedRequests)
{
this._wrappedRequests[wrappedRequest.Key] = wrappedRequest.Value;
}
}
}
@@ -1,7 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
@@ -24,7 +23,6 @@ internal class WorkflowHostExecutor : Executor, IAsyncDisposable
private InProcessRunner? _activeRunner;
private InMemoryCheckpointManager? _checkpointManager;
private readonly ExecutorOptions _options;
private readonly ConcurrentDictionary<string, RequestPortInfo> _pendingResponsePorts = new(StringComparer.Ordinal);
private ISuperStepJoinContext? _joinContext;
private string? _joinId;
@@ -165,11 +163,6 @@ internal class WorkflowHostExecutor : Executor, IAsyncDisposable
private ExternalResponse? CheckAndUnqualifyResponse([DisallowNull] ExternalResponse response)
{
if (this._pendingResponsePorts.TryRemove(response.RequestId, out RequestPortInfo? originalPort))
{
return response with { PortInfo = originalPort };
}
if (!Throw.IfNull(response).PortInfo.PortId.StartsWith($"{this.Id}.", StringComparison.Ordinal))
{
return null;
@@ -200,7 +193,6 @@ internal class WorkflowHostExecutor : Executor, IAsyncDisposable
break;
case RequestInfoEvent requestInfoEvt:
ExternalRequest request = requestInfoEvt.Request;
this._pendingResponsePorts[request.RequestId] = request.PortInfo;
resultTask = this._joinContext?.SendMessageAsync(this.Id, this.QualifyRequestPortId(request)).AsTask() ?? Task.CompletedTask;
break;
case WorkflowErrorEvent errorEvent:
@@ -254,13 +246,9 @@ internal class WorkflowHostExecutor : Executor, IAsyncDisposable
}
private const string CheckpointManagerStateKey = nameof(CheckpointManager);
private const string PendingResponsePortsStateKey = nameof(PendingResponsePortsStateKey);
protected internal override async ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default)
{
await context.QueueStateUpdateAsync(CheckpointManagerStateKey, this._checkpointManager, cancellationToken: cancellationToken).ConfigureAwait(false);
await context.QueueStateUpdateAsync(PendingResponsePortsStateKey,
new Dictionary<string, RequestPortInfo>(this._pendingResponsePorts, StringComparer.Ordinal),
cancellationToken: cancellationToken).ConfigureAwait(false);
await base.OnCheckpointingAsync(context, cancellationToken).ConfigureAwait(false);
}
@@ -281,15 +269,6 @@ internal class WorkflowHostExecutor : Executor, IAsyncDisposable
await this.ResetAsync().ConfigureAwait(false);
}
this._pendingResponsePorts.Clear();
Dictionary<string, RequestPortInfo> pendingResponsePorts =
await context.ReadStateAsync<Dictionary<string, RequestPortInfo>>(PendingResponsePortsStateKey, cancellationToken: cancellationToken)
.ConfigureAwait(false) ?? [];
foreach (KeyValuePair<string, RequestPortInfo> pendingResponsePort in pendingResponsePorts)
{
this._pendingResponsePorts[pendingResponsePort.Key] = pendingResponsePort.Value;
}
await this.EnsureRunSendMessageAsync(resume: true, cancellationToken: cancellationToken).ConfigureAwait(false);
}
@@ -301,8 +280,6 @@ internal class WorkflowHostExecutor : Executor, IAsyncDisposable
this._run = null;
}
this._pendingResponsePorts.Clear();
if (this._activeRunner != null)
{
this._activeRunner.OutgoingEvents.EventRaised -= this.ForwardWorkflowEventAsync;
@@ -19,14 +19,7 @@ namespace Microsoft.Agents.AI.Workflows;
internal sealed class WorkflowSession : AgentSession
{
private readonly Workflow _workflow;
/// <summary>
/// The execution environment for this session. Concrete type is required because
/// <see cref="CreateOrResumeRunAsync"/> uses the internal
/// <see cref="InProcessExecutionEnvironment.ResumeStreamingInternalAsync"/> API.
/// </summary>
private readonly InProcessExecutionEnvironment _inProcEnvironment;
private readonly IWorkflowExecutionEnvironment _executionEnvironment;
private readonly bool _includeExceptionDetails;
private readonly bool _includeWorkflowOutputsInResponse;
@@ -70,22 +63,17 @@ internal sealed class WorkflowSession : AgentSession
public WorkflowSession(Workflow workflow, string sessionId, IWorkflowExecutionEnvironment executionEnvironment, bool includeExceptionDetails = false, bool includeWorkflowOutputsInResponse = false)
{
this._workflow = Throw.IfNull(workflow);
this._executionEnvironment = Throw.IfNull(executionEnvironment);
this._includeExceptionDetails = includeExceptionDetails;
this._includeWorkflowOutputsInResponse = includeWorkflowOutputsInResponse;
IWorkflowExecutionEnvironment env = Throw.IfNull(executionEnvironment);
if (VerifyCheckpointingConfiguration(env, out InProcessExecutionEnvironment? inProcEnv))
if (VerifyCheckpointingConfiguration(executionEnvironment, out InProcessExecutionEnvironment? inProcEnv))
{
// We have an InProcessExecutionEnvironment which is not configured for checkpointing. Ensure it has an externalizable checkpoint manager,
// since we are responsible for maintaining the state.
env = inProcEnv.WithCheckpointing(this.EnsureExternalizedInMemoryCheckpointing());
this._executionEnvironment = inProcEnv.WithCheckpointing(this.EnsureExternalizedInMemoryCheckpointing());
}
this._inProcEnvironment = env as InProcessExecutionEnvironment
?? throw new InvalidOperationException(
$"WorkflowSession requires an {nameof(InProcessExecutionEnvironment)}, " +
$"but received {env.GetType().Name}.");
this.SessionId = Throw.IfNullOrEmpty(sessionId);
this.ChatHistoryProvider = new WorkflowChatHistoryProvider();
}
@@ -98,30 +86,24 @@ internal sealed class WorkflowSession : AgentSession
public WorkflowSession(Workflow workflow, JsonElement serializedSession, IWorkflowExecutionEnvironment executionEnvironment, bool includeExceptionDetails = false, bool includeWorkflowOutputsInResponse = false, JsonSerializerOptions? jsonSerializerOptions = null)
{
this._workflow = Throw.IfNull(workflow);
this._executionEnvironment = Throw.IfNull(executionEnvironment);
this._includeExceptionDetails = includeExceptionDetails;
this._includeWorkflowOutputsInResponse = includeWorkflowOutputsInResponse;
IWorkflowExecutionEnvironment env = Throw.IfNull(executionEnvironment);
JsonMarshaller marshaller = new(jsonSerializerOptions);
SessionState sessionState = marshaller.Marshal<SessionState>(serializedSession);
this._inMemoryCheckpointManager = sessionState.CheckpointManager;
if (this._inMemoryCheckpointManager != null &&
VerifyCheckpointingConfiguration(env, out InProcessExecutionEnvironment? inProcEnv))
VerifyCheckpointingConfiguration(executionEnvironment, out InProcessExecutionEnvironment? inProcEnv))
{
env = inProcEnv.WithCheckpointing(this.EnsureExternalizedInMemoryCheckpointing());
this._executionEnvironment = inProcEnv.WithCheckpointing(this.EnsureExternalizedInMemoryCheckpointing());
}
else if (this._inMemoryCheckpointManager != null)
{
throw new ArgumentException("The session was saved with an externalized checkpoint manager, but the incoming execution environment does not support it.", nameof(executionEnvironment));
}
this._inProcEnvironment = env as InProcessExecutionEnvironment
?? throw new InvalidOperationException(
$"WorkflowSession requires an {nameof(InProcessExecutionEnvironment)}, " +
$"but received {env.GetType().Name}.");
this.SessionId = sessionState.SessionId;
this.ChatHistoryProvider = new WorkflowChatHistoryProvider();
@@ -178,15 +160,10 @@ internal sealed class WorkflowSession : AgentSession
// and does not need to be checked again here.
if (this.LastCheckpoint is not null)
{
// Use the internal resume path that suppresses pending request republishing.
// WorkflowSession handles pending requests itself by converting matching responses
// via SendMessagesWithResponseConversionAsync, so event-stream republishing would
// cause unwanted duplicate events visible to the consumer.
StreamingRun run =
await this._inProcEnvironment
.ResumeStreamingInternalAsync(this._workflow,
await this._executionEnvironment
.ResumeStreamingAsync(this._workflow,
this.LastCheckpoint,
republishPendingEvents: false,
cancellationToken)
.ConfigureAwait(false);
@@ -195,7 +172,7 @@ internal sealed class WorkflowSession : AgentSession
return new ResumeRunResult(run, dispatchInfo);
}
StreamingRun newRun = await this._inProcEnvironment
StreamingRun newRun = await this._executionEnvironment
.RunStreamingAsync(this._workflow,
messages,
this.SessionId,
@@ -126,13 +126,6 @@ internal sealed class WorkflowHarness(Workflow workflow, string runId)
{
hasRequest = true;
}
else
{
// This is a republished event for the request we're already responding to
// (emitted by RepublishUnservicedRequestsAsync during checkpoint resume).
// Skip yielding it so downstream code doesn't treat it as a new pending request.
continue;
}
break;
case ConversationUpdateEvent conversationEvent:
@@ -1,445 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using FluentAssertions;
using Microsoft.Agents.AI.Workflows.InProc;
using Microsoft.Agents.AI.Workflows.Sample;
namespace Microsoft.Agents.AI.Workflows.UnitTests;
/// <summary>
/// Regression tests for GH-2485: pending <see cref="RequestInfoEvent"/> objects must be
/// re-emitted after resuming a workflow from a checkpoint.
/// </summary>
public class CheckpointResumeTests
{
/// <summary>
/// Verifies that a resumed workflow re-emits <see cref="RequestInfoEvent"/>s for
/// pending external requests that existed at the time of the checkpoint.
/// </summary>
[Theory]
[InlineData(ExecutionEnvironment.InProcess_OffThread)]
[InlineData(ExecutionEnvironment.InProcess_Lockstep)]
internal async Task Checkpoint_Resume_WithPendingRequests_RepublishesRequestInfoEventsAsync(ExecutionEnvironment environment)
{
// Arrange
RequestPort<string, string> requestPort = RequestPort.Create<string, string>("TestPort");
ForwardMessageExecutor<string> processor = new("Processor");
Workflow workflow = new WorkflowBuilder(requestPort)
.AddEdge(requestPort, processor)
.Build();
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
// Act 1: Run workflow, collect pending requests and a checkpoint.
List<ExternalRequest> originalRequests = [];
CheckpointInfo? checkpoint = null;
await using (StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
.RunStreamingAsync(workflow, "Hello"))
{
await foreach (WorkflowEvent evt in firstRun.WatchStreamAsync(blockOnPendingRequest: false))
{
if (evt is RequestInfoEvent requestInfo)
{
originalRequests.Add(requestInfo.Request);
}
if (evt is SuperStepCompletedEvent step && step.CompletionInfo?.Checkpoint is { } cp)
{
checkpoint = cp;
}
}
originalRequests.Should().NotBeEmpty("the workflow should have created at least one external request");
checkpoint.Should().NotBeNull("a checkpoint should have been created");
}
// Act 2: Resume from the checkpoint.
await using StreamingRun resumed = await env.WithCheckpointing(checkpointManager)
.ResumeStreamingAsync(workflow, checkpoint!);
// Assert: The pending requests should be re-emitted.
List<ExternalRequest> reEmittedRequests = [];
using CancellationTokenSource cts = new(TimeSpan.FromSeconds(10));
await foreach (WorkflowEvent evt in resumed.WatchStreamAsync(blockOnPendingRequest: false, cts.Token))
{
if (evt is RequestInfoEvent requestInfo)
{
reEmittedRequests.Add(requestInfo.Request);
}
}
reEmittedRequests.Should().HaveCount(originalRequests.Count,
"all pending requests from the checkpoint should be re-emitted after resume");
reEmittedRequests.Select(r => r.RequestId)
.Should().BeEquivalentTo(originalRequests.Select(r => r.RequestId),
"the re-emitted request IDs should match the original pending request IDs");
}
/// <summary>
/// Verifies that <see cref="RunStatus"/> transitions to <see cref="RunStatus.PendingRequests"/>
/// after resuming from a checkpoint with pending external requests (not stuck at NotStarted).
/// </summary>
[Theory]
[InlineData(ExecutionEnvironment.InProcess_OffThread)]
[InlineData(ExecutionEnvironment.InProcess_Lockstep)]
internal async Task Checkpoint_Resume_WithPendingRequests_RunStatusIsPendingRequestsAsync(ExecutionEnvironment environment)
{
// Arrange
RequestPort<string, string> requestPort = RequestPort.Create<string, string>("TestPort");
ForwardMessageExecutor<string> processor = new("Processor");
Workflow workflow = new WorkflowBuilder(requestPort)
.AddEdge(requestPort, processor)
.Build();
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
// First run: collect a checkpoint with pending requests.
CheckpointInfo? checkpoint = null;
await using (StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
.RunStreamingAsync(workflow, "Hello"))
{
await foreach (WorkflowEvent evt in firstRun.WatchStreamAsync(blockOnPendingRequest: false))
{
if (evt is SuperStepCompletedEvent step && step.CompletionInfo?.Checkpoint is { } cp)
{
checkpoint = cp;
}
}
checkpoint.Should().NotBeNull();
}
// Act: Resume from the checkpoint and consume events so the run loop processes.
await using StreamingRun resumed = await env.WithCheckpointing(checkpointManager)
.ResumeStreamingAsync(workflow, checkpoint!);
using CancellationTokenSource cts = new(TimeSpan.FromSeconds(10));
await foreach (WorkflowEvent _ in resumed.WatchStreamAsync(blockOnPendingRequest: false, cts.Token))
{
// Consume all events until the stream completes.
}
// Assert
RunStatus status = await resumed.GetStatusAsync();
status.Should().Be(RunStatus.PendingRequests,
"the resumed workflow should report PendingRequests after rehydration");
}
/// <summary>
/// Verifies the full roundtrip: resume from checkpoint, observe the re-emitted request,
/// send a response, and verify the workflow completes without duplicating the request.
/// </summary>
[Theory]
[InlineData(ExecutionEnvironment.InProcess_OffThread)]
[InlineData(ExecutionEnvironment.InProcess_Lockstep)]
internal async Task Checkpoint_Resume_RespondToPendingRequest_CompletesWithoutDuplicateAsync(ExecutionEnvironment environment)
{
// Arrange
RequestPort<string, string> requestPort = RequestPort.Create<string, string>("TestPort");
ForwardMessageExecutor<string> processor = new("Processor");
Workflow workflow = new WorkflowBuilder(requestPort)
.AddEdge(requestPort, processor)
.Build();
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
// First run: collect checkpoint + pending request.
ExternalRequest? pendingRequest = null;
CheckpointInfo? checkpoint = null;
await using (StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
.RunStreamingAsync(workflow, "Hello"))
{
await foreach (WorkflowEvent evt in firstRun.WatchStreamAsync(blockOnPendingRequest: false))
{
if (evt is RequestInfoEvent requestInfo)
{
pendingRequest = requestInfo.Request;
}
if (evt is SuperStepCompletedEvent step && step.CompletionInfo?.Checkpoint is { } cp)
{
checkpoint = cp;
}
}
pendingRequest.Should().NotBeNull();
checkpoint.Should().NotBeNull();
}
// Act: Resume and respond to the restored request.
await using StreamingRun resumed = await env.WithCheckpointing(checkpointManager)
.ResumeStreamingAsync(workflow, checkpoint!);
int requestEventCount = 0;
using CancellationTokenSource cts = new(TimeSpan.FromSeconds(10));
// Use blockOnPendingRequest: false for the first pass to see the re-emitted requests.
await foreach (WorkflowEvent evt in resumed.WatchStreamAsync(blockOnPendingRequest: false, cts.Token))
{
if (evt is RequestInfoEvent requestInfo)
{
requestEventCount++;
requestInfo.Request.RequestId.Should().Be(pendingRequest!.RequestId,
"the re-emitted request should match the original");
}
}
requestEventCount.Should().Be(1,
"the pending request should be emitted exactly once (no duplicates)");
// Assert intermediate state before responding: the run should be in PendingRequests
// and we should have observed the re-emitted request. If the first WatchStreamAsync
// didn't complete or yielded nothing, these assertions catch it with a clear message.
RunStatus statusBeforeResponse = await resumed.GetStatusAsync();
statusBeforeResponse.Should().Be(RunStatus.PendingRequests,
"the run should be in PendingRequests state before we send a response");
// Now send the response and verify the workflow processes it.
ExternalResponse response = pendingRequest!.CreateResponse("World");
await resumed.SendResponseAsync(response);
// Consume the resulting events to verify the workflow progresses without errors.
List<WorkflowEvent> postResponseEvents = [];
using CancellationTokenSource cts2 = new(TimeSpan.FromSeconds(10));
await foreach (WorkflowEvent evt in resumed.WatchStreamAsync(blockOnPendingRequest: false, cts2.Token))
{
postResponseEvents.Add(evt);
}
postResponseEvents.Should().NotBeEmpty(
"the workflow should process the response and produce events");
postResponseEvents.OfType<WorkflowErrorEvent>().Should().BeEmpty(
"no errors should occur when processing the restored request's response");
}
/// <summary>
/// Verifies that restoring a live run to a checkpoint re-emits pending requests and allows
/// the workflow to continue from that restored point.
/// </summary>
[Theory]
[InlineData(ExecutionEnvironment.InProcess_OffThread)]
[InlineData(ExecutionEnvironment.InProcess_Lockstep)]
internal async Task Checkpoint_Restore_WithPendingRequests_RepublishesRequestInfoEventsAsync(ExecutionEnvironment environment)
{
// Arrange
Workflow workflow = CreateSimpleRequestWorkflow();
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
await using StreamingRun run = await env.WithCheckpointing(checkpointManager)
.RunStreamingAsync(workflow, "Hello");
(ExternalRequest pendingRequest, CheckpointInfo checkpoint) = await CapturePendingRequestAndCheckpointAsync(run);
// Advance the run past the checkpoint so the restore has meaningful work to undo.
await run.SendResponseAsync(pendingRequest.CreateResponse("World"));
List<WorkflowEvent> firstCompletionEvents = await ReadToHaltAsync(run);
firstCompletionEvents.OfType<WorkflowErrorEvent>().Should().BeEmpty(
"the workflow should continue cleanly before we restore");
RunStatus statusAfterFirstResponse = await run.GetStatusAsync();
statusAfterFirstResponse.Should().Be(RunStatus.Idle,
"the workflow should finish processing the first response before we restore");
// Act
await run.RestoreCheckpointAsync(checkpoint);
// Assert
List<WorkflowEvent> restoredEvents = await ReadToHaltAsync(run);
ExternalRequest[] replayedRequests = [.. restoredEvents.OfType<RequestInfoEvent>().Select(evt => evt.Request)];
replayedRequests.Should().ContainSingle("runtime restore should re-emit the restored pending request");
replayedRequests[0].RequestId.Should().Be(pendingRequest.RequestId,
"the replayed request should match the request captured at the checkpoint");
await run.SendResponseAsync(replayedRequests[0].CreateResponse("Again"));
List<WorkflowEvent> secondCompletionEvents = await ReadToHaltAsync(run);
secondCompletionEvents.OfType<WorkflowErrorEvent>().Should().BeEmpty(
"runtime restore replay should not introduce workflow errors");
RunStatus statusAfterRestoreResponse = await run.GetStatusAsync();
statusAfterRestoreResponse.Should().Be(RunStatus.Idle,
"the workflow should be able to continue after the runtime restore replay");
}
/// <summary>
/// Verifies that a resumed parent workflow re-emits pending requests that originated in a subworkflow.
/// </summary>
[Theory]
[InlineData(ExecutionEnvironment.InProcess_OffThread)]
[InlineData(ExecutionEnvironment.InProcess_Lockstep)]
internal async Task Checkpoint_Resume_SubworkflowWithPendingRequests_RepublishesQualifiedRequestInfoEventsAsync(ExecutionEnvironment environment)
{
// Arrange
Workflow workflow = CreateCheckpointedSubworkflowRequestWorkflow();
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
ExternalRequest pendingRequest;
CheckpointInfo checkpoint;
await using (StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
.RunStreamingAsync(workflow, "Hello"))
{
(pendingRequest, checkpoint) = await CapturePendingRequestAndCheckpointAsync(firstRun);
}
// Act
await using StreamingRun resumed = await env.WithCheckpointing(checkpointManager)
.ResumeStreamingAsync(workflow, checkpoint);
// Assert
List<WorkflowEvent> resumedEvents = await ReadToHaltAsync(resumed);
ExternalRequest[] replayedRequests = [.. resumedEvents.OfType<RequestInfoEvent>().Select(evt => evt.Request)];
replayedRequests.Should().ContainSingle("the resumed parent workflow should surface the subworkflow request once");
replayedRequests[0].RequestId.Should().Be(pendingRequest.RequestId,
"the replayed subworkflow request should match the checkpointed request");
replayedRequests[0].PortInfo.PortId.Should().Be(pendingRequest.PortInfo.PortId,
"the replayed request should remain qualified through the subworkflow boundary");
await resumed.SendResponseAsync(replayedRequests[0].CreateResponse("World"));
List<WorkflowEvent> completionEvents = await ReadToHaltAsync(resumed);
completionEvents.OfType<RequestInfoEvent>().Should().BeEmpty(
"the resumed subworkflow request should not be replayed twice");
completionEvents.OfType<WorkflowErrorEvent>().Should().BeEmpty(
"subworkflow replay should not introduce workflow errors");
RunStatus statusAfterSubworkflowResponse = await resumed.GetStatusAsync();
statusAfterSubworkflowResponse.Should().Be(RunStatus.Idle,
"the resumed subworkflow should continue after responding to the replayed request");
}
/// <summary>
/// Verifies that when <c>republishPendingEvents</c> is <see langword="false"/>,
/// no <see cref="RequestInfoEvent"/> is re-emitted after resuming from a checkpoint.
/// </summary>
[Theory]
[InlineData(ExecutionEnvironment.InProcess_OffThread)]
[InlineData(ExecutionEnvironment.InProcess_Lockstep)]
internal async Task Checkpoint_Resume_WithRepublishDisabled_DoesNotEmitRequestInfoEventsAsync(ExecutionEnvironment environment)
{
// Arrange
RequestPort<string, string> requestPort = RequestPort.Create<string, string>("TestPort");
ForwardMessageExecutor<string> processor = new("Processor");
Workflow workflow = new WorkflowBuilder(requestPort)
.AddEdge(requestPort, processor)
.Build();
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment();
// First run: collect a checkpoint with pending requests.
CheckpointInfo? checkpoint = null;
await using (StreamingRun firstRun = await env.WithCheckpointing(checkpointManager)
.RunStreamingAsync(workflow, "Hello"))
{
await foreach (WorkflowEvent evt in firstRun.WatchStreamAsync(blockOnPendingRequest: false))
{
if (evt is SuperStepCompletedEvent step && step.CompletionInfo?.Checkpoint is { } cp)
{
checkpoint = cp;
}
}
checkpoint.Should().NotBeNull();
}
// Act: Resume with republishPendingEvents: false via the internal API.
await using StreamingRun resumed = await env.WithCheckpointing(checkpointManager)
.ResumeStreamingInternalAsync(workflow, checkpoint!, republishPendingEvents: false);
// Assert: No RequestInfoEvent should appear in the event stream.
int requestEventCount = 0;
using CancellationTokenSource cts = new(TimeSpan.FromSeconds(10));
await foreach (WorkflowEvent evt in resumed.WatchStreamAsync(blockOnPendingRequest: false, cts.Token))
{
if (evt is RequestInfoEvent)
{
requestEventCount++;
}
}
requestEventCount.Should().Be(0,
"no RequestInfoEvent should be emitted when republishPendingEvents is false");
}
private static Workflow CreateSimpleRequestWorkflow(
string requestPortId = "TestPort",
string processorId = "Processor")
{
RequestPort<string, string> requestPort = RequestPort.Create<string, string>(requestPortId);
ForwardMessageExecutor<string> processor = new(processorId);
return new WorkflowBuilder(requestPort)
.AddEdge(requestPort, processor)
.Build();
}
private static Workflow CreateCheckpointedSubworkflowRequestWorkflow()
{
ExecutorBinding subworkflow = CreateSimpleRequestWorkflow(
requestPortId: "InnerTestPort",
processorId: "InnerProcessor")
.BindAsExecutor("Subworkflow");
return new WorkflowBuilder(subworkflow)
.AddExternalRequest<string, string>(subworkflow, id: "ForwardedSubworkflowRequest")
.Build();
}
private static async ValueTask<(ExternalRequest PendingRequest, CheckpointInfo Checkpoint)> CapturePendingRequestAndCheckpointAsync(StreamingRun run)
{
ExternalRequest? pendingRequest = null;
CheckpointInfo? checkpoint = null;
await foreach (WorkflowEvent evt in run.WatchStreamAsync(blockOnPendingRequest: false))
{
if (evt is RequestInfoEvent requestInfo)
{
pendingRequest ??= requestInfo.Request;
}
if (evt is SuperStepCompletedEvent step && step.CompletionInfo?.Checkpoint is { } cp)
{
checkpoint = cp;
}
}
pendingRequest.Should().NotBeNull("the workflow should have emitted a pending request");
checkpoint.Should().NotBeNull("the workflow should have produced a checkpoint");
return (pendingRequest!, checkpoint!);
}
private static async ValueTask<List<WorkflowEvent>> ReadToHaltAsync(StreamingRun run)
{
List<WorkflowEvent> events = [];
using CancellationTokenSource cts = new(TimeSpan.FromSeconds(10));
await foreach (WorkflowEvent evt in run.WatchStreamAsync(blockOnPendingRequest: false, cts.Token))
{
events.Add(evt);
}
return events;
}
}
@@ -132,53 +132,6 @@ public class InProcessExecutionTests
"both versions should produce the same number of agent events");
}
/// <summary>
/// This test checks that the logic around waiting for input and halting appropriately works right when the
/// workflow runs to halting before the EventStream is watched by the user.
/// </summary>
[Fact]
public async Task RunStreamingAsyncWaitToTakeStreamAsync()
{
// Arrange: Create a simple agent that responds to messages
var agent = new SimpleTestAgent("test-agent");
var workflow = AgentWorkflowBuilder.BuildSequential(agent);
var inputMessage = new ChatMessage(ChatRole.User, "Hello");
// Act: Execute using streaming version with TurnToken
await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, new List<ChatMessage> { inputMessage });
// Send TurnToken to actually trigger execution (this is the key step)
bool messageSent = await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
messageSent.Should().BeTrue("TurnToken should be accepted");
while (await run.GetStatusAsync() != RunStatus.Idle)
{
await Task.Delay(200);
}
// Collect events
List<WorkflowEvent> events = [];
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
events.Add(evt);
}
// Assert: The workflow should have executed and produced events
RunStatus status = await run.GetStatusAsync();
status.Should().Be(RunStatus.Idle, "workflow should complete execution");
events.Should().NotBeEmpty("workflow should produce events during execution");
// Check that we have agent execution events
var agentEvents = events.OfType<AgentResponseUpdateEvent>().ToList();
agentEvents.Should().NotBeEmpty("agent should have executed and produced update events");
// Check that we have output events
var outputEvents = events.OfType<WorkflowOutputEvent>().ToList();
outputEvents.Should().NotBeEmpty("workflow should produce output events");
}
/// <summary>
/// Simple test agent that echoes back the input message.
/// </summary>
@@ -35,9 +35,9 @@ from agent_framework import (
AgentResponseUpdate,
AgentSession,
BaseAgent,
BaseHistoryProvider,
Content,
ContinuationToken,
HistoryProvider,
Message,
ResponseStream,
SessionContext,
@@ -353,7 +353,7 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
# Run before_run providers (forward order)
for provider in self.context_providers:
if isinstance(provider, HistoryProvider) and not provider.load_messages:
if isinstance(provider, BaseHistoryProvider) and not provider.load_messages:
continue
if session is None:
raise RuntimeError("Provider session must be available when context providers are configured.")
+2 -2
View File
@@ -24,8 +24,8 @@ from agent_framework import (
AgentResponse,
AgentResponseUpdate,
AgentSession,
BaseContextProvider,
Content,
ContextProvider,
Message,
SessionContext,
)
@@ -869,7 +869,7 @@ async def test_poll_task_completed(a2a_agent: A2AAgent, mock_a2a_client: MockA2A
# region Context Provider Tests
class TrackingContextProvider(ContextProvider):
class TrackingContextProvider(BaseContextProvider):
"""A context provider that records when before_run and after_run are called."""
def __init__(self) -> None:
@@ -1,9 +1,9 @@
# Copyright (c) Microsoft. All rights reserved.
"""New-pattern Azure AI Search context provider using ContextProvider.
"""New-pattern Azure AI Search context provider using BaseContextProvider.
This module provides ``AzureAISearchContextProvider``, built on the new
:class:`ContextProvider` hooks pattern.
:class:`BaseContextProvider` hooks pattern.
"""
from __future__ import annotations
@@ -11,21 +11,11 @@ from __future__ import annotations
import logging
import sys
from collections.abc import Awaitable, Callable
from typing import TYPE_CHECKING, Any, ClassVar, Literal, TypedDict, overload
from typing import TYPE_CHECKING, Any, ClassVar, Literal, TypedDict
from agent_framework import (
AGENT_FRAMEWORK_USER_AGENT,
AgentSession,
Annotation,
Content,
ContextProvider,
Message,
SecretString,
SessionContext,
SupportsGetEmbeddings,
load_settings,
)
from agent_framework.exceptions import SettingNotFoundError
from agent_framework import AGENT_FRAMEWORK_USER_AGENT, Annotation, Content, Message, SupportsGetEmbeddings
from agent_framework._sessions import AgentSession, BaseContextProvider, SessionContext
from agent_framework._settings import SecretString, load_settings
from azure.core.credentials import AzureKeyCredential, TokenCredential
from azure.core.credentials_async import AsyncTokenCredential
from azure.core.exceptions import ResourceNotFoundError
@@ -121,9 +111,6 @@ except ImportError:
_agentic_retrieval_available = False
AzureCredentialTypes = TokenCredential | AsyncTokenCredential
EmbeddingFunction = Callable[[str], Awaitable[list[float]]] | SupportsGetEmbeddings[str, list[float], Any]
KnowledgeBaseOutputModeLiteral = Literal["extractive_data", "answer_synthesis"]
RetrievalReasoningEffortLiteral = Literal["minimal", "medium", "low"]
logger = logging.getLogger("agent_framework.azure_ai_search")
@@ -154,8 +141,8 @@ class AzureAISearchSettings(TypedDict, total=False):
api_key: SecretString | None
class AzureAISearchContextProvider(ContextProvider):
"""Azure AI Search context provider using the new ContextProvider hooks pattern.
class AzureAISearchContextProvider(BaseContextProvider):
"""Azure AI Search context provider using the new BaseContextProvider hooks pattern.
Retrieves relevant context from Azure AI Search using semantic or agentic search
modes.
@@ -164,230 +151,6 @@ class AzureAISearchContextProvider(ContextProvider):
_DEFAULT_SEARCH_CONTEXT_PROMPT: ClassVar[str] = "Use the following context to answer the question:"
DEFAULT_SOURCE_ID: ClassVar[str] = "azure_ai_search"
@overload
def __init__(
self,
source_id: str = DEFAULT_SOURCE_ID,
endpoint: str | None = None,
index_name: str | None = None,
api_key: str | AzureKeyCredential | None = None,
credential: AzureCredentialTypes | None = None,
*,
mode: Literal["semantic"] = "semantic",
top_k: int = 5,
semantic_configuration_name: str | None = None,
vector_field_name: str | None = None,
embedding_function: EmbeddingFunction | None = None,
context_prompt: str | None = None,
azure_openai_resource_url: str | None = None,
model_deployment_name: str | None = None,
model_name: str | None = None,
knowledge_base_name: None = None,
retrieval_instructions: str | None = None,
azure_openai_api_key: str | None = None,
knowledge_base_output_mode: KnowledgeBaseOutputModeLiteral = "extractive_data",
retrieval_reasoning_effort: RetrievalReasoningEffortLiteral = "minimal",
agentic_message_history_count: int = _DEFAULT_AGENTIC_MESSAGE_HISTORY_COUNT,
env_file_path: str | None = None,
env_file_encoding: str | None = None,
) -> None:
"""Initialize a semantic Azure AI Search context provider.
Keyword Args:
source_id: Unique identifier for this provider instance.
endpoint: Azure AI Search endpoint URL.
index_name: Name of the search index to query.
api_key: API key for authentication.
credential: Azure credential for managed identity authentication.
mode: Must be ``"semantic"`` for this overload.
top_k: Maximum number of documents to retrieve.
semantic_configuration_name: Name of the semantic configuration in the index.
vector_field_name: Name of the vector field in the index.
embedding_function: Embedding provider used for vector search.
context_prompt: Custom prompt to prepend to retrieved context.
azure_openai_resource_url: Unused in semantic mode.
model_deployment_name: Unused in semantic mode.
model_name: Unused in semantic mode.
knowledge_base_name: Must be ``None`` for this overload.
retrieval_instructions: Unused in semantic mode.
azure_openai_api_key: Unused in semantic mode.
knowledge_base_output_mode: Unused in semantic mode.
retrieval_reasoning_effort: Unused in semantic mode.
agentic_message_history_count: Unused in semantic mode.
env_file_path: Optional ``.env`` file checked before process environment variables.
env_file_encoding: Encoding for the ``.env`` file.
"""
...
@overload
def __init__(
self,
source_id: str = DEFAULT_SOURCE_ID,
endpoint: str | None = None,
index_name: str | None = None,
api_key: str | AzureKeyCredential | None = None,
credential: AzureCredentialTypes | None = None,
*,
mode: Literal["agentic"],
top_k: int = 5,
semantic_configuration_name: str | None = None,
vector_field_name: str | None = None,
embedding_function: EmbeddingFunction | None = None,
context_prompt: str | None = None,
azure_openai_resource_url: str,
model_deployment_name: str,
model_name: str | None = None,
knowledge_base_name: None = None,
retrieval_instructions: str | None = None,
azure_openai_api_key: str | None = None,
knowledge_base_output_mode: KnowledgeBaseOutputModeLiteral = "extractive_data",
retrieval_reasoning_effort: RetrievalReasoningEffortLiteral = "minimal",
agentic_message_history_count: int = _DEFAULT_AGENTIC_MESSAGE_HISTORY_COUNT,
env_file_path: str | None = None,
env_file_encoding: str | None = None,
) -> None:
"""Initialize an agentic provider that creates a Knowledge Base from an index.
Keyword Args:
source_id: Unique identifier for this provider instance.
endpoint: Azure AI Search endpoint URL.
index_name: Name of the search index used to create the Knowledge Base.
api_key: API key for authentication.
credential: Azure credential for managed identity authentication.
mode: Must be ``"agentic"`` for this overload.
top_k: Maximum number of documents to retrieve.
semantic_configuration_name: Semantic configuration name used by hybrid search operations.
vector_field_name: Vector field name used by hybrid search operations.
embedding_function: Embedding provider used for vector search.
context_prompt: Custom prompt to prepend to retrieved context.
azure_openai_resource_url: Azure OpenAI resource URL for Knowledge Base creation.
model_deployment_name: Azure OpenAI deployment used by the generated Knowledge Base.
model_name: Underlying model name for the Knowledge Base model configuration.
knowledge_base_name: Must be ``None`` for this overload.
retrieval_instructions: Custom instructions for Knowledge Base retrieval.
azure_openai_api_key: Optional Azure OpenAI API key for Knowledge Base creation.
knowledge_base_output_mode: Output mode for Knowledge Base retrieval.
retrieval_reasoning_effort: Reasoning effort for query planning.
agentic_message_history_count: Number of recent messages included in retrieval.
env_file_path: Optional ``.env`` file checked before process environment variables.
env_file_encoding: Encoding for the ``.env`` file.
"""
...
@overload
def __init__(
self,
source_id: str = DEFAULT_SOURCE_ID,
endpoint: str | None = None,
index_name: None = None,
api_key: str | AzureKeyCredential | None = None,
credential: AzureCredentialTypes | None = None,
*,
mode: Literal["agentic"],
top_k: int = 5,
semantic_configuration_name: str | None = None,
vector_field_name: str | None = None,
embedding_function: EmbeddingFunction | None = None,
context_prompt: str | None = None,
azure_openai_resource_url: str | None = None,
model_deployment_name: str | None = None,
model_name: str | None = None,
knowledge_base_name: str,
retrieval_instructions: str | None = None,
azure_openai_api_key: str | None = None,
knowledge_base_output_mode: KnowledgeBaseOutputModeLiteral = "extractive_data",
retrieval_reasoning_effort: RetrievalReasoningEffortLiteral = "minimal",
agentic_message_history_count: int = _DEFAULT_AGENTIC_MESSAGE_HISTORY_COUNT,
env_file_path: str | None = None,
env_file_encoding: str | None = None,
) -> None:
"""Initialize an agentic provider that connects to an existing Knowledge Base.
Keyword Args:
source_id: Unique identifier for this provider instance.
endpoint: Azure AI Search endpoint URL.
index_name: Must be ``None`` for this overload.
knowledge_base_name: Name of the existing Knowledge Base to use.
api_key: API key for authentication.
credential: Azure credential for managed identity authentication.
mode: Must be ``"agentic"`` for this overload.
top_k: Maximum number of documents to retrieve.
semantic_configuration_name: Semantic configuration name used by hybrid search operations.
vector_field_name: Vector field name used by hybrid search operations.
embedding_function: Embedding provider used for vector search.
context_prompt: Custom prompt to prepend to retrieved context.
azure_openai_resource_url: Unused when connecting to an existing Knowledge Base.
model_deployment_name: Unused when connecting to an existing Knowledge Base.
model_name: Unused when connecting to an existing Knowledge Base.
retrieval_instructions: Custom instructions for Knowledge Base retrieval.
azure_openai_api_key: Unused when connecting to an existing Knowledge Base.
knowledge_base_output_mode: Output mode for Knowledge Base retrieval.
retrieval_reasoning_effort: Reasoning effort for query planning.
agentic_message_history_count: Number of recent messages included in retrieval.
env_file_path: Optional ``.env`` file checked before process environment variables.
env_file_encoding: Encoding for the ``.env`` file.
"""
...
@overload
def __init__(
self,
source_id: str = DEFAULT_SOURCE_ID,
endpoint: str | None = None,
index_name: None = None,
api_key: str | AzureKeyCredential | None = None,
credential: AzureCredentialTypes | None = None,
*,
mode: Literal["agentic"],
top_k: int = 5,
semantic_configuration_name: str | None = None,
vector_field_name: str | None = None,
embedding_function: EmbeddingFunction | None = None,
context_prompt: str | None = None,
azure_openai_resource_url: str | None = None,
model_deployment_name: str | None = None,
model_name: str | None = None,
knowledge_base_name: None = None,
retrieval_instructions: str | None = None,
azure_openai_api_key: str | None = None,
knowledge_base_output_mode: KnowledgeBaseOutputModeLiteral = "extractive_data",
retrieval_reasoning_effort: RetrievalReasoningEffortLiteral = "minimal",
agentic_message_history_count: int = _DEFAULT_AGENTIC_MESSAGE_HISTORY_COUNT,
env_file_path: str | None = None,
env_file_encoding: str | None = None,
) -> None:
"""Initialize an agentic provider using environment-resolved setup.
This overload is for agentic initialization where ``index_name`` or
``knowledge_base_name`` is supplied by ``env_file_path`` or the
``AZURE_SEARCH_*`` environment variables.
Keyword Args:
source_id: Unique identifier for this provider instance.
endpoint: Azure AI Search endpoint URL.
index_name: Resolved from ``env_file_path`` or ``AZURE_SEARCH_INDEX_NAME``.
api_key: API key for authentication.
credential: Azure credential for managed identity authentication.
mode: Must be ``"agentic"`` for this overload.
top_k: Maximum number of documents to retrieve.
semantic_configuration_name: Semantic configuration name used by hybrid search operations.
vector_field_name: Vector field name used by hybrid search operations.
embedding_function: Embedding provider used for vector search.
context_prompt: Custom prompt to prepend to retrieved context.
azure_openai_resource_url: Azure OpenAI resource URL when creating a Knowledge Base from an index.
model_deployment_name: Azure OpenAI deployment when creating a Knowledge Base from an index.
model_name: Underlying model name for Knowledge Base model configuration.
knowledge_base_name: Resolved from ``env_file_path`` or ``AZURE_SEARCH_KNOWLEDGE_BASE_NAME``.
retrieval_instructions: Custom instructions for Knowledge Base retrieval.
azure_openai_api_key: Optional Azure OpenAI API key for Knowledge Base creation.
knowledge_base_output_mode: Output mode for Knowledge Base retrieval.
retrieval_reasoning_effort: Reasoning effort for query planning.
agentic_message_history_count: Number of recent messages included in retrieval.
env_file_path: Optional ``.env`` file checked before process environment variables.
env_file_encoding: Encoding for the ``.env`` file.
"""
...
def __init__(
self,
source_id: str = DEFAULT_SOURCE_ID,
@@ -400,7 +163,9 @@ class AzureAISearchContextProvider(ContextProvider):
top_k: int = 5,
semantic_configuration_name: str | None = None,
vector_field_name: str | None = None,
embedding_function: EmbeddingFunction | None = None,
embedding_function: Callable[[str], Awaitable[list[float]]]
| SupportsGetEmbeddings[str, list[float], Any]
| None = None,
context_prompt: str | None = None,
azure_openai_resource_url: str | None = None,
model_deployment_name: str | None = None,
@@ -408,8 +173,8 @@ class AzureAISearchContextProvider(ContextProvider):
knowledge_base_name: str | None = None,
retrieval_instructions: str | None = None,
azure_openai_api_key: str | None = None,
knowledge_base_output_mode: KnowledgeBaseOutputModeLiteral = "extractive_data",
retrieval_reasoning_effort: RetrievalReasoningEffortLiteral = "minimal",
knowledge_base_output_mode: Literal["extractive_data", "answer_synthesis"] = "extractive_data",
retrieval_reasoning_effort: Literal["minimal", "medium", "low"] = "minimal",
agentic_message_history_count: int = _DEFAULT_AGENTIC_MESSAGE_HISTORY_COUNT,
env_file_path: str | None = None,
env_file_encoding: str | None = None,
@@ -419,9 +184,7 @@ class AzureAISearchContextProvider(ContextProvider):
Args:
source_id: Unique identifier for this provider instance.
endpoint: Azure AI Search endpoint URL.
index_name: Name of the search index to query. In agentic mode, providing this
explicitly selects the index-backed setup and ignores any environment-provided
knowledge base name.
index_name: Name of the search index to query.
api_key: API key for authentication.
credential: Azure credential for managed identity authentication.
Accepts a TokenCredential, AsyncTokenCredential, or a callable token provider.
@@ -434,9 +197,7 @@ class AzureAISearchContextProvider(ContextProvider):
azure_openai_resource_url: Azure OpenAI resource URL for Knowledge Base.
model_deployment_name: Model deployment name in Azure OpenAI.
model_name: The underlying model name.
knowledge_base_name: Name of an existing Knowledge Base to use. In agentic mode,
providing this explicitly selects the Knowledge Base-backed setup and ignores any
environment-provided index name.
knowledge_base_name: Name of an existing Knowledge Base to use.
retrieval_instructions: Custom instructions for Knowledge Base retrieval.
azure_openai_api_key: Azure OpenAI API key.
knowledge_base_output_mode: Output mode for Knowledge Base retrieval.
@@ -447,26 +208,12 @@ class AzureAISearchContextProvider(ContextProvider):
"""
super().__init__(source_id)
required: list[str | tuple[str, ...]]
ignored_agentic_field: Literal["index_name", "knowledge_base_name"] | None = None
explicit_index_name = index_name is not None
explicit_knowledge_base_name = knowledge_base_name is not None
# Determine which fields are required based on mode
required: list[str | tuple[str, ...]] = ["endpoint"]
if mode == "semantic":
required = ["endpoint", "index_name"]
elif explicit_index_name and explicit_knowledge_base_name:
raise SettingNotFoundError(
"Only one of 'index_name', 'knowledge_base_name' may be provided, "
"but multiple were set: 'index_name', 'knowledge_base_name'."
)
elif explicit_index_name:
required = ["endpoint", "index_name"]
ignored_agentic_field = "knowledge_base_name"
elif explicit_knowledge_base_name:
required = ["endpoint", "knowledge_base_name"]
ignored_agentic_field = "index_name"
else:
required = ["endpoint", ("index_name", "knowledge_base_name")]
required.append("index_name")
elif mode == "agentic":
required.append(("index_name", "knowledge_base_name"))
# Load settings from environment/file
settings = load_settings(
@@ -480,8 +227,6 @@ class AzureAISearchContextProvider(ContextProvider):
env_file_path=env_file_path,
env_file_encoding=env_file_encoding,
)
if ignored_agentic_field is not None:
settings[ignored_agentic_field] = None
if mode == "agentic" and settings.get("index_name") and not model_deployment_name:
raise ValueError(
@@ -283,37 +283,6 @@ class TestInitAgenticValidation:
assert provider._use_existing_knowledge_base is False
assert provider.knowledge_base_name == "idx-kb"
def test_agentic_explicit_kb_ignores_env_index_name(self) -> None:
with patch.dict(os.environ, {"AZURE_SEARCH_INDEX_NAME": "env-index"}, clear=False):
provider = AzureAISearchContextProvider(
source_id="s",
endpoint="https://test.search.windows.net",
knowledge_base_name="my-kb",
api_key="key",
mode="agentic",
)
assert provider.index_name is None
assert provider.knowledge_base_name == "my-kb"
assert provider._use_existing_knowledge_base is True
assert provider._search_client is None
def test_agentic_explicit_index_ignores_env_kb_name(self) -> None:
with patch.dict(os.environ, {"AZURE_SEARCH_KNOWLEDGE_BASE_NAME": "env-kb"}, clear=False):
provider = AzureAISearchContextProvider(
source_id="s",
endpoint="https://test.search.windows.net",
index_name="idx",
api_key="key",
mode="agentic",
model_deployment_name="deploy",
azure_openai_resource_url="https://aoai.openai.azure.com",
)
assert provider.index_name == "idx"
assert provider.knowledge_base_name == "idx-kb"
assert provider._use_existing_knowledge_base is False
# -- __aenter__ / __aexit__ ---------------------------------------------------
@@ -11,7 +11,7 @@ from collections.abc import Sequence
from typing import Any, ClassVar, TypedDict
from agent_framework import AGENT_FRAMEWORK_USER_AGENT, Message
from agent_framework._sessions import HistoryProvider
from agent_framework._sessions import BaseHistoryProvider
from agent_framework._settings import SecretString, load_settings
from azure.core.credentials import TokenCredential
from azure.core.credentials_async import AsyncTokenCredential
@@ -32,8 +32,8 @@ class AzureCosmosHistorySettings(TypedDict, total=False):
key: SecretString | None
class CosmosHistoryProvider(HistoryProvider):
"""Azure Cosmos DB-backed history provider using HistoryProvider hooks."""
class CosmosHistoryProvider(BaseHistoryProvider):
"""Azure Cosmos DB-backed history provider using BaseHistoryProvider hooks."""
DEFAULT_SOURCE_ID: ClassVar[str] = "azure_cosmos_history"
_BATCH_OPERATION_LIMIT: ClassVar[int] = 100
@@ -16,8 +16,8 @@ from agent_framework import (
AgentRunInputs,
AgentSession,
BaseAgent,
BaseContextProvider,
Content,
ContextProvider,
FunctionTool,
Message,
ResponseStream,
@@ -223,7 +223,7 @@ class RawClaudeAgent(BaseAgent, Generic[OptionsT]):
id: str | None = None,
name: str | None = None,
description: str | None = None,
context_providers: Sequence[ContextProvider] | None = None,
context_providers: Sequence[BaseContextProvider] | None = None,
middleware: Sequence[AgentMiddlewareTypes] | None = None,
tools: ToolTypes | Callable[..., Any] | str | Sequence[ToolTypes | Callable[..., Any] | str] | None = None,
default_options: OptionsT | MutableMapping[str, Any] | None = None,
@@ -11,8 +11,8 @@ from agent_framework import (
AgentResponseUpdate,
AgentSession,
BaseAgent,
BaseContextProvider,
Content,
ContextProvider,
Message,
ResponseStream,
normalize_messages,
@@ -60,7 +60,7 @@ class CopilotStudioAgent(BaseAgent):
id: str | None = None,
name: str | None = None,
description: str | None = None,
context_providers: Sequence[ContextProvider] | None = None,
context_providers: Sequence[BaseContextProvider] | None = None,
middleware: list[AgentMiddlewareTypes] | None = None,
environment_id: str | None = None,
agent_identifier: str | None = None,
+3 -3
View File
@@ -61,8 +61,8 @@ agent_framework/
- **`AgentSession`** - Manages conversation state and session metadata
- **`SessionContext`** - Context object for session-scoped data during agent runs
- **`ContextProvider`** - Base class for context providers (RAG, memory systems)
- **`HistoryProvider`** - Base class for conversation history storage
- **`BaseContextProvider`** - Base class for context providers (RAG, memory systems)
- **`BaseHistoryProvider`** - Base class for conversation history storage
### Skills (`_skills.py`)
@@ -70,7 +70,7 @@ agent_framework/
- **`SkillResource`** - Named supplementary content attached to a skill; holds either static `content` or a dynamic `function` (sync or async). Exactly one must be provided.
- **`SkillScript`** - An executable script attached to a skill; holds either an inline `function` (code-defined, runs in-process) or a `path` to a file on disk (file-based, delegated to a runner). Exactly one must be provided.
- **`SkillScriptRunner`** - Protocol for file-based script execution. Any callable matching `(skill, script, args) -> Any` satisfies it. Code-defined scripts do not use a runner.
- **`SkillsProvider`** - Context provider (extends `ContextProvider`) that discovers file-based skills from `SKILL.md` files and/or accepts code-defined `Skill` instances. Follows progressive disclosure: advertise → load → read resources / run scripts.
- **`SkillsProvider`** - Context provider (extends `BaseContextProvider`) that discovers file-based skills from `SKILL.md` files and/or accepts code-defined `Skill` instances. Follows progressive disclosure: advertise → load → read resources / run scripts.
### Workflows (`_workflows/`)
@@ -102,10 +102,8 @@ from ._middleware import (
)
from ._sessions import (
AgentSession,
BaseContextProvider, # type: ignore[reportDeprecated]
BaseHistoryProvider, # type: ignore[reportDeprecated]
ContextProvider,
HistoryProvider,
BaseContextProvider,
BaseHistoryProvider,
InMemoryHistoryProvider,
SessionContext,
register_state_type,
@@ -298,7 +296,6 @@ __all__ = [
"CompactionProvider",
"CompactionStrategy",
"Content",
"ContextProvider",
"ContinuationToken",
"ConversationSplit",
"ConversationSplitter",
@@ -334,7 +331,6 @@ __all__ = [
"FunctionTool",
"GeneratedEmbeddings",
"GraphConnectivityError",
"HistoryProvider",
"InMemoryCheckpointStorage",
"InMemoryHistoryProvider",
"InProcRunnerContext",
+137 -275
View File
@@ -29,16 +29,14 @@ from . import _tools as _tool_utils # pyright: ignore[reportPrivateUsage]
from ._clients import BaseChatClient, SupportsChatGetResponse
from ._docstrings import apply_layered_docstring
from ._mcp import LOG_LEVEL_MAPPING, MCPTool
from ._middleware import AgentMiddlewareLayer, FunctionInvocationContext, MiddlewareTypes, categorize_middleware
from ._middleware import AgentMiddlewareLayer, FunctionInvocationContext, MiddlewareTypes
from ._serialization import SerializationMixin
from ._sessions import (
AgentSession,
ContextProvider,
HistoryProvider,
BaseContextProvider,
BaseHistoryProvider,
InMemoryHistoryProvider,
PerServiceCallHistoryPersistingMiddleware,
SessionContext,
is_local_history_conversation_id,
)
from ._tools import FunctionInvocationLayer, FunctionTool, ToolTypes, normalize_tools
from ._types import (
@@ -52,7 +50,7 @@ from ._types import (
map_chat_to_agent_update,
normalize_messages,
)
from .exceptions import AgentInvalidRequestException, AgentInvalidResponseException, UserInputRequiredException
from .exceptions import AgentInvalidResponseException, UserInputRequiredException
from .observability import AgentTelemetryLayer
if sys.version_info >= (3, 13):
@@ -168,7 +166,6 @@ class _RunContext(TypedDict):
input_messages: Sequence[Message]
session_messages: Sequence[Message]
agent_name: str
suppress_response_id: bool
chat_options: MutableMapping[str, Any]
compaction_strategy: CompactionStrategy | None
tokenizer: TokenizerProtocol | None
@@ -369,7 +366,6 @@ class BaseAgent(SerializationMixin):
"""
DEFAULT_EXCLUDE: ClassVar[set[str]] = {"additional_properties"}
require_per_service_call_history_persistence: bool = False
def __init__(
self,
@@ -377,7 +373,7 @@ class BaseAgent(SerializationMixin):
id: str | None = None,
name: str | None = None,
description: str | None = None,
context_providers: Sequence[ContextProvider] | None = None,
context_providers: Sequence[BaseContextProvider] | None = None,
middleware: Sequence[MiddlewareTypes] | None = None,
additional_properties: MutableMapping[str, Any] | None = None,
) -> None:
@@ -397,7 +393,7 @@ class BaseAgent(SerializationMixin):
self.id = id
self.name = name
self.description = description
self.context_providers: list[ContextProvider] = list(context_providers or [])
self.context_providers: list[BaseContextProvider] = list(context_providers or [])
self.middleware: list[MiddlewareTypes] | None = (
cast(list[MiddlewareTypes], middleware) if middleware is not None else None
)
@@ -459,12 +455,7 @@ class BaseAgent(SerializationMixin):
if provider_session is None and self.context_providers:
provider_session = AgentSession()
per_service_call_history_required = self.require_per_service_call_history_persistence and any(
isinstance(provider, HistoryProvider) for provider in self.context_providers
)
for provider in reversed(self.context_providers):
if per_service_call_history_required and isinstance(provider, HistoryProvider):
continue
if provider_session is None:
raise RuntimeError("Provider session must be available when context providers are configured.")
await provider.after_run(
@@ -665,9 +656,8 @@ class RawAgent(BaseAgent, Generic[OptionsCoT]): # type: ignore[misc]
description: str | None = None,
tools: ToolTypes | Callable[..., Any] | Sequence[ToolTypes | Callable[..., Any]] | None = None,
default_options: OptionsCoT | None = None,
context_providers: Sequence[ContextProvider] | None = None,
context_providers: Sequence[BaseContextProvider] | None = None,
middleware: Sequence[MiddlewareTypes] | None = None,
require_per_service_call_history_persistence: bool = False,
compaction_strategy: CompactionStrategy | None = None,
tokenizer: TokenizerProtocol | None = None,
additional_properties: MutableMapping[str, Any] | None = None,
@@ -685,11 +675,6 @@ class RawAgent(BaseAgent, Generic[OptionsCoT]): # type: ignore[misc]
description: A brief description of the agent's purpose.
context_providers: Context providers to include during agent invocation.
middleware: List of middleware to intercept agent and function invocations.
require_per_service_call_history_persistence: When True, history providers are invoked
around each model call instead of once per ``run()`` when the service
is not already storing history. If service-side storage is active for
the run, the agent skips local history providers and relies on the
service-managed conversation instead.
default_options: A TypedDict containing chat options. When using a typed agent like
``Agent[OpenAIChatOptions]``, this enables IDE autocomplete for
provider-specific options including temperature, max_tokens, model_id,
@@ -721,7 +706,6 @@ class RawAgent(BaseAgent, Generic[OptionsCoT]): # type: ignore[misc]
)
self.client = client
self.compaction_strategy = compaction_strategy
self.require_per_service_call_history_persistence = require_per_service_call_history_persistence
self.tokenizer = tokenizer
# Get tools from options or named parameter (named param takes precedence)
@@ -780,35 +764,6 @@ class RawAgent(BaseAgent, Generic[OptionsCoT]): # type: ignore[misc]
await self._async_exit_stack.enter_async_context(context_manager)
return self
def _get_history_providers(self) -> list[HistoryProvider]:
return [provider for provider in self.context_providers if isinstance(provider, HistoryProvider)]
def _resolve_per_service_call_history_providers(
self,
*,
session: AgentSession | None,
options: Mapping[str, Any] | None,
service_stores_history: bool,
) -> list[HistoryProvider]:
history_providers = self._get_history_providers()
if not self.require_per_service_call_history_persistence or not history_providers:
return []
conversation_id = (
session.service_session_id
if session and session.service_session_id
else cast(str | None, (options or {}).get("conversation_id") or self.default_options.get("conversation_id"))
)
if service_stores_history:
return []
if conversation_id is not None:
raise AgentInvalidRequestException(
"require_per_service_call_history_persistence cannot be used "
"with an existing service-managed conversation."
)
return history_providers
async def __aexit__(
self,
exc_type: type[BaseException] | None,
@@ -930,9 +885,97 @@ class RawAgent(BaseAgent, Generic[OptionsCoT]): # type: ignore[misc]
When stream=True: A ResponseStream of AgentResponseUpdate items with
``get_final_response()`` for the final AgentResponse.
"""
if not stream:
async def _prepare_run_context() -> _RunContext:
return await self._prepare_run_context(
async def _run_non_streaming() -> AgentResponse[Any]:
ctx = await self._prepare_run_context(
messages=messages,
session=session,
tools=tools,
options=options,
compaction_strategy=compaction_strategy,
tokenizer=tokenizer,
function_invocation_kwargs=function_invocation_kwargs,
client_kwargs=client_kwargs,
)
response = cast(
ChatResponse[Any],
await self.client.get_response( # type: ignore
messages=ctx["session_messages"],
stream=False,
options=ctx["chat_options"], # type: ignore[reportArgumentType]
compaction_strategy=ctx["compaction_strategy"],
tokenizer=ctx["tokenizer"],
function_invocation_kwargs=ctx["function_invocation_kwargs"],
client_kwargs=ctx["client_kwargs"],
),
)
if not response:
raise AgentInvalidResponseException("Chat client did not return a response.")
await self._finalize_response(
response=response,
agent_name=ctx["agent_name"],
session=ctx["session"],
session_context=ctx["session_context"],
)
response_format = ctx["chat_options"].get("response_format")
if not (
response_format is not None
and isinstance(response_format, type)
and issubclass(response_format, BaseModel)
):
response_format = None
return AgentResponse(
messages=response.messages,
response_id=response.response_id,
created_at=response.created_at,
usage_details=response.usage_details,
value=response.value,
response_format=response_format,
continuation_token=response.continuation_token,
raw_representation=response,
additional_properties=response.additional_properties,
)
return _run_non_streaming()
# Use a holder to capture the context created during stream initialization
ctx_holder: dict[str, _RunContext | None] = {"ctx": None}
async def _post_hook(response: AgentResponse) -> None:
ctx = ctx_holder["ctx"]
if ctx is None:
return # No context available (shouldn't happen in normal flow)
# Update thread with conversation_id derived from streaming raw updates.
# Using response_id here can break function-call continuation for APIs
# where response IDs are not valid conversation handles.
conversation_id = self._extract_conversation_id_from_streaming_response(response)
# Ensure author names are set for all messages
for message in response.messages:
if message.author_name is None:
message.author_name = ctx["agent_name"]
# Propagate conversation_id back to session from streaming updates.
# For Responses-style APIs this can rotate every turn (response_id-based continuation),
# so refresh when a newer value is returned.
sess = ctx["session"]
if sess and conversation_id and sess.service_session_id != conversation_id:
sess.service_session_id = conversation_id
# Run after_run providers (reverse order)
session_context = ctx["session_context"]
session_context._response = AgentResponse( # type: ignore[assignment]
messages=response.messages,
response_id=response.response_id,
)
await self._run_after_providers(session=ctx["session"], context=session_context)
async def _get_stream() -> ResponseStream[ChatResponseUpdate, ChatResponse[Any]]:
ctx_holder["ctx"] = await self._prepare_run_context(
messages=messages,
session=session,
tools=tools,
@@ -942,177 +985,55 @@ class RawAgent(BaseAgent, Generic[OptionsCoT]): # type: ignore[misc]
function_invocation_kwargs=function_invocation_kwargs,
client_kwargs=client_kwargs,
)
if not stream:
async def _run_non_streaming() -> AgentResponse[Any]:
ctx = await _prepare_run_context()
response = await self._call_chat_client(ctx, stream=False)
return await self._parse_non_streaming_response(ctx, response)
return _run_non_streaming()
async def _run_streaming() -> ResponseStream[AgentResponseUpdate, AgentResponse[Any]]:
ctx = await _prepare_run_context()
stream_response = self._call_chat_client(ctx, stream=True)
return self._parse_streaming_response(ctx, stream_response)
return cast(
ResponseStream[AgentResponseUpdate, AgentResponse[Any]],
cast(Any, ResponseStream).from_awaitable(_run_streaming()),
)
@overload
def _call_chat_client(
self,
context: _RunContext,
*,
stream: Literal[False],
) -> Awaitable[ChatResponse[Any]]: ...
@overload
def _call_chat_client(
self,
context: _RunContext,
*,
stream: Literal[True],
) -> ResponseStream[ChatResponseUpdate, ChatResponse[Any]]: ...
def _call_chat_client(
self,
context: _RunContext,
*,
stream: bool,
) -> Awaitable[ChatResponse[Any]] | ResponseStream[ChatResponseUpdate, ChatResponse[Any]]:
"""Invoke the downstream chat client for a prepared run context."""
if stream:
ctx: _RunContext = ctx_holder["ctx"] # type: ignore[assignment] # Safe: we just assigned it
return self.client.get_response( # type: ignore[call-overload, no-any-return]
messages=context["session_messages"],
messages=ctx["session_messages"],
stream=True,
options=context["chat_options"], # type: ignore[reportArgumentType]
compaction_strategy=context["compaction_strategy"],
tokenizer=context["tokenizer"],
function_invocation_kwargs=context["function_invocation_kwargs"],
client_kwargs=context["client_kwargs"],
options=ctx["chat_options"], # type: ignore[reportArgumentType]
compaction_strategy=ctx["compaction_strategy"],
tokenizer=ctx["tokenizer"],
function_invocation_kwargs=ctx["function_invocation_kwargs"],
client_kwargs=ctx["client_kwargs"],
)
return self.client.get_response( # type: ignore[call-overload, no-any-return]
messages=context["session_messages"],
stream=False,
options=context["chat_options"], # type: ignore[reportArgumentType]
compaction_strategy=context["compaction_strategy"],
tokenizer=context["tokenizer"],
function_invocation_kwargs=context["function_invocation_kwargs"],
client_kwargs=context["client_kwargs"],
)
def _propagate_conversation_id(
update: AgentResponseUpdate,
) -> AgentResponseUpdate:
"""Eagerly propagate conversation_id to session as updates arrive.
async def _parse_non_streaming_response(
self,
context: _RunContext,
response: ChatResponse[Any],
) -> AgentResponse[Any]:
"""Finalize a non-streaming chat response into an AgentResponse."""
if not response:
raise AgentInvalidResponseException("Chat client did not return a response.")
await self._finalize_response(
response=response,
agent_name=context["agent_name"],
session=context["session"],
session_context=context["session_context"],
suppress_response_id=context["suppress_response_id"],
)
response_format = context["chat_options"].get("response_format")
if not (
response_format is not None and isinstance(response_format, type) and issubclass(response_format, BaseModel)
):
response_format = None
return AgentResponse(
messages=response.messages,
response_id=None if context["suppress_response_id"] else response.response_id,
created_at=response.created_at,
usage_details=response.usage_details,
value=response.value,
response_format=response_format,
continuation_token=response.continuation_token,
raw_representation=response,
additional_properties=response.additional_properties,
)
def _parse_streaming_response(
self,
context: _RunContext,
stream_response: ResponseStream[ChatResponseUpdate, ChatResponse[Any]],
) -> ResponseStream[AgentResponseUpdate, AgentResponse[Any]]:
"""Finalize a streaming chat response into an agent response stream."""
async def _post_hook(response: AgentResponse) -> None:
# Update thread with conversation_id derived from streaming raw updates.
# Using response_id here can break function-call continuation for APIs
# where response IDs are not valid conversation handles.
conversation_id = self._extract_conversation_id_from_streaming_response(response)
for message in response.messages:
if message.author_name is None:
message.author_name = context["agent_name"]
session = context["session"]
if (
session
and conversation_id
and not is_local_history_conversation_id(conversation_id)
and session.service_session_id != conversation_id
):
session.service_session_id = conversation_id
suppress_response_id = context["suppress_response_id"]
session_context = context["session_context"]
session_context._response = AgentResponse( # type: ignore[assignment]
messages=response.messages,
response_id=None if suppress_response_id else response.response_id,
)
await self._run_after_providers(session=session, context=session_context)
def _propagate_conversation_id(update: AgentResponseUpdate) -> AgentResponseUpdate:
"""Eagerly propagate conversation_id to session as updates arrive."""
session = context["session"]
This ensures session.service_session_id is set even when the user
only iterates the stream without calling get_final_response().
"""
if session is None:
return update
raw = update.raw_representation
conversation_id = getattr(raw, "conversation_id", None) if raw else None
if (
isinstance(conversation_id, str)
and conversation_id
and not is_local_history_conversation_id(conversation_id)
and session.service_session_id != conversation_id
):
session.service_session_id = conversation_id
return update
def _suppress_response_id(update: AgentResponseUpdate) -> AgentResponseUpdate:
"""Hide raw service response ids when local per-service-call persistence owns continuation."""
update.response_id = None
conv_id = getattr(raw, "conversation_id", None) if raw else None
if isinstance(conv_id, str) and conv_id and session.service_session_id != conv_id:
session.service_session_id = conv_id
return update
def _finalizer(updates: Sequence[AgentResponseUpdate]) -> AgentResponse[Any]:
return self._finalize_response_updates(
updates,
response_format=context["chat_options"].get("response_format"),
ctx = ctx_holder["ctx"]
rf = (
ctx.get("chat_options", {}).get("response_format")
if ctx
else (options.get("response_format") if options else None) # type: ignore[union-attr]
)
return self._finalize_response_updates(updates, response_format=rf)
stream = stream_response.map(
transform=partial(
map_chat_to_agent_update,
agent_name=self.name,
),
finalizer=_finalizer,
return (
ResponseStream
.from_awaitable(_get_stream()) # type: ignore[reportUnknownMemberType]
.map(
transform=partial(
map_chat_to_agent_update,
agent_name=self.name,
),
finalizer=_finalizer,
)
.with_transform_hook(_propagate_conversation_id)
.with_result_hook(_post_hook)
)
if context["suppress_response_id"]:
stream = stream.with_transform_hook(_suppress_response_id)
return stream.with_transform_hook(_propagate_conversation_id).with_result_hook(_post_hook)
def _finalize_response_updates(
self,
@@ -1190,12 +1111,6 @@ class RawAgent(BaseAgent, Generic[OptionsCoT]): # type: ignore[misc]
if active_session is None and self.context_providers:
active_session = AgentSession()
per_service_call_history_providers = self._resolve_per_service_call_history_providers(
session=active_session,
options=opts,
service_stores_history=bool(store_),
)
session_context, chat_options = await self._prepare_session_and_messages(
session=active_session,
input_messages=input_messages,
@@ -1276,43 +1191,6 @@ class RawAgent(BaseAgent, Generic[OptionsCoT]): # type: ignore[misc]
effective_client_kwargs = dict(client_kwargs) if client_kwargs is not None else {}
if active_session is not None:
effective_client_kwargs["session"] = active_session
if per_service_call_history_providers and active_session is not None:
per_service_call_history_middleware = PerServiceCallHistoryPersistingMiddleware(
agent=self,
session=active_session,
providers=per_service_call_history_providers,
)
existing_middleware = effective_client_kwargs.get("middleware")
if isinstance(existing_middleware, Sequence) and not isinstance(existing_middleware, (str, bytes)):
effective_client_kwargs["middleware"] = [per_service_call_history_middleware, *existing_middleware]
elif existing_middleware is not None:
effective_client_kwargs["middleware"] = [
per_service_call_history_middleware,
cast(MiddlewareTypes, existing_middleware),
]
else:
effective_client_kwargs["middleware"] = [per_service_call_history_middleware]
provider_middleware = session_context.get_middleware()
if provider_middleware:
middleware_list = categorize_middleware(provider_middleware)
provider_function_chat_middleware = [
*middleware_list["function"],
*middleware_list["chat"],
]
if provider_function_chat_middleware:
existing_middleware = effective_client_kwargs.get("middleware")
if isinstance(existing_middleware, Sequence) and not isinstance(existing_middleware, (str, bytes)):
effective_client_kwargs["middleware"] = [
*existing_middleware,
*provider_function_chat_middleware,
]
elif existing_middleware is not None:
effective_client_kwargs["middleware"] = [
cast(MiddlewareTypes, existing_middleware),
*provider_function_chat_middleware,
]
else:
effective_client_kwargs["middleware"] = provider_function_chat_middleware
return {
"session": active_session,
@@ -1320,7 +1198,6 @@ class RawAgent(BaseAgent, Generic[OptionsCoT]): # type: ignore[misc]
"input_messages": input_messages,
"session_messages": session_messages,
"agent_name": agent_name,
"suppress_response_id": bool(per_service_call_history_providers),
"chat_options": co,
"compaction_strategy": compaction_strategy or self.compaction_strategy,
"tokenizer": tokenizer or self.tokenizer,
@@ -1334,7 +1211,6 @@ class RawAgent(BaseAgent, Generic[OptionsCoT]): # type: ignore[misc]
agent_name: str,
session: AgentSession | None,
session_context: SessionContext,
suppress_response_id: bool = False,
) -> None:
"""Finalize response by setting author names and running after_run providers.
@@ -1343,7 +1219,6 @@ class RawAgent(BaseAgent, Generic[OptionsCoT]): # type: ignore[misc]
agent_name: The name of the agent to set as author.
session: The conversation session.
session_context: The invocation context.
suppress_response_id: When True, omit the raw service response ID from the public response.
"""
# Ensure that the author name is set for each message in the response.
for message in response.messages:
@@ -1353,18 +1228,13 @@ class RawAgent(BaseAgent, Generic[OptionsCoT]): # type: ignore[misc]
# Propagate conversation_id back to session (e.g. thread ID from Assistants API).
# For Responses-style APIs this can rotate every turn (response_id-based continuation),
# so refresh when a newer value is returned.
if (
session
and response.conversation_id
and not is_local_history_conversation_id(response.conversation_id)
and session.service_session_id != response.conversation_id
):
if session and response.conversation_id and session.service_session_id != response.conversation_id:
session.service_session_id = response.conversation_id
# Set the response on the context for after_run providers
session_context._response = AgentResponse( # type: ignore[assignment]
messages=response.messages,
response_id=None if suppress_response_id else response.response_id,
response_id=response.response_id,
)
# Run after_run providers (reverse order)
@@ -1414,15 +1284,9 @@ class RawAgent(BaseAgent, Generic[OptionsCoT]): # type: ignore[misc]
options=options or {},
)
per_service_call_history_required = self.require_per_service_call_history_persistence and bool(
self._get_history_providers()
)
# Run before_run providers (forward order, skip HistoryProvider when per-service-call persistence owns history)
# Run before_run providers (forward order, skip BaseHistoryProvider with load_messages=False)
for provider in self.context_providers:
if per_service_call_history_required and isinstance(provider, HistoryProvider):
continue
if isinstance(provider, HistoryProvider) and not provider.load_messages:
if isinstance(provider, BaseHistoryProvider) and not provider.load_messages:
continue
if provider_session is None:
raise RuntimeError("Provider session must be available when context providers are configured.")
@@ -1687,9 +1551,8 @@ class Agent(
description: str | None = None,
tools: ToolTypes | Callable[..., Any] | Sequence[ToolTypes | Callable[..., Any]] | None = None,
default_options: OptionsCoT | None = None,
context_providers: Sequence[ContextProvider] | None = None,
context_providers: Sequence[BaseContextProvider] | None = None,
middleware: Sequence[MiddlewareTypes] | None = None,
require_per_service_call_history_persistence: bool = False,
compaction_strategy: CompactionStrategy | None = None,
tokenizer: TokenizerProtocol | None = None,
additional_properties: MutableMapping[str, Any] | None = None,
@@ -1705,7 +1568,6 @@ class Agent(
default_options=default_options,
context_providers=context_providers,
middleware=middleware,
require_per_service_call_history_persistence=require_per_service_call_history_persistence,
compaction_strategy=compaction_strategy,
tokenizer=tokenizer,
additional_properties=additional_properties,
@@ -572,7 +572,6 @@ class BaseChatClient(SerializationMixin, ABC, Generic[OptionsCoT]):
default_options: OptionsCoT | Mapping[str, Any] | None = None,
context_providers: Sequence[Any] | None = None,
middleware: Sequence[MiddlewareTypes] | None = None,
require_per_service_call_history_persistence: bool = False,
function_invocation_configuration: FunctionInvocationConfiguration | None = None,
compaction_strategy: CompactionStrategy | None = None,
tokenizer: TokenizerProtocol | None = None,
@@ -597,10 +596,6 @@ class BaseChatClient(SerializationMixin, ABC, Generic[OptionsCoT]):
and dict literals are accepted without specialized option typing.
context_providers: Context providers to include during agent invocation.
middleware: List of middleware to intercept agent and function invocations.
require_per_service_call_history_persistence: Whether to require per-service-call
chat history persistence. When enabled, history providers are invoked around
each model call instead of once per ``run()`` when the service is not already
storing history.
function_invocation_configuration: Optional function invocation configuration override.
compaction_strategy: Optional agent-level compaction override. When omitted,
client-level compaction defaults remain in effect for each call.
@@ -641,7 +636,6 @@ class BaseChatClient(SerializationMixin, ABC, Generic[OptionsCoT]):
"default_options": cast(Any, default_options),
"context_providers": context_providers,
"middleware": middleware,
"require_per_service_call_history_persistence": require_per_service_call_history_persistence,
"compaction_strategy": compaction_strategy,
"tokenizer": tokenizer,
"additional_properties": dict(additional_properties) if additional_properties is not None else None,
@@ -814,21 +808,22 @@ class SupportsFileSearchTool(Protocol):
# region SupportsGetEmbeddings Protocol
# TypeVars for the Protocol
# Contravariant TypeVars for the Protocol
EmbeddingInputContraT = TypeVar(
"EmbeddingInputContraT",
default="str",
contravariant=True,
)
EmbeddingProtocolOptionsT = TypeVar(
"EmbeddingProtocolOptionsT",
EmbeddingOptionsContraT = TypeVar(
"EmbeddingOptionsContraT",
bound=TypedDict, # type: ignore[valid-type]
default="EmbeddingGenerationOptions",
contravariant=True,
)
@runtime_checkable
class SupportsGetEmbeddings(Protocol[EmbeddingInputContraT, EmbeddingT, EmbeddingProtocolOptionsT]):
class SupportsGetEmbeddings(Protocol[EmbeddingInputContraT, EmbeddingT, EmbeddingOptionsContraT]):
"""Protocol for an embedding client that can generate embeddings.
This protocol enables duck-typing for embedding generation. Any class that
@@ -855,8 +850,8 @@ class SupportsGetEmbeddings(Protocol[EmbeddingInputContraT, EmbeddingT, Embeddin
self,
values: Sequence[EmbeddingInputContraT],
*,
options: EmbeddingProtocolOptionsT | None = None,
) -> Awaitable[GeneratedEmbeddings[EmbeddingT, EmbeddingProtocolOptionsT]]:
options: EmbeddingOptionsContraT | None = None,
) -> Awaitable[GeneratedEmbeddings[EmbeddingT]]:
"""Generate embeddings for the given values.
Args:
@@ -15,7 +15,7 @@ from typing import (
runtime_checkable,
)
from ._sessions import ContextProvider
from ._sessions import BaseContextProvider
from ._types import ChatResponse, Content, Message
if TYPE_CHECKING:
@@ -1152,7 +1152,7 @@ async def apply_compaction(
COMPACTION_STATE_KEY: Final[str] = "_compaction_messages"
class CompactionProvider(ContextProvider):
class CompactionProvider(BaseContextProvider):
"""Context provider that compacts messages before and after agent runs.
This provider accepts two separate strategies:
@@ -96,7 +96,6 @@ class ConversationSplitter(Protocol):
# Fallback: split at last user message
return EvalItem._split_last_turn_static(conversation)
item.split_messages(split=split_before_memory)
"""
@@ -469,7 +468,10 @@ class EvalResults:
"""
if not self.all_passed:
errored = (self.result_counts or {}).get("errored", 0)
detail = msg or (f"Eval run {self.run_id} {self.status}: {self.passed} passed, {self.failed} failed.")
detail = msg or (
f"Eval run {self.run_id} {self.status}: "
f"{self.passed} passed, {self.failed} failed."
)
if errored:
detail += f" {errored} errored."
if self.report_url:
@@ -1186,7 +1188,8 @@ def _coerce_result(value: Any, check_name: str) -> CheckResult:
score = float(d["score"])
except (TypeError, ValueError) as exc:
raise TypeError(
f"Function evaluator '{check_name}' returned dict with non-numeric 'score' value: {d['score']!r}"
f"Function evaluator '{check_name}' returned dict with non-numeric 'score' value:"
f" {d['score']!r}"
) from exc
# Honour an explicit 'passed' override; otherwise threshold-based.
passed = bool(d["passed"]) if "passed" in d else score >= float(d.get("threshold", 0.5))
+19 -274
View File
@@ -4,8 +4,8 @@
This module provides the core types for the context provider pipeline:
- SessionContext: Per-invocation state passed through providers
- ContextProvider: Base class for context providers
- HistoryProvider: Base class for history storage providers
- BaseContextProvider: Base class for context providers (renamed to ContextProvider in PR2)
- BaseHistoryProvider: Base class for history storage providers (renamed to HistoryProvider in PR2)
- AgentSession: Lightweight session state container
- InMemoryHistoryProvider: Built-in in-memory history provider
"""
@@ -13,42 +13,21 @@ This module provides the core types for the context provider pipeline:
from __future__ import annotations
import copy
import sys
import uuid
from abc import abstractmethod
from collections.abc import Awaitable, Callable, Mapping, Sequence
from typing import TYPE_CHECKING, Any, ClassVar, TypeGuard, cast
from collections.abc import Sequence
from typing import TYPE_CHECKING, Any, ClassVar, cast
if sys.version_info >= (3, 13):
from warnings import deprecated # type: ignore # pragma: no cover
else:
from typing_extensions import deprecated # type: ignore # pragma: no cover
from ._middleware import ChatContext, ChatMiddleware
from ._types import AgentResponse, ChatResponse, Message, ResponseStream
from .exceptions import ChatClientInvalidResponseException
from ._types import AgentResponse, Message
if TYPE_CHECKING:
from ._agents import SupportsAgentRun
from ._middleware import MiddlewareTypes
# Registry of known types for state deserialization
_STATE_TYPE_REGISTRY: dict[str, type] = {}
def _is_middleware_sequence(
middleware: MiddlewareTypes | Sequence[MiddlewareTypes],
) -> TypeGuard[Sequence[MiddlewareTypes]]:
return isinstance(middleware, Sequence) and not isinstance(middleware, (str, bytes))
def _is_single_middleware(
middleware: MiddlewareTypes | Sequence[MiddlewareTypes],
) -> TypeGuard[MiddlewareTypes]:
return not _is_middleware_sequence(middleware)
def register_state_type(cls: type) -> None:
"""Register a type for automatic deserialization in session state.
@@ -152,8 +131,6 @@ class SessionContext:
Maintains insertion order (provider execution order).
instructions: Additional instructions added by providers.
tools: Additional tools added by providers.
middleware: Dict mapping source_id -> chat/function middleware added by that provider.
Maintains insertion order (provider execution order).
response: After invocation, contains the full AgentResponse, should not be changed.
options: Options passed to agent.run() - read-only, for reflection only.
metadata: Shared metadata dictionary for cross-provider communication.
@@ -168,7 +145,6 @@ class SessionContext:
context_messages: dict[str, list[Message]] | None = None,
instructions: list[str] | None = None,
tools: list[Any] | None = None,
middleware: dict[str, list[MiddlewareTypes]] | None = None,
options: dict[str, Any] | None = None,
metadata: dict[str, Any] | None = None,
):
@@ -181,7 +157,6 @@ class SessionContext:
context_messages: Pre-populated context messages by source.
instructions: Pre-populated instructions.
tools: Pre-populated tools.
middleware: Pre-populated chat/function middleware by source.
options: Options from agent.run() - read-only for providers.
metadata: Shared metadata for cross-provider communication.
"""
@@ -191,10 +166,6 @@ class SessionContext:
self.context_messages: dict[str, list[Message]] = context_messages or {}
self.instructions: list[str] = instructions or []
self.tools: list[Any] = tools or []
self.middleware: dict[str, list[MiddlewareTypes]] = {}
if middleware:
for source_id, provider_middleware in middleware.items():
self.extend_middleware(source_id, provider_middleware)
self._response: AgentResponse | None = None
self.options: dict[str, Any] = options or {}
self.metadata: dict[str, Any] = metadata or {}
@@ -265,40 +236,6 @@ class SessionContext:
additional_properties["context_source"] = source_id
self.tools.extend(tools)
def extend_middleware(
self,
source_id: str,
middleware: MiddlewareTypes | Sequence[MiddlewareTypes],
) -> None:
"""Add middleware to be applied for this invocation.
Args:
source_id: The provider source_id adding this middleware.
middleware: A single chat/function middleware object/callable or sequence of middleware.
"""
from ._middleware import categorize_middleware
from .exceptions import MiddlewareException
if _is_middleware_sequence(middleware):
middleware_items = list(middleware)
elif _is_single_middleware(middleware):
middleware_items = [middleware]
else:
raise TypeError("middleware must be a middleware object or a sequence of middleware objects.")
middleware_list = categorize_middleware(middleware_items)
if middleware_list["agent"]:
raise MiddlewareException("Context providers may only add chat or function middleware.")
if source_id not in self.middleware:
self.middleware[source_id] = []
self.middleware[source_id].extend(middleware_items)
def get_middleware(self) -> list[MiddlewareTypes]:
"""Get provider-added chat/function middleware in provider execution order."""
result: list[MiddlewareTypes] = []
for middleware_items in self.middleware.values():
result.extend(middleware_items)
return result
def get_messages(
self,
*,
@@ -335,12 +272,17 @@ class SessionContext:
return result
class ContextProvider:
"""Base class for context providers.
class BaseContextProvider:
"""Base class for context providers (hooks pattern).
Context providers participate in the context engineering pipeline,
adding context before model invocation and processing responses after.
Note:
This class uses a temporary name prefixed with ``_`` to avoid collision
with the existing ``ContextProvider`` in ``_memory.py``. It will be
renamed to ``ContextProvider`` in PR2 when the old class is removed.
Attributes:
source_id: Unique identifier for this provider instance (required).
Used for message/tool attribution so other providers can filter.
@@ -370,7 +312,7 @@ class ContextProvider:
Args:
agent: The agent running this invocation.
session: The current session.
context: The invocation context - add messages/instructions/tools/chat/function middleware here.
context: The invocation context - add messages/instructions/tools here.
state: The provider-scoped mutable state dict for this provider.
Full cross-provider state remains available at ``session.state``.
"""
@@ -397,7 +339,7 @@ class ContextProvider:
"""
class HistoryProvider(ContextProvider):
class BaseHistoryProvider(BaseContextProvider):
"""Base class for conversation history storage providers.
A single class configurable for different use cases:
@@ -405,6 +347,10 @@ class HistoryProvider(ContextProvider):
- Audit/logging storage (stores only, doesn't load)
- Evaluation storage (stores only for later analysis)
Note:
This class uses a temporary name prefixed with ``_`` to avoid collision
with existing types. It will be renamed to ``HistoryProvider`` in PR2.
Subclasses only need to implement ``get_messages()`` and ``save_messages()``.
The default ``before_run``/``after_run`` handle loading and storing based on
configuration flags. Override them for custom behavior.
@@ -521,207 +467,6 @@ class HistoryProvider(ContextProvider):
await self.save_messages(context.session_id, messages_to_store, state=state)
LOCAL_HISTORY_CONVERSATION_ID = "agent_framework_local_history_persistence"
def is_local_history_conversation_id(conversation_id: str | None) -> bool:
"""Return whether a conversation id is the local history-persistence sentinel."""
return conversation_id == LOCAL_HISTORY_CONVERSATION_ID
def _response_contains_follow_up_request(response: ChatResponse) -> bool:
"""Return whether a response requires another model call in the current run."""
return any(
item.type in {"function_call", "function_approval_request"}
for message in response.messages
for item in message.contents
)
def _split_service_call_messages(messages: Sequence[Message]) -> tuple[list[Message], dict[str, list[Message]]]:
"""Split service-call messages into input messages and attributed context messages."""
input_messages: list[Message] = []
context_messages: dict[str, list[Message]] = {}
for message in messages:
attribution = message.additional_properties.get("_attribution")
if isinstance(attribution, Mapping):
attribution_mapping = cast(Mapping[str, Any], attribution)
source_id = attribution_mapping.get("source_id")
if isinstance(source_id, str):
context_messages.setdefault(source_id, []).append(message)
continue
input_messages.append(message)
return input_messages, context_messages
class PerServiceCallHistoryPersistingMiddleware(ChatMiddleware):
"""Persist local chat history after each service call when history is framework-managed.
This middleware runs around each model call when
``require_per_service_call_history_persistence`` is enabled. It loads history providers
before the model call, persists them after the model call, and uses a local
sentinel conversation id so the function loop follows the existing
service-managed branch without forwarding that sentinel to the leaf client.
"""
def __init__(
self,
*,
agent: SupportsAgentRun,
session: AgentSession,
providers: Sequence[HistoryProvider],
) -> None:
"""Initialize the middleware.
Args:
agent: The agent that owns the history providers.
session: The active session for the current run.
providers: The history providers participating in per-service-call persistence.
"""
self._agent = agent
self._session = session
self._providers = list(providers)
async def _prepare_service_call_context(self, messages: Sequence[Message]) -> SessionContext:
"""Create a per-call SessionContext and load history providers into it."""
input_messages, context_messages = _split_service_call_messages(messages)
service_call_context = SessionContext(
session_id=self._session.session_id,
service_session_id=None,
input_messages=list(input_messages),
)
for source_id, source_messages in context_messages.items():
service_call_context.extend_messages(source_id, source_messages)
for provider in self._providers:
if not provider.load_messages:
continue
await provider.before_run(
agent=self._agent,
session=self._session,
context=service_call_context,
state=self._session.state.setdefault(provider.source_id, {}),
)
return service_call_context
async def _persist_service_call_response(
self,
*,
service_call_context: SessionContext,
response: ChatResponse,
) -> None:
"""Persist a single model-call response through the configured history providers."""
service_call_context._response = AgentResponse( # type: ignore[assignment]
messages=response.messages,
response_id=None,
)
for provider in reversed(self._providers):
await provider.after_run(
agent=self._agent,
session=self._session,
context=service_call_context,
state=self._session.state.setdefault(provider.source_id, {}),
)
def _strip_local_conversation_id(self, context: ChatContext) -> None:
"""Remove the local sentinel before the leaf chat client is invoked."""
if is_local_history_conversation_id(cast(str | None, context.kwargs.get("conversation_id"))):
context.kwargs.pop("conversation_id", None)
if context.options is None:
return
mutable_options = dict(context.options)
if is_local_history_conversation_id(cast(str | None, mutable_options.get("conversation_id"))):
mutable_options.pop("conversation_id", None)
context.options = mutable_options
async def _finalize_response(
self,
*,
service_call_context: SessionContext,
response: ChatResponse,
) -> ChatResponse:
"""Persist a model response and apply the local follow-up sentinel when needed."""
if response.conversation_id is not None and not is_local_history_conversation_id(response.conversation_id):
raise ChatClientInvalidResponseException(
"require_per_service_call_history_persistence cannot be used "
"when the chat client returns a real conversation_id."
)
await self._persist_service_call_response(
service_call_context=service_call_context,
response=response,
)
if _response_contains_follow_up_request(response):
response.mark_internal_conversation_id()
response.conversation_id = LOCAL_HISTORY_CONVERSATION_ID
return response
async def process(self, context: ChatContext, call_next: Callable[[], Awaitable[None]]) -> None:
"""Load and persist history providers around a single model call.
Args:
context: The chat invocation context for the current model call.
call_next: The next middleware or the leaf chat client.
Raises:
ChatClientInvalidResponseException: If the leaf client returns a real
service-managed conversation id while local per-service-call persistence is enabled.
ValueError: If the downstream middleware contract returns the wrong
result type for streaming or non-streaming execution.
"""
service_call_context = await self._prepare_service_call_context(context.messages)
context.messages = service_call_context.get_messages(include_input=True)
self._strip_local_conversation_id(context)
await call_next()
if context.result is None:
return
if context.stream:
if not isinstance(context.result, ResponseStream):
raise ValueError("Streaming chat middleware requires a ResponseStream result.")
context.result = context.result.with_result_hook(
lambda response: self._finalize_response(
service_call_context=service_call_context,
response=response,
)
)
return
if isinstance(context.result, ResponseStream):
raise ValueError("Non-streaming chat middleware requires a ChatResponse result.")
context.result = await self._finalize_response(
service_call_context=service_call_context,
response=context.result,
)
@deprecated(
"BaseContextProvider is deprecated. Use ContextProvider instead.",
category=DeprecationWarning,
)
class BaseContextProvider(ContextProvider):
"""Deprecated alias for :class:`ContextProvider`.
.. deprecated::
BaseContextProvider is deprecated. Use :class:`ContextProvider` instead.
"""
@deprecated(
"BaseHistoryProvider is deprecated. Use HistoryProvider instead.",
category=DeprecationWarning,
)
class BaseHistoryProvider(HistoryProvider):
"""Deprecated alias for :class:`HistoryProvider`.
.. deprecated::
BaseHistoryProvider is deprecated. Use :class:`HistoryProvider` instead.
"""
class AgentSession:
"""A conversation session with an agent.
@@ -790,7 +535,7 @@ class AgentSession:
return session
class InMemoryHistoryProvider(HistoryProvider):
class InMemoryHistoryProvider(BaseHistoryProvider):
"""Built-in history provider that stores messages in session.state.
Messages are stored in ``state["messages"]`` as a list of
@@ -36,7 +36,7 @@ from pathlib import Path, PurePosixPath
from typing import TYPE_CHECKING, Any, ClassVar, Final, Protocol, runtime_checkable
from ._feature_stage import ExperimentalFeature, experimental
from ._sessions import ContextProvider
from ._sessions import BaseContextProvider
from ._tools import FunctionTool
if TYPE_CHECKING:
@@ -519,7 +519,7 @@ SCRIPT_RUNNER_INSTRUCTIONS: Final[str] = (
@experimental(feature_id=ExperimentalFeature.SKILLS)
class SkillsProvider(ContextProvider):
class SkillsProvider(BaseContextProvider):
"""Context provider that advertises skills and exposes skill tools.
Supports both **file-based** skills (discovered from ``SKILL.md`` files)
+5 -55
View File
@@ -1688,34 +1688,6 @@ def _update_conversation_id(
options["conversation_id"] = conversation_id
def _update_continuation_state(
kwargs: dict[str, Any],
response: ChatResponse[Any],
*,
session: AgentSession | None,
options: dict[str, Any] | None = None,
) -> None:
"""Update in-flight and persisted continuation state from a response."""
conversation_id = response.conversation_id
if conversation_id is None:
return
_update_conversation_id(kwargs, conversation_id, options)
if (
session is not None
and not response.has_internal_conversation_id()
and session.service_session_id != conversation_id
):
session.service_session_id = conversation_id
def _clear_internal_conversation_id(response: ChatResponse[Any]) -> ChatResponse[Any]:
if response.has_internal_conversation_id():
response.conversation_id = None
response.clear_internal_conversation_id()
return response
def _extract_tools(
options: dict[str, Any] | None,
) -> ToolTypes | Callable[..., Any] | Sequence[ToolTypes | Callable[..., Any]] | None:
@@ -2234,14 +2206,9 @@ class FunctionInvocationLayer(Generic[OptionsCoT]):
),
)
aggregated_usage = add_usage_details(aggregated_usage, response.usage_details)
_update_continuation_state(
filtered_kwargs,
response,
session=invocation_session,
options=mutable_options,
)
if response.conversation_id is not None:
_update_conversation_id(filtered_kwargs, response.conversation_id, mutable_options)
prepped_messages = []
result = await _process_function_requests(
@@ -2256,7 +2223,7 @@ class FunctionInvocationLayer(Generic[OptionsCoT]):
)
if result.get("action") == "return":
response.usage_details = aggregated_usage
return _clear_internal_conversation_id(response)
return response
total_function_calls += result.get("function_call_count", 0)
if result.get("action") == "stop":
# Error threshold reached: force a final non-tool turn so
@@ -2312,17 +2279,11 @@ class FunctionInvocationLayer(Generic[OptionsCoT]):
),
)
aggregated_usage = add_usage_details(aggregated_usage, response.usage_details)
_update_continuation_state(
filtered_kwargs,
response,
session=invocation_session,
options=mutable_options,
)
response.usage_details = aggregated_usage
if fcc_messages:
for msg in reversed(fcc_messages):
response.messages.insert(0, msg)
return _clear_internal_conversation_id(response)
return response
return _get_response()
@@ -2382,12 +2343,6 @@ class FunctionInvocationLayer(Generic[OptionsCoT]):
# Get the finalized response from the inner stream
# This triggers the inner stream's finalizer and result hooks
response = await inner_stream.get_final_response()
_update_continuation_state(
filtered_kwargs,
response,
session=invocation_session,
options=mutable_options,
)
if not any(
item.type in ("function_call", "function_approval_request")
@@ -2397,6 +2352,7 @@ class FunctionInvocationLayer(Generic[OptionsCoT]):
return
if response.conversation_id is not None:
_update_conversation_id(filtered_kwargs, response.conversation_id, mutable_options)
prepped_messages = []
result = await _process_function_requests(
@@ -2474,13 +2430,7 @@ class FunctionInvocationLayer(Generic[OptionsCoT]):
async for update in final_inner_stream:
yield update
# Finalize the inner stream to trigger its hooks
final_response = await final_inner_stream.get_final_response()
_update_continuation_state(
filtered_kwargs,
final_response,
session=invocation_session,
options=mutable_options,
)
await final_inner_stream.get_final_response()
def _finalize(updates: Sequence[ChatResponseUpdate]) -> ChatResponse[Any]:
# Note: stream_result_hooks are already run via inner stream's get_final_response()
+1 -27
View File
@@ -1380,13 +1380,6 @@ class Content:
def _add_text_reasoning_content(self, other: Content) -> Content:
"""Add two TextReasoningContent instances."""
# Ensure we do not silently merge contents with conflicting ids
if self.id and other.id and self.id != other.id:
raise AdditionItemMismatch(
f"Cannot add text_reasoning content with different ids: {self.id!r} != {other.id!r}"
)
combined_id = self.id or other.id
# Concatenate text, handling None values
self_text = self.text or "" # type: ignore[attr-defined]
other_text = other.text or "" # type: ignore[attr-defined]
@@ -1397,7 +1390,6 @@ class Content:
return Content(
"text_reasoning",
id=combined_id,
text=combined_text,
protected_data=protected_data,
annotations=_combine_annotations(self.annotations, other.annotations),
@@ -1888,12 +1880,7 @@ def _coalesce_text_content(contents: list[Content], type_str: Literal["text", "t
if first_new_content is None:
first_new_content = deepcopy(content)
else:
try:
first_new_content += content
except AdditionItemMismatch:
# Different IDs means a new logical segment; flush the current one
coalesced_contents.append(first_new_content)
first_new_content = deepcopy(content)
first_new_content += content
else:
# skip this content, it is not of the right type
# so write the existing one to the list and start a new one,
@@ -2001,7 +1988,6 @@ class ChatResponse(SerializationMixin, Generic[ResponseModelT]):
"""
DEFAULT_EXCLUDE: ClassVar[set[str]] = {"raw_representation", "additional_properties"}
_INTERNAL_CONVERSATION_ID_KEY: ClassVar[str] = "_agent_framework_internal_conversation_id"
def __init__(
self,
@@ -2070,18 +2056,6 @@ class ChatResponse(SerializationMixin, Generic[ResponseModelT]):
self.continuation_token = continuation_token
self.raw_representation: Any | list[Any] | None = raw_representation
def mark_internal_conversation_id(self) -> None:
"""Mark the current conversation_id as internal control-flow state."""
self.additional_properties[self._INTERNAL_CONVERSATION_ID_KEY] = True
def clear_internal_conversation_id(self) -> None:
"""Remove the internal conversation-id marker."""
self.additional_properties.pop(self._INTERNAL_CONVERSATION_ID_KEY, None)
def has_internal_conversation_id(self) -> bool:
"""Return whether conversation_id is internal control-flow state."""
return bool(self.additional_properties.get(self._INTERNAL_CONVERSATION_ID_KEY, False))
@property
def model_id(self) -> str | None:
"""Deprecated alias for :attr:`model`."""
@@ -14,8 +14,8 @@ from typing import TYPE_CHECKING, Any, ClassVar, Literal, cast, overload
from .._agents import BaseAgent
from .._sessions import (
AgentSession,
ContextProvider,
HistoryProvider,
BaseContextProvider,
BaseHistoryProvider,
InMemoryHistoryProvider,
SessionContext,
)
@@ -86,7 +86,7 @@ class WorkflowAgent(BaseAgent):
id: str | None = None,
name: str | None = None,
description: str | None = None,
context_providers: Sequence[ContextProvider] | None = None,
context_providers: Sequence[BaseContextProvider] | None = None,
**kwargs: Any,
) -> None:
"""Initialize the WorkflowAgent.
@@ -249,7 +249,7 @@ class WorkflowAgent(BaseAgent):
options={},
)
for provider in self.context_providers:
if isinstance(provider, HistoryProvider) and not provider.load_messages:
if isinstance(provider, BaseHistoryProvider) and not provider.load_messages:
continue
if provider_session is None:
raise RuntimeError("Provider session must be available when context providers are configured.")
@@ -314,7 +314,7 @@ class WorkflowAgent(BaseAgent):
options={},
)
for provider in self.context_providers:
if isinstance(provider, HistoryProvider) and not provider.load_messages:
if isinstance(provider, BaseHistoryProvider) and not provider.load_messages:
continue
if provider_session is None:
raise RuntimeError("Provider session must be available when context providers are configured.")
@@ -15,10 +15,6 @@ _IMPORTS: dict[str, tuple[str, str]] = {
"AzureAISearchContextProvider": ("agent_framework_azure_ai_search", "agent-framework-azure-ai-search"),
"AzureAISearchSettings": ("agent_framework_azure_ai_search", "agent-framework-azure-ai-search"),
"AzureAISettings": ("agent_framework_azure_ai", "agent-framework-azure-ai"),
"AzureAIInferenceEmbeddingClient": ("agent_framework_azure_ai", "agent-framework-azure-ai"),
"AzureAIInferenceEmbeddingOptions": ("agent_framework_azure_ai", "agent-framework-azure-ai"),
"AzureAIInferenceEmbeddingSettings": ("agent_framework_azure_ai", "agent-framework-azure-ai"),
"RawAzureAIInferenceEmbeddingClient": ("agent_framework_azure_ai", "agent-framework-azure-ai"),
"AzureCredentialTypes": ("agent_framework_azure_ai", "agent-framework-azure-ai"),
"AzureTokenProvider": ("agent_framework_azure_ai", "agent-framework-azure-ai"),
"DurableAIAgent": ("agent_framework_durabletask", "agent-framework-durabletask"),
@@ -4,13 +4,9 @@
# Install the relevant packages for full type support.
from agent_framework_azure_ai import (
AzureAIInferenceEmbeddingClient,
AzureAIInferenceEmbeddingOptions,
AzureAIInferenceEmbeddingSettings,
AzureAISettings,
AzureCredentialTypes,
AzureTokenProvider,
RawAzureAIInferenceEmbeddingClient,
)
from agent_framework_azure_ai_search import (
AzureAISearchContextProvider,
@@ -30,9 +26,6 @@ __all__ = [
"AgentCallbackContext",
"AgentFunctionApp",
"AgentResponseCallbackProtocol",
"AzureAIInferenceEmbeddingClient",
"AzureAIInferenceEmbeddingOptions",
"AzureAIInferenceEmbeddingSettings",
"AzureAISearchContextProvider",
"AzureAISearchSettings",
"AzureAISettings",
@@ -42,5 +35,4 @@ __all__ = [
"DurableAIAgentClient",
"DurableAIAgentOrchestrationContext",
"DurableAIAgentWorker",
"RawAzureAIInferenceEmbeddingClient",
]
@@ -1502,161 +1502,6 @@ class AgentTelemetryLayer:
self.token_usage_histogram = _get_token_usage_histogram()
self.duration_histogram = _get_duration_histogram()
def _trace_agent_invocation(
self,
*,
messages: AgentRunInputs | None,
session: AgentSession | None,
merged_options: Mapping[str, Any],
client_kwargs: Mapping[str, Any] | None,
stream: bool,
execute: Callable[[], Awaitable[AgentResponse[Any]] | ResponseStream[AgentResponseUpdate, AgentResponse[Any]]],
) -> Awaitable[AgentResponse[Any]] | ResponseStream[AgentResponseUpdate, AgentResponse[Any]]:
"""Trace an agent invocation while delegating execution to ``execute``."""
global OBSERVABILITY_SETTINGS
from ._types import ResponseStream
if not OBSERVABILITY_SETTINGS.ENABLED:
return execute()
provider_name = str(self.otel_provider_name)
merged_client_kwargs = dict(client_kwargs) if client_kwargs is not None else {}
attributes = _get_span_attributes(
operation_name=OtelAttr.AGENT_INVOKE_OPERATION,
provider_name=provider_name,
agent_id=getattr(self, "id", "unknown"),
agent_name=getattr(self, "name", None) or getattr(self, "id", "unknown"),
agent_description=getattr(self, "description", None),
thread_id=session.service_session_id if session else None,
all_options=dict(merged_options),
**merged_client_kwargs,
)
inner_response_telemetry_captured_fields: set[str] = set()
inner_response_telemetry_captured_fields_token = INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.set(
inner_response_telemetry_captured_fields
)
inner_accumulated_usage_token = INNER_ACCUMULATED_USAGE.set({})
if stream:
try:
run_result: object = execute()
if isinstance(run_result, ResponseStream):
result_stream: ResponseStream[AgentResponseUpdate, AgentResponse[Any]] = run_result # pyright: ignore[reportUnknownVariableType]
elif isinstance(run_result, Awaitable):
result_stream = ResponseStream.from_awaitable(run_result) # type: ignore[arg-type] # pyright: ignore[reportArgumentType]
else:
raise RuntimeError("Streaming telemetry requires a ResponseStream result.")
except Exception:
INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.reset(inner_response_telemetry_captured_fields_token)
INNER_ACCUMULATED_USAGE.reset(inner_accumulated_usage_token)
raise
operation = attributes.get(OtelAttr.OPERATION, "operation")
span_name = attributes.get(OtelAttr.AGENT_NAME, "unknown")
span = get_tracer().start_span(f"{operation} {span_name}")
span.set_attributes(attributes)
if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED and messages:
_capture_messages(
span=span,
provider_name=provider_name,
messages=messages,
system_instructions=_get_instructions_from_options(dict(merged_options)),
)
span_state = {"closed": False}
duration_state: dict[str, float] = {}
start_time = perf_counter()
def _close_span() -> None:
if span_state["closed"]:
return
span_state["closed"] = True
span.end()
def _record_duration() -> None:
duration_state["duration"] = perf_counter() - start_time
async def _finalize_stream() -> None:
from ._types import AgentResponse
try:
response: AgentResponse[Any] = await result_stream.get_final_response()
duration = duration_state.get("duration")
response_attributes = _get_response_attributes(
attributes,
response,
capture_response_id=INNER_RESPONSE_ID_CAPTURED_FIELD
not in inner_response_telemetry_captured_fields,
capture_usage=INNER_USAGE_CAPTURED_FIELD not in inner_response_telemetry_captured_fields,
)
_apply_accumulated_usage(response_attributes, inner_response_telemetry_captured_fields)
_capture_response(span=span, attributes=response_attributes, duration=duration)
if (
OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED
and isinstance(response, AgentResponse)
and response.messages
):
_capture_messages(
span=span,
provider_name=provider_name,
messages=response.messages,
output=True,
)
except Exception as exception:
capture_exception(span=span, exception=exception, timestamp=time_ns())
finally:
INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.reset(inner_response_telemetry_captured_fields_token)
INNER_ACCUMULATED_USAGE.reset(inner_accumulated_usage_token)
_close_span()
wrapped_stream: ResponseStream[AgentResponseUpdate, AgentResponse[Any]] = result_stream.with_cleanup_hook(
_record_duration
).with_cleanup_hook(_finalize_stream)
weakref.finalize(wrapped_stream, _close_span)
return wrapped_stream
async def _run() -> AgentResponse[Any]:
try:
with _get_span(attributes=attributes, span_name_attribute=OtelAttr.AGENT_NAME) as span:
if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED and messages:
_capture_messages(
span=span,
provider_name=provider_name,
messages=messages,
system_instructions=_get_instructions_from_options(dict(merged_options)),
)
start_time_stamp = perf_counter()
try:
response: AgentResponse[Any] = await execute()
except Exception as exception:
capture_exception(span=span, exception=exception, timestamp=time_ns())
raise
duration = perf_counter() - start_time_stamp
if response:
response_attributes = _get_response_attributes(
attributes,
response,
capture_response_id=INNER_RESPONSE_ID_CAPTURED_FIELD
not in inner_response_telemetry_captured_fields,
capture_usage=INNER_USAGE_CAPTURED_FIELD not in inner_response_telemetry_captured_fields,
)
_apply_accumulated_usage(response_attributes, inner_response_telemetry_captured_fields)
_capture_response(span=span, attributes=response_attributes, duration=duration)
if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED and response.messages:
_capture_messages(
span=span,
provider_name=provider_name,
messages=response.messages,
output=True,
)
return response # type: ignore[return-value,no-any-return]
finally:
INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.reset(inner_response_telemetry_captured_fields_token)
INNER_ACCUMULATED_USAGE.reset(inner_accumulated_usage_token)
return _run()
@overload
def run(
self,
@@ -1720,12 +1565,14 @@ class AgentTelemetryLayer:
client_kwargs: Mapping[str, Any] | None = None,
) -> Awaitable[AgentResponse[Any]] | ResponseStream[AgentResponseUpdate, AgentResponse[Any]]:
"""Trace agent runs with OpenTelemetry spans and metrics."""
from ._types import merge_chat_options
global OBSERVABILITY_SETTINGS
from ._types import ResponseStream, merge_chat_options
super_run = cast(
"Callable[..., Awaitable[AgentResponse[Any]] | ResponseStream[AgentResponseUpdate, AgentResponse[Any]]]",
super().run, # type: ignore[misc]
)
provider_name = str(self.otel_provider_name)
super_run_kwargs: dict[str, Any] = {
"messages": messages,
"stream": stream,
@@ -1739,21 +1586,156 @@ class AgentTelemetryLayer:
}
if middleware is not None:
super_run_kwargs["middleware"] = middleware
if not OBSERVABILITY_SETTINGS.ENABLED:
return super_run(**super_run_kwargs) # type: ignore[no-any-return]
default_options = dict(getattr(self, "default_options", {}))
merged_client_kwargs = dict(client_kwargs) if client_kwargs is not None else {}
merged_options: dict[str, Any] = merge_chat_options(
default_options, dict(options) if options is not None else {}
)
return self._trace_agent_invocation(
messages=messages,
session=session,
merged_options=merged_options,
client_kwargs=merged_client_kwargs,
stream=stream,
execute=lambda: super_run(**super_run_kwargs),
attributes = _get_span_attributes(
operation_name=OtelAttr.AGENT_INVOKE_OPERATION,
provider_name=provider_name,
agent_id=getattr(self, "id", "unknown"),
agent_name=getattr(self, "name", None) or getattr(self, "id", "unknown"),
agent_description=getattr(self, "description", None),
thread_id=session.service_session_id if session else None,
all_options=merged_options,
**merged_client_kwargs,
)
inner_response_telemetry_captured_fields: set[str] = set()
inner_response_telemetry_captured_fields_token = INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.set(
inner_response_telemetry_captured_fields
)
inner_accumulated_usage_token = INNER_ACCUMULATED_USAGE.set({})
if stream:
try:
run_result: object = super_run(**super_run_kwargs)
if isinstance(run_result, ResponseStream):
result_stream: ResponseStream[AgentResponseUpdate, AgentResponse[Any]] = run_result # pyright: ignore[reportUnknownVariableType]
elif isinstance(run_result, Awaitable):
result_stream = ResponseStream.from_awaitable(run_result) # type: ignore[arg-type] # pyright: ignore[reportArgumentType]
else:
raise RuntimeError("Streaming telemetry requires a ResponseStream result.")
except Exception:
INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.reset(inner_response_telemetry_captured_fields_token)
INNER_ACCUMULATED_USAGE.reset(inner_accumulated_usage_token)
raise
# Create span directly without trace.use_span() context attachment.
# Streaming spans are closed asynchronously in cleanup hooks, which run
# in a different async context than creation — using use_span() would
# cause "Failed to detach context" errors from OpenTelemetry.
operation = attributes.get(OtelAttr.OPERATION, "operation")
span_name = attributes.get(OtelAttr.AGENT_NAME, "unknown")
span = get_tracer().start_span(f"{operation} {span_name}")
span.set_attributes(attributes)
if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED and messages:
_capture_messages(
span=span,
provider_name=provider_name,
messages=messages,
system_instructions=_get_instructions_from_options(merged_options),
)
span_state = {"closed": False}
duration_state: dict[str, float] = {}
start_time = perf_counter()
def _close_span() -> None:
if span_state["closed"]:
return
span_state["closed"] = True
span.end()
def _record_duration() -> None:
duration_state["duration"] = perf_counter() - start_time
async def _finalize_stream() -> None:
from ._types import AgentResponse
try:
response: AgentResponse[Any] = await result_stream.get_final_response()
duration = duration_state.get("duration")
response_attributes = _get_response_attributes(
attributes,
response,
capture_response_id=INNER_RESPONSE_ID_CAPTURED_FIELD
not in inner_response_telemetry_captured_fields,
capture_usage=INNER_USAGE_CAPTURED_FIELD not in inner_response_telemetry_captured_fields,
)
_apply_accumulated_usage(response_attributes, inner_response_telemetry_captured_fields)
_capture_response(span=span, attributes=response_attributes, duration=duration)
if (
OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED
and isinstance(response, AgentResponse)
and response.messages
):
_capture_messages(
span=span,
provider_name=provider_name,
messages=response.messages,
output=True,
)
except Exception as exception:
capture_exception(span=span, exception=exception, timestamp=time_ns())
finally:
INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.reset(inner_response_telemetry_captured_fields_token)
INNER_ACCUMULATED_USAGE.reset(inner_accumulated_usage_token)
_close_span()
# Register a weak reference callback to close the span if stream is garbage collected
# without being consumed. This ensures spans don't leak if users don't consume streams.
wrapped_stream: ResponseStream[AgentResponseUpdate, AgentResponse[Any]] = result_stream.with_cleanup_hook(
_record_duration
).with_cleanup_hook(_finalize_stream)
weakref.finalize(wrapped_stream, _close_span)
return wrapped_stream
async def _run() -> AgentResponse:
try:
with _get_span(attributes=attributes, span_name_attribute=OtelAttr.AGENT_NAME) as span:
if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED and messages:
_capture_messages(
span=span,
provider_name=provider_name,
messages=messages,
system_instructions=_get_instructions_from_options(merged_options),
)
start_time_stamp = perf_counter()
try:
response: AgentResponse[Any] = await super_run(**super_run_kwargs)
except Exception as exception:
capture_exception(span=span, exception=exception, timestamp=time_ns())
raise
duration = perf_counter() - start_time_stamp
if response:
response_attributes = _get_response_attributes(
attributes,
response,
capture_response_id=INNER_RESPONSE_ID_CAPTURED_FIELD
not in inner_response_telemetry_captured_fields,
capture_usage=INNER_USAGE_CAPTURED_FIELD not in inner_response_telemetry_captured_fields,
)
_apply_accumulated_usage(response_attributes, inner_response_telemetry_captured_fields)
_capture_response(span=span, attributes=response_attributes, duration=duration)
if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED and response.messages:
_capture_messages(
span=span,
provider_name=provider_name,
messages=response.messages,
output=True,
)
return response # type: ignore[return-value,no-any-return]
finally:
INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.reset(inner_response_telemetry_captured_fields_token)
INNER_ACCUMULATED_USAGE.reset(inner_accumulated_usage_token)
return _run()
# region Otel Helpers
+7 -491
View File
@@ -3,8 +3,8 @@
import contextlib
import inspect
import json
from collections.abc import AsyncIterable, Awaitable, Callable, MutableSequence, Sequence
from typing import Any, cast
from collections.abc import AsyncIterable, MutableSequence
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
from uuid import uuid4
@@ -18,29 +18,22 @@ from agent_framework import (
AgentResponse,
AgentResponseUpdate,
AgentSession,
ChatContext,
BaseContextProvider,
ChatOptions,
ChatResponse,
ChatResponseUpdate,
Content,
ContextProvider,
FunctionTool,
HistoryProvider,
InMemoryHistoryProvider,
Message,
ResponseStream,
SessionContext,
SlidingWindowStrategy,
SupportsAgentRun,
SupportsChatGetResponse,
TruncationStrategy,
chat_middleware,
tool,
)
from agent_framework._agents import _get_tool_name, _merge_options, _sanitize_agent_name
from agent_framework._mcp import MCPTool, _build_prefixed_mcp_name, _normalize_mcp_name
from agent_framework._middleware import FunctionInvocationContext
from agent_framework.exceptions import AgentInvalidRequestException, ChatClientInvalidResponseException
class _FixedTokenizer:
@@ -75,49 +68,6 @@ class _ConnectedMCPTool(MCPTool):
raise NotImplementedError
class _RecordingHistoryProvider(HistoryProvider):
def __init__(self, source_id: str = "recording_history") -> None:
super().__init__(source_id=source_id)
async def get_messages(
self,
session_id: str | None,
*,
state: dict[str, Any] | None = None,
**kwargs: Any,
) -> list[Message]:
if state is None:
return []
state["get_call_count"] = state.get("get_call_count", 0) + 1
return list(cast(list[Message], state.get("messages", [])))
async def save_messages(
self,
session_id: str | None,
messages: Sequence[Message],
*,
state: dict[str, Any] | None = None,
**kwargs: Any,
) -> None:
if state is None:
return
state["save_call_count"] = state.get("save_call_count", 0) + 1
state.setdefault("messages", []).extend(messages)
class _ResponseIdRecordingHistoryProvider(_RecordingHistoryProvider):
async def after_run(
self,
*,
agent: SupportsAgentRun,
session: AgentSession,
context: SessionContext,
state: dict[str, Any],
) -> None:
state.setdefault("response_ids", []).append(context.response.response_id if context.response else None)
await super().after_run(agent=agent, session=session, context=context, state=state)
def test_agent_session_type(agent_session: AgentSession) -> None:
assert isinstance(agent_session, AgentSession)
@@ -364,413 +314,6 @@ async def test_prepare_run_context_handles_function_kwargs(
assert ctx["client_kwargs"]["session"] is session
async def test_chat_agent_persists_history_per_service_call(
chat_client_base: SupportsChatGetResponse,
) -> None:
provider = _RecordingHistoryProvider()
@tool(name="lookup_weather", approval_mode="never_require")
def lookup_weather(location: str) -> str:
return f"Weather in {location}: sunny"
session = AgentSession()
session.state[provider.source_id] = {
"messages": [
Message(role="user", text="Earlier question"),
Message(role="assistant", text="Earlier answer"),
]
}
chat_client_base.run_responses = [
ChatResponse(
messages=Message(
role="assistant",
contents=[
Content.from_function_call(
call_id="call_1",
name="lookup_weather",
arguments='{"location": "Seattle"}',
)
],
),
response_id="resp_call_1",
),
ChatResponse(messages=Message(role="assistant", text="It is sunny in Seattle."), response_id="resp_call_2"),
]
agent = Agent(
client=chat_client_base,
tools=[lookup_weather],
context_providers=[provider],
require_per_service_call_history_persistence=True,
)
result = await agent.run("What's the weather in Seattle?", session=session)
provider_state = session.state[provider.source_id]
stored_messages = cast(list[Message], provider_state["messages"])
assert result.text == "It is sunny in Seattle."
assert result.response_id is None
assert chat_client_base.call_count == 2
assert provider_state["get_call_count"] == 2
assert provider_state["save_call_count"] == 2
assert stored_messages[-1].text == "It is sunny in Seattle."
assert session.service_session_id is None
async def test_chat_agent_persists_history_per_service_call_streaming(
chat_client_base: SupportsChatGetResponse,
) -> None:
provider = _RecordingHistoryProvider()
@tool(name="lookup_weather", approval_mode="never_require")
def lookup_weather(location: str) -> str:
return f"Weather in {location}: sunny"
session = AgentSession()
session.state[provider.source_id] = {
"messages": [
Message(role="user", text="Earlier question"),
Message(role="assistant", text="Earlier answer"),
]
}
chat_client_base.streaming_responses = [
[
ChatResponseUpdate(
contents=[
Content.from_function_call(
call_id="call_1",
name="lookup_weather",
arguments='{"location": "Seattle"}',
)
],
role="assistant",
finish_reason="stop",
response_id="resp_call_1",
)
],
[
ChatResponseUpdate(
contents=[Content.from_text("It is sunny in Seattle.")],
role="assistant",
finish_reason="stop",
response_id="resp_call_2",
)
],
]
agent = Agent(
client=chat_client_base,
tools=[lookup_weather],
context_providers=[provider],
require_per_service_call_history_persistence=True,
)
stream = agent.run("What's the weather in Seattle?", session=session, stream=True)
async for _ in stream:
pass
result = await stream.get_final_response()
provider_state = session.state[provider.source_id]
stored_messages = cast(list[Message], provider_state["messages"])
assert result.text == "It is sunny in Seattle."
assert result.response_id is None
assert chat_client_base.call_count == 2
assert provider_state["get_call_count"] == 2
assert provider_state["save_call_count"] == 2
assert stored_messages[-1].text == "It is sunny in Seattle."
assert session.service_session_id is None
async def test_streaming_per_service_call_persistence_hides_response_id_from_after_run(
chat_client_base: SupportsChatGetResponse,
) -> None:
provider = _ResponseIdRecordingHistoryProvider()
@tool(name="lookup_weather", approval_mode="never_require")
def lookup_weather(location: str) -> str:
return f"Weather in {location}: sunny"
session = AgentSession()
session.state[provider.source_id] = {"messages": []}
chat_client_base.streaming_responses = [
[
ChatResponseUpdate(
contents=[
Content.from_function_call(
call_id="call_1",
name="lookup_weather",
arguments='{"location": "Seattle"}',
)
],
role="assistant",
finish_reason="stop",
response_id="resp_call_1",
)
],
[
ChatResponseUpdate(
contents=[Content.from_text("It is sunny in Seattle.")],
role="assistant",
finish_reason="stop",
response_id="resp_call_2",
)
],
]
agent = Agent(
client=chat_client_base,
tools=[lookup_weather],
context_providers=[provider],
require_per_service_call_history_persistence=True,
)
stream = agent.run("What's the weather in Seattle?", session=session, stream=True)
async for _ in stream:
pass
result = await stream.get_final_response()
provider_state = session.state[provider.source_id]
assert result.response_id is None
assert provider_state["response_ids"] == [None, None]
async def test_per_service_call_persistence_uses_real_service_storage_when_client_stores_by_default(
chat_client_base: SupportsChatGetResponse,
) -> None:
provider = _RecordingHistoryProvider()
@tool(name="lookup_weather", approval_mode="never_require")
def lookup_weather(location: str) -> str:
return f"Weather in {location}: sunny"
chat_client_base.STORES_BY_DEFAULT = True # type: ignore[attr-defined]
session = AgentSession()
session.state[provider.source_id] = {"messages": []}
chat_client_base.run_responses = [
ChatResponse(
messages=Message(
role="assistant",
contents=[
Content.from_function_call(
call_id="call_1",
name="lookup_weather",
arguments='{"location": "Seattle"}',
)
],
),
conversation_id="resp_service_managed",
response_id="resp_call_1",
),
ChatResponse(
messages=Message(role="assistant", text="It is sunny in Seattle."),
conversation_id="resp_service_managed",
response_id="resp_call_2",
),
]
agent = Agent(
client=chat_client_base,
tools=[lookup_weather],
context_providers=[provider],
require_per_service_call_history_persistence=True,
)
result = await agent.run("What's the weather in Seattle?", session=session)
provider_state = session.state[provider.source_id]
assert result.text == "It is sunny in Seattle."
assert result.response_id == "resp_call_2"
assert chat_client_base.call_count == 2
assert "get_call_count" not in provider_state
assert "save_call_count" not in provider_state
assert session.service_session_id == "resp_service_managed"
async def test_service_storage_updates_session_handle_per_service_call_before_non_streaming_failure(
chat_client_base: SupportsChatGetResponse,
) -> None:
provider = _RecordingHistoryProvider()
@tool(name="lookup_weather", approval_mode="never_require")
def lookup_weather(location: str) -> str:
return f"Weather in {location}: sunny"
chat_client_base.STORES_BY_DEFAULT = True # type: ignore[attr-defined]
session = AgentSession()
session.state[provider.source_id] = {"messages": []}
first_response = ChatResponse(
messages=Message(
role="assistant",
contents=[
Content.from_function_call(
call_id="call_1",
name="lookup_weather",
arguments='{"location": "Seattle"}',
)
],
),
conversation_id="resp_call_1",
response_id="resp_call_1",
)
mock_get_non_streaming_response = AsyncMock(
side_effect=[first_response, RuntimeError("service down")],
)
agent = Agent(
client=chat_client_base,
tools=[lookup_weather],
context_providers=[provider],
require_per_service_call_history_persistence=True,
)
with (
patch.object(chat_client_base, "_get_non_streaming_response", new=mock_get_non_streaming_response),
pytest.raises(RuntimeError, match="service down"),
):
await agent.run("What's the weather in Seattle?", session=session)
assert mock_get_non_streaming_response.await_count == 2
assert session.service_session_id == "resp_call_1"
async def test_service_storage_updates_session_handle_per_service_call_before_streaming_failure(
chat_client_base: SupportsChatGetResponse,
) -> None:
provider = _RecordingHistoryProvider()
@tool(name="lookup_weather", approval_mode="never_require")
def lookup_weather(location: str) -> str:
return f"Weather in {location}: sunny"
chat_client_base.STORES_BY_DEFAULT = True # type: ignore[attr-defined]
session = AgentSession()
session.state[provider.source_id] = {"messages": []}
async def _first_stream_updates() -> AsyncIterable[ChatResponseUpdate]:
yield ChatResponseUpdate(
contents=[
Content.from_function_call(
call_id="call_1",
name="lookup_weather",
arguments='{"location": "Seattle"}',
)
],
role="assistant",
finish_reason="stop",
)
def _finalize_first_stream(_updates: Sequence[ChatResponseUpdate]) -> ChatResponse[Any]:
return ChatResponse(
messages=Message(
role="assistant",
contents=[
Content.from_function_call(
call_id="call_1",
name="lookup_weather",
arguments='{"location": "Seattle"}',
)
],
),
conversation_id="resp_call_1",
response_id="resp_call_1",
)
first_stream = ResponseStream(_first_stream_updates(), finalizer=_finalize_first_stream)
mock_get_streaming_response = MagicMock(side_effect=[first_stream, RuntimeError("service down")])
agent = Agent(
client=chat_client_base,
tools=[lookup_weather],
context_providers=[provider],
require_per_service_call_history_persistence=True,
)
with (
patch.object(chat_client_base, "_get_streaming_response", new=mock_get_streaming_response),
pytest.raises(RuntimeError, match="service down"),
):
stream = agent.run("What's the weather in Seattle?", session=session, stream=True)
async for _ in stream:
pass
assert mock_get_streaming_response.call_count == 2
assert session.service_session_id == "resp_call_1"
async def test_chat_agent_without_per_service_call_persistence_preserves_response_id(
chat_client_base: SupportsChatGetResponse,
) -> None:
chat_client_base.run_responses = [
ChatResponse(
messages=Message(role="assistant", text="Hello"),
response_id="resp_call_1",
)
]
agent = Agent(
client=chat_client_base,
context_providers=[InMemoryHistoryProvider()],
)
result = await agent.run("Hello", session=AgentSession(), options={"store": False})
assert result.response_id == "resp_call_1"
async def test_per_service_call_persistence_rejects_real_service_conversation_id(
chat_client_base: SupportsChatGetResponse,
) -> None:
provider = _RecordingHistoryProvider()
chat_client_base.STORES_BY_DEFAULT = True # type: ignore[attr-defined]
session = AgentSession()
session.state[provider.source_id] = {"messages": []}
chat_client_base.run_responses = [
ChatResponse(
messages=Message(role="assistant", text="Hello"),
conversation_id="resp_service_managed",
)
]
agent = Agent(
client=chat_client_base,
context_providers=[provider],
require_per_service_call_history_persistence=True,
)
with pytest.raises(
ChatClientInvalidResponseException,
match="require_per_service_call_history_persistence cannot be used",
):
await agent.run("Hello", session=session, options={"store": False})
async def test_per_service_call_persistence_rejects_existing_conversation_id_when_service_not_storing_history(
chat_client_base: SupportsChatGetResponse,
) -> None:
provider = _RecordingHistoryProvider()
session = AgentSession()
session.state[provider.source_id] = {"messages": []}
agent = Agent(
client=chat_client_base,
context_providers=[provider],
require_per_service_call_history_persistence=True,
)
with pytest.raises(
AgentInvalidRequestException,
match="require_per_service_call_history_persistence cannot be used",
):
await agent.run("Hello", session=session, options={"store": False, "conversation_id": "existing_conversation"})
async def test_chat_client_agent_run_with_session(chat_client_base: SupportsChatGetResponse) -> None:
mock_response = ChatResponse(
messages=[Message(role="assistant", contents=[Content.from_text("test response")])],
@@ -1043,7 +586,7 @@ async def test_chat_client_agent_author_name_is_used_from_response(
# Mock context provider for testing
class MockContextProvider(ContextProvider):
class MockContextProvider(BaseContextProvider):
def __init__(self, messages: list[Message] | None = None) -> None:
super().__init__(source_id="mock")
self.context_messages = messages
@@ -2180,7 +1723,7 @@ async def test_agent_create_session_with_context_providers(
):
"""Test that create_session works when context_providers are set on the agent."""
class TestContextProvider(ContextProvider):
class TestContextProvider(BaseContextProvider):
def __init__(self):
super().__init__(source_id="test")
@@ -2255,7 +1798,7 @@ async def test_chat_agent_context_provider_adds_tools_when_agent_has_none(
"""A tool provided by context."""
return text
class ToolContextProvider(ContextProvider):
class ToolContextProvider(BaseContextProvider):
def __init__(self):
super().__init__(source_id="tool-context")
@@ -2284,7 +1827,7 @@ async def test_chat_agent_context_provider_adds_instructions_when_agent_has_none
):
"""Test that context provider instructions are used when agent has no default instructions."""
class InstructionContextProvider(ContextProvider):
class InstructionContextProvider(BaseContextProvider):
def __init__(self):
super().__init__(source_id="instruction-context")
@@ -2306,33 +1849,6 @@ async def test_chat_agent_context_provider_adds_instructions_when_agent_has_none
assert options.get("instructions") == "Context-provided instructions"
async def test_chat_agent_context_provider_adds_middleware_when_agent_has_none(
chat_client_base: SupportsChatGetResponse,
) -> None:
"""Test that context provider middleware is collected during preparation."""
@chat_middleware
async def context_chat_middleware(context: ChatContext, call_next: Callable[[], Awaitable[None]]) -> None:
await call_next()
class MiddlewareContextProvider(ContextProvider):
def __init__(self) -> None:
super().__init__(source_id="middleware-context")
async def before_run(self, *, agent, session, context, state) -> None:
context.extend_middleware("middleware-context", context_chat_middleware)
agent = Agent(client=chat_client_base, context_providers=[MiddlewareContextProvider()])
session_context, _ = await agent._prepare_session_and_messages( # type: ignore[reportPrivateUsage]
session=None,
input_messages=[Message(role="user", text="Hello")],
)
assert session_context.middleware["middleware-context"] == [context_chat_middleware]
assert session_context.get_middleware() == [context_chat_middleware]
# region STORES_BY_DEFAULT tests
@@ -15,7 +15,6 @@ from agent_framework import (
ChatResponse,
ChatResponseUpdate,
Content,
ContextProvider,
FunctionInvocationContext,
FunctionMiddleware,
FunctionTool,
@@ -465,31 +464,6 @@ class TestChatAgentMultipleMiddlewareOrdering:
expected_order = ["class_agent_before", "function_agent_before", "function_agent_after", "class_agent_after"]
assert execution_order == expected_order
async def test_provider_added_agent_middleware_is_rejected(self, chat_client_base: "MockBaseChatClient") -> None:
"""Test provider-added agent middleware is rejected explicitly."""
@agent_middleware
async def provider_middleware(context: AgentContext, call_next: Callable[[], Awaitable[None]]) -> None:
await call_next()
class ProviderMiddlewareContextProvider(ContextProvider):
def __init__(self) -> None:
super().__init__(source_id="provider-middleware")
async def before_run(self, *, agent, session, context, state) -> None:
context.extend_middleware(self.source_id, provider_middleware)
agent = Agent(
client=chat_client_base,
context_providers=[ProviderMiddlewareContextProvider()],
)
with pytest.raises(
MiddlewareException,
match="Context providers may only add chat or function middleware",
):
await agent.run([Message(role="user", text="test message")])
# region Tool Functions for Testing
@@ -2092,121 +2066,6 @@ class TestChatAgentChatMiddleware:
"agent_middleware_after",
]
async def test_provider_added_chat_and_function_middleware_are_forwarded(
self, chat_client_base: "MockBaseChatClient"
) -> None:
"""Test provider-added chat and function middleware forwarding and ordering."""
execution_order: list[str] = []
@chat_middleware
async def constructor_chat_middleware(context: ChatContext, call_next: Callable[[], Awaitable[None]]) -> None:
execution_order.append("constructor_chat_before")
await call_next()
execution_order.append("constructor_chat_after")
@chat_middleware
async def provider_chat_middleware(context: ChatContext, call_next: Callable[[], Awaitable[None]]) -> None:
execution_order.append("provider_chat_before")
await call_next()
execution_order.append("provider_chat_after")
@chat_middleware
async def run_chat_middleware(context: ChatContext, call_next: Callable[[], Awaitable[None]]) -> None:
execution_order.append("run_chat_before")
await call_next()
execution_order.append("run_chat_after")
@function_middleware
async def constructor_function_middleware(
context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]]
) -> None:
execution_order.append("constructor_function_before")
await call_next()
execution_order.append("constructor_function_after")
@function_middleware
async def provider_function_middleware(
context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]]
) -> None:
execution_order.append("provider_function_before")
await call_next()
execution_order.append("provider_function_after")
@function_middleware
async def run_function_middleware(
context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]]
) -> None:
execution_order.append("run_function_before")
await call_next()
execution_order.append("run_function_after")
class ProviderMiddlewareContextProvider(ContextProvider):
def __init__(self) -> None:
super().__init__(source_id="provider-middleware")
async def before_run(self, *, agent, session, context, state) -> None:
context.extend_middleware(
self.source_id,
[
provider_chat_middleware,
provider_function_middleware,
],
)
chat_client_base.run_responses = [
ChatResponse(
messages=[
Message(
role="assistant",
contents=[
Content.from_function_call(
call_id="call_provider",
name="sample_tool_function",
arguments='{"location": "Seattle"}',
)
],
)
]
),
ChatResponse(messages=[Message(role="assistant", text="Final response")]),
]
agent = Agent(
client=chat_client_base,
middleware=[constructor_chat_middleware, constructor_function_middleware],
context_providers=[ProviderMiddlewareContextProvider()],
tools=[sample_tool_function],
)
response = await agent.run(
[Message(role="user", text="Get weather for Seattle")],
middleware=[run_chat_middleware, run_function_middleware],
)
assert response is not None
assert chat_client_base.call_count == 2
assert response.messages[-1].text == "Final response"
assert execution_order == [
"constructor_chat_before",
"run_chat_before",
"provider_chat_before",
"provider_chat_after",
"run_chat_after",
"constructor_chat_after",
"constructor_function_before",
"run_function_before",
"provider_function_before",
"provider_function_after",
"run_function_after",
"constructor_function_after",
"constructor_chat_before",
"run_chat_before",
"provider_chat_before",
"provider_chat_after",
"run_chat_after",
"constructor_chat_after",
]
async def test_agent_middleware_can_access_and_override_options(self) -> None:
"""Test that agent middleware can access and override runtime options."""
captured_options: dict[str, Any] = {}
@@ -1,26 +1,16 @@
# Copyright (c) Microsoft. All rights reserved.
import json
from collections.abc import Awaitable, Callable, Sequence
from collections.abc import Sequence
import pytest
from agent_framework import (
AgentContext,
from agent_framework import Message
from agent_framework._sessions import (
AgentSession,
BaseContextProvider,
BaseHistoryProvider,
ChatContext,
ContextProvider,
HistoryProvider,
InMemoryHistoryProvider,
Message,
SessionContext,
agent_middleware,
chat_middleware,
)
from agent_framework._sessions import LOCAL_HISTORY_CONVERSATION_ID, is_local_history_conversation_id
from agent_framework.exceptions import MiddlewareException
# ---------------------------------------------------------------------------
# SessionContext tests
@@ -112,50 +102,6 @@ class TestSessionContext:
ctx.extend_instructions("sys", ["Be helpful", "Be concise"])
assert ctx.instructions == ["Be helpful", "Be concise"]
def test_extend_middleware_creates_key_and_appends(self) -> None:
ctx = SessionContext(input_messages=[])
@chat_middleware
async def first_middleware(context: ChatContext, call_next: Callable[[], Awaitable[None]]) -> None:
await call_next()
@chat_middleware
async def second_middleware(context: ChatContext, call_next: Callable[[], Awaitable[None]]) -> None:
await call_next()
ctx.extend_middleware("rag", first_middleware)
ctx.extend_middleware("rag", [second_middleware])
assert ctx.middleware["rag"] == [first_middleware, second_middleware]
assert ctx.get_middleware() == [first_middleware, second_middleware]
def test_extend_middleware_preserves_source_order(self) -> None:
ctx = SessionContext(input_messages=[])
@chat_middleware
async def first_middleware(context: ChatContext, call_next: Callable[[], Awaitable[None]]) -> None:
await call_next()
@chat_middleware
async def second_middleware(context: ChatContext, call_next: Callable[[], Awaitable[None]]) -> None:
await call_next()
ctx.extend_middleware("a", first_middleware)
ctx.extend_middleware("b", second_middleware)
assert list(ctx.middleware.keys()) == ["a", "b"]
assert ctx.get_middleware() == [first_middleware, second_middleware]
def test_extend_middleware_rejects_agent_middleware(self) -> None:
ctx = SessionContext(input_messages=[])
@agent_middleware
async def provider_agent_middleware(context: AgentContext, call_next: Callable[[], Awaitable[None]]) -> None:
await call_next()
with pytest.raises(MiddlewareException, match="Context providers may only add chat or function middleware"):
ctx.extend_middleware("rag", provider_agent_middleware)
def test_get_messages_all(self) -> None:
ctx = SessionContext(input_messages=[])
ctx.extend_messages("a", [Message(role="user", contents=["a"])])
@@ -208,58 +154,37 @@ class TestSessionContext:
ctx._response = resp
assert ctx.response is resp
def test_local_history_conversation_id_sentinel(self) -> None:
assert is_local_history_conversation_id(LOCAL_HISTORY_CONVERSATION_ID) is True
assert is_local_history_conversation_id("some_other_id") is False
# ---------------------------------------------------------------------------
# ContextProvider tests
# BaseContextProvider tests
# ---------------------------------------------------------------------------
class TestContextProvider:
class TestContextProviderBase:
def test_source_id_required(self) -> None:
provider = ContextProvider(source_id="test")
provider = BaseContextProvider(source_id="test")
assert provider.source_id == "test"
async def test_before_run_is_noop(self) -> None:
provider = ContextProvider(source_id="test")
provider = BaseContextProvider(source_id="test")
session = AgentSession()
ctx = SessionContext(input_messages=[])
# Should not raise
await provider.before_run(agent=None, session=session, context=ctx, state={}) # type: ignore[arg-type]
async def test_after_run_is_noop(self) -> None:
provider = ContextProvider(source_id="test")
provider = BaseContextProvider(source_id="test")
session = AgentSession()
ctx = SessionContext(input_messages=[])
await provider.after_run(agent=None, session=session, context=ctx, state={}) # type: ignore[arg-type]
# ---------------------------------------------------------------------------
# Deprecated provider alias tests
# BaseHistoryProvider tests
# ---------------------------------------------------------------------------
class TestDeprecatedProviderAliases:
def test_base_context_provider_warns_and_is_compatible(self) -> None:
with pytest.warns(DeprecationWarning, match="BaseContextProvider is deprecated. Use ContextProvider instead."):
provider = BaseContextProvider(source_id="test")
assert isinstance(provider, ContextProvider)
def test_base_provider_aliases_preserve_subtyping(self) -> None:
assert issubclass(BaseContextProvider, ContextProvider)
assert issubclass(BaseHistoryProvider, HistoryProvider)
# ---------------------------------------------------------------------------
# HistoryProvider tests
# ---------------------------------------------------------------------------
class ConcreteHistoryProvider(HistoryProvider):
class ConcreteHistoryProvider(BaseHistoryProvider):
"""Concrete test implementation."""
def __init__(self, source_id: str, stored_messages: list[Message] | None = None, **kwargs) -> None:
+1 -83
View File
@@ -43,7 +43,7 @@ from agent_framework._types import (
add_usage_details,
validate_tool_mode,
)
from agent_framework.exceptions import AdditionItemMismatch, ContentError
from agent_framework.exceptions import ContentError
@fixture
@@ -1526,88 +1526,6 @@ def test_text_reasoning_content_iadd_coverage():
assert t1.text == "Thinking 1 Thinking 2"
def test_text_reasoning_content_add_preserves_id():
"""Test that coalescing text_reasoning Content preserves the id field."""
t1 = Content.from_text_reasoning(id="rs_abc123", text="Thinking part 1")
t2 = Content.from_text_reasoning(id="rs_abc123", text=" part 2")
result = t1 + t2
assert result.text == "Thinking part 1 part 2"
assert result.id == "rs_abc123"
def test_text_reasoning_content_add_id_fallback_to_other():
"""Test that coalescing falls back to other's id when self has no id."""
t1 = Content.from_text_reasoning(text="Thinking part 1")
t2 = Content.from_text_reasoning(id="rs_abc123", text=" part 2")
result = t1 + t2
assert result.id == "rs_abc123"
def test_text_reasoning_content_add_preserves_id_with_encrypted_content():
"""Test that id and encrypted_content both survive coalescing for round-trip."""
t1 = Content.from_text_reasoning(
id="rs_abc123",
text="Thinking",
additional_properties={"encrypted_content": "enc_blob_data"},
)
t2 = Content.from_text_reasoning(id="rs_abc123", text=" more")
result = t1 + t2
assert result.text == "Thinking more"
assert result.id == "rs_abc123"
assert result.additional_properties.get("encrypted_content") == "enc_blob_data"
def test_text_reasoning_content_add_conflicting_ids_raises():
"""Test that coalescing text_reasoning Content with different ids raises AdditionItemMismatch."""
t1 = Content.from_text_reasoning(id="rs_abc123", text="Thinking part 1")
t2 = Content.from_text_reasoning(id="rs_xyz789", text=" part 2")
with pytest.raises(AdditionItemMismatch, match="different ids"):
t1 + t2
def test_text_reasoning_content_add_neither_has_id():
"""Test that coalescing text_reasoning Content when neither has an id results in None id."""
t1 = Content.from_text_reasoning(text="Thinking part 1")
t2 = Content.from_text_reasoning(text=" part 2")
result = t1 + t2
assert result.text == "Thinking part 1 part 2"
assert result.id is None
def test_coalesce_text_reasoning_with_different_ids():
"""Test that _coalesce_text_content keeps separate text_reasoning items when IDs differ.
Regression test: streaming responses can produce multiple text_reasoning
segments with distinct IDs. These must not be merged into one.
"""
from agent_framework._types import _coalesce_text_content
contents = [
Content.from_text_reasoning(id="rs_aaa", text="Thinking A1"),
Content.from_text_reasoning(id="rs_aaa", text=" A2"),
Content.from_text_reasoning(id="rs_bbb", text="Thinking B1"),
Content.from_text_reasoning(id="rs_bbb", text=" B2"),
]
_coalesce_text_content(contents, "text_reasoning")
assert len(contents) == 2
assert contents[0].id == "rs_aaa"
assert contents[0].text == "Thinking A1 A2"
assert contents[1].id == "rs_bbb"
assert contents[1].text == "Thinking B1 B2"
def test_comprehensive_to_dict_exclude_options():
"""Test to_dict methods with various exclude options for better coverage."""
@@ -17,9 +17,9 @@ from typing import TYPE_CHECKING, Any, ClassVar, Generic, cast
from agent_framework import (
AGENT_FRAMEWORK_USER_AGENT,
AgentMiddlewareLayer,
BaseContextProvider,
ChatAndFunctionMiddlewareTypes,
ChatMiddlewareLayer,
ContextProvider,
FunctionInvocationConfiguration,
FunctionInvocationLayer,
FunctionTool,
@@ -50,8 +50,8 @@ else:
if TYPE_CHECKING:
from agent_framework import (
Agent,
BaseContextProvider,
ChatAndFunctionMiddlewareTypes,
ContextProvider,
MiddlewareTypes,
ToolTypes,
)
@@ -224,9 +224,8 @@ class RawFoundryAgentChatClient( # type: ignore[misc]
instructions: str | None = None,
tools: ToolTypes | Callable[..., Any] | Sequence[ToolTypes | Callable[..., Any]] | None = None,
default_options: FoundryAgentOptionsT | Mapping[str, Any] | None = None,
context_providers: Sequence[ContextProvider] | None = None,
context_providers: Sequence[BaseContextProvider] | None = None,
middleware: Sequence[MiddlewareTypes] | None = None,
require_per_service_call_history_persistence: bool = False,
function_invocation_configuration: FunctionInvocationConfiguration | None = None,
compaction_strategy: CompactionStrategy | None = None,
tokenizer: TokenizerProtocol | None = None,
@@ -247,7 +246,6 @@ class RawFoundryAgentChatClient( # type: ignore[misc]
tools=function_tools,
context_providers=context_providers,
middleware=middleware,
require_per_service_call_history_persistence=require_per_service_call_history_persistence,
client_type=cast(type[RawFoundryAgentChatClient], self.__class__),
id=id,
name=self.agent_name if name is None else name,
@@ -470,7 +468,7 @@ class RawFoundryAgent( # type: ignore[misc]
project_client: AIProjectClient | None = None,
allow_preview: bool | None = None,
tools: FunctionTool | Callable[..., Any] | Sequence[FunctionTool | Callable[..., Any]] | None = None,
context_providers: Sequence[ContextProvider] | None = None,
context_providers: Sequence[BaseContextProvider] | None = None,
middleware: Sequence[MiddlewareTypes] | None = None,
client_type: type[RawFoundryAgentChatClient] | None = None,
env_file_path: str | None = None,
@@ -480,7 +478,6 @@ class RawFoundryAgent( # type: ignore[misc]
description: str | None = None,
instructions: str | None = None,
default_options: FoundryAgentOptionsT | Mapping[str, Any] | None = None,
require_per_service_call_history_persistence: bool = False,
function_invocation_configuration: FunctionInvocationConfiguration | None = None,
compaction_strategy: CompactionStrategy | None = None,
tokenizer: TokenizerProtocol | None = None,
@@ -510,8 +507,6 @@ class RawFoundryAgent( # type: ignore[misc]
description: Optional local description for the local agent wrapper.
instructions: Optional instructions for the local agent wrapper.
default_options: Default chat options for the local agent wrapper.
require_per_service_call_history_persistence: Whether to require per-service-call
chat history persistence when using local history providers.
function_invocation_configuration: Optional function invocation configuration override.
compaction_strategy: Optional agent-level in-run compaction override.
tokenizer: Optional agent-level tokenizer override.
@@ -553,7 +548,6 @@ class RawFoundryAgent( # type: ignore[misc]
default_options=cast(FoundryAgentOptionsT | None, default_options),
context_providers=context_providers,
middleware=middleware,
require_per_service_call_history_persistence=require_per_service_call_history_persistence,
compaction_strategy=compaction_strategy,
tokenizer=tokenizer,
additional_properties=dict(additional_properties) if additional_properties is not None else None,
@@ -667,7 +661,7 @@ class FoundryAgent( # type: ignore[misc]
project_client: AIProjectClient | None = None,
allow_preview: bool | None = None,
tools: FunctionTool | Callable[..., Any] | Sequence[FunctionTool | Callable[..., Any]] | None = None,
context_providers: Sequence[ContextProvider] | None = None,
context_providers: Sequence[BaseContextProvider] | None = None,
middleware: Sequence[MiddlewareTypes] | None = None,
client_type: type[RawFoundryAgentChatClient] | None = None,
env_file_path: str | None = None,
@@ -677,7 +671,6 @@ class FoundryAgent( # type: ignore[misc]
description: str | None = None,
instructions: str | None = None,
default_options: FoundryAgentOptionsT | Mapping[str, Any] | None = None,
require_per_service_call_history_persistence: bool = False,
function_invocation_configuration: FunctionInvocationConfiguration | None = None,
compaction_strategy: CompactionStrategy | None = None,
tokenizer: TokenizerProtocol | None = None,
@@ -703,8 +696,6 @@ class FoundryAgent( # type: ignore[misc]
description: Optional local description for the local agent wrapper.
instructions: Optional instructions for the local agent wrapper.
default_options: Default chat options for the local agent wrapper.
require_per_service_call_history_persistence: Whether to require per-service-call
chat history persistence when using local history providers.
function_invocation_configuration: Optional function invocation configuration override.
compaction_strategy: Optional agent-level in-run compaction override.
tokenizer: Optional agent-level tokenizer override.
@@ -728,7 +719,6 @@ class FoundryAgent( # type: ignore[misc]
description=description,
instructions=instructions,
default_options=default_options,
require_per_service_call_history_persistence=require_per_service_call_history_persistence,
function_invocation_configuration=function_invocation_configuration,
compaction_strategy=compaction_strategy,
tokenizer=tokenizer,
@@ -1,9 +1,9 @@
# Copyright (c) Microsoft. All rights reserved.
"""Foundry Memory Context Provider using ContextProvider.
"""Foundry Memory Context Provider using BaseContextProvider.
This module provides ``FoundryMemoryProvider``, built on
:class:`ContextProvider`.
:class:`BaseContextProvider`.
"""
from __future__ import annotations
@@ -16,7 +16,7 @@ from typing import TYPE_CHECKING, Any, ClassVar
from agent_framework import (
AGENT_FRAMEWORK_USER_AGENT,
AgentSession,
ContextProvider,
BaseContextProvider,
Message,
SessionContext,
load_settings,
@@ -46,8 +46,8 @@ class FoundryProjectSettings(TypedDict, total=False):
project_endpoint: str | None
class FoundryMemoryProvider(ContextProvider):
"""Foundry Memory context provider using the new ContextProvider hooks pattern.
class FoundryMemoryProvider(BaseContextProvider):
"""Foundry Memory context provider using the new BaseContextProvider hooks pattern.
Integrates Azure AI Foundry Memory Store for persistent semantic memory,
searching and storing memories via the Azure AI Projects SDK.

Some files were not shown because too many files have changed in this diff Show More