Compare commits

..
Author SHA1 Message Date
Tao Chen 20ac21c780 Fix tests 2026-06-01 13:25:54 -07:00
Tao Chen de5b4d619a Make unused tool as comment 2026-06-01 13:15:08 -07:00
Tao Chen 98fbaf2481 Resolve conflict 2026-06-01 13:08:33 -07:00
Tao Chen 0fc5600ae2 Fix toolbox consent flow in hosted agent 2026-06-01 12:59:01 -07:00
103 changed files with 1060 additions and 5317 deletions
@@ -1,64 +0,0 @@
name: Free runner disk space
description: |
Reclaims disk space on GitHub-hosted Ubuntu runners by removing
pre-installed toolchains we do not use (Android SDK, GHC/Haskell,
CodeQL bundle), Docker images, and swap. Also relocates the
NuGet package cache to /mnt (which has ~75 GB free vs ~14 GB
on /). No-op on non-Linux runners.
runs:
using: composite
steps:
- name: Free disk space (Linux only)
if: runner.os == 'Linux'
shell: bash
run: |
set -euo pipefail
echo "::group::Disk usage before cleanup"
df -h /
echo "::endgroup::"
# Remove pre-installed toolchains we never use on this repo's
# dotnet/python jobs. These reclaim ~25-30 GB on ubuntu-latest.
sudo rm -rf \
/usr/local/lib/android \
/usr/share/dotnet/sdk/NuGetFallbackFolder \
/opt/ghc \
/usr/local/.ghcup \
/opt/hostedtoolcache/CodeQL \
/opt/hostedtoolcache/PyPy \
/opt/hostedtoolcache/Ruby \
/opt/hostedtoolcache/go \
/usr/local/share/boost \
/usr/local/share/powershell \
/usr/local/share/chromium \
/usr/local/share/vcpkg \
/usr/local/lib/heroku \
"${AGENT_TOOLSDIRECTORY:-/opt/hostedtoolcache}/PyPy" \
"${AGENT_TOOLSDIRECTORY:-/opt/hostedtoolcache}/Ruby" \
"${AGENT_TOOLSDIRECTORY:-/opt/hostedtoolcache}/go" || true
# Drop docker images shipped on the runner; jobs that need
# docker pull what they need fresh.
if command -v docker >/dev/null 2>&1; then
sudo docker image prune --all --force >/dev/null 2>&1 || true
fi
# Disable swap to free its backing file.
sudo swapoff -a || true
sudo rm -f /mnt/swapfile /swapfile || true
echo "::group::Disk usage after cleanup"
df -h /
echo "::endgroup::"
- name: Relocate NuGet package cache to /mnt (Linux only)
if: runner.os == 'Linux'
shell: bash
run: |
set -euo pipefail
sudo mkdir -p /mnt/nuget
sudo chown -R "$USER":"$USER" /mnt/nuget
echo "NUGET_PACKAGES=/mnt/nuget" >> "$GITHUB_ENV"
echo "Relocated NuGet package cache to /mnt/nuget"
df -h /mnt || true
+9 -12
View File
@@ -63,22 +63,19 @@ function buildLimitMessage({ author, exemptLabelName, maxOpenPrs, openPrCount })
}
async function getOpenPrCount({ github, owner, repo, author, pullRequestNumber }) {
const openPullRequests = await github.paginate(github.rest.pulls.list, {
owner,
repo,
state: 'open',
const query = `repo:${owner}/${repo} is:pr is:open author:${author}`;
const response = await github.rest.search.issuesAndPullRequests({
q: query,
per_page: 100,
});
const authorOpenPullRequestNumbers = openPullRequests
.filter((pullRequest) => pullRequest.user?.login === author)
.map((pullRequest) => pullRequest.number);
const currentPrIsOpen = authorOpenPullRequestNumbers.includes(pullRequestNumber);
const existingOpenPrCount = currentPrIsOpen
? authorOpenPullRequestNumbers.length - 1
: authorOpenPullRequestNumbers.length;
const indexedPrNumbers = response.data.items.map((item) => item.number);
const currentPrIsIndexed = indexedPrNumbers.includes(pullRequestNumber);
if (currentPrIsIndexed || response.data.total_count >= 100) {
return response.data.total_count;
}
return existingOpenPrCount + 1;
return response.data.total_count + 1;
}
async function enforcePrLimit({ github, context, core, exemptLabelName, maxOpenPrs, labelName }) {
+27 -57
View File
@@ -44,20 +44,23 @@ function createCore() {
};
}
function createGithub({
itemNumbers,
labelExists = true,
pullRequests = createPullRequestPage({ numbers: itemNumbers }),
}) {
function createGithub({ totalCount, itemNumbers, labelExists = true }) {
const calls = [];
return {
calls,
async paginate(method, params) {
calls.push({ api: 'paginate', method, params });
return pullRequests;
},
rest: {
search: {
async issuesAndPullRequests(params) {
calls.push({ api: 'search.issuesAndPullRequests', params });
return {
data: {
total_count: totalCount,
items: itemNumbers.map((number) => ({ number })),
},
};
},
},
issues: {
async getLabel(params) {
calls.push({ api: 'issues.getLabel', params });
@@ -82,10 +85,6 @@ function createGithub({
},
},
pulls: {
async list(params) {
calls.push({ api: 'pulls.list', params });
return { data: pullRequests };
},
async update(params) {
calls.push({ api: 'pulls.update', params });
return { data: { state: params.state } };
@@ -95,15 +94,6 @@ function createGithub({
};
}
function createPullRequestPage({ author = 'community-user', numbers }) {
return numbers.map((number) => ({
number,
user: {
login: author,
},
}));
}
// ---------------------------------------------------------------------------
// PR limit enforcement
@@ -112,6 +102,7 @@ function createPullRequestPage({ author = 'community-user', numbers }) {
describe('PR limit enforcement', () => {
it('does not close the PR when the author is at the open PR limit', async () => {
const github = createGithub({
totalCount: 10,
itemNumbers: [1, 2, 3, 4, 5, 6, 7, 8, 9, 123],
});
@@ -128,13 +119,14 @@ describe('PR limit enforcement', () => {
assert.equal(result.openPrCount, 10);
assert.deepEqual(
github.calls.map((call) => call.api),
['paginate'],
['search.issuesAndPullRequests'],
);
});
it('counts the new PR when the pull list includes it', async () => {
it('counts the new PR when search has not indexed it yet', async () => {
const github = createGithub({
itemNumbers: [123, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
totalCount: 10,
itemNumbers: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
});
const result = await enforcePrLimit({
@@ -151,7 +143,7 @@ describe('PR limit enforcement', () => {
assert.deepEqual(
github.calls.map((call) => call.api),
[
'paginate',
'search.issuesAndPullRequests',
'issues.getLabel',
'issues.addLabels',
'issues.createComment',
@@ -160,31 +152,9 @@ describe('PR limit enforcement', () => {
);
});
it('counts the current PR on top of existing open PRs', async () => {
const github = createGithub({
itemNumbers: [123, ...Array.from({ length: 24 }, (_, index) => index + 1)],
pullRequests: createPullRequestPage({
numbers: [123, ...Array.from({ length: 25 }, (_, index) => index + 1)],
}),
});
const result = await enforcePrLimit({
github,
context: createContext(),
core: createCore(),
exemptLabelName: 'pr-limit-exempt',
maxOpenPrs: 10,
labelName: 'too-many-prs',
});
assert.equal(result.closed, true);
assert.equal(result.openPrCount, 26);
const comment = github.calls.find((call) => call.api === 'issues.createComment').params.body;
assert.match(comment, /This PR would put you at 26 open pull requests/);
});
it('creates the label when it does not already exist', async () => {
const github = createGithub({
totalCount: 11,
itemNumbers: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 123],
labelExists: false,
});
@@ -202,7 +172,7 @@ describe('PR limit enforcement', () => {
assert.deepEqual(
github.calls.map((call) => call.api),
[
'paginate',
'search.issuesAndPullRequests',
'issues.getLabel',
'issues.createLabel',
'issues.addLabels',
@@ -218,6 +188,7 @@ describe('PR limit enforcement', () => {
it('tolerates a 422 race when creating the label', async () => {
const github = createGithub({
totalCount: 11,
itemNumbers: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 123],
labelExists: false,
});
@@ -241,7 +212,7 @@ describe('PR limit enforcement', () => {
assert.deepEqual(
github.calls.map((call) => call.api),
[
'paginate',
'search.issuesAndPullRequests',
'issues.getLabel',
'issues.createLabel',
'issues.addLabels',
@@ -253,11 +224,8 @@ describe('PR limit enforcement', () => {
it('uses a diplomatic close message with the configured limit', async () => {
const github = createGithub({
totalCount: 11,
itemNumbers: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 123],
pullRequests: createPullRequestPage({
author: 'octo-contributor',
numbers: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 123],
}),
});
await enforcePrLimit({
@@ -278,6 +246,7 @@ describe('PR limit enforcement', () => {
it('does not close an exempt PR when it is reopened', async () => {
const github = createGithub({
totalCount: 11,
itemNumbers: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 123],
});
@@ -296,9 +265,10 @@ describe('PR limit enforcement', () => {
assert.deepEqual(github.calls, []);
});
it('counts the current PR when the author has more than one page of open PRs', async () => {
it('does not over-count when the current PR is not on the first search page', async () => {
const github = createGithub({
itemNumbers: [123, ...Array.from({ length: 100 }, (_, index) => index + 1)],
totalCount: 101,
itemNumbers: Array.from({ length: 100 }, (_, index) => index + 1),
});
const result = await enforcePrLimit({
@@ -121,9 +121,6 @@ jobs:
python
declarative-agents
- name: Free runner disk space
uses: ./.github/actions/free-runner-disk-space
- name: Setup dotnet
uses: actions/setup-dotnet@c2fa09f4bde5ebb9d1777cf28262a3eb3db3ced7 # v5.2.0
with:
@@ -194,9 +191,6 @@ jobs:
python
declarative-agents
- name: Free runner disk space
uses: ./.github/actions/free-runner-disk-space
# Start Cosmos DB Emulator for all integration tests and only for unit tests when CosmosDB changes happened)
- name: Start Azure Cosmos DB Emulator
if: ${{ runner.os == 'Windows' && (needs.paths-filter.outputs.cosmosDbChanges == 'true' || (github.event_name != 'pull_request' && matrix.integration-tests)) }}
@@ -371,9 +365,6 @@ jobs:
dotnet
python
- name: Free runner disk space
uses: ./.github/actions/free-runner-disk-space
- name: Setup dotnet
uses: actions/setup-dotnet@c2fa09f4bde5ebb9d1777cf28262a3eb3db3ced7 # v5.2.0
with:
@@ -461,9 +452,6 @@ jobs:
python
declarative-agents
- name: Free runner disk space
uses: ./.github/actions/free-runner-disk-space
- name: Setup dotnet
uses: actions/setup-dotnet@c2fa09f4bde5ebb9d1777cf28262a3eb3db3ced7 # v5.2.0
with:
@@ -8,7 +8,6 @@ on:
permissions:
contents: read
actions: read
pull-requests: write
jobs:
@@ -24,7 +23,7 @@ jobs:
- name: Download coverage report
uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8
with:
github-token: ${{ github.token }}
github-token: ${{ secrets.GH_ACTIONS_PR_WRITE }}
run-id: ${{ github.event.workflow_run.id }}
path: ./python
merge-multiple: true
@@ -39,9 +38,9 @@ jobs:
echo "PR number file 'pr_number' is missing or empty"
exit 1
fi
PR_NUMBER=$(cat pr_number)
if ! [[ "$PR_NUMBER" =~ ^[0-9]+$ ]]; then
echo "::error::PR number file contains invalid content"
PR_NUMBER=$(head -1 pr_number | tr -dc '0-9')
if [ -z "$PR_NUMBER" ]; then
echo "PR number file 'pr_number' does not contain a valid PR number"
exit 1
fi
echo "PR_NUMBER=$PR_NUMBER" >> "$GITHUB_ENV"
@@ -49,7 +48,7 @@ jobs:
id: coverageComment
uses: MishaKav/pytest-coverage-comment@26f986d2599c288bb62f623d29c2da98609e9cd4 # v1.6.0
with:
github-token: ${{ github.token }}
github-token: ${{ secrets.GH_ACTIONS_PR_WRITE }}
issue-number: ${{ env.PR_NUMBER }}
pytest-xml-coverage-path: python/python-coverage.xml
title: "Python Test Coverage Report"
-1
View File
@@ -248,4 +248,3 @@ dotnet/filtered-*.slnx
.omx/
**/issues/
.test_*
+17 -17
View File
@@ -1,17 +1,17 @@
# Support
## How to file issues and get help
This project uses GitHub Issues to track bugs and feature requests. Please search the existing
issues before filing new issues to avoid duplicates. For new issues, file your bug or
feature request as a new Issue.
For help and questions about using this project, please create a GitHub issue.
AI Support team will support Microsoft Agent Framework issues for customers under a **Unified support agreement when the issue arises from usage of Azure AI services** (Foundry Models, Foundry Agents etc.) in conjunction with the SDK. Conversely, if customer has any other / non unified support agreement and/or Agent Framework SDK is used in a way **not involving an Azure service**, it is treated as a purely open-source tool – Microsoft’s support organization will not handle it, and users should use GitHub or forums for assistance
For Copilot Studio SDK implementation issues, customers should use GitHub Issues for assistance, as outlined above. Conversely, for prerequisites managed within the Copilot Studio portal, customers can rely on the standard Microsoft Copilot Studio support channels.
## Microsoft Support Policy
Support for this **PROJECT or PRODUCT** is limited to the resources listed above.
# Support
## How to file issues and get help
This project uses GitHub Issues to track bugs and feature requests. Please search the existing
issues before filing new issues to avoid duplicates. For new issues, file your bug or
feature request as a new Issue.
For help and questions about using this project, please create a GitHub issue.
AI Support team will support Microsoft Agent Framework issues for customers under a **Unified support agreement when the issue arises from usage of Azure AI services** (Foundry Models, Foundry Agents etc.) in conjunction with the SDK. Conversely, if customer has any other / non unified support agreement and/or Agent Framework SDK is used in a way **not involving an Azure service**, it is treated as a purely open-source tool – Microsoft’s support organization will not handle it, and users should use GitHub or forums for assistance
For Copilot Studio SDK implementation issues, customers should use GitHub Issues for assistance, as outlined above. Conversely, for prerequisites managed within the Copilot Studio portal, customers can rely on the standard Microsoft Copilot Studio support channels.
## Microsoft Support Policy
Support for this **PROJECT or PRODUCT** is limited to the resources listed above.
-3
View File
@@ -344,9 +344,6 @@
<Folder Name="/Samples/04-hosting/FoundryHostedAgents/responses/Hosted-Toolbox/">
<Project Path="samples/04-hosting/FoundryHostedAgents/responses/Hosted-Toolbox/HostedToolbox.csproj" />
</Folder>
<Folder Name="/Samples/04-hosting/FoundryHostedAgents/responses/Hosted-ToolboxMcpSkills/">
<Project Path="samples/04-hosting/FoundryHostedAgents/responses/Hosted-ToolboxMcpSkills/HostedToolboxMcpSkills.csproj" />
</Folder>
<Folder Name="/Samples/04-hosting/FoundryHostedAgents/responses/Hosted-AzureSearchRag/">
<Project Path="samples/04-hosting/FoundryHostedAgents/responses/Hosted-AzureSearchRag/HostedAzureSearchRag.csproj" />
</Folder>
@@ -0,0 +1,10 @@
{
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "dotnet-isolated",
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
"AZURE_OPENAI_ENDPOINT": "<AZURE_OPENAI_ENDPOINT>",
"AZURE_OPENAI_DEPLOYMENT_NAME": "<AZURE_OPENAI_DEPLOYMENT_NAME>"
}
}
@@ -0,0 +1,10 @@
{
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "dotnet-isolated",
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
"AZURE_OPENAI_ENDPOINT": "<AZURE_OPENAI_ENDPOINT>",
"AZURE_OPENAI_DEPLOYMENT_NAME": "<AZURE_OPENAI_DEPLOYMENT_NAME>"
}
}
@@ -0,0 +1,10 @@
{
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "dotnet-isolated",
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
"AZURE_OPENAI_ENDPOINT": "<AZURE_OPENAI_ENDPOINT>",
"AZURE_OPENAI_DEPLOYMENT_NAME": "<AZURE_OPENAI_DEPLOYMENT_NAME>"
}
}
@@ -0,0 +1,10 @@
{
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "dotnet-isolated",
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
"AZURE_OPENAI_ENDPOINT": "<AZURE_OPENAI_ENDPOINT>",
"AZURE_OPENAI_DEPLOYMENT_NAME": "<AZURE_OPENAI_DEPLOYMENT_NAME>"
}
}
@@ -0,0 +1,10 @@
{
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "dotnet-isolated",
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
"AZURE_OPENAI_ENDPOINT": "<AZURE_OPENAI_ENDPOINT>",
"AZURE_OPENAI_DEPLOYMENT_NAME": "<AZURE_OPENAI_DEPLOYMENT_NAME>"
}
}
@@ -0,0 +1,10 @@
{
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "dotnet-isolated",
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
"AZURE_OPENAI_ENDPOINT": "<AZURE_OPENAI_ENDPOINT>",
"AZURE_OPENAI_DEPLOYMENT_NAME": "<AZURE_OPENAI_DEPLOYMENT_NAME>"
}
}
@@ -0,0 +1,10 @@
{
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "dotnet-isolated",
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
"AZURE_OPENAI_ENDPOINT": "<AZURE_OPENAI_ENDPOINT>",
"AZURE_OPENAI_DEPLOYMENT_NAME": "<AZURE_OPENAI_DEPLOYMENT_NAME>"
}
}
@@ -0,0 +1,12 @@
{
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "dotnet-isolated",
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
"AZURE_OPENAI_ENDPOINT": "<AZURE_OPENAI_ENDPOINT>",
"AZURE_OPENAI_DEPLOYMENT_NAME": "<AZURE_OPENAI_DEPLOYMENT_NAME>",
"REDIS_CONNECTION_STRING": "localhost:6379",
"REDIS_STREAM_TTL_MINUTES": "10"
}
}
@@ -0,0 +1,10 @@
{
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "dotnet-isolated",
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
"AZURE_OPENAI_ENDPOINT": "<AZURE_OPENAI_ENDPOINT>",
"AZURE_OPENAI_DEPLOYMENT_NAME": "<AZURE_OPENAI_DEPLOYMENT_NAME>"
}
}
@@ -0,0 +1,10 @@
{
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "dotnet-isolated",
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
"AZURE_OPENAI_ENDPOINT": "<AZURE_OPENAI_ENDPOINT>",
"AZURE_OPENAI_DEPLOYMENT_NAME": "<AZURE_OPENAI_DEPLOYMENT_NAME>"
}
}
@@ -0,0 +1,10 @@
{
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "dotnet-isolated",
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
"AZURE_OPENAI_ENDPOINT": "<AZURE_OPENAI_ENDPOINT>",
"AZURE_OPENAI_DEPLOYMENT_NAME": "<AZURE_OPENAI_DEPLOYMENT_NAME>"
}
}
@@ -0,0 +1,8 @@
{
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "dotnet-isolated",
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"
}
}
@@ -0,0 +1,10 @@
{
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "dotnet-isolated",
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
"AZURE_OPENAI_ENDPOINT": "<AZURE_OPENAI_ENDPOINT>",
"AZURE_OPENAI_DEPLOYMENT_NAME": "<AZURE_OPENAI_DEPLOYMENT_NAME>"
}
}
@@ -1,6 +0,0 @@
AZURE_AI_PROJECT_ENDPOINT=<your-azure-ai-project-endpoint>
ASPNETCORE_URLS=http://+:8088
ASPNETCORE_ENVIRONMENT=Development
AZURE_AI_MODEL_DEPLOYMENT_NAME=gpt-5
FOUNDRY_TOOLBOX_NAME=<your-toolbox-name>
AZURE_BEARER_TOKEN=DefaultAzureCredential
@@ -1,26 +0,0 @@
# Dockerfile for end-users consuming the Agent Framework via NuGet packages.
#
# This Dockerfile performs a full `dotnet restore` and `dotnet publish` inside the container,
# which only succeeds when the project references its dependencies via PackageReference (see the
# commented-out section in HostedToolboxMcpSkills.csproj). Contributors building from the
# agent-framework repository source must use Dockerfile.contributor instead because
# ProjectReference dependencies live outside this folder and cannot be restored from inside
# this build context.
#
# Use the official .NET 10.0 ASP.NET runtime as a parent image
FROM mcr.microsoft.com/dotnet/aspnet:10.0 AS base
WORKDIR /app
FROM mcr.microsoft.com/dotnet/sdk:10.0 AS build
WORKDIR /src
COPY . .
RUN dotnet restore
RUN dotnet publish -c Release -o /app/publish
# Final stage
FROM base AS final
WORKDIR /app
COPY --from=build /app/publish .
EXPOSE 8088
ENV ASPNETCORE_URLS=http://+:8088
ENTRYPOINT ["dotnet", "HostedToolboxMcpSkills.dll"]
@@ -1,18 +0,0 @@
# Dockerfile for contributors building from the agent-framework repository source.
#
# This project uses ProjectReference to the local source, which means a standard
# multi-stage Docker build cannot resolve dependencies outside this folder.
# Pre-publish the app targeting the container runtime and copy the output:
#
# dotnet publish -c Debug -f net10.0 -r linux-musl-x64 --self-contained false -o out
# docker build -f Dockerfile.contributor -t hosted-toolbox-mcp-skills .
# docker run --rm -p 8088:8088 -e AGENT_NAME=hosted-toolbox-mcp-skills -e AZURE_BEARER_TOKEN=$AZURE_BEARER_TOKEN --env-file .env hosted-toolbox-mcp-skills
#
# For end-users consuming the NuGet package (not ProjectReference), use the standard
# Dockerfile which performs a full dotnet restore + publish inside the container.
FROM mcr.microsoft.com/dotnet/aspnet:10.0-alpine AS final
WORKDIR /app
COPY out/ .
EXPOSE 8088
ENV ASPNETCORE_URLS=http://+:8088
ENTRYPOINT ["dotnet", "HostedToolboxMcpSkills.dll"]
@@ -1,36 +0,0 @@
<Project Sdk="Microsoft.NET.Sdk.Web">
<PropertyGroup>
<TargetFrameworks>net10.0</TargetFrameworks>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<CentralPackageTransitivePinningEnabled>false</CentralPackageTransitivePinningEnabled>
<RootNamespace>HostedToolboxMcpSkills</RootNamespace>
<AssemblyName>HostedToolboxMcpSkills</AssemblyName>
<NoWarn>$(NoWarn);</NoWarn>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Azure.AI.Projects" />
<PackageReference Include="Azure.Identity" />
<PackageReference Include="ModelContextProtocol" VersionOverride="1.2.0" />
<PackageReference Include="DotNetEnv" />
</ItemGroup>
<!-- For contributors: uses ProjectReference to build against local source -->
<ItemGroup>
<ProjectReference Include="..\..\..\..\..\src\Microsoft.Agents.AI.Foundry\Microsoft.Agents.AI.Foundry.csproj" />
<ProjectReference Include="..\..\..\..\..\src\Microsoft.Agents.AI.Foundry.Hosting\Microsoft.Agents.AI.Foundry.Hosting.csproj" />
<ProjectReference Include="..\..\..\..\..\src\Microsoft.Agents.AI.Mcp\Microsoft.Agents.AI.Mcp.csproj" />
<ProjectReference Include="..\Hosted_Shared_Contributor_Setup\Hosted_Shared_Contributor_Setup.csproj" />
</ItemGroup>
<!-- For end-users: uncomment the PackageReference below and remove the ProjectReference above
<ItemGroup>
<PackageReference Include="Microsoft.Agents.AI.Foundry" Version="1.6.1-preview.260514.1" />
<PackageReference Include="Microsoft.Agents.AI.Foundry.Hosting" Version="1.6.1-preview.260514.1" />
<PackageReference Include="Microsoft.Agents.AI.Mcp" Version="1.6.1-preview.260514.1" />
</ItemGroup>
-->
</Project>
@@ -1,109 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
// Hosted Toolbox MCP Skills Agent
//
// Demonstrates how to host an agent that discovers MCP-based skills from a
// Foundry Toolbox MCP endpoint and injects them as AIContextProviders using
// AgentSkillsProviderBuilder.UseMcpSkills().
//
// Required environment variables:
// AZURE_AI_PROJECT_ENDPOINT - Azure AI Foundry project endpoint
// FOUNDRY_TOOLBOX_NAME - Name of the Foundry Toolbox to connect to
// AZURE_AI_MODEL_DEPLOYMENT_NAME - Model deployment name (default: gpt-5)
using System.Net.Http.Headers;
using Azure.AI.Projects;
using Azure.Core;
using Azure.Identity;
using DotNetEnv;
using Hosted_Shared_Contributor_Setup;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Foundry.Hosting;
using ModelContextProtocol.Client;
// Load .env file if present (for local development)
Env.TraversePath().Load();
var projectEndpoint = Environment.GetEnvironmentVariable("AZURE_AI_PROJECT_ENDPOINT")
?? throw new InvalidOperationException("AZURE_AI_PROJECT_ENDPOINT is not set.");
var deployment = Environment.GetEnvironmentVariable("AZURE_AI_MODEL_DEPLOYMENT_NAME") ?? "gpt-5";
var toolboxName = Environment.GetEnvironmentVariable("FOUNDRY_TOOLBOX_NAME")
?? throw new InvalidOperationException("FOUNDRY_TOOLBOX_NAME is not set.");
// Build the Toolbox MCP URL from the project endpoint and toolbox name.
var toolboxMcpServerUrl = $"{projectEndpoint.TrimEnd('/')}/toolboxes/{toolboxName}/mcp?api-version=v1";
// Use a chained credential: try a temporary dev token first (for local Docker debugging),
// then fall back to DefaultAzureCredential (for local dev via dotnet run / managed identity in production).
TokenCredential credential = new ChainedTokenCredential(
new DevTemporaryTokenCredential(),
new DefaultAzureCredential());
// ── Connect to the Foundry Toolbox MCP endpoint ─────────────────────────────
// Create an HttpClient that attaches a fresh Foundry bearer token to every request.
using var httpClient = new HttpClient(new BearerTokenHandler(credential, "https://ai.azure.com/.default") { CheckCertificateRevocationList = true });
Console.WriteLine($"Connecting to Foundry Toolbox '{toolboxName}' MCP server...");
await using var mcpClient = await McpClient.CreateAsync(
new HttpClientTransport(
new HttpClientTransportOptions
{
Endpoint = new Uri(toolboxMcpServerUrl),
Name = toolboxName,
TransportMode = HttpTransportMode.StreamableHttp,
AdditionalHeaders = new Dictionary<string, string>
{
["Foundry-Features"] = "Toolboxes=V1Preview",
},
},
httpClient));
// ── Configure MCP-based skills provider ──────────────────────────────────────
var skillsProvider = new AgentSkillsProviderBuilder()
.UseMcpSkills(mcpClient)
.Build();
// ── Create the agent ─────────────────────────────────────────────────────────
AIAgent agent = new AIProjectClient(new Uri(projectEndpoint), credential)
.AsAIAgent(new ChatClientAgentOptions
{
Name = Environment.GetEnvironmentVariable("AGENT_NAME") ?? "hosted-toolbox-mcp-skills",
Description = "Hosted agent with MCP skills discovered from a Foundry Toolbox",
ChatOptions = new()
{
ModelId = deployment,
Instructions = "You are a helpful assistant.",
},
AIContextProviders = [skillsProvider],
});
// ── Build the host ───────────────────────────────────────────────────────────
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddFoundryResponses(agent);
builder.Services.AddDevTemporaryLocalContributorSetup(); // Local Docker debugging only - must not be used in production.
var app = builder.Build();
app.MapFoundryResponses();
// Contributor-only: in Development, also map the per-agent OpenAI route shape that live Foundry uses
// so a local REPL client can target this server via AIProjectClient.AsAIAgent(Uri agentEndpoint).
// Do not use this in production. Hosted Foundry agents only support the agent-endpoint path.
app.MapDevTemporaryLocalAgentEndpoint();
app.Run();
// ---------------------------------------------------------------------------
// HttpClientHandler: attaches a fresh Foundry bearer token to every request
// ---------------------------------------------------------------------------
internal sealed class BearerTokenHandler(TokenCredential credential, string scope) : HttpClientHandler
{
private readonly TokenRequestContext _tokenContext = new([scope]);
protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
{
AccessToken token = await credential.GetTokenAsync(this._tokenContext, cancellationToken).ConfigureAwait(false);
request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", token.Token);
return await base.SendAsync(request, cancellationToken).ConfigureAwait(false);
}
}
@@ -1,103 +0,0 @@
# Hosted-ToolboxMcpSkills
A hosted agent that discovers **MCP-based skills from a Foundry Toolbox** and makes them available to the agent using `AgentSkillsProviderBuilder.UseMcpSkills(mcpClient)`.
The `AgentSkillsProvider` is attached to the agent as a context provider and implements the [Agent Skills](https://agentskills.io/) progressive-disclosure pattern. When the agent is prompted, it discovers available skills in the Foundry Toolbox via the provider:
1. **Advertise** - skill names and descriptions are injected into the system prompt so the agent knows what is available.
2. **Load** - when the agent decides a skill is relevant, it retrieves the full skill body with detailed instructions via the provider.
3. **Read resources** - if a skill includes supplementary content (reference documents, assets), the agent reads them on demand via the provider.
This way the full skill body and resources are only loaded when the agent actually needs them, reducing token usage.
## Prerequisites
- [.NET 10 SDK](https://dotnet.microsoft.com/download/dotnet/10.0)
- An Azure AI Foundry project with a deployed model (e.g., `gpt-5`)
- A Foundry Toolbox already configured with skills provisioned
- Azure CLI logged in (`az login`)
## Configuration
Copy the template and fill in your values:
```bash
cp .env.example .env
```
Edit `.env` and set your Azure AI Foundry project endpoint and toolbox name:
```env
AZURE_AI_PROJECT_ENDPOINT=https://<your-account>.services.ai.azure.com/api/projects/<your-project>
ASPNETCORE_URLS=http://+:8088
ASPNETCORE_ENVIRONMENT=Development
AZURE_AI_MODEL_DEPLOYMENT_NAME=gpt-5
FOUNDRY_TOOLBOX_NAME=my-toolbox
```
> **Note:** `.env` is gitignored. The `.env.example` template is checked in as a reference.
## Running directly (contributors)
This project uses `ProjectReference` to build against the local Agent Framework source.
```bash
cd dotnet/samples/04-hosting/FoundryHostedAgents/responses/Hosted-ToolboxMcpSkills
dotnet run
```
The agent will start on `http://localhost:8088`.
### Test it
Using the Azure Developer CLI:
```bash
azd ai agent invoke --local "What skills do you have available?"
```
## Running with Docker
Since this project uses `ProjectReference`, use `Dockerfile.contributor` which takes a pre-published output.
### 1. Publish for the container runtime (Linux Alpine)
```bash
dotnet publish -c Debug -f net10.0 -r linux-musl-x64 --self-contained false -o out
```
### 2. Build the Docker image
```bash
docker build -f Dockerfile.contributor -t hosted-toolbox-mcp-skills .
```
### 3. Run the container
Generate a bearer token on your host and pass it to the container:
```bash
# Generate token (expires in ~1 hour)
export AZURE_BEARER_TOKEN=$(az account get-access-token --resource https://ai.azure.com --query accessToken -o tsv)
# Run with token
docker run --rm -p 8088:8088 \
-e AGENT_NAME=hosted-toolbox-mcp-skills \
-e AZURE_BEARER_TOKEN=$AZURE_BEARER_TOKEN \
--env-file .env \
hosted-toolbox-mcp-skills
```
> **Note:** `AGENT_NAME` is passed via `-e` to simulate the platform injection. `AZURE_BEARER_TOKEN` provides Azure credentials to the container (tokens expire after ~1 hour). The `.env` file provides the remaining configuration.
### 4. Test it
Using the Azure Developer CLI:
```bash
azd ai agent invoke --local "What skills do you have available?"
```
## NuGet package users
If you are consuming the Agent Framework as a NuGet package (not building from source), use the standard `Dockerfile` instead of `Dockerfile.contributor`. See the commented section in `HostedToolboxMcpSkills.csproj` for the `PackageReference` alternative.
@@ -1,43 +0,0 @@
# yaml-language-server: $schema=https://raw.githubusercontent.com/microsoft/AgentSchema/refs/heads/main/schemas/v1.0/AgentManifest.yaml
name: hosted-toolbox-mcp-skills
displayName: "Hosted Toolbox MCP Skills Agent"
description: >
A hosted agent that discovers MCP-based skills from a Foundry Toolbox
and makes them available to the agent via the agent skills provider.
metadata:
tags:
- AI Agent Hosting
- Azure AI AgentServer
- Responses Protocol
- Agent Framework
- MCP
- Model Context Protocol
- Agent Skills
- Foundry Toolbox
- Foundry Toolbox Skills
template:
name: hosted-toolbox-mcp-skills
kind: hosted
protocols:
- protocol: responses
version: 1.0.0
resources:
cpu: "0.25"
memory: 0.5Gi
environment_variables:
- name: AZURE_AI_MODEL_DEPLOYMENT_NAME
value: "{{AZURE_AI_MODEL_DEPLOYMENT_NAME}}"
- name: FOUNDRY_TOOLBOX_NAME
value: "{{FOUNDRY_TOOLBOX_NAME}}"
parameters:
properties:
- name: FOUNDRY_TOOLBOX_NAME
secret: false
description: Name of the Foundry Toolbox to connect to for MCP skill discovery
resources:
- kind: model
id: gpt-5
name: AZURE_AI_MODEL_DEPLOYMENT_NAME
@@ -1,14 +0,0 @@
# yaml-language-server: $schema=https://raw.githubusercontent.com/microsoft/AgentSchema/refs/heads/main/schemas/v1.0/ContainerAgent.yaml
kind: hosted
name: hosted-toolbox-mcp-skills
protocols:
- protocol: responses
version: 1.0.0
resources:
cpu: "0.25"
memory: 0.5Gi
environment_variables:
- name: AZURE_AI_MODEL_DEPLOYMENT_NAME
value: ${AZURE_AI_MODEL_DEPLOYMENT_NAME}
- name: FOUNDRY_TOOLBOX_NAME
value: ${FOUNDRY_TOOLBOX_NAME}
@@ -281,19 +281,14 @@ internal static class OutputConverter
var outputText = EncodeFunctionResultAsJsonStringPayload(functionResult.Result);
// Use the SDK's convenience method so the OutputItemFunctionToolCallOutput
// is constructed with a populated Id. The public OutputItemFunctionToolCallOutput
// ctor only sets CallId/Output (Id is read-only), and AddOutputItem<T>+EmitAdded
// does not auto-stamp Id — only ResponseId/AgentReference. Without this, the
// serialized item arrives at the Foundry storage layer with id=null and is
// rejected with "ID cannot be null or empty (Parameter 'id')".
foreach (var evt in stream.OutputItemFunctionCallOutput(
var itemId = GenerateItemId("fc");
var outputItem = new OutputItemFunctionToolCallOutput(
functionResult.CallId,
BinaryData.FromString(outputText)))
{
yield return evt;
}
BinaryData.FromString(outputText));
var outputBuilder = stream.AddOutputItem<OutputItemFunctionToolCallOutput>(itemId);
yield return outputBuilder.EmitAdded(outputItem);
yield return outputBuilder.EmitDone(outputItem);
break;
}
@@ -24,13 +24,11 @@
<ItemGroup>
<PackageReference Include="Azure.AI.Projects" />
<PackageReference Include="Azure.Core" />
<PackageReference Include="Microsoft.Extensions.AI" />
<PackageReference Include="Microsoft.Extensions.AI.Abstractions" />
<PackageReference Include="Microsoft.Extensions.AI.OpenAI" />
<PackageReference Include="Microsoft.Extensions.Compliance.Abstractions" />
<PackageReference Include="OpenAI" />
<PackageReference Include="System.ClientModel" />
</ItemGroup>
<!-- Evaluation support requires net8.0+ (MEAI.Evaluation does not support legacy TFMs) -->
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<!-- Preview while Microsoft.Agents.AI.Foundry is preview (blocked by Azure.AI.Projects 2.1.0-beta). Flip to IsReleased=true once that ships stable. -->
<IsReleaseCandidate>true</IsReleaseCandidate>
<NoWarn>$(NoWarn);MEAI001;OPENAI001</NoWarn>
</PropertyGroup>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<IsReleased>true</IsReleased>
<IsReleaseCandidate>true</IsReleaseCandidate>
<NoWarn>$(NoWarn);MEAI001;OPENAI001</NoWarn>
</PropertyGroup>
@@ -13,11 +13,9 @@
<Import Project="$(RepoRoot)/dotnet/nuget/nuget-package.props" />
<!-- First Stable release after the RC milestone. Baseline against the latest
published RC so package validation catches accidental breaking changes.
Future releases should bump this to the previous stable version. -->
<!-- Package not yet published to NuGet — disable baseline validation until first release -->
<PropertyGroup>
<PackageValidationBaselineVersion>1.8.0-rc1</PackageValidationBaselineVersion>
<EnablePackageValidation>false</EnablePackageValidation>
</PropertyGroup>
<PropertyGroup>
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<IsReleased>true</IsReleased>
<IsReleaseCandidate>true</IsReleaseCandidate>
<NoWarn>$(NoWarn);MEAI001;OPENAI001</NoWarn>
</PropertyGroup>
@@ -13,13 +13,6 @@
<Import Project="$(RepoRoot)/dotnet/nuget/nuget-package.props" />
<!-- First Stable release after the RC milestone. Baseline against the latest
published RC so package validation catches accidental breaking changes.
Future releases should bump this to the previous stable version. -->
<PropertyGroup>
<PackageValidationBaselineVersion>1.8.0-rc1</PackageValidationBaselineVersion>
</PropertyGroup>
<PropertyGroup>
<!-- NuGet Package Settings -->
<Title>Microsoft Agent Framework Declarative Workflows</Title>
@@ -6,11 +6,11 @@
.DESCRIPTION
The IT fixture targets stable, scenario-keyed agent names (e.g. it-happy-path) and only
manages versions on each test run. The agent itself must already exist AND its managed
identity must hold the Foundry User role on the project scope, otherwise inbound
identity must hold the Azure AI User role on the project scope, otherwise inbound
inference calls fail with HTTP 500 PermissionDenied.
This script idempotently creates each scenario agent (with a placeholder version) and
grants Foundry User on the project to its managed identity. Re-run it safely; existing
grants Azure AI User on the project to its managed identity. Re-run it safely; existing
agents and role assignments are left in place.
.PARAMETER ProjectEndpoint
@@ -135,20 +135,20 @@ foreach ($scenario in $Scenarios) {
-Body $patchBody | Out-Null
}
# 3. Grant Foundry User on the project scope to the agent MI (idempotent).
# 3. Grant Azure AI User on the project scope to the agent MI (idempotent).
$existing = az role assignment list --assignee $principalId --scope $projectScope `
--query "[?roleDefinitionName=='Foundry User']" 2>$null | ConvertFrom-Json
--query "[?roleDefinitionName=='Azure AI User']" 2>$null | ConvertFrom-Json
if ($existing) {
Write-Host " role already assigned"
} else {
Write-Host " granting Foundry User..."
Write-Host " granting Azure AI User..."
$maxAttempts = 12
$granted = $false
for ($i = 1; $i -le $maxAttempts; $i++) {
$output = az role assignment create `
--assignee-object-id $principalId `
--assignee-principal-type ServicePrincipal `
--role 'Foundry User' `
--role 'Azure AI User' `
--scope $projectScope 2>&1
if ($LASTEXITCODE -eq 0) {
$granted = $true
@@ -0,0 +1,12 @@
{
"profiles": {
"Microsoft.Agents.AI.DevUI.UnitTests": {
"commandName": "Project",
"launchBrowser": true,
"environmentVariables": {
"ASPNETCORE_ENVIRONMENT": "Development"
},
"applicationUrl": "https://localhost:63009;http://localhost:63010"
}
}
}
@@ -704,35 +704,6 @@ public class OutputConverterTests
Assert.Equal("[{\"id\":1}]", inner);
}
// K-06e: Regression — the OutputItemFunctionToolCallOutput must have a populated Id
// and a matching wire id on the added/done events. The Foundry storage layer extracts
// a partition id from this field and throws "ID cannot be null or empty (Parameter 'id')"
// when it is missing.
[Fact]
public async Task ConvertUpdatesToEventsAsync_FunctionResult_OutputItemHasIdAsync()
{
var (stream, _) = CreateTestStream();
var update = new AgentResponseUpdate { Contents = [new FunctionResultContent("call_1", "sunny")] };
var events = new List<ResponseStreamEvent>();
await foreach (var evt in OutputConverter.ConvertUpdatesToEventsAsync(ToAsync(new[] { update }), stream))
{
events.Add(evt);
}
var added = Assert.Single(events.OfType<ResponseOutputItemAddedEvent>());
var done = Assert.Single(events.OfType<ResponseOutputItemDoneEvent>());
var addedOutput = Assert.IsType<OutputItemFunctionToolCallOutput>(added.Item);
var doneOutput = Assert.IsType<OutputItemFunctionToolCallOutput>(done.Item);
Assert.False(string.IsNullOrEmpty(addedOutput.Id));
Assert.False(string.IsNullOrEmpty(doneOutput.Id));
Assert.Equal(addedOutput.Id, doneOutput.Id);
Assert.Equal("call_1", addedOutput.CallId);
Assert.Equal("call_1", doneOutput.CallId);
}
// L-01
[Fact]
public async Task ConvertUpdatesToEventsAsync_ExecutorInvokedEvent_EmitsWorkflowActionItemAsync()
@@ -0,0 +1,12 @@
{
"profiles": {
"Microsoft.Agents.AI.Hosting.A2A.UnitTests": {
"commandName": "Project",
"launchBrowser": true,
"environmentVariables": {
"ASPNETCORE_ENVIRONMENT": "Development"
},
"applicationUrl": "https://localhost:52186;http://localhost:52187"
}
}
}
@@ -0,0 +1,12 @@
{
"profiles": {
"Microsoft.Agents.AI.Hosting.OpenAI.UnitTests": {
"commandName": "Project",
"launchBrowser": true,
"environmentVariables": {
"ASPNETCORE_ENVIRONMENT": "Development"
},
"applicationUrl": "https://localhost:60491;http://localhost:60492"
}
}
}
+1 -8
View File
@@ -27,7 +27,7 @@ Status is grouped into these buckets:
| `agent-framework-claude` | `python/packages/claude` | `beta` |
| `agent-framework-copilotstudio` | `python/packages/copilotstudio` | `beta` |
| `agent-framework-core` | `python/packages/core` | `released` |
| `agent-framework-declarative` | `python/packages/declarative` | `rc` |
| `agent-framework-declarative` | `python/packages/declarative` | `beta` |
| `agent-framework-devui` | `python/packages/devui` | `beta` |
| `agent-framework-durabletask` | `python/packages/durabletask` | `beta` |
| `agent-framework-foundry` | `python/packages/foundry` | `released` |
@@ -58,13 +58,6 @@ listed below.
### Experimental features
#### `DECLARATIVE_AGENTS`
- `agent-framework-declarative`: declarative agent loading APIs from
`agent_framework_declarative`, including `AgentFactory`,
`DeclarativeLoaderError`, `ProviderLookupError`, and `ProviderTypeMapping`
from `agent_framework_declarative/_loader.py`
#### `EVALS`
- `agent-framework-core`: exported evaluation APIs from `agent_framework`, including
@@ -287,7 +287,9 @@ class A2AExecutor(AgentExecutor):
artifact_id=artifact_id,
metadata=metadata,
append=(
True if streamed_artifact_ids is not None and artifact_id in streamed_artifact_ids else None
True
if streamed_artifact_ids is not None and artifact_id in streamed_artifact_ids
else None
),
)
if artifact_id and streamed_artifact_ids is not None:
@@ -803,15 +803,6 @@ class RawAnthropicClient(
}
a_content.append(mcp_result)
case "text_reasoning":
if content.text is None:
if (
content.protected_data
and a_content
and a_content[-1].get("type") == "thinking"
and "signature" not in a_content[-1]
):
a_content[-1]["signature"] = content.protected_data
continue
thinking_block: dict[str, Any] = {"type": "thinking", "thinking": content.text}
if content.protected_data:
thinking_block["signature"] = content.protected_data
@@ -485,48 +485,6 @@ def test_prepare_message_for_anthropic_text_reasoning_with_signature(
assert result["content"][0]["signature"] == "sig_abc123"
def test_prepare_message_for_anthropic_attaches_signature_only_reasoning(
mock_anthropic_client: MagicMock,
) -> None:
client = create_test_anthropic_client(mock_anthropic_client)
message = Message(
role="assistant",
contents=[
Content.from_text_reasoning(text="Let me think about this..."),
Content.from_text_reasoning(text=None, protected_data="sig_abc123"),
],
)
result = client._prepare_message_for_anthropic(message)
assert result["content"] == [
{"type": "thinking", "thinking": "Let me think about this...", "signature": "sig_abc123"}
]
def test_prepare_message_for_anthropic_skips_orphan_signature_only_reasoning(
mock_anthropic_client: MagicMock,
) -> None:
client = create_test_anthropic_client(mock_anthropic_client)
message = Message(
role="assistant",
contents=[
Content.from_text_reasoning(text=None, protected_data="sig_abc123"),
Content.from_function_call(
call_id="call_123",
name="get_weather",
arguments={"location": "San Francisco"},
),
],
)
result = client._prepare_message_for_anthropic(message)
assert len(result["content"]) == 1
assert result["content"][0]["type"] == "tool_use"
assert result["content"][0]["id"] == "call_123"
def test_prepare_message_for_anthropic_mcp_server_tool_call(
mock_anthropic_client: MagicMock,
) -> None:
@@ -4,7 +4,6 @@
from __future__ import annotations
import asyncio
import copy
import json
import logging
import sys
@@ -37,7 +36,6 @@ from agent_framework.observability import ChatTelemetryLayer
from boto3.session import Session as Boto3Session
from botocore.client import BaseClient
from botocore.config import Config as BotoConfig
from botocore.exceptions import ClientError
from pydantic import BaseModel
if sys.version_info >= (3, 13):
@@ -117,20 +115,13 @@ class BedrockChatOptions(ChatOptions[ResponseModelT], Generic[ResponseModelT], t
translates to ``toolConfig.tools``.
tool_choice: How the model should use tools,
translates to ``toolConfig.toolChoice``.
response_format: Structured output format. Accepts a Pydantic BaseModel
subclass or an OpenAI-style dict schema
(``{"json_schema": {"name": ..., "schema": ...}}``).
When provided, the Converse API request includes
``outputConfig.textFormat`` with the schema serialized as a JSON
string. ``ChatResponse.value`` will be populated with the parsed
model instance. Only supported on models that support
``outputConfig.textFormat``. Unsupported models raise a ValueError.
# Options not supported in Bedrock Converse API:
seed: Not supported.
frequency_penalty: Not supported.
presence_penalty: Not supported.
allow_multiple_tool_calls: Not supported (models handle parallel calls automatically).
response_format: Not directly supported (use model-specific prompting).
user: Not supported.
store: Not supported.
logit_bias: Not supported.
@@ -170,6 +161,9 @@ class BedrockChatOptions(ChatOptions[ResponseModelT], Generic[ResponseModelT], t
allow_multiple_tool_calls: None # type: ignore[misc]
"""Not supported. Bedrock models handle parallel tool calls automatically."""
response_format: None # type: ignore[misc]
"""Not directly supported. Use model-specific prompting for JSON output."""
user: None # type: ignore[misc]
"""Not supported in Bedrock Converse API."""
@@ -330,28 +324,10 @@ class BedrockChatClient(
return Boto3Session(**session_kwargs)
def _invoke_converse(self, request: Mapping[str, Any]) -> dict[str, Any]:
try:
response = self._bedrock_client.converse(**request)
if not isinstance(response, Mapping):
raise ChatClientInvalidResponseException("Bedrock converse response must be a mapping.")
return response
except ClientError as e:
error_details = e.response.get("Error", {})
error_code = error_details.get("Code", "")
error_message = error_details.get("Message", "")
# "outputConfig" in error_message catches cases where Bedrock explicitly
# rejects the outputConfig field (unsupported model). Other ValidationExceptions
# (e.g. malformed schema shape, invalid property values) will not mention
# "outputConfig" and will bubble up as raw ClientError without being misdiagnosed.
if error_code == "ValidationException" and (
"outputconfig" in error_message.lower() or "outputconfig" in str(e).lower()
):
raise ValueError(
f"Model '{self.model}' does not support structured output via outputConfig.textFormat. "
"Check the model's Bedrock Converse outputConfig/textFormat support. "
f"AWS error Code: {error_code}. AWS error Message: {error_message}"
) from e
raise
response = self._bedrock_client.converse(**request)
if not isinstance(response, Mapping):
raise ChatClientInvalidResponseException("Bedrock converse response must be a mapping.")
return response
@override
def _inner_get_response(
@@ -368,7 +344,7 @@ class BedrockChatClient(
# Streaming mode - simulate streaming by yielding a single update
async def _stream() -> AsyncIterable[ChatResponseUpdate]:
response = await asyncio.to_thread(self._invoke_converse, request)
parsed_response = self._process_converse_response(response, options)
parsed_response = self._process_converse_response(response)
contents = list(parsed_response.messages[0].contents if parsed_response.messages else [])
if parsed_response.usage_details:
contents.append(Content.from_usage(usage_details=parsed_response.usage_details)) # type: ignore[arg-type]
@@ -384,12 +360,12 @@ class BedrockChatClient(
raw_representation=parsed_response.raw_representation,
)
return self._build_response_stream(_stream(), response_format=options.get("response_format"))
return self._build_response_stream(_stream())
# Non-streaming mode
async def _get_response() -> ChatResponse:
raw_response = await asyncio.to_thread(self._invoke_converse, request)
return self._process_converse_response(raw_response, options)
return self._process_converse_response(raw_response)
return _get_response()
@@ -454,9 +430,6 @@ class BedrockChatClient(
if tool_config:
run_options["toolConfig"] = tool_config
if output_config := self._prepare_output_config(options.get("response_format")):
run_options["outputConfig"] = output_config
return run_options
def _prepare_bedrock_messages(
@@ -655,9 +628,7 @@ class BedrockChatClient(
def _generate_tool_call_id() -> str:
return f"tool-call-{uuid4().hex}"
def _process_converse_response(
self, response: dict[str, Any], options: Mapping[str, Any] | None = None
) -> ChatResponse:
def _process_converse_response(self, response: dict[str, Any]) -> ChatResponse:
"""Convert Bedrock Converse API response to ChatResponse."""
output = response.get("output") or {}
message = output.get("message") or {}
@@ -675,7 +646,6 @@ class BedrockChatClient(
usage_details=usage_details,
model=model,
finish_reason=finish_reason,
response_format=options.get("response_format") if options else None,
raw_representation=response,
)
@@ -758,108 +728,6 @@ class BedrockChatClient(
return None
return FINISH_REASON_MAP.get(reason.lower())
def _prepare_output_config(self, response_format: Any | None) -> dict[str, Any] | None:
"""Convert response_format into the AWS Bedrock outputConfig wire format.
Args:
response_format: A Pydantic model class or a dict schema, or None.
Returns:
A dict for the Converse API ``outputConfig`` parameter, or None if
response_format is not set.
"""
if response_format is None:
return None
if isinstance(response_format, Mapping):
if "json_schema" in response_format:
# Shape A — OpenAI-style wrapper
json_schema_config = response_format["json_schema"]
schema_src = json_schema_config.get("schema", {})
name = json_schema_config.get("name", "output_schema")
elif "schema" in response_format:
# Shape B — inner shape directly {"name": ..., "schema": ...}
schema_src = response_format["schema"]
name = response_format.get("name", "output_schema")
else:
# Shape C — assume entire dict is the raw schema
logger.warning(
"response_format dict has no 'json_schema' or 'schema' key; "
"treating entire dict as raw JSON schema."
)
schema_src = dict(response_format)
name = "output_schema"
if isinstance(schema_src, str):
schema_src = json.loads(schema_src)
schema = copy.deepcopy(schema_src)
else:
if not isinstance(response_format, type) or not issubclass(response_format, BaseModel):
raise TypeError(
"response_format must be None, a dict JSON schema, "
"or a Pydantic BaseModel subclass."
)
# response_format is a Pydantic model class
schema = response_format.model_json_schema()
name = response_format.__name__
self._set_additional_properties_false(schema)
json_schema: dict[str, Any] = {
"name": name,
"schema": json.dumps(schema),
}
description = getattr(response_format, "__doc__", None) if not isinstance(response_format, Mapping) else None
if description and isinstance(description, str) and description.strip():
json_schema["description"] = description.strip()
return {
"textFormat": {
"type": "json_schema",
"structure": {
"jsonSchema": json_schema
},
}
}
def _set_additional_properties_false(self, schema: dict[str, Any]) -> None:
"""Recursively set additionalProperties: false on all object types in a JSON schema.
AWS requires strict schema enforcement. This mirrors the approach used by
AnthropicChatClient._prepare_response_format().
Args:
schema: The JSON schema dict to modify in-place.
"""
visited: set[int] = set()
def walk(node: Any) -> None:
if isinstance(node, dict):
node_id = id(node)
if node_id in visited:
return
visited.add(node_id)
if node.get("type") == "object" or (
"properties" in node and "type" not in node
):
existing = node.get("additionalProperties")
if existing is None or existing is True:
node["additionalProperties"] = False
for value in node.values():
if isinstance(value, (dict, list)):
walk(value)
elif isinstance(node, list):
node_id = id(node)
if node_id in visited:
return
visited.add(node_id)
for item in node:
if isinstance(item, (dict, list)):
walk(item)
walk(schema)
def service_url(self) -> str:
"""Returns the service URL for the Bedrock runtime in the configured AWS region.
@@ -1,382 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
from __future__ import annotations
import copy
import json
from typing import Any
from unittest.mock import patch
import pytest
from agent_framework import Content, Message
from botocore.exceptions import ClientError
from pydantic import BaseModel
from agent_framework_bedrock import BedrockChatClient
# region Test models
class WeatherReport(BaseModel):
city: str
temperature: float
summary: str
class NestedAddress(BaseModel):
street: str
city: str
zip_code: str
class Person(BaseModel):
name: str
age: int
address: NestedAddress
# endregion
# region Helpers
class _StubBedrockRuntime:
"""Stub that records calls and returns a canned response."""
def __init__(self, response_text: str = "Bedrock says hi") -> None:
self.calls: list[dict[str, Any]] = []
self._response_text = response_text
def converse(self, **kwargs: Any) -> dict[str, Any]:
self.calls.append(kwargs)
return {
"modelId": kwargs["modelId"],
"responseId": "resp-structured",
"usage": {"inputTokens": 10, "outputTokens": 20, "totalTokens": 30},
"output": {
"completionReason": "end_turn",
"message": {
"id": "msg-structured",
"role": "assistant",
"content": [{"text": self._response_text}],
},
},
}
def _make_client(response_text: str = "Bedrock says hi") -> tuple[BedrockChatClient, _StubBedrockRuntime]:
stub = _StubBedrockRuntime(response_text)
client = BedrockChatClient(
model="us.anthropic.claude-haiku-4-5-v1:0",
region="us-east-1",
client=stub,
)
return client, stub
def _user_messages() -> list[Message]:
return [Message(role="user", contents=[Content.from_text(text="Give me a weather report")])]
# endregion
# region Tests
def test_prepare_output_config_correct_wire_shape() -> None:
"""_prepare_output_config(WeatherReport) must produce the correct
textFormat → structure → jsonSchema shape with type: 'json_schema'."""
client, _ = _make_client()
output_config = client._prepare_output_config(WeatherReport)
assert output_config is not None
text_format = output_config["textFormat"]
assert text_format["type"] == "json_schema"
assert "structure" in text_format
json_schema = text_format["structure"]["jsonSchema"]
assert json_schema["name"] == "WeatherReport"
assert "schema" in json_schema
def test_prepare_output_config_schema_is_json_string() -> None:
"""The schema value inside jsonSchema must be a JSON string, not a dict."""
client, _ = _make_client()
output_config = client._prepare_output_config(WeatherReport)
assert output_config is not None
schema_value = output_config["textFormat"]["structure"]["jsonSchema"]["schema"]
assert isinstance(schema_value, str), f"Expected str, got {type(schema_value)}"
# Verify it's valid JSON
parsed = json.loads(schema_value)
assert isinstance(parsed, dict)
assert parsed["type"] == "object"
def test_additional_properties_false_set_recursively() -> None:
"""additionalProperties: false must be set on all nested object types."""
client, _ = _make_client()
output_config = client._prepare_output_config(Person)
assert output_config is not None
schema_str = output_config["textFormat"]["structure"]["jsonSchema"]["schema"]
schema = json.loads(schema_str)
# Top-level object
assert schema.get("additionalProperties") is False
# Check $defs for NestedAddress
defs = schema.get("$defs", {})
assert "NestedAddress" in defs, "Expected NestedAddress to be present in $defs"
assert defs["NestedAddress"].get("additionalProperties") is False, (
"Expected additionalProperties=False on nested NestedAddress schema"
)
def test_no_output_config_when_response_format_none() -> None:
"""When response_format is None, no outputConfig key should appear in the request."""
client, stub = _make_client()
messages = _user_messages()
request = client._prepare_options(messages, {"max_tokens": 100})
assert "outputConfig" not in request, (
f"outputConfig should not be present when response_format is None, got: {request.get('outputConfig')}"
)
async def test_chat_response_value_populated() -> None:
"""After a mocked response with response_format, .value should be a populated Pydantic model."""
json_response = json.dumps({"city": "Seattle", "temperature": 72.5, "summary": "Sunny and warm"})
client, stub = _make_client(response_text=json_response)
messages = _user_messages()
response = await client.get_response(
messages=messages,
options={"max_tokens": 100, "response_format": WeatherReport},
)
assert response.text == json_response
assert response.value is not None
assert isinstance(response.value, WeatherReport)
assert response.value.city == "Seattle"
assert response.value.temperature == 72.5
assert response.value.summary == "Sunny and warm"
# Verify outputConfig was sent to the API
assert len(stub.calls) == 1
api_request = stub.calls[0]
assert "outputConfig" in api_request
assert api_request["outputConfig"]["textFormat"]["type"] == "json_schema"
def test_dict_schema_response_format() -> None:
"""_prepare_output_config should work when response_format is a dict, not just a Pydantic class."""
client, _ = _make_client()
dict_schema = {
"json_schema": {
"name": "weather_output",
"schema": {
"type": "object",
"properties": {
"city": {"type": "string"},
"temp": {"type": "number"},
},
},
}
}
output_config = client._prepare_output_config(dict_schema)
assert output_config is not None
json_schema = output_config["textFormat"]["structure"]["jsonSchema"]
assert json_schema["name"] == "weather_output"
schema_parsed = json.loads(json_schema["schema"])
assert schema_parsed["type"] == "object"
assert "city" in schema_parsed["properties"]
def test_prepare_output_config_none_returns_none() -> None:
"""_prepare_output_config(None) must return None."""
client, _ = _make_client()
result = client._prepare_output_config(None)
assert result is None
async def test_chat_response_value_populated_streaming() -> None:
"""In streaming mode, .value should also be populated on the final response."""
json_response = json.dumps({"city": "Portland", "temperature": 68.0, "summary": "Cloudy"})
client, stub = _make_client(response_text=json_response)
messages = _user_messages()
stream = client.get_response(
messages=messages,
stream=True,
options={"max_tokens": 100, "response_format": WeatherReport},
)
# Consume stream and get final response
async for _ in stream:
pass
response = await stream.get_final_response()
assert response.value is not None
assert isinstance(response.value, WeatherReport)
assert response.value.city == "Portland"
# Verify outputConfig was sent
assert len(stub.calls) == 1
assert "outputConfig" in stub.calls[0]
async def test_unsupported_model_validation_exception() -> None:
"""When a model doesn't support outputConfig, a clear error should be raised."""
class _FailingStubBedrockRuntime:
def converse(self, **kwargs: Any) -> dict[str, Any]:
# Simulate botocore ClientError for ValidationException
error_response = {"Error": {"Code": "ValidationException", "Message": "Invalid field outputConfig"}}
raise ClientError(error_response, "Converse")
client = BedrockChatClient(
model="us.anthropic.claude-v2",
region="us-east-1",
client=_FailingStubBedrockRuntime(),
)
with pytest.raises(ValueError) as exc:
await client.get_response(
messages=_user_messages(),
options={"response_format": WeatherReport},
)
assert "does not support structured output via outputConfig.textFormat" in str(exc.value)
assert "Check the model's Bedrock Converse outputConfig/textFormat support." in str(exc.value)
def test_invalid_response_format_type_raises() -> None:
"""Non-dict, non-BaseModel response_format should raise TypeError."""
client, _ = _make_client()
with pytest.raises(TypeError, match="Pydantic BaseModel subclass"):
client._prepare_output_config("not_a_valid_format")
def test_mapping_response_format_accepted() -> None:
"""A non-dict Mapping response_format must be accepted and produce
correct outputConfig, not raise TypeError."""
from collections.abc import MutableMapping
class _WrappedMapping(MutableMapping):
def __init__(self, data):
self._data = dict(data)
def __getitem__(self, key):
return self._data[key]
def __setitem__(self, key, value):
self._data[key] = value
def __delitem__(self, key):
del self._data[key]
def __iter__(self):
return iter(self._data)
def __len__(self):
return len(self._data)
client, _ = _make_client()
mapping_format = _WrappedMapping({
"json_schema": {
"name": "test_output",
"schema": {
"type": "object",
"properties": {"result": {"type": "string"}},
},
}
})
output_config = client._prepare_output_config(mapping_format)
assert output_config is not None
json_schema = output_config["textFormat"]["structure"]["jsonSchema"]
assert json_schema["name"] == "test_output"
schema = json.loads(json_schema["schema"])
assert schema.get("additionalProperties") is False
def test_shape_b_dict_schema_wire_format() -> None:
"""Dict response_format in Shape B (inner shape directly) should
produce correct outputConfig."""
client, _ = _make_client()
response_format = {
"name": "weather_output",
"schema": {
"type": "object",
"properties": {
"city": {"type": "string"},
"temperature": {"type": "number"},
},
},
}
output_config = client._prepare_output_config(response_format)
assert output_config is not None
text_format = output_config["textFormat"]
assert text_format["type"] == "json_schema"
json_schema = text_format["structure"]["jsonSchema"]
assert json_schema["name"] == "weather_output"
schema = json.loads(json_schema["schema"])
assert schema.get("additionalProperties") is False
def test_dict_schema_not_mutated() -> None:
"""Caller's dict schema must not be mutated by _prepare_output_config."""
client, _ = _make_client()
original_schema = {
"json_schema": {
"name": "test",
"schema": {
"type": "object",
"properties": {"a": {"type": "string"}},
},
}
}
snapshot = copy.deepcopy(original_schema)
client._prepare_output_config(original_schema)
assert original_schema == snapshot, "Original dict schema was mutated"
async def test_non_outputconfig_validation_exception_propagates() -> None:
"""ValidationException unrelated to outputConfig must propagate
as raw ClientError, not be caught and reclassified."""
client, _ = _make_client()
error_response = {
"Error": {
"Code": "ValidationException",
"Message": "Invalid message format",
}
}
with (
patch.object(
client,
"_bedrock_client",
**{"converse.side_effect": ClientError(error_response, "Converse")},
),
pytest.raises(ClientError),
):
await client.get_response(
messages=_user_messages(),
options={"max_tokens": 100},
)
# endregion
@@ -71,7 +71,6 @@ from ._evaluation import (
Evaluator,
ExpectedToolCall,
LocalEvaluator,
RubricScore,
evaluate_agent,
evaluate_workflow,
evaluator,
@@ -461,7 +460,6 @@ __all__ = [
"ResponseStream",
"Role",
"RoleLiteral",
"RubricScore",
"RunContext",
"Runner",
"RunnerContext",
@@ -311,15 +311,12 @@ class EvalScoreResult:
score: Numeric score from the evaluator.
passed: Whether the item passed this evaluator's threshold.
sample: Optional raw evaluator output (rationale, metadata).
dimensions: Per-dimension scores when this evaluator is a rubric
evaluator. ``None`` for non-rubric (e.g. built-in) evaluators.
"""
name: str
score: float
passed: bool | None = None
sample: dict[str, Any] | None = None
dimensions: list[RubricScore] | None = None
@experimental(feature_id=ExperimentalFeature.EVALS)
@@ -499,179 +496,6 @@ class EvalResults:
detail += f" Errored items: {', '.join(summaries)}."
raise EvalNotPassedError(detail)
def assert_score_at_least(
self,
min_score: float,
*,
evaluator: str | None = None,
msg: str | None = None,
) -> None:
"""Assert every item's score (optionally filtered by evaluator) is ``>= min_score``.
Designed for CI gates on generated rubric evaluators (e.g.
``results.assert_score_at_least(0.80)``). Includes any
sub-results from workflow evaluations.
Args:
min_score: Minimum acceptable score (inclusive).
evaluator: When set, only check scores from the evaluator
whose ``EvalScoreResult.name`` matches.
msg: Optional custom failure message.
Raises:
EvalNotPassedError: When any matching score is below the threshold.
"""
offenders: list[str] = []
def _check(results: EvalResults) -> None:
for item in results.items:
for score in item.scores:
if evaluator is not None and score.name != evaluator:
continue
if score.score < min_score:
offenders.append(f"{item.item_id}/{score.name}={score.score:.3f}")
for sub in results.sub_results.values():
_check(sub)
_check(self)
if offenders:
detail = msg or (
f"{len(offenders)} score(s) below threshold {min_score}"
f"{' for ' + evaluator if evaluator else ''}: {', '.join(offenders[:5])}"
+ (f" (+{len(offenders) - 5} more)" if len(offenders) > 5 else "")
)
raise EvalNotPassedError(detail)
def assert_dimension_score_at_least(
self,
dimension_id: str,
min_score: float,
*,
evaluator: str | None = None,
require_applicable: bool = False,
msg: str | None = None,
) -> None:
"""Assert every item's score for a rubric *dimension* is ``>= min_score``.
Walks ``EvalScoreResult.dimensions`` looking for the named
dimension across all items (and sub-results). Non-applicable
dimensions are skipped by default; pass
``require_applicable=True`` to fail when no applicable score is
produced.
Args:
dimension_id: Dimension id (matches the rubric definition).
min_score: Minimum acceptable dimension score (inclusive).
evaluator: When set, only consider scores from the evaluator
whose ``EvalScoreResult.name`` matches.
require_applicable: When ``True``, missing or non-applicable
dimension scores raise. Defaults to ``False`` (skip).
msg: Optional custom failure message.
Raises:
EvalNotPassedError: When the dimension fails the threshold.
"""
offenders: list[str] = []
missing_items: list[str] = []
def _check(results: EvalResults) -> None:
for item in results.items:
found_applicable = False
for score in item.scores:
if evaluator is not None and score.name != evaluator:
continue
if not score.dimensions:
continue
for rs in score.dimensions:
if rs.id != dimension_id:
continue
if not rs.applicable:
continue
found_applicable = True
if rs.score is None or rs.score < min_score:
offenders.append(
f"{item.item_id}/{score.name}/{dimension_id}="
f"{rs.score if rs.score is not None else 'None'}"
)
if require_applicable and not found_applicable:
missing_items.append(item.item_id)
for sub in results.sub_results.values():
_check(sub)
_check(self)
problems: list[str] = []
if offenders:
problems.append(
f"{len(offenders)} dimension score(s) for '{dimension_id}' below {min_score}: "
f"{', '.join(offenders[:5])}" + (f" (+{len(offenders) - 5} more)" if len(offenders) > 5 else "")
)
if missing_items:
problems.append(
f"Dimension '{dimension_id}' not applicable on {len(missing_items)} item(s): "
f"{', '.join(missing_items[:5])}"
)
if problems:
raise EvalNotPassedError(msg or "; ".join(problems))
def assert_no_failed_items(self, msg: str | None = None) -> None:
"""Assert no item ended in ``fail`` or ``error`` status.
Includes any sub-results from workflow evaluations.
Args:
msg: Optional custom failure message.
Raises:
EvalNotPassedError: When any item failed or errored.
"""
bad: list[str] = []
def _check(results: EvalResults) -> None:
for item in results.items:
if item.is_failed or item.is_error:
bad.append(f"{item.item_id}:{item.status}")
for sub in results.sub_results.values():
_check(sub)
_check(self)
if bad:
detail = msg or (
f"{len(bad)} item(s) failed or errored: {', '.join(bad[:5])}"
+ (f" (+{len(bad) - 5} more)" if len(bad) > 5 else "")
)
raise EvalNotPassedError(detail)
# endregion
# region Generated rubric evaluators
@experimental(feature_id=ExperimentalFeature.EVALS)
@dataclass(frozen=True)
class RubricScore:
"""A single dimension's score from a rubric-based evaluator run.
Rubric evaluators emit one ``RubricScore`` per dimension per item.
Attached to :class:`EvalScoreResult` as a typed view of the raw
``properties.rubric_scores`` payload returned by providers such as
Foundry's generated rubric evaluators.
Attributes:
id: Dimension id (matches the rubric definition).
score: Numeric score, or ``None`` when the dimension was marked
non-applicable for this item.
applicable: Whether the dimension applied to this item.
weight: Dimension weight (mirrors the rubric definition).
reason: Short rationale produced by the evaluator.
"""
id: str
score: int | None
applicable: bool
weight: int
reason: str
# endregion
@@ -50,7 +50,6 @@ class ExperimentalFeature(str, Enum):
on enum membership or attribute presence over time.
"""
DECLARATIVE_AGENTS = "DECLARATIVE_AGENTS"
EVALS = "EVALS"
FILE_HISTORY = "FILE_HISTORY"
FIDES = "FIDES"
@@ -14,13 +14,12 @@ import logging
from collections.abc import Callable, Sequence
from typing import TYPE_CHECKING, Any
from .._agents import Agent, SupportsAgentRun
from .._agents import Agent
from .._clients import SupportsWebSearchTool
from .._compaction import CompactionProvider, ContextWindowCompactionStrategy, ToolResultCompactionStrategy
from .._feature_stage import ExperimentalFeature, experimental
from .._sessions import ContextProvider, HistoryProvider, InMemoryHistoryProvider
from .._skills import SkillsProvider
from ._background_agents import BackgroundAgentsProvider
from ._memory import MemoryContextProvider, MemoryStore
from ._mode import AgentModeProvider
from ._todo import TodoProvider
@@ -104,8 +103,6 @@ def _assemble_context_providers(
memory_store: MemoryStore | None,
skills_provider: SkillsProvider | None,
skills_paths: Sequence[str] | None,
background_agents: Sequence[SupportsAgentRun] | None,
background_agents_instructions: str | None,
extra_context_providers: Sequence[ContextProvider] | None,
) -> list[ContextProvider]:
"""Assemble the ordered list of context providers."""
@@ -133,10 +130,6 @@ def _assemble_context_providers(
if skills_paths:
providers.append(SkillsProvider.from_paths(*skills_paths))
# Background agents are opt-in: only added when agents are provided.
if background_agents:
providers.append(BackgroundAgentsProvider(background_agents, instructions=background_agents_instructions))
# Append any user-supplied additional providers.
if extra_context_providers:
providers.extend(extra_context_providers)
@@ -172,8 +165,6 @@ def create_harness_agent(
memory_store: MemoryStore | None = None,
skills_provider: SkillsProvider | None = None,
skills_paths: Sequence[str] | None = None,
background_agents: Sequence[SupportsAgentRun] | None = None,
background_agents_instructions: str | None = None,
disable_web_search: bool = False,
otel_provider_name: str | None = None,
context_providers: Sequence[ContextProvider] | None = None,
@@ -191,7 +182,6 @@ def create_harness_agent(
- **AgentModeProvider** — plan/execute mode tracking
- **MemoryContextProvider** — file-based durable memory (when ``memory_store`` provided)
- **SkillsProvider** — skill discovery and progressive loading
- **BackgroundAgentsProvider** — delegate work to background sub-agents
- **OpenTelemetry** — observability via ``AgentTelemetryLayer``
Each feature can be disabled or customized via keyword arguments.
@@ -263,13 +253,6 @@ def create_harness_agent(
skills_paths: Paths for file-based skill discovery (looks for SKILL.md files).
Can be combined with ``skills_provider``. When neither ``skills_provider``
nor ``skills_paths`` is provided, no SkillsProvider is added.
background_agents: Collection of agents available for background task delegation.
When provided, a ``BackgroundAgentsProvider`` is automatically included,
enabling the agent to start, monitor, and retrieve results from background tasks.
Each agent must have a non-empty, unique name (case-insensitive).
background_agents_instructions: Optional instruction override for the
``BackgroundAgentsProvider``. May include ``{background_agents}`` placeholder
which will be replaced with the agent listing.
disable_web_search: When True, skip automatic web search tool inclusion.
When False (default), the web search tool is automatically added if the
client implements SupportsWebSearchTool. A warning is logged if the client
@@ -319,8 +302,6 @@ def create_harness_agent(
memory_store=memory_store,
skills_provider=skills_provider,
skills_paths=skills_paths,
background_agents=background_agents,
background_agents_instructions=background_agents_instructions,
extra_context_providers=context_providers,
)
@@ -36,10 +36,11 @@ if TYPE_CHECKING:
from pydantic import BaseModel
from ._agents import SupportsAgentRun
from ._clients import SupportsChatGetResponse
from ._compaction import CompactionStrategy, TokenizerProtocol
from ._sessions import AgentSession
from ._tools import FunctionTool, ToolTypes
from ._types import ChatOptions
from ._types import ChatOptions, ChatResponse, ChatResponseUpdate
ResponseModelBoundT = TypeVar("ResponseModelBoundT", bound=BaseModel)
@@ -7,8 +7,6 @@ import json
import logging
import re
from collections.abc import Mapping, MutableMapping
from dataclasses import asdict, is_dataclass
from datetime import date, datetime
from typing import Any, ClassVar, Protocol, TypeVar, runtime_checkable
logger = logging.getLogger("agent_framework")
@@ -616,46 +614,3 @@ class SerializationMixin:
# Fallback and default
# Convert class name to snake_case
return _CAMEL_TO_SNAKE_PATTERN.sub("_", cls.__name__).lower()
def make_json_safe(obj: Any) -> Any:
"""Recursively convert an object to a JSON-serializable form.
Handles dataclasses, Pydantic models, objects with ``to_dict``/``dict``/``__dict__``,
datetimes, lists, dicts, and primitives. Falls back to ``str()`` for any remaining
non-serializable value so that ``json.dumps`` never raises a ``TypeError``.
Args:
obj: Object to make JSON safe.
Returns:
A JSON-serializable version of the object.
"""
if obj is None or isinstance(obj, (str, int, float, bool)):
return obj
if isinstance(obj, (datetime, date)):
return obj.isoformat()
if is_dataclass(obj) and not isinstance(obj, type):
return make_json_safe(asdict(obj)) # type: ignore[arg-type]
if callable(getattr(obj, "model_dump", None)):
try:
return make_json_safe(obj.model_dump()) # type: ignore[no-any-return]
except TypeError:
pass
if callable(getattr(obj, "to_dict", None)):
try:
return make_json_safe(obj.to_dict()) # type: ignore[no-any-return]
except TypeError:
pass
if callable(getattr(obj, "dict", None)):
try:
return make_json_safe(obj.dict()) # type: ignore[no-any-return]
except TypeError:
pass
if isinstance(obj, dict):
return {str(key): make_json_safe(value) for key, value in obj.items()} # type: ignore[misc]
if isinstance(obj, (list, tuple)):
return [make_json_safe(item) for item in obj] # type: ignore[misc]
if hasattr(obj, "__dict__"):
return {key: make_json_safe(value) for key, value in vars(obj).items()} # type: ignore[misc]
return str(obj)
@@ -2134,7 +2134,9 @@ class SkillsProvider(ContextProvider):
),
FunctionTool(
name="read_skill_resource",
description=("Reads a resource associated with a skill, such as references, assets, or dynamic data."),
description=(
"Reads a resource associated with a skill, such as references, assets, or dynamic data."
),
func=_read_resource,
input_model={
"type": "object",
@@ -2171,7 +2173,8 @@ class SkillsProvider(ContextProvider):
"type": "object",
"additionalProperties": True,
"description": (
'Named arguments as key-value pairs (e.g. {"length": 24, "uppercase": true}).'
"Named arguments as key-value pairs "
'(e.g. {"length": 24, "uppercase": true}).'
),
},
{
@@ -2,6 +2,7 @@
from __future__ import annotations
import json
import logging
import sys
import uuid
@@ -28,12 +29,11 @@ from .._types import (
UsageDetails,
add_usage_details,
)
from ..exceptions import AgentException, AgentInvalidRequestException, AgentInvalidResponseException
from ..exceptions import AgentInvalidRequestException, AgentInvalidResponseException
from ._checkpoint import CheckpointStorage
from ._events import (
AGENT_FORWARDED_EVENT_TYPES,
WorkflowEvent,
WorkflowRunState,
)
from ._message_utils import normalize_messages_input
from ._typing_utils import is_instance_of, is_type_compatible
@@ -58,24 +58,27 @@ class WorkflowAgent(BaseAgent):
@dataclass
class RequestInfoFunctionArgs:
request_id: str
request_event: WorkflowEvent
data: Any
def to_dict(self) -> dict[str, Any]:
return {"request_id": self.request_id, "request_event": self.request_event.to_dict()}
return {"request_id": self.request_id, "data": self.data}
def to_json(self) -> str:
return json.dumps(self.to_dict())
@classmethod
def from_dict(cls, payload: dict[str, Any]) -> WorkflowAgent.RequestInfoFunctionArgs:
if "request_id" not in payload or "request_event" not in payload:
raise ValueError(
"Invalid payload for RequestInfoFunctionArgs. 'request_id' and 'request_event' are required."
)
if not payload["request_id"]:
raise ValueError("request_id cannot be empty.")
return cls(request_id=payload.get("request_id", ""), data=payload.get("data"))
return cls(
request_id=payload.get("request_id", ""),
request_event=WorkflowEvent.from_dict(payload.get("request_event", {})),
)
@classmethod
def from_json(cls, raw: str) -> WorkflowAgent.RequestInfoFunctionArgs:
try:
parsed: Any = json.loads(raw)
except json.JSONDecodeError as exc:
raise ValueError(f"RequestInfoFunctionArgs JSON payload is malformed: {exc}") from exc
if not isinstance(parsed, dict):
raise ValueError("RequestInfoFunctionArgs JSON payload must decode to a mapping")
return cls.from_dict(cast(dict[str, Any], parsed))
def __init__(
self,
@@ -125,11 +128,16 @@ class WorkflowAgent(BaseAgent):
**kwargs,
)
self._workflow: Workflow = workflow
self._pending_requests: dict[str, WorkflowEvent[Any]] = {}
@property
def workflow(self) -> Workflow:
return self._workflow
@property
def pending_requests(self) -> dict[str, WorkflowEvent[Any]]:
return self._pending_requests
# region Run Methods
@overload
@@ -173,7 +181,7 @@ class WorkflowAgent(BaseAgent):
Args:
messages: The message(s) to send to the workflow. Required for new runs,
could be None if only restoring the underlying workflow from a checkpoint.
should be None when resuming from checkpoint.
Keyword Args:
stream: If True, returns an async iterable of updates. If False (default),
@@ -407,44 +415,16 @@ class WorkflowAgent(BaseAgent):
Yields:
WorkflowEvent objects from the workflow execution.
"""
# Restore the workflow state if a checkpoint is provided
if checkpoint_id is not None:
if checkpoint_storage is None:
raise AgentInvalidRequestException("checkpoint_storage must be provided when checkpoint_id is provided")
logger.debug(f"Restoring workflow from checkpoint {checkpoint_id}")
# Restore the workflow from checkpoint
if streaming:
async for _ in self.workflow.run(
stream=True,
checkpoint_id=checkpoint_id,
checkpoint_storage=checkpoint_storage,
):
pass
else:
_ = await self.workflow.run(
checkpoint_id=checkpoint_id,
checkpoint_storage=checkpoint_storage,
)
if not input_messages:
logger.info("No input messages provided; the workflow has been restored to the checkpoint state.")
return
final_state = self._workflow.status
logger.debug(f"Workflow state: {final_state}")
if final_state == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS:
# Extract function responses from input messages, and ensure that
# only function responses are present in messages if there is any
# pending request.
# NOTE: It is possible that some pending requests are not fulfilled,
# and we will let the workflow to handle this -- the agent does not
# have an opinion on this.
function_responses = self._extract_function_responses(input_messages)
# Determine the execution mode based on state.
# The streaming flag controls the workflow's internal streaming mode,
# which affects executor behavior (e.g. AgentExecutor emits different event
# types in streaming vs non-streaming mode).
if bool(self.pending_requests):
function_responses = self._process_pending_requests(input_messages)
if streaming:
async for event in self.workflow.run(
responses=function_responses,
stream=True,
checkpoint_storage=checkpoint_storage,
function_invocation_kwargs=function_invocation_kwargs,
client_kwargs=client_kwargs,
):
@@ -452,16 +432,23 @@ class WorkflowAgent(BaseAgent):
else:
for event in await self.workflow.run(
responses=function_responses,
checkpoint_storage=checkpoint_storage,
function_invocation_kwargs=function_invocation_kwargs,
client_kwargs=client_kwargs,
):
yield event
elif final_state == WorkflowRunState.IDLE:
elif checkpoint_id is not None:
# Restore the prior workflow state from the checkpoint. Shared
# state (e.g. accumulated conversation history maintained by the
# workflow's executors) survives across turns because Workflow.run
# no longer wipes state per call. Callers who want to deliver a
# new user message after restore should make a second
# `workflow.run(message=...)` call - they are NOT mutually
# exclusive on the same instance, but each must be its own call.
if streaming:
async for event in self.workflow.run(
message=input_messages,
stream=True,
checkpoint_id=checkpoint_id,
checkpoint_storage=checkpoint_storage,
function_invocation_kwargs=function_invocation_kwargs,
client_kwargs=client_kwargs,
@@ -469,17 +456,60 @@ class WorkflowAgent(BaseAgent):
yield event
else:
for event in await self.workflow.run(
message=input_messages,
checkpoint_id=checkpoint_id,
checkpoint_storage=checkpoint_storage,
function_invocation_kwargs=function_invocation_kwargs,
client_kwargs=client_kwargs,
):
yield event
else:
raise AgentException(f"The underlying workflow is in an invalid state to restart: {final_state}.")
if streaming:
async for event in self.workflow.run(
message=input_messages,
stream=True,
checkpoint_storage=checkpoint_storage,
function_invocation_kwargs=function_invocation_kwargs,
client_kwargs=client_kwargs,
):
yield event
else:
for event in await self.workflow.run(
message=input_messages,
checkpoint_storage=checkpoint_storage,
function_invocation_kwargs=function_invocation_kwargs,
client_kwargs=client_kwargs,
):
yield event
# endregion Run Methods
def _process_pending_requests(self, input_messages: Sequence[Message]) -> dict[str, Any]:
"""Process pending requests by extracting function responses and updating state.
Args:
input_messages: Input messages that may contain function responses.
Returns:
A dictionary mapping request IDs to their response data.
"""
logger.info(f"Continuing workflow to address {len(self.pending_requests)} requests")
# Extract function responses from input messages, and ensure that
# only function responses are present in messages if there is any
# pending request.
function_responses = self._extract_function_responses(input_messages)
# Pop pending requests if fulfilled.
for request_id in list(self.pending_requests.keys()):
if request_id in function_responses:
self.pending_requests.pop(request_id)
# NOTE: It is possible that some pending requests are not fulfilled,
# and we will let the workflow to handle this -- the agent does not
# have an opinion on this.
return function_responses
def _convert_workflow_events_to_agent_response(
self,
response_id: str,
@@ -497,10 +527,10 @@ class WorkflowAgent(BaseAgent):
for output_event in output_events:
if output_event.type == "request_info":
request_content = self._process_request_info_event(output_event)
function_call, approval_request = self._process_request_info_event(output_event)
messages.append(
Message(
contents=[request_content],
contents=[function_call, approval_request],
role="assistant",
author_name=output_event.source_executor_id,
message_id=str(uuid.uuid4()),
@@ -567,6 +597,38 @@ class WorkflowAgent(BaseAgent):
raw_representation=raw_representations,
)
def _process_request_info_event(
self,
event: WorkflowEvent[Any],
) -> tuple[Content, Content]:
"""Convert a request_info event to FunctionCallContent and FunctionApprovalRequestContent.
Args:
event: A WorkflowEvent with type='request_info'.
Returns:
A tuple of (FunctionCallContent, FunctionApprovalRequestContent).
"""
request_id = event.request_id
if not request_id:
raise ValueError("request_info event must have a request_id")
self.pending_requests[request_id] = event
args = self.RequestInfoFunctionArgs(request_id=request_id, data=event.data).to_dict()
function_call = Content.from_function_call(
call_id=request_id,
name=self.REQUEST_INFO_FUNCTION_NAME,
arguments=args,
)
approval_request = Content.from_function_approval_request(
id=request_id,
function_call=function_call,
additional_properties={"request_id": request_id},
)
return function_call, approval_request
def _convert_workflow_event_to_agent_response_updates(
self,
response_id: str,
@@ -668,72 +730,85 @@ class WorkflowAgent(BaseAgent):
]
if event.type == "request_info":
request_content = self._process_request_info_event(event)
# Store the pending request for later correlation
request_id = event.request_id
if not request_id:
raise ValueError("request_info event must have a request_id")
self.pending_requests[request_id] = event
args = self.RequestInfoFunctionArgs(request_id=request_id, data=event.data).to_dict()
function_call = Content.from_function_call(
call_id=request_id,
name=self.REQUEST_INFO_FUNCTION_NAME,
arguments=args,
)
approval_request = Content.from_function_approval_request(
id=request_id,
function_call=function_call,
additional_properties={"request_id": request_id},
)
return [
AgentResponseUpdate(
contents=[request_content],
contents=[function_call, approval_request],
role="assistant",
author_name=self.name,
response_id=response_id,
message_id=str(uuid.uuid4()),
created_at=datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
raw_representation=event,
)
]
# Ignore workflow-internal events
return []
def _process_request_info_event(
self,
event: WorkflowEvent[Any],
) -> Content:
"""Convert a request_info event to FunctionApprovalRequestContent.
Args:
event: A WorkflowEvent with type='request_info'.
Returns:
A content object representing the request info. The content can be a `function_approval_request`
or a `function_call` depending on the structure of the event data.
Note:
If the event data is already a FunctionApprovalRequestContent, it will be returned as-is.
"""
if isinstance(event.data, Content) and event.data.user_input_request:
# Return the event data as-is if it's already a properly formed FunctionApprovalRequestContent
return event.data
request_id = event.request_id
args = self.RequestInfoFunctionArgs(request_id=request_id, request_event=event).to_dict()
return Content.from_function_call(
call_id=request_id,
name=self.REQUEST_INFO_FUNCTION_NAME,
arguments=args,
)
def _extract_function_responses(self, input_messages: Sequence[Message]) -> dict[str, Any]:
"""Extract function responses from input messages.
The responses are for pending requests that the workflow is waiting on, and
will be passed to the workflow. The pending requests are processed to either
`function_approval_request` or `function_call` content by `_process_request_info_event`.
"""
"""Extract function responses from input messages."""
function_responses: dict[str, Any] = {}
for message in input_messages:
for content in message.contents:
if content.type == "function_approval_response":
request_id: str = content.id # type: ignore[assignment]
function_responses[request_id] = content
elif content.type == "function_result":
response_data = content.result if hasattr(content, "result") else str(content) # type: ignore[attr-defined]
function_responses[content.call_id] = response_data # type: ignore
else:
raise AgentInvalidResponseException(
"Unexpected content type while awaiting request info responses."
)
# Parse the function arguments to recover request payload
arguments_payload = content.function_call.arguments # type: ignore[attr-defined, union-attr]
if isinstance(arguments_payload, str):
try:
parsed_args = self.RequestInfoFunctionArgs.from_json(arguments_payload)
except ValueError as exc:
raise AgentInvalidResponseException(
"FunctionApprovalResponseContent arguments must decode to a mapping."
) from exc
elif isinstance(arguments_payload, dict):
parsed_args = self.RequestInfoFunctionArgs.from_dict(arguments_payload)
else:
raise AgentInvalidResponseException(
"FunctionApprovalResponseContent arguments must be a mapping or JSON string."
)
request_id = parsed_args.request_id or content.id # type: ignore[attr-defined]
if not content.approved: # type: ignore[attr-defined]
raise AgentInvalidResponseException(f"Request '{request_id}' was not approved by the caller.")
if request_id in self.pending_requests:
function_responses[request_id] = parsed_args.data
elif bool(self.pending_requests):
raise AgentInvalidRequestException(
"Only responses for pending requests are allowed when there are outstanding approvals."
)
elif content.type == "function_result":
request_id = content.call_id # type: ignore[attr-defined]
if request_id in self.pending_requests:
response_data = content.result if hasattr(content, "result") else str(content) # type: ignore[attr-defined]
function_responses[request_id] = response_data
elif bool(self.pending_requests):
raise AgentInvalidRequestException(
"Only function responses for pending requests are allowed while requests are outstanding."
)
else:
if bool(self.pending_requests):
raise AgentInvalidResponseException(
"Unexpected content type while awaiting request info responses."
)
return function_responses
def _extract_contents(self, data: Any) -> list[Content]:
@@ -429,30 +429,15 @@ class AgentExecutor(Executor):
function_invocation_kwargs=function_invocation_kwargs,
client_kwargs=client_kwargs,
)
await ctx.yield_output(response)
# Handle any user input requests
if response.user_input_requests:
user_input_request_count = len(response.user_input_requests)
total_message_content_count = sum(len(msg.contents) for msg in response.messages)
if user_input_request_count != total_message_content_count:
logger.warning(
"Response %s contains %d user input requests but total message contents are %d. "
"This indicates the response contains both user input requests and message contents. "
"Double check if this is the intended behavior, as non user input request contents in "
"this response will not be emitted.",
response.response_id,
user_input_request_count,
total_message_content_count,
)
for user_input_request in response.user_input_requests:
self._pending_agent_requests[user_input_request.id] = user_input_request # type: ignore[index]
await ctx.request_info(user_input_request, Content, request_id=user_input_request.id)
await ctx.request_info(user_input_request, Content)
return None
# Only yield output if the response is complete and not waiting for user input.
# This is to avoid emitting two events of different types ('output' and 'request_info')
# that carry the same payload.
await ctx.yield_output(response)
return response
async def _run_agent_streaming(self, ctx: WorkflowContext[Never, AgentResponseUpdate]) -> AgentResponse | None:
@@ -487,25 +472,9 @@ class AgentExecutor(Executor):
)
async for update in stream:
updates.append(update)
await ctx.yield_output(update)
if update.user_input_requests:
user_input_request_count = len(update.user_input_requests)
total_message_content_count = len(update.contents)
if user_input_request_count != total_message_content_count:
logger.warning(
"Response update %s contains %d user input requests but total message contents are %d. "
"This indicates the response update contains both user input requests and message contents. "
"Double check if this is the intended behavior, as non user input request contents will "
"not be emitted.",
update.response_id,
user_input_request_count,
total_message_content_count,
)
streamed_user_input_requests.extend(update.user_input_requests)
else:
# Only yield output events for updates that do not contain user input requests.
# This is to avoid emitting two events of different types ('output' and 'request_info')
# that carry the same payload.
await ctx.yield_output(update)
# Prefer stream finalization when available so result hooks run
# (e.g., thread conversation updates). Fall back to reconstructing from updates
@@ -540,7 +509,7 @@ class AgentExecutor(Executor):
if user_input_requests:
for user_input_request in user_input_requests:
self._pending_agent_requests[user_input_request.id] = user_input_request # type: ignore[index]
await ctx.request_info(user_input_request, Content, request_id=user_input_request.id)
await ctx.request_info(user_input_request, Content)
return None
return response
@@ -47,7 +47,6 @@ from copy import deepcopy
from typing import Any, Generic, Literal, TypeVar, overload
from .._feature_stage import ExperimentalFeature, experimental
from .._serialization import make_json_safe
from .._types import AgentResponse, AgentResponseUpdate, ResponseStream
from ..observability import OtelAttr, capture_exception, create_workflow_span
from ._checkpoint import CheckpointStorage, WorkflowCheckpoint
@@ -1516,7 +1515,7 @@ class FunctionalWorkflowAgent:
function_call = Content.from_function_call(
call_id=request_id,
name=self.REQUEST_INFO_FUNCTION_NAME,
arguments={"request_id": request_id, "data": make_json_safe(event.data)},
arguments={"request_id": request_id, "data": event.data},
)
return Content.from_function_approval_request(
id=request_id,
@@ -360,22 +360,6 @@ class Workflow(DictConvertible):
# Flag to prevent concurrent workflow executions
self._is_running = False
# Current run-level status of this workflow instance. Updated in lockstep with
# the status events emitted from `_run_workflow_with_tracing`. Defaults to IDLE
# for a freshly built workflow that has not yet been run.
self._status: WorkflowRunState = WorkflowRunState.IDLE
@property
def status(self) -> WorkflowRunState:
"""Return the current run-level status of this workflow instance.
Mirrors the most recent status event emitted by the workflow. Safe to read at
any time: workflows run on a single asyncio event loop, and the underlying
attribute is a single enum reference whose assignment is atomic under the
CPython GIL, so no locking is required.
"""
return self._status
def _ensure_not_running(self) -> None:
"""Ensure the workflow is not already running."""
if self._is_running:
@@ -529,9 +513,8 @@ class Workflow(DictConvertible):
with _framework_event_origin():
started = WorkflowEvent.started()
yield started # noqa: RUF070
self._status = WorkflowRunState.IN_PROGRESS
with _framework_event_origin():
in_progress = WorkflowEvent.status(self._status)
in_progress = WorkflowEvent.status(WorkflowRunState.IN_PROGRESS)
yield in_progress # noqa: RUF070
# Per-run reset for fresh-message runs only. We deliberately
@@ -586,20 +569,17 @@ class Workflow(DictConvertible):
if event.type == "request_info" and not emitted_in_progress_pending:
emitted_in_progress_pending = True
self._status = WorkflowRunState.IN_PROGRESS_PENDING_REQUESTS
with _framework_event_origin():
pending_status = WorkflowEvent.status(self._status)
pending_status = WorkflowEvent.status(WorkflowRunState.IN_PROGRESS_PENDING_REQUESTS)
yield pending_status # noqa: RUF070
# Workflow runs until idle - emit final status based on whether requests are pending
if saw_request:
self._status = WorkflowRunState.IDLE_WITH_PENDING_REQUESTS
with _framework_event_origin():
terminal_status = WorkflowEvent.status(self._status)
terminal_status = WorkflowEvent.status(WorkflowRunState.IDLE_WITH_PENDING_REQUESTS)
yield terminal_status
else:
self._status = WorkflowRunState.IDLE
with _framework_event_origin():
terminal_status = WorkflowEvent.status(self._status)
terminal_status = WorkflowEvent.status(WorkflowRunState.IDLE)
yield terminal_status
span.add_event(OtelAttr.WORKFLOW_COMPLETED)
@@ -613,7 +593,6 @@ class Workflow(DictConvertible):
with _framework_event_origin():
failed_event = WorkflowEvent.failed(details)
yield failed_event # noqa: RUF070
self._status = WorkflowRunState.FAILED
with _framework_event_origin():
failed_status = WorkflowEvent.status(WorkflowRunState.FAILED)
yield failed_status # noqa: RUF070
@@ -34,7 +34,6 @@ _IMPORTS: dict[str, tuple[str, str]] = {
"FoundryLocalChatOptions": ("agent_framework_foundry_local", "agent-framework-foundry-local"),
"FoundryLocalClient": ("agent_framework_foundry_local", "agent-framework-foundry-local"),
"FoundryLocalSettings": ("agent_framework_foundry_local", "agent-framework-foundry-local"),
"GeneratedEvaluatorRef": ("agent_framework_foundry", "agent-framework-foundry"),
"RawAnthropicFoundryClient": ("agent_framework_anthropic", "agent-framework-anthropic"),
"RawFoundryAgent": ("agent_framework_foundry", "agent-framework-foundry"),
"RawFoundryAgentChatClient": ("agent_framework_foundry", "agent-framework-foundry"),
@@ -20,7 +20,6 @@ from agent_framework_foundry import (
FoundryEmbeddingSettings,
FoundryEvals,
FoundryMemoryProvider,
GeneratedEvaluatorRef,
RawFoundryAgent,
RawFoundryAgentChatClient,
RawFoundryChatClient,
@@ -53,7 +52,6 @@ __all__ = [
"FoundryLocalClient",
"FoundryLocalSettings",
"FoundryMemoryProvider",
"GeneratedEvaluatorRef",
"RawAnthropicFoundryClient",
"RawFoundryAgent",
"RawFoundryAgentChatClient",
@@ -498,34 +498,14 @@ def _get_exporters_from_env(
# Get base endpoint
base_endpoint = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT")
# Get signal-specific endpoints (these override base endpoint and are used verbatim)
traces_endpoint_specific = os.getenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT")
metrics_endpoint_specific = os.getenv("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT")
logs_endpoint_specific = os.getenv("OTEL_EXPORTER_OTLP_LOGS_ENDPOINT")
# Get signal-specific endpoints (these override base endpoint)
traces_endpoint = os.getenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT") or base_endpoint
metrics_endpoint = os.getenv("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT") or base_endpoint
logs_endpoint = os.getenv("OTEL_EXPORTER_OTLP_LOGS_ENDPOINT") or base_endpoint
# Get protocol (default is grpc)
protocol = os.getenv("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc").lower()
# Per the OTel spec, OTEL_EXPORTER_OTLP_ENDPOINT is a *base* URL for HTTP — the SDK
# auto-appends /v1/{traces,metrics,logs} when it reads the env var directly. The
# signal-specific endpoint env vars are *full* URLs used verbatim. Because we read
# the env vars here and forward them as the ``endpoint=`` constructor argument
# (which the SDK always treats as a full URL), we must replicate the auto-append
# ourselves for HTTP when falling back to the base endpoint. For gRPC, the base
# endpoint is used as-is.
traces_endpoint: str | None
metrics_endpoint: str | None
logs_endpoint: str | None
if protocol in ("http/protobuf", "http") and base_endpoint:
base_for_http = base_endpoint.rstrip("/")
traces_endpoint = traces_endpoint_specific or f"{base_for_http}/v1/traces"
metrics_endpoint = metrics_endpoint_specific or f"{base_for_http}/v1/metrics"
logs_endpoint = logs_endpoint_specific or f"{base_for_http}/v1/logs"
else:
traces_endpoint = traces_endpoint_specific or base_endpoint
metrics_endpoint = metrics_endpoint_specific or base_endpoint
logs_endpoint = logs_endpoint_specific or base_endpoint
# Get base headers
base_headers_str = os.getenv("OTEL_EXPORTER_OTLP_HEADERS", "")
base_headers = _parse_headers(base_headers_str)
@@ -394,94 +394,3 @@ def test_create_harness_agent_logs_warning_when_no_web_search(caplog: pytest.Log
max_output_tokens=16_384,
)
assert any("SupportsWebSearchTool" in msg for msg in caplog.messages)
# --- Background Agents Tests ---
class _FakeBackgroundAgent:
"""Minimal agent stub satisfying SupportsAgentRun for background agents tests."""
def __init__(self, name: str, description: str | None = None):
self.id = f"agent-{name}"
self.name = name
self.description = description
def create_session(self, *, session_id: str | None = None) -> AgentSession:
return AgentSession(session_id=session_id)
def get_session(self, service_session_id: str, *, session_id: str | None = None) -> AgentSession:
return AgentSession(service_session_id=service_session_id, session_id=session_id)
async def run(self, messages: Any = None, *, stream: bool = False, session: Any = None, **kwargs: Any) -> Any:
from agent_framework import AgentResponse
return AgentResponse(messages=[], response_id="fake-bg-response")
def test_create_harness_agent_no_background_agents_by_default() -> None:
"""No BackgroundAgentsProvider should be included when background_agents is not provided."""
from agent_framework._harness._background_agents import BackgroundAgentsProvider
agent = create_harness_agent(
client=_FakeChatClient(), # type: ignore[arg-type]
max_context_window_tokens=128_000,
max_output_tokens=16_384,
disable_web_search=True,
)
providers = agent.context_providers or []
assert not any(isinstance(p, BackgroundAgentsProvider) for p in providers)
def test_create_harness_agent_adds_background_agents_provider() -> None:
"""BackgroundAgentsProvider should be included when background_agents are provided."""
from agent_framework._harness._background_agents import BackgroundAgentsProvider
bg_agent = _FakeBackgroundAgent("WebSearcher", "Searches the web")
agent = create_harness_agent(
client=_FakeChatClient(), # type: ignore[arg-type]
max_context_window_tokens=128_000,
max_output_tokens=16_384,
disable_web_search=True,
background_agents=[bg_agent],
)
providers = agent.context_providers or []
bg_providers = [p for p in providers if isinstance(p, BackgroundAgentsProvider)]
assert len(bg_providers) == 1
def test_create_harness_agent_background_agents_custom_instructions() -> None:
"""Custom instructions should be passed to BackgroundAgentsProvider."""
from agent_framework._harness._background_agents import BackgroundAgentsProvider
custom_instructions = "## Custom\n\nUse agents wisely.\n\n{background_agents}"
bg_agent = _FakeBackgroundAgent("Helper", "A helper agent")
agent = create_harness_agent(
client=_FakeChatClient(), # type: ignore[arg-type]
max_context_window_tokens=128_000,
max_output_tokens=16_384,
disable_web_search=True,
background_agents=[bg_agent],
background_agents_instructions=custom_instructions,
)
providers = agent.context_providers or []
bg_providers = [p for p in providers if isinstance(p, BackgroundAgentsProvider)]
assert len(bg_providers) == 1
# Verify the custom instructions were used (placeholder replaced with agent list).
assert "Custom" in bg_providers[0]._instructions
assert "Helper" in bg_providers[0]._instructions
def test_create_harness_agent_empty_background_agents_list() -> None:
"""An empty background_agents list should NOT add a BackgroundAgentsProvider."""
from agent_framework._harness._background_agents import BackgroundAgentsProvider
agent = create_harness_agent(
client=_FakeChatClient(), # type: ignore[arg-type]
max_context_window_tokens=128_000,
max_output_tokens=16_384,
disable_web_search=True,
background_agents=[],
)
providers = agent.context_providers or []
assert not any(isinstance(p, BackgroundAgentsProvider) for p in providers)
@@ -11,13 +11,8 @@ import pytest
from agent_framework._evaluation import (
CheckResult,
EvalItem,
EvalItemResult,
EvalNotPassedError,
EvalResults,
EvalScoreResult,
ExpectedToolCall,
LocalEvaluator,
RubricScore,
_coerce_result,
evaluator,
keyword_check,
@@ -1015,101 +1010,19 @@ class TestAllPassedSubResults:
# ---------------------------------------------------------------------------
# Rubric assertions (EvalResults.assert_*)
# r5 review: _build_overall_item with empty outputs
# ---------------------------------------------------------------------------
def _rubric_results(*scores_per_item: list[EvalScoreResult]) -> EvalResults:
items = [
EvalItemResult(item_id=f"item-{i}", status="pass", scores=scores) for i, scores in enumerate(scores_per_item)
]
return EvalResults(
provider="test",
eval_id="ev1",
run_id="run1",
result_counts={"passed": len(items), "failed": 0, "errored": 0, "total": len(items)},
items=items,
)
class TestBuildOverallItemEmpty:
"""Test _build_overall_item returns None for empty workflow outputs."""
def test_returns_none_for_empty_outputs(self):
from unittest.mock import MagicMock
class TestRubricAssertions:
"""Tests for EvalResults.assert_dimension_score_at_least."""
from agent_framework._evaluation import _build_overall_item
def test_dimension_at_or_above_threshold_passes(self) -> None:
results = _rubric_results(
[
EvalScoreResult(
name="policy",
score=0.9,
dimensions=[RubricScore(id="clarity", score=4, applicable=True, weight=1, reason="")],
)
],
)
# Should not raise.
results.assert_dimension_score_at_least("clarity", 3)
def test_dimension_below_threshold_raises(self) -> None:
results = _rubric_results(
[
EvalScoreResult(
name="policy",
score=0.5,
dimensions=[RubricScore(id="clarity", score=2, applicable=True, weight=1, reason="")],
)
],
)
with pytest.raises(EvalNotPassedError):
results.assert_dimension_score_at_least("clarity", 3)
def test_non_applicable_skipped_by_default(self) -> None:
results = _rubric_results(
[
EvalScoreResult(
name="policy",
score=1.0,
dimensions=[RubricScore(id="clarity", score=None, applicable=False, weight=1, reason="n/a")],
)
],
)
# No applicable scores; default behaviour is to skip silently.
results.assert_dimension_score_at_least("clarity", 3)
def test_require_applicable_raises_when_dimension_absent(self) -> None:
results = _rubric_results(
[EvalScoreResult(name="policy", score=1.0, dimensions=[])],
)
with pytest.raises(EvalNotPassedError, match="not applicable"):
results.assert_dimension_score_at_least("clarity", 3, require_applicable=True)
def test_require_applicable_raises_when_filtered_evaluator_missing(self) -> None:
# Regression: previously the (not evaluator or found_any) guard caused
# this case to silently pass even with require_applicable=True.
results = _rubric_results(
[
EvalScoreResult(
name="other",
score=0.9,
dimensions=[RubricScore(id="clarity", score=4, applicable=True, weight=1, reason="")],
)
],
)
with pytest.raises(EvalNotPassedError, match="not applicable"):
results.assert_dimension_score_at_least("clarity", 3, evaluator="policy", require_applicable=True)
def test_evaluator_filter_isolates_offenders(self) -> None:
results = _rubric_results(
[
EvalScoreResult(
name="other",
score=0.1,
dimensions=[RubricScore(id="clarity", score=1, applicable=True, weight=1, reason="")],
),
EvalScoreResult(
name="policy",
score=0.9,
dimensions=[RubricScore(id="clarity", score=4, applicable=True, weight=1, reason="")],
),
],
)
# The low-scoring "other" evaluator is filtered out; "policy" passes.
results.assert_dimension_score_at_least("clarity", 3, evaluator="policy")
mock_result = MagicMock()
mock_result.get_outputs.return_value = []
item = _build_overall_item("Hello", mock_result)
assert item is None
@@ -25,7 +25,6 @@ from agent_framework import (
prepend_agent_framework_to_user_agent,
tool,
)
from agent_framework._serialization import make_json_safe
from agent_framework.observability import (
ROLE_EVENT_MAP,
AgentTelemetryLayer,
@@ -762,115 +761,6 @@ def test_get_exporters_from_env_missing_grpc_dependency(monkeypatch):
_get_exporters_from_env()
# region Test OTLP endpoint computation (base-URL auto-append for HTTP)
def test_get_exporters_from_env_http_base_endpoint_appends_signal_paths(monkeypatch):
"""OTEL_EXPORTER_OTLP_ENDPOINT is a base URL for HTTP; SDK auto-appends
/v1/{traces,metrics,logs}. Because we read the env var and forward it as the
constructor ``endpoint=`` arg (which the SDK treats as a full URL), we must
replicate the auto-append ourselves.
"""
from unittest.mock import patch
from agent_framework import observability
monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4318")
monkeypatch.setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "http/protobuf")
for key in (
"OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",
"OTEL_EXPORTER_OTLP_METRICS_ENDPOINT",
"OTEL_EXPORTER_OTLP_LOGS_ENDPOINT",
):
monkeypatch.delenv(key, raising=False)
with patch.object(observability, "_create_otlp_exporters", return_value=[]) as create:
observability._get_exporters_from_env()
kwargs = create.call_args.kwargs
assert kwargs["protocol"] == "http/protobuf"
assert kwargs["traces_endpoint"] == "http://localhost:4318/v1/traces"
assert kwargs["metrics_endpoint"] == "http://localhost:4318/v1/metrics"
assert kwargs["logs_endpoint"] == "http://localhost:4318/v1/logs"
def test_get_exporters_from_env_http_base_endpoint_trailing_slash(monkeypatch):
"""A trailing slash on the base endpoint should not produce a doubled slash."""
from unittest.mock import patch
from agent_framework import observability
monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4318/")
monkeypatch.setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "http/protobuf")
for key in (
"OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",
"OTEL_EXPORTER_OTLP_METRICS_ENDPOINT",
"OTEL_EXPORTER_OTLP_LOGS_ENDPOINT",
):
monkeypatch.delenv(key, raising=False)
with patch.object(observability, "_create_otlp_exporters", return_value=[]) as create:
observability._get_exporters_from_env()
kwargs = create.call_args.kwargs
assert kwargs["traces_endpoint"] == "http://localhost:4318/v1/traces"
assert kwargs["metrics_endpoint"] == "http://localhost:4318/v1/metrics"
assert kwargs["logs_endpoint"] == "http://localhost:4318/v1/logs"
def test_get_exporters_from_env_http_signal_specific_used_verbatim(monkeypatch):
"""Signal-specific endpoint env vars are full URLs and must be used verbatim,
even when a base endpoint is also set.
"""
from unittest.mock import patch
from agent_framework import observability
monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4318")
monkeypatch.setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "http://traces.example.com/custom/path")
monkeypatch.setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "http/protobuf")
for key in (
"OTEL_EXPORTER_OTLP_METRICS_ENDPOINT",
"OTEL_EXPORTER_OTLP_LOGS_ENDPOINT",
):
monkeypatch.delenv(key, raising=False)
with patch.object(observability, "_create_otlp_exporters", return_value=[]) as create:
observability._get_exporters_from_env()
kwargs = create.call_args.kwargs
# Signal-specific is verbatim — no path appended
assert kwargs["traces_endpoint"] == "http://traces.example.com/custom/path"
# Others fall back to base, with path appended
assert kwargs["metrics_endpoint"] == "http://localhost:4318/v1/metrics"
assert kwargs["logs_endpoint"] == "http://localhost:4318/v1/logs"
def test_get_exporters_from_env_grpc_base_endpoint_unchanged(monkeypatch):
"""For gRPC, the base endpoint applies to all signals as-is (no path append)."""
from unittest.mock import patch
from agent_framework import observability
monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317")
monkeypatch.setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc")
for key in (
"OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",
"OTEL_EXPORTER_OTLP_METRICS_ENDPOINT",
"OTEL_EXPORTER_OTLP_LOGS_ENDPOINT",
):
monkeypatch.delenv(key, raising=False)
with patch.object(observability, "_create_otlp_exporters", return_value=[]) as create:
observability._get_exporters_from_env()
kwargs = create.call_args.kwargs
assert kwargs["protocol"] == "grpc"
assert kwargs["traces_endpoint"] == "http://localhost:4317"
assert kwargs["metrics_endpoint"] == "http://localhost:4317"
assert kwargs["logs_endpoint"] == "http://localhost:4317"
# region Test create_resource
@@ -1801,65 +1691,6 @@ def test_to_otel_part_function_call():
}
def test_to_otel_part_function_call_reuses_prepared_arguments():
"""Test _to_otel_part does not re-serialize function-call arguments in the observability hot path."""
from agent_framework import Content
from agent_framework.observability import _to_otel_part
arguments = {"payload": object()}
content = Content(type="function_call", call_id="call_789", name="handoff", arguments=arguments)
result = _to_otel_part(content)
assert result is not None
assert result["arguments"] is arguments
def test_make_json_safe_non_callable_method_attribute():
"""Test make_json_safe handles objects where model_dump/to_dict/dict are non-callable attributes."""
from agent_framework._serialization import make_json_safe
class ObjWithNonCallableModelDump:
model_dump = 42 # not callable
obj = ObjWithNonCallableModelDump()
result = make_json_safe(obj)
assert result == {}
def test_make_json_safe_callable_method_type_error_falls_through():
"""Test make_json_safe falls through when serializer-like methods require arguments."""
from agent_framework._serialization import make_json_safe
class ObjWithRequiredArgModelDump:
def __init__(self) -> None:
self.value = "fallback"
def model_dump(self, required: str) -> dict[str, str]:
return {"required": required}
obj = ObjWithRequiredArgModelDump()
result = make_json_safe(obj)
assert result == {"value": "fallback"}
def test_make_json_safe_dict_with_non_string_keys():
"""Test make_json_safe converts non-primitive dict keys to strings."""
import json
from datetime import datetime
from agent_framework._serialization import make_json_safe
dt_key = datetime(2024, 1, 1)
obj = {dt_key: "value", 42: "num_value", "str_key": "normal"}
result = make_json_safe(obj)
# json.dumps must not raise TypeError
serialized = json.dumps(result)
parsed = json.loads(serialized)
assert parsed[str(dt_key)] == "value"
assert parsed["42"] == "num_value"
assert parsed["str_key"] == "normal"
def test_to_otel_part_function_result():
"""Test _to_otel_part with function_result content."""
from agent_framework import Content
@@ -3188,47 +3019,6 @@ async def test_system_instructions_preserves_non_ascii_characters(span_exporter:
assert [msg.get("role") for msg in input_messages] == ["user"]
@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True)
def test_capture_messages_with_prepared_request_info_function_call_arguments(span_exporter: InMemorySpanExporter):
"""Test _capture_messages handles request-info function-call arguments prepared at Content creation."""
import dataclasses
import json
from opentelemetry import trace
@dataclasses.dataclass
class HandoffRequest:
target_agent: str
reason: str
arguments = {
"request_id": "call_dc",
"data": make_json_safe(HandoffRequest(target_agent="helper", reason="overflow")),
}
msg = Message(
role="assistant",
contents=[
Content(
type="function_call",
call_id="call_dc",
name="request_info",
arguments=arguments,
)
],
)
span_exporter.clear()
tracer = trace.get_tracer("test")
with tracer.start_as_current_span("test_span") as span:
_capture_messages(span=span, provider_name="test_provider", messages=[msg])
spans = span_exporter.get_finished_spans()
span = spans[0]
input_messages = json.loads(span.attributes[OtelAttr.INPUT_MESSAGES])
tool_part = input_messages[0]["parts"][0]
assert tool_part["type"] == "tool_call"
assert tool_part["arguments"]["data"] == {"target_agent": "helper", "reason": "overflow"}
def test_capture_messages_keeps_framework_instructions_out_of_logs_and_span_messages(
span_exporter: InMemorySpanExporter,
):
@@ -4086,8 +4086,8 @@ class TestClassSkill:
async def test_content_is_cached(self) -> None:
skill = _MinimalClassSkill()
content1 = await skill.get_content()
content2 = await skill.get_content()
content1 = (await skill.get_content())
content2 = (await skill.get_content())
assert content1 is content2
def test_resources_are_lazy_cached(self) -> None:
@@ -5587,8 +5587,8 @@ class TestInlineSkillContentCaching:
async def test_content_cached_after_first_access(self) -> None:
"""InlineSkill.content returns the same object on subsequent accesses."""
skill = InlineSkill(frontmatter=SkillFrontmatter(name="test-skill", description="Test"), instructions="Body")
first = await skill.get_content()
second = await skill.get_content()
first = (await skill.get_content())
second = (await skill.get_content())
assert first is second # Same object (cached)
assert "<name>test-skill</name>" in first
@@ -699,171 +699,3 @@ async def test_resolve_executor_kwargs_empty_per_executor_does_not_fallback_to_g
resolved = {"exec_a": {}, GLOBAL_KWARGS_KEY: {"global_key": "global_val"}}
result = executor._resolve_executor_kwargs(resolved) # pyright: ignore[reportPrivateUsage]
assert result == {}
# region Tool approval emission
class _ApprovalEmittingAgent(BaseAgent):
"""Agent that returns a single ``function_approval_request`` Content.
Used to verify that ``AgentExecutor`` does *not* surface the approval
payload via both an ``output`` event and a ``request_info`` event in the
same superstep — only the ``request_info`` event must carry it.
"""
def __init__(
self,
*,
approval_request_id: str = "apr_1",
tool_name: str = "delete_file",
tool_arguments: dict[str, Any] | None = None,
**kwargs: Any,
):
super().__init__(**kwargs)
self._approval_request_id = approval_request_id
self._tool_name = tool_name
self._tool_arguments: dict[str, Any] = tool_arguments or {"path": "/tmp/secret.txt"}
self.run_count = 0
def _build_approval_content(self) -> Content:
function_call = Content.from_function_call(
call_id=self._approval_request_id,
name=self._tool_name,
arguments=self._tool_arguments,
)
return Content.from_function_approval_request(id=self._approval_request_id, function_call=function_call)
@overload
def run(
self,
messages: AgentRunInputs | None = ...,
*,
stream: Literal[False] = ...,
session: AgentSession | None = ...,
**kwargs: Any,
) -> Awaitable[AgentResponse[Any]]: ...
@overload
def run(
self,
messages: AgentRunInputs | None = ...,
*,
stream: Literal[True],
session: AgentSession | None = ...,
**kwargs: Any,
) -> ResponseStream[AgentResponseUpdate, AgentResponse[Any]]: ...
def run(
self,
messages: AgentRunInputs | None = None,
*,
stream: bool = False,
session: AgentSession | None = None,
**kwargs: Any,
) -> Awaitable[AgentResponse[Any]] | ResponseStream[AgentResponseUpdate, AgentResponse[Any]]:
self.run_count += 1
approval = self._build_approval_content()
if stream:
async def _stream() -> AsyncIterable[AgentResponseUpdate]:
yield AgentResponseUpdate(contents=[approval], role="assistant")
return ResponseStream(_stream(), finalizer=AgentResponse.from_updates)
async def _run() -> AgentResponse:
return AgentResponse(messages=[Message("assistant", [approval])])
return _run()
def _has_approval_payload(event: WorkflowEvent[Any]) -> bool:
"""Return True if the event's data carries a ``function_approval_request`` content."""
data: Any = event.data
def _contents_of(value: Any) -> list[Content]:
if isinstance(value, AgentResponseUpdate):
return list(value.contents)
if isinstance(value, AgentResponse):
return [c for m in value.messages for c in m.contents]
if isinstance(value, AgentExecutorResponse):
return [c for m in value.agent_response.messages for c in m.contents]
if isinstance(value, Message):
return list(value.contents)
if isinstance(value, Content):
return [value]
return []
return any(c.type == "function_approval_request" for c in _contents_of(data))
async def test_agent_executor_does_not_double_emit_approval_non_streaming() -> None:
"""Non-streaming: approval payload must only appear in the ``request_info`` event.
Regression test for the bug where ``AgentExecutor._run_agent`` first
``yield_output``-ed the response (carrying the approval Content) and then
additionally emitted a ``request_info`` event for the same payload.
"""
agent = _ApprovalEmittingAgent(id="approve_agent", name="ApproveAgent", approval_request_id="apr_ns_1")
executor = AgentExecutor(agent, id="approve_exec")
workflow = WorkflowBuilder(start_executor=executor).build()
request_info_events: list[WorkflowEvent[Any]] = []
output_events: list[WorkflowEvent[Any]] = []
for event in await workflow.run("please delete it"):
if event.type == "request_info":
request_info_events.append(event)
elif event.type == "output":
output_events.append(event)
assert len(request_info_events) == 1
assert _has_approval_payload(request_info_events[0])
# The approval payload must not also be surfaced as a workflow output.
assert not any(_has_approval_payload(e) for e in output_events)
assert agent.run_count == 1
async def test_agent_executor_does_not_double_emit_approval_streaming() -> None:
"""Streaming: per-update approval payload must not be ``yield_output``-ed."""
agent = _ApprovalEmittingAgent(id="approve_agent_s", name="ApproveAgentS", approval_request_id="apr_st_1")
executor = AgentExecutor(agent, id="approve_exec_s")
workflow = WorkflowBuilder(start_executor=executor).build()
request_info_events: list[WorkflowEvent[Any]] = []
output_events: list[WorkflowEvent[Any]] = []
async for event in workflow.run("please delete it", stream=True):
if event.type == "request_info":
request_info_events.append(event)
elif event.type == "output":
output_events.append(event)
assert len(request_info_events) == 1
assert _has_approval_payload(request_info_events[0])
assert not any(_has_approval_payload(e) for e in output_events)
assert agent.run_count == 1
async def test_agent_executor_request_info_uses_user_input_request_id() -> None:
"""``ctx.request_info`` must register the request under the agent's approval id.
This makes the workflow's pending-request id round-trip with the
``function_approval_response.id`` the caller echoes back, so
``Workflow._send_responses_internal`` can look it up directly.
"""
agent = _ApprovalEmittingAgent(id="approve_agent_id", name="ApproveAgentId", approval_request_id="apr_match")
executor = AgentExecutor(agent, id="approve_exec_id")
workflow = WorkflowBuilder(start_executor=executor).build()
request_info_events: list[WorkflowEvent[Any]] = []
async for event in workflow.run("please delete it", stream=True):
if event.type == "request_info":
request_info_events.append(event)
assert len(request_info_events) == 1
assert request_info_events[0].request_id == "apr_match"
# endregion Tool approval emission
@@ -5,7 +5,6 @@
from __future__ import annotations
import asyncio
import json
import logging
from collections.abc import Iterator
from contextlib import contextmanager
@@ -1643,37 +1642,6 @@ class TestFunctionalWorkflowAgentHITL:
break
assert approval_found, "expected FunctionApprovalRequestContent in agent response"
async def test_request_info_dataclass_arguments_are_serialized_for_agent(self):
@dataclass
class HandoffRequest:
target_agent: str
reason: str
@workflow
async def wf(x: str, ctx: RunContext) -> str:
answer = await ctx.request_info(
HandoffRequest(target_agent=x, reason="overflow"),
response_type=str,
request_id="rid-1",
)
return f"got:{answer}"
agent = wf.as_agent()
response = await agent.run("helper")
function_call_arguments = None
for message in response.messages:
for content in message.contents:
if getattr(content, "type", None) == "function_approval_request" and content.function_call is not None:
function_call_arguments = content.function_call.arguments
break
assert function_call_arguments == {
"request_id": "rid-1",
"data": {"target_agent": "helper", "reason": "overflow"},
}
assert json.loads(json.dumps(function_call_arguments)) == function_call_arguments
async def test_resume_via_agent_responses_kwarg(self):
@workflow
async def wf(x: str, ctx: RunContext) -> str:
@@ -2,7 +2,6 @@
import uuid
from collections.abc import Awaitable, Sequence
from dataclasses import dataclass
from typing import Any, Literal, overload
import pytest
@@ -24,25 +23,10 @@ from agent_framework import (
WorkflowAgent,
WorkflowBuilder,
WorkflowContext,
WorkflowEvent,
executor,
handler,
response_handler,
)
from agent_framework._workflows._typing_utils import deserialize_type
@dataclass
class HandoffRequest:
"""Module-level dataclass used by request_info tests.
Defined at module scope (not nested inside a test method) so
``serialize_type``/``deserialize_type`` can round-trip the request_type via
the importable qualified name ``tests.workflow.test_workflow_agent.HandoffRequest``.
"""
target_agent: str
reason: str
class SimpleExecutor(Executor):
@@ -253,45 +237,52 @@ class TestWorkflowAgent:
# Should have received an approval request for the request info
assert len(updates) > 0
request_update: AgentResponseUpdate | None = None
approval_update: AgentResponseUpdate | None = None
for update in updates:
if any(content.type == "function_call" for content in update.contents):
request_update = update
if any(content.type == "function_approval_request" for content in update.contents):
approval_update = update
break
assert request_update is not None, "Should have received a request_info wrapped in a function_call content"
assert approval_update is not None, "Should have received a request_info approval request"
request_function_call = next(content for content in request_update.contents if content.type == "function_call")
assert request_function_call.call_id is not None
# Verify the function call has expected structure
assert request_function_call.name == WorkflowAgent.REQUEST_INFO_FUNCTION_NAME
assert isinstance(request_function_call.arguments, dict)
assert request_function_call.arguments.get("request_id") is not None
assert request_function_call.arguments.get("request_event") is not None
request_event = request_function_call.arguments["request_event"]
assert request_event.get("type") == "request_info"
assert deserialize_type(request_event.get("response_type")) is str
deserialized_args = WorkflowAgent.RequestInfoFunctionArgs.from_dict(request_function_call.arguments)
assert deserialized_args.request_id == request_function_call.call_id
assert isinstance(deserialized_args.request_event, WorkflowEvent)
assert deserialized_args.request_event.type == "request_info"
assert deserialized_args.request_event.data == "Mock request data"
assert deserialized_args.request_event.response_type is str
# Verify the request is tracked in pending_requests
pending_requests = await workflow._runner_context.get_pending_request_info_events()
assert len(pending_requests) == 1
assert request_function_call.call_id in pending_requests
# Now provide a function result response with updated arguments to test continuation
function_result = Content.from_function_result(
call_id=request_function_call.call_id,
result="Mock response to request info",
function_call = next(content for content in approval_update.contents if content.type == "function_call")
approval_request = next(
content for content in approval_update.contents if content.type == "function_approval_request"
)
response_message = Message(role="user", contents=[function_result])
# Verify the function call has expected structure
assert function_call.call_id is not None
assert function_call.name == "request_info"
assert isinstance(function_call.arguments, dict)
assert function_call.arguments.get("request_id") == approval_request.id
# Approval request should reference the same function call
assert approval_request.id is not None
assert approval_request.function_call is not None
assert approval_request.function_call.call_id == function_call.call_id
assert approval_request.function_call.name == function_call.name
# Verify the request is tracked in pending_requests
assert len(agent.pending_requests) == 1
assert function_call.call_id in agent.pending_requests
# Now provide an approval response with updated arguments to test continuation
response_args = WorkflowAgent.RequestInfoFunctionArgs(
request_id=approval_request.id,
data="User provided answer",
).to_dict()
approval_response = Content.from_function_approval_response(
approved=True,
id=approval_request.id,
function_call=Content.from_function_call(
call_id=function_call.call_id,
name=function_call.name,
arguments=response_args,
),
)
response_message = Message(role="user", contents=[approval_response])
# Continue the workflow with the response
continuation_result = await agent.run(response_message)
@@ -300,382 +291,7 @@ class TestWorkflowAgent:
assert isinstance(continuation_result, AgentResponse)
# Verify cleanup - pending requests should be cleared after function response handling
pending_requests = await workflow._runner_context.get_pending_request_info_events()
assert len(pending_requests) == 0
def test_request_info_dataclass_arguments_are_serialized_when_content_is_created(self) -> None:
"""Test WorkflowAgent prepares request_info arguments before observability captures messages."""
executor = SimpleExecutor(id="executor1", response_text="Response")
workflow = WorkflowBuilder(start_executor=executor).build()
agent = WorkflowAgent(workflow=workflow, name="Request Test Agent")
event = WorkflowEvent.request_info(
request_id="request_123",
source_executor_id="executor1",
request_data=HandoffRequest(target_agent="helper", reason="overflow"),
response_type=str,
)
request_function_call = agent._process_request_info_event(event) # pyright: ignore[reportPrivateUsage]
assert request_function_call.call_id == "request_123"
assert isinstance(request_function_call.arguments, dict)
assert request_function_call.arguments.get("request_event") is not None
request_event = request_function_call.arguments["request_event"]
assert request_event.get("type") == "request_info"
assert request_event.get("request_id") == "request_123"
assert request_event.get("source_executor_id") == "executor1"
assert deserialize_type(request_event.get("response_type")) is str
assert request_event.get("data") == HandoffRequest(target_agent="helper", reason="overflow")
deserialized_args = WorkflowAgent.RequestInfoFunctionArgs.from_dict(request_function_call.arguments)
assert deserialized_args.request_id == "request_123"
assert isinstance(deserialized_args.request_event, WorkflowEvent)
assert deserialized_args.request_event.type == "request_info"
assert deserialized_args.request_event.data == HandoffRequest(target_agent="helper", reason="overflow")
assert deserialized_args.request_event.response_type is str
def test_process_request_info_event_passes_through_function_approval_request(self) -> None:
"""If the event data is already a function approval request, it is forwarded unchanged.
Tool-approval requests emitted by an inner agent surface as ``Content``
objects with ``user_input_request=True``. ``WorkflowAgent`` must not
re-wrap these inside a synthesized ``request_info`` function call;
instead it should return the original content as-is so callers can
respond with a matching ``function_approval_response``.
"""
executor = SimpleExecutor(id="executor1", response_text="Response")
workflow = WorkflowBuilder(start_executor=executor).build()
agent = WorkflowAgent(workflow=workflow, name="Approval Passthrough Agent")
approval_id = "approval-passthrough-1"
inner_function_call = Content.from_function_call(
call_id="tool-call-1",
name="delete_file",
arguments={"path": "/tmp/x"},
)
approval_request = Content.from_function_approval_request(
id=approval_id,
function_call=inner_function_call,
)
event = WorkflowEvent.request_info(
request_id=approval_id,
source_executor_id="executor1",
request_data=approval_request,
response_type=Content,
)
result = agent._process_request_info_event(event) # pyright: ignore[reportPrivateUsage]
# The original FunctionApprovalRequestContent is returned as-is — same
# instance, with the original tool name preserved (NOT replaced by the
# synthesized REQUEST_INFO_FUNCTION_NAME).
assert result is approval_request
assert result.type == "function_approval_request"
assert result.id == approval_id
assert result.user_input_request is True
assert result.function_call is inner_function_call # type: ignore[attr-defined]
assert result.function_call.name == "delete_file" # type: ignore[attr-defined]
assert result.function_call.name != WorkflowAgent.REQUEST_INFO_FUNCTION_NAME # type: ignore[attr-defined]
def test_extract_function_responses_passes_through_approval_response_approved(self) -> None:
"""A function_approval_response with approved=True is keyed by content.id and forwarded as-is.
After the refactor, ``WorkflowAgent`` no longer unwraps a synthesized
``request_info`` function call from approval responses — the response
content is routed straight back to the workflow under its own ``id``,
which matches the pending request id surfaced by
``_process_request_info_event``.
"""
executor = SimpleExecutor(id="executor1", response_text="Response")
workflow = WorkflowBuilder(start_executor=executor).build()
agent = WorkflowAgent(workflow=workflow, name="Approval Response Agent")
approval_id = "approval-response-approved-1"
inner_function_call = Content.from_function_call(
call_id="tool-call-1",
name="delete_file",
arguments={"path": "/tmp/x"},
)
approval_request = Content.from_function_approval_request(
id=approval_id,
function_call=inner_function_call,
)
approval_response = approval_request.to_function_approval_response(approved=True) # type: ignore[attr-defined]
message = Message(role="user", contents=[approval_response])
responses = agent._extract_function_responses([message]) # pyright: ignore[reportPrivateUsage]
assert set(responses.keys()) == {approval_id}
assert responses[approval_id] is approval_response
assert responses[approval_id].approved is True # type: ignore[attr-defined]
def test_extract_function_responses_passes_through_approval_response_denied(self) -> None:
"""A function_approval_response with approved=False is forwarded the same way as an approval.
Only the ``approved`` flag changes — routing back to the workflow is
identical for accept and reject paths.
"""
executor = SimpleExecutor(id="executor1", response_text="Response")
workflow = WorkflowBuilder(start_executor=executor).build()
agent = WorkflowAgent(workflow=workflow, name="Approval Response Agent")
approval_id = "approval-response-denied-1"
inner_function_call = Content.from_function_call(
call_id="tool-call-2",
name="send_email",
arguments={"to": "alice@example.com"},
)
approval_request = Content.from_function_approval_request(
id=approval_id,
function_call=inner_function_call,
)
approval_response = approval_request.to_function_approval_response(approved=False) # type: ignore[attr-defined]
message = Message(role="user", contents=[approval_response])
responses = agent._extract_function_responses([message]) # pyright: ignore[reportPrivateUsage]
assert set(responses.keys()) == {approval_id}
assert responses[approval_id] is approval_response
assert responses[approval_id].approved is False # type: ignore[attr-defined]
async def test_function_approval_request_flows_end_to_end_approved(self) -> None:
"""End-to-end: an executor emits a function_approval_request, the agent
forwards it unchanged, and an ``approved=True`` response resumes the workflow.
This exercises the full pass-through path:
``ctx.request_info(approval_content, ...)`` -> ``WorkflowAgent`` surfaces
the original ``FunctionApprovalRequestContent`` -> caller responds with a
``FunctionApprovalResponseContent`` -> ``WorkflowAgent`` routes it back
to the workflow which delivers it to the executor's ``@response_handler``.
"""
approval_id = "e2e-approval-1"
inner_function_call = Content.from_function_call(
call_id="tool-call-e2e-1",
name="delete_file",
arguments={"path": "/tmp/x"},
)
approval_request = Content.from_function_approval_request(
id=approval_id,
function_call=inner_function_call,
)
class ApprovalRequestingExecutor(Executor):
@handler
async def handle_message(self, _: list[Message], ctx: WorkflowContext) -> None:
await ctx.request_info(approval_request, Content, request_id=approval_id)
@response_handler
async def handle_response(
self,
original_request: Content,
response: Content,
ctx: WorkflowContext[Never, AgentResponse],
) -> None:
assert response.type == "function_approval_response"
assert response.id == approval_id # type: ignore[attr-defined]
approved = bool(response.approved) # type: ignore[attr-defined]
tool_name = original_request.function_call.name # type: ignore[attr-defined]
await ctx.yield_output(
AgentResponse(
messages=[
Message(
role="assistant",
contents=[Content.from_text(text=f"{tool_name} approved={approved}")],
)
]
)
)
executor = ApprovalRequestingExecutor(id="approval_requester")
workflow = WorkflowBuilder(start_executor=executor).build()
agent = WorkflowAgent(workflow=workflow, name="E2E Approval Agent")
# First run: workflow pauses with the approval request.
first = await agent.run("please delete it")
assert isinstance(first, AgentResponse)
forwarded = next(
(
c
for m in first.messages
for c in m.contents
if c.type == "function_approval_request" and c.id == approval_id
),
None,
)
assert forwarded is approval_request, "Approval request must surface unchanged"
pending = await workflow._runner_context.get_pending_request_info_events()
assert approval_id in pending
# Respond with approved=True.
approval_response = approval_request.to_function_approval_response(approved=True) # type: ignore[attr-defined]
final = await agent.run(Message(role="user", contents=[approval_response]))
assert isinstance(final, AgentResponse)
final_text = " ".join(m.text or "" for m in final.messages)
assert "delete_file approved=True" in final_text
pending = await workflow._runner_context.get_pending_request_info_events()
assert approval_id not in pending
async def test_function_approval_request_flows_end_to_end_denied(self) -> None:
"""End-to-end denied path: ``approved=False`` is delivered to the executor's
response handler so the workflow can branch on the rejection."""
approval_id = "e2e-approval-deny-1"
inner_function_call = Content.from_function_call(
call_id="tool-call-e2e-deny-1",
name="send_email",
arguments={"to": "alice@example.com"},
)
approval_request = Content.from_function_approval_request(
id=approval_id,
function_call=inner_function_call,
)
class ApprovalRequestingExecutor(Executor):
@handler
async def handle_message(self, _: list[Message], ctx: WorkflowContext) -> None:
await ctx.request_info(approval_request, Content, request_id=approval_id)
@response_handler
async def handle_response(
self,
original_request: Content,
response: Content,
ctx: WorkflowContext[Never, AgentResponse],
) -> None:
assert response.type == "function_approval_response"
assert response.id == approval_id # type: ignore[attr-defined]
approved = bool(response.approved) # type: ignore[attr-defined]
tool_name = original_request.function_call.name # type: ignore[attr-defined]
await ctx.yield_output(
AgentResponse(
messages=[
Message(
role="assistant",
contents=[Content.from_text(text=f"{tool_name} approved={approved}")],
)
]
)
)
executor = ApprovalRequestingExecutor(id="approval_requester_deny")
workflow = WorkflowBuilder(start_executor=executor).build()
agent = WorkflowAgent(workflow=workflow, name="E2E Approval Deny Agent")
first = await agent.run("please send")
assert isinstance(first, AgentResponse)
forwarded = next(
(
c
for m in first.messages
for c in m.contents
if c.type == "function_approval_request" and c.id == approval_id
),
None,
)
assert forwarded is approval_request
# Respond with approved=False.
approval_response = approval_request.to_function_approval_response(approved=False) # type: ignore[attr-defined]
final = await agent.run(Message(role="user", contents=[approval_response]))
assert isinstance(final, AgentResponse)
final_text = " ".join(m.text or "" for m in final.messages)
assert "send_email approved=False" in final_text
pending = await workflow._runner_context.get_pending_request_info_events()
assert approval_id not in pending
async def test_request_info_non_approval_flows_end_to_end(self) -> None:
"""End-to-end: when request data is not a function approval content, the
agent surfaces a synthesized ``function_call`` (name=REQUEST_INFO_FUNCTION_NAME)
and routes a matching ``function_result`` back to the executor.
"""
captured: dict[str, Any] = {}
class HandoffRequestingExecutor(Executor):
@handler
async def handle_message(self, _: list[Message], ctx: WorkflowContext) -> None:
await ctx.request_info(
HandoffRequest(target_agent="helper", reason="overflow"),
str,
)
@response_handler
async def handle_response(
self,
original_request: HandoffRequest,
response: str,
ctx: WorkflowContext[Never, AgentResponse],
) -> None:
captured["original"] = original_request
captured["response"] = response
await ctx.yield_output(
AgentResponse(
messages=[
Message(
role="assistant",
contents=[
Content.from_text(text=f"handoff to {original_request.target_agent}: {response}")
],
)
]
)
)
executor = HandoffRequestingExecutor(id="handoff_requester")
workflow = WorkflowBuilder(start_executor=executor).build()
agent = WorkflowAgent(workflow=workflow, name="E2E Handoff Agent")
# First run: workflow pauses with a synthesized request_info function_call.
first = await agent.run("start handoff")
assert isinstance(first, AgentResponse)
function_call = next(
(
c
for m in first.messages
for c in m.contents
if c.type == "function_call" and c.name == WorkflowAgent.REQUEST_INFO_FUNCTION_NAME
),
None,
)
assert function_call is not None, "Expected a synthesized request_info function_call"
assert function_call.call_id is not None
assert isinstance(function_call.arguments, dict)
request_id = function_call.arguments["request_id"]
assert function_call.call_id == request_id
request_payload = function_call.arguments["request_event"]
assert request_payload.get("type") == "request_info"
assert request_payload.get("data") == HandoffRequest(target_agent="helper", reason="overflow")
deserialized_args = WorkflowAgent.RequestInfoFunctionArgs.from_dict(function_call.arguments)
assert deserialized_args.request_id == request_id
assert isinstance(deserialized_args.request_event, WorkflowEvent)
assert deserialized_args.request_event.type == "request_info"
assert deserialized_args.request_event.data == HandoffRequest(target_agent="helper", reason="overflow")
assert deserialized_args.request_event.response_type is str
pending = await workflow._runner_context.get_pending_request_info_events()
assert request_id in pending
# Respond with a function_result keyed by the call_id.
function_result = Content.from_function_result(call_id=request_id, result="ok-do-it")
final = await agent.run(Message(role="user", contents=[function_result]))
assert isinstance(final, AgentResponse)
final_text = " ".join(m.text or "" for m in final.messages)
assert "handoff to helper: ok-do-it" in final_text
# The executor's response handler received the original request and the response.
assert isinstance(captured.get("original"), HandoffRequest)
assert captured["original"].target_agent == "helper"
assert captured["response"] == "ok-do-it"
pending = await workflow._runner_context.get_pending_request_info_events()
assert request_id not in pending
assert len(agent.pending_requests) == 0
def test_workflow_as_agent_method(self) -> None:
"""Test that Workflow.as_agent() creates a properly configured WorkflowAgent."""
@@ -1946,406 +1562,3 @@ class TestWorkflowAgentMergeUpdates:
# Order: text (user), text (assistant), function_result (orphan at end)
assert content_types == ["text", "text", "function_result"]
class _ToolApprovalMockAgent(SupportsAgentRun):
"""Mock agent whose first run returns a FunctionApprovalRequestContent.
Subsequent runs (after receiving an approval response in the input messages)
return a final assistant text response that echoes the approved arguments.
This mirrors a real agent whose tool invocation requires user approval.
"""
def __init__(
self,
name: str,
*,
tool_name: str = "delete_file",
tool_arguments: dict[str, Any] | None = None,
approval_request_ids: Sequence[str] | None = None,
) -> None:
self.id = str(uuid.uuid4())
self.name = name
self.description: str | None = None
self._tool_name = tool_name
self._tool_arguments = tool_arguments or {"path": "/tmp/example"}
# Pre-allocated request ids so the test can verify what the WorkflowAgent forwards.
self._approval_request_ids: list[str] = list(approval_request_ids) if approval_request_ids else []
self.run_count = 0
# Inputs received on the most recent (continuation) run, for assertions.
self.last_run_messages: list[Message] = []
def create_session(self, **kwargs: Any) -> AgentSession:
return AgentSession()
def get_session(self, *, service_session_id: str, **kwargs: Any) -> AgentSession:
return AgentSession()
def _next_request_id(self) -> str:
if self._approval_request_ids:
return self._approval_request_ids.pop(0)
return str(uuid.uuid4())
def _build_approval_request(self) -> Content:
request_id = self._next_request_id()
function_call = Content.from_function_call(
call_id=request_id,
name=self._tool_name,
arguments=self._tool_arguments,
)
return Content.from_function_approval_request(id=request_id, function_call=function_call)
@overload
def run(
self,
messages: str | Content | Message | Sequence[str | Content | Message] | None = ...,
*,
stream: Literal[False] = ...,
session: AgentSession | None = ...,
**kwargs: Any,
) -> Awaitable[AgentResponse[Any]]: ...
@overload
def run(
self,
messages: str | Content | Message | Sequence[str | Content | Message] | None = ...,
*,
stream: Literal[True],
session: AgentSession | None = ...,
**kwargs: Any,
) -> ResponseStream[AgentResponseUpdate, AgentResponse[Any]]: ...
def run(
self,
messages: str | Content | Message | Sequence[str | Content | Message] | None = None,
*,
stream: bool = False,
session: AgentSession | None = None,
**kwargs: Any,
) -> Awaitable[AgentResponse] | ResponseStream[AgentResponseUpdate, AgentResponse]:
if stream:
return self._run_stream(messages=messages, session=session, **kwargs)
return self._run(messages=messages, session=session, **kwargs)
def _normalize(
self,
messages: str | Content | Message | Sequence[str | Content | Message] | None,
) -> list[Message]:
if messages is None:
return []
if isinstance(messages, str):
return [Message(role="user", contents=[Content.from_text(text=messages)])]
if isinstance(messages, Message):
return [messages]
if isinstance(messages, Content):
return [Message(role="user", contents=[messages])]
result: list[Message] = []
for item in messages:
if isinstance(item, Message):
result.append(item)
elif isinstance(item, Content):
result.append(Message(role="user", contents=[item]))
else:
result.append(Message(role="user", contents=[Content.from_text(text=item)]))
return result
def _approval_responses_in(self, messages: list[Message]) -> list[Content]:
approvals: list[Content] = []
for msg in messages:
for content in msg.contents:
if content.type == "function_approval_response":
approvals.append(content)
return approvals
async def _run(
self,
messages: str | Content | Message | Sequence[str | Content | Message] | None = None,
*,
session: AgentSession | None = None,
**kwargs: Any,
) -> AgentResponse:
normalized = self._normalize(messages)
self.last_run_messages = normalized
self.run_count += 1
approvals = self._approval_responses_in(normalized)
if approvals:
# Continuation: reflect approved arguments in the final response text.
approved_text = "; ".join(
f"approved={a.approved} id={a.id}" # type: ignore[attr-defined]
for a in approvals
)
return AgentResponse(messages=[Message("assistant", [Content.from_text(text=f"done ({approved_text})")])])
# First run: ask for tool approval.
approval = self._build_approval_request()
return AgentResponse(messages=[Message("assistant", [approval])])
def _run_stream(
self,
messages: str | Content | Message | Sequence[str | Content | Message] | None = None,
*,
session: AgentSession | None = None,
**kwargs: Any,
) -> ResponseStream[AgentResponseUpdate, AgentResponse]:
normalized = self._normalize(messages)
self.last_run_messages = normalized
self.run_count += 1
approvals = self._approval_responses_in(normalized)
async def _iter():
if approvals:
approved_text = "; ".join(
f"approved={a.approved} id={a.id}" # type: ignore[attr-defined]
for a in approvals
)
yield AgentResponseUpdate(
contents=[Content.from_text(text=f"done ({approved_text})")],
role="assistant",
author_name=self.name,
)
return
approval = self._build_approval_request()
yield AgentResponseUpdate(
contents=[approval],
role="assistant",
author_name=self.name,
)
return ResponseStream(_iter(), finalizer=AgentResponse.from_updates)
class TestWorkflowAgentToolApproval:
"""Tests for tool-approval requests bubbling through WorkflowAgent.
Covers the case where a workflow contains an AgentExecutor whose underlying
agent emits a FunctionApprovalRequestContent (tool needing user approval).
The WorkflowAgent must:
* forward the original FunctionApprovalRequestContent unchanged (no
wrapping inside a synthesized 'request_info' function call), and
* route a subsequent FunctionApprovalResponseContent back to the
AgentExecutor so the agent can resume.
"""
def _find_approval_request(
self,
contents: Sequence[Content],
tool_name: str,
) -> Content | None:
for content in contents:
if (
content.type == "function_approval_request"
and getattr(content.function_call, "name", None) == tool_name # type: ignore[attr-defined]
):
return content
return None
async def test_tool_approval_request_forwarded_unchanged(self) -> None:
"""The agent's FunctionApprovalRequestContent surfaces verbatim (not re-wrapped)."""
approval_id = "approval-abc-123"
mock_agent = _ToolApprovalMockAgent(
name="approval-agent",
tool_name="delete_file",
tool_arguments={"path": "/tmp/secret.txt"},
approval_request_ids=[approval_id],
)
@executor
async def start(messages: list[Message], ctx: WorkflowContext[AgentExecutorRequest]) -> None:
await ctx.send_message(AgentExecutorRequest(messages=messages, should_respond=True))
workflow = WorkflowBuilder(start_executor=start).add_edge(start, mock_agent).build()
agent = WorkflowAgent(workflow=workflow, name="Approval Test Agent")
result = await agent.run("please delete the file")
assert isinstance(result, AgentResponse)
# Locate the approval request emitted by the WorkflowAgent.
all_contents: list[Content] = [c for m in result.messages for c in m.contents]
approval = self._find_approval_request(all_contents, tool_name="delete_file")
assert approval is not None, "WorkflowAgent did not forward the tool approval request"
# The id and inner function_call must match what the underlying agent produced
# — i.e. the WorkflowAgent must NOT have re-wrapped it inside a synthesized
# 'request_info' approval request.
assert approval.id == approval_id
function_call = approval.function_call # type: ignore[attr-defined]
assert function_call is not None
assert function_call.name == "delete_file"
assert function_call.name != WorkflowAgent.REQUEST_INFO_FUNCTION_NAME
assert function_call.arguments == {"path": "/tmp/secret.txt"}
# The agent must be paused awaiting the approval response.
pending = await workflow._runner_context.get_pending_request_info_events()
assert approval_id in pending
async def test_tool_approval_request_forwarded_unchanged_streaming(self) -> None:
"""Streaming variant: the approval request is forwarded as-is in updates."""
approval_id = "approval-stream-1"
mock_agent = _ToolApprovalMockAgent(
name="approval-agent-stream",
tool_name="send_email",
tool_arguments={"to": "alice@example.com"},
approval_request_ids=[approval_id],
)
@executor
async def start(messages: list[Message], ctx: WorkflowContext[AgentExecutorRequest]) -> None:
await ctx.send_message(AgentExecutorRequest(messages=messages, should_respond=True))
workflow = WorkflowBuilder(start_executor=start).add_edge(start, mock_agent).build()
agent = WorkflowAgent(workflow=workflow, name="Approval Stream Agent")
updates: list[AgentResponseUpdate] = []
async for update in agent.run("hi", stream=True):
updates.append(update)
approval_updates = [u for u in updates if any(c.type == "function_approval_request" for c in u.contents)]
assert approval_updates, "Streaming did not surface a tool approval request"
approval = self._find_approval_request(approval_updates[-1].contents, tool_name="send_email")
assert approval is not None
assert approval.id == approval_id
function_call = approval.function_call # type: ignore[attr-defined]
assert function_call is not None
assert function_call.name == "send_email"
assert function_call.name != WorkflowAgent.REQUEST_INFO_FUNCTION_NAME
assert function_call.arguments == {"to": "alice@example.com"}
async def test_tool_approval_response_resumes_agent(self) -> None:
"""Sending the approval response back resumes the agent and clears pending requests."""
approval_id = "approval-resume-1"
mock_agent = _ToolApprovalMockAgent(
name="approval-resume-agent",
tool_name="delete_file",
tool_arguments={"path": "/tmp/x"},
approval_request_ids=[approval_id],
)
@executor
async def start(messages: list[Message], ctx: WorkflowContext[AgentExecutorRequest]) -> None:
await ctx.send_message(AgentExecutorRequest(messages=messages, should_respond=True))
workflow = WorkflowBuilder(start_executor=start).add_edge(start, mock_agent).build()
agent = WorkflowAgent(workflow=workflow, name="Approval Resume Agent")
first_result = await agent.run("delete it")
approval = self._find_approval_request(
[c for m in first_result.messages for c in m.contents],
tool_name="delete_file",
)
assert approval is not None
assert mock_agent.run_count == 1
# Build the approval response. NOTE: the inner function_call's name is the
# original tool name ('delete_file'), NOT 'request_info'. This exercises the
# branch in WorkflowAgent._extract_function_responses that routes raw
# tool-approval responses straight through using content.id.
approval_response = approval.to_function_approval_response(approved=True) # type: ignore[attr-defined]
response_message = Message(role="user", contents=[approval_response])
final_result = await agent.run(response_message)
assert isinstance(final_result, AgentResponse)
# The mock agent should have been invoked a second time and seen the
# approval response in its inputs.
assert mock_agent.run_count == 2
approvals_seen = [
c for m in mock_agent.last_run_messages for c in m.contents if c.type == "function_approval_response"
]
assert len(approvals_seen) == 1
assert approvals_seen[0].id == approval_id # type: ignore[attr-defined]
assert approvals_seen[0].approved is True # type: ignore[attr-defined]
# The pending approval should now be cleared.
pending = await workflow._runner_context.get_pending_request_info_events()
assert approval_id not in pending
# The final assistant message reflects the resumption.
final_text = " ".join(m.text or "" for m in final_result.messages)
assert "done" in final_text
assert approval_id in final_text
async def test_tool_approval_response_rejected_resumes_agent(self) -> None:
"""Rejection path: ``approved=False`` is forwarded to the inner agent and clears the pending request.
The WorkflowAgent must route a rejection response back to the paused
``AgentExecutor`` exactly the same way as an approval — only the
``approved`` flag differs. The inner agent decides what to do with it.
"""
approval_id = "approval-reject-1"
mock_agent = _ToolApprovalMockAgent(
name="approval-reject-agent",
tool_name="delete_file",
tool_arguments={"path": "/tmp/x"},
approval_request_ids=[approval_id],
)
@executor
async def start(messages: list[Message], ctx: WorkflowContext[AgentExecutorRequest]) -> None:
await ctx.send_message(AgentExecutorRequest(messages=messages, should_respond=True))
workflow = WorkflowBuilder(start_executor=start).add_edge(start, mock_agent).build()
agent = WorkflowAgent(workflow=workflow, name="Approval Reject Agent")
first_result = await agent.run("delete it")
approval = self._find_approval_request(
[c for m in first_result.messages for c in m.contents],
tool_name="delete_file",
)
assert approval is not None
assert mock_agent.run_count == 1
# Reject the tool invocation.
approval_response = approval.to_function_approval_response(approved=False) # type: ignore[attr-defined]
response_message = Message(role="user", contents=[approval_response])
final_result = await agent.run(response_message)
assert isinstance(final_result, AgentResponse)
# The inner agent must have been resumed and seen ``approved=False``.
assert mock_agent.run_count == 2
approvals_seen = [
c for m in mock_agent.last_run_messages for c in m.contents if c.type == "function_approval_response"
]
assert len(approvals_seen) == 1
assert approvals_seen[0].id == approval_id # type: ignore[attr-defined]
assert approvals_seen[0].approved is False # type: ignore[attr-defined]
# Pending approval cleared regardless of approve/reject.
pending = await workflow._runner_context.get_pending_request_info_events()
assert approval_id not in pending
# The final assistant message reflects the rejection.
final_text = " ".join(m.text or "" for m in final_result.messages)
assert "approved=False" in final_text
assert approval_id in final_text
async def test_tool_approval_request_id_matches_pending_request(self) -> None:
"""The approval request id surfaced by WorkflowAgent matches the workflow's pending request id.
This guards the AgentExecutor change that forwards
request_id=user_input_request.id to ctx.request_info(...), which is what
allows the response routed back via WorkflowAgent to resolve the pending
request without an id-mismatch error.
"""
approval_id = "approval-id-match-1"
mock_agent = _ToolApprovalMockAgent(
name="approval-id-match-agent",
approval_request_ids=[approval_id],
)
@executor
async def start(messages: list[Message], ctx: WorkflowContext[AgentExecutorRequest]) -> None:
await ctx.send_message(AgentExecutorRequest(messages=messages, should_respond=True))
workflow = WorkflowBuilder(start_executor=start).add_edge(start, mock_agent).build()
agent = WorkflowAgent(workflow=workflow, name="Approval Id Agent")
await agent.run("go")
pending = await workflow._runner_context.get_pending_request_info_events()
# The agent's approval id is used as the workflow's pending request id.
assert list(pending.keys()) == [approval_id]
@@ -1,149 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
"""Tests for the ``Workflow.status`` property."""
from dataclasses import dataclass
import pytest
from agent_framework import (
Executor,
Workflow,
WorkflowBuilder,
WorkflowContext,
WorkflowEvent,
WorkflowRunState,
handler,
response_handler,
)
from agent_framework._workflows._executor import Executor as _Executor
from agent_framework._workflows._request_info_mixin import RequestInfoMixin
class PassThroughExecutor(Executor):
"""Executor that yields its input as a workflow output and stops."""
@handler
async def passthrough(self, msg: str, ctx: WorkflowContext[str, str]) -> None:
await ctx.yield_output(msg)
class FailingExecutor(Executor):
"""Executor that raises at runtime to drive the FAILED status."""
@handler
async def fail(self, msg: int, ctx: WorkflowContext) -> None: # pragma: no cover - invoked via workflow
raise RuntimeError("boom")
@dataclass
class _ApprovalRequest:
prompt: str
request_id: str = ""
def __post_init__(self) -> None:
if not self.request_id:
import uuid
self.request_id = str(uuid.uuid4())
class ApprovalExecutor(_Executor, RequestInfoMixin):
"""Executor that issues a single request_info call and finalizes on response."""
def __init__(self, id: str = "approval"):
super().__init__(id=id)
@handler
async def start(self, message: str, ctx: WorkflowContext[str, str]) -> None:
await ctx.request_info(_ApprovalRequest(prompt=message), bool)
@response_handler
async def on_response(
self, original_request: _ApprovalRequest, approved: bool, ctx: WorkflowContext[str, str]
) -> None:
await ctx.yield_output(f"approved={approved}")
def _build_passthrough_workflow() -> Workflow:
executor = PassThroughExecutor(id="p")
return WorkflowBuilder(start_executor=executor, output_from=[executor]).build()
def _build_failing_workflow() -> Workflow:
# FailingExecutor has no workflow_output_types, so we leave designation
# implicit; the deprecation warning is filtered at call sites that need it.
return WorkflowBuilder(start_executor=FailingExecutor(id="f")).build()
def _build_approval_workflow() -> Workflow:
executor = ApprovalExecutor(id="approval")
return WorkflowBuilder(start_executor=executor, output_from=[executor]).build()
async def test_status_default_is_idle_before_first_run():
wf = _build_passthrough_workflow()
assert wf.status is WorkflowRunState.IDLE
async def test_status_is_idle_after_successful_run():
wf = _build_passthrough_workflow()
await wf.run("hello")
assert wf.status is WorkflowRunState.IDLE
async def test_status_is_failed_after_failure():
wf = _build_failing_workflow()
with pytest.raises(RuntimeError, match="boom"):
await wf.run(0)
assert wf.status is WorkflowRunState.FAILED
async def test_status_transitions_during_streaming_run():
"""Workflow.status mirrors the most recent emitted status event."""
wf = _build_passthrough_workflow()
observed: list[WorkflowRunState] = []
async for event in wf.run("hi", stream=True):
if isinstance(event, WorkflowEvent) and event.type == "status":
# By the time a status event surfaces to the consumer, the property
# must already reflect that state (updated in lockstep with emission).
assert wf.status == event.state
observed.append(event.state) # type: ignore
# IN_PROGRESS must precede IDLE; both must appear.
assert WorkflowRunState.IN_PROGRESS in observed
assert observed[-1] is WorkflowRunState.IDLE
assert wf.status is WorkflowRunState.IDLE
async def test_status_idle_with_pending_requests_then_resolves_to_idle():
wf = _build_approval_workflow()
request_event: WorkflowEvent | None = None
async for event in wf.run("please approve", stream=True):
if isinstance(event, WorkflowEvent) and event.type == "request_info":
request_event = event
assert request_event is not None
assert wf.status is WorkflowRunState.IDLE_WITH_PENDING_REQUESTS
async for _ in wf.run(stream=True, responses={request_event.request_id: True}):
pass
assert wf.status is WorkflowRunState.IDLE
async def test_status_in_progress_pending_requests_observed_mid_run():
"""While streaming, status reaches IN_PROGRESS_PENDING_REQUESTS after a request_info event."""
wf = _build_approval_workflow()
seen_states: list[WorkflowRunState] = []
async for event in wf.run("please approve", stream=True):
if isinstance(event, WorkflowEvent) and event.type == "status":
seen_states.append(event.state) # type: ignore
assert WorkflowRunState.IN_PROGRESS in seen_states
assert WorkflowRunState.IN_PROGRESS_PENDING_REQUESTS in seen_states
assert seen_states[-1] is WorkflowRunState.IDLE_WITH_PENDING_REQUESTS
assert wf.status is WorkflowRunState.IDLE_WITH_PENDING_REQUESTS
-12
View File
@@ -6,18 +6,6 @@ Please install this package via pip:
pip install agent-framework-declarative --pre
```
## Release stage
This package ships at two different stability levels:
- **Declarative workflows** (`WorkflowFactory`, executors, handlers, and the
`_workflows` surface) are at **release-candidate** stability and may receive only
minor refinements before GA.
- **Declarative agents** (`AgentFactory` and the YAML agent loading/parsing path:
`DeclarativeLoaderError`, `ProviderLookupError`, `ProviderTypeMapping`) are
**experimental** and may change or be removed in future versions without notice.
Using any of these symbols emits an `ExperimentalWarning` on first use.
## Declarative features
The declarative packages provides support for building agents based on a declarative yaml specification.
@@ -1,18 +1,5 @@
# Copyright (c) Microsoft. All rights reserved.
"""Declarative specification support for Microsoft Agent Framework.
Release stage:
* The declarative-workflows surface (``WorkflowFactory``, executors, handlers,
etc.) is at release-candidate stability.
* The declarative-agents surface (``AgentFactory`` and the YAML agent
loading/parsing path: ``DeclarativeLoaderError``, ``ProviderLookupError``,
``ProviderTypeMapping``) is *experimental* and may change or be removed in
future versions without notice. Using these symbols emits an
``ExperimentalWarning`` on first use.
"""
from importlib import metadata
from ._loader import AgentFactory, DeclarativeLoaderError, ProviderLookupError, ProviderTypeMapping
@@ -15,10 +15,6 @@ from agent_framework import (
from agent_framework import (
FunctionTool as AFFunctionTool,
)
from agent_framework._feature_stage import ( # type: ignore[reportPrivateUsage]
ExperimentalFeature,
experimental,
)
from agent_framework.exceptions import AgentException
from dotenv import load_dotenv
@@ -47,7 +43,6 @@ else:
from typing_extensions import TypedDict # type: ignore # pragma: no cover
@experimental(feature_id=ExperimentalFeature.DECLARATIVE_AGENTS)
class ProviderTypeMapping(TypedDict, total=True):
package: str
name: str
@@ -123,21 +118,18 @@ PROVIDER_TYPE_OBJECT_MAPPING: dict[str, ProviderTypeMapping] = {
}
@experimental(feature_id=ExperimentalFeature.DECLARATIVE_AGENTS)
class DeclarativeLoaderError(AgentException):
"""Exception raised for errors in the declarative loader."""
pass
@experimental(feature_id=ExperimentalFeature.DECLARATIVE_AGENTS)
class ProviderLookupError(DeclarativeLoaderError):
"""Exception raised for errors in provider type lookup."""
pass
@experimental(feature_id=ExperimentalFeature.DECLARATIVE_AGENTS)
class AgentFactory:
"""Factory for creating Agent instances from declarative YAML definitions.
+3 -4
View File
@@ -4,7 +4,7 @@ description = "Declarative specification support for Microsoft Agent Framework."
authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
readme = "README.md"
requires-python = ">=3.10"
version = "1.0.0rc1"
version = "1.0.0b260528"
license-files = ["LICENSE"]
urls.homepage = "https://aka.ms/agent-framework"
urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -29,7 +29,7 @@ dependencies = [
]
[dependency-groups]
dev = [
"types-PyYaml==6.0.12.20260518"
"types-PyYaml==6.0.12.20250915"
]
[tool.uv]
@@ -49,8 +49,7 @@ addopts = "-ra -q -r fEX"
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"
filterwarnings = [
"ignore:Support for class-based `config` is deprecated:DeprecationWarning:pydantic.*",
"ignore::agent_framework._feature_stage.ExperimentalWarning",
"ignore:Support for class-based `config` is deprecated:DeprecationWarning:pydantic.*"
]
timeout = 120
markers = [
@@ -375,15 +375,13 @@ class DevServer:
logger.info("Starting Agent Framework Server")
await self._ensure_executor()
await self._ensure_openai_executor() # Initialize OpenAI executor
try:
yield
finally:
# Shutdown
logger.info("Shutting down Agent Framework Server")
yield
# Shutdown
logger.info("Shutting down Agent Framework Server")
# Cleanup entity resources (e.g., close credentials, clients)
if self.executor:
await self._cleanup_entities()
# Cleanup entity resources (e.g., close credentials, clients)
if self.executor:
await self._cleanup_entities()
app = FastAPI(
title="Agent Framework Server",
+1 -1
View File
@@ -30,7 +30,7 @@ dependencies = [
[dependency-groups]
dev = [
"types-python-dateutil==2.9.0.20260518",
"types-python-dateutil==2.9.0.20260402",
]
[tool.uv]
@@ -12,7 +12,6 @@ from ._embedding_client import (
)
from ._foundry_evals import (
FoundryEvals,
GeneratedEvaluatorRef,
evaluate_foundry_target,
evaluate_traces,
)
@@ -34,7 +33,6 @@ __all__ = [
"FoundryEmbeddingSettings",
"FoundryEvals",
"FoundryMemoryProvider",
"GeneratedEvaluatorRef",
"RawFoundryAgent",
"RawFoundryAgentChatClient",
"RawFoundryChatClient",
@@ -57,6 +57,8 @@ if TYPE_CHECKING:
from agent_framework import (
Agent,
AgentRunInputs,
ChatAndFunctionMiddlewareTypes,
ContextProvider,
MiddlewareTypes,
ToolTypes,
)
@@ -351,7 +353,6 @@ class RawFoundryAgentChatClient( # type: ignore[misc]
if _uses_foundry_agent_session(conversation_id):
run_options.pop("previous_response_id", None)
run_options.pop("conversation", None)
run_options.pop("model", None)
extra_body["agent_session_id"] = conversation_id
# Non-preview Prompt/Hosted Agent calls need agent_reference in the request body to
# tell the Responses API which Foundry agent (and version) is in use, since ``model``
@@ -367,6 +368,7 @@ class RawFoundryAgentChatClient( # type: ignore[misc]
# Strip tools from request body - Foundry API rejects requests with both
# agent endpoint and tools present. FunctionTools are invoked client-side
# by the function invocation layer, not sent to the service.
run_options.pop("model", None)
if not self.allow_preview:
run_options.pop("tools", None)
run_options.pop("tool_choice", None)
@@ -28,9 +28,8 @@ from __future__ import annotations
import asyncio
import logging
from collections.abc import Iterable, Sequence
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, cast
from collections.abc import Sequence
from typing import TYPE_CHECKING, Any
from agent_framework._evaluation import (
AgentEvalConverter,
@@ -40,7 +39,6 @@ from agent_framework._evaluation import (
EvalItemResult,
EvalResults,
EvalScoreResult,
RubricScore,
)
from agent_framework._feature_stage import ExperimentalFeature, experimental
from openai import AsyncOpenAI
@@ -53,54 +51,6 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
# region Generated rubric evaluator references
@experimental(feature_id=ExperimentalFeature.EVALS)
@dataclass(frozen=True)
class GeneratedEvaluatorRef:
"""A reference to a rubric evaluator that already exists in Foundry.
Pass instances of this class to :class:`FoundryEvals` to score items
with a pre-existing rubric evaluator (manually authored or
auto-generated through the Foundry portal). agent-framework is a
consumer here: it does not create or modify the evaluator definition;
it only references the persisted version by name.
Pinning ``version`` is strongly recommended so evaluation runs are
reproducible. ``version=None`` resolves to whichever version is
current at execution time; :class:`FoundryEvals` emits a warning when
a versionless reference is used. CI gates should always pass a
concrete version.
Attributes:
name: Evaluator name as stored in the Foundry project (for
example ``"reservation-policy-rubric"``). Distinct from
built-in evaluators such as ``"builtin.relevance"``.
version: Pinned evaluator version. ``None`` means "latest" —
this is discouraged for CI/repro and :class:`FoundryEvals`
will emit a warning when used.
display_name: Optional human-readable name used in result
summaries. Defaults to ``name`` when unset.
"""
name: str
version: str | None = None
display_name: str | None = None
@classmethod
def latest(cls, name: str, *, display_name: str | None = None) -> GeneratedEvaluatorRef:
"""Construct a versionless reference (resolves to the latest version at run time).
Discouraged for reproducible runs. Prefer the constructor with
an explicit ``version`` so CI and replay evaluations stay stable
when the evaluator is updated in Foundry.
"""
return cls(name=name, version=None, display_name=display_name)
# endregion
# Agent evaluators that accept query/response as conversation arrays.
# Maintained manually — check https://learn.microsoft.com/en-us/azure/ai-studio/how-to/develop/evaluate-sdk
# for the latest evaluator list. These are the evaluators that need conversation-format input.
@@ -216,7 +166,7 @@ def _resolve_evaluator(name: str) -> str:
def _build_testing_criteria(
evaluators: Sequence[str | GeneratedEvaluatorRef],
evaluators: Sequence[str],
model: str,
*,
include_data_mapping: bool = False,
@@ -225,9 +175,7 @@ def _build_testing_criteria(
"""Build ``testing_criteria`` for ``evals.create()``.
Args:
evaluators: Evaluator names (built-in shorts / fully-qualified
``builtin.*`` names) or :class:`GeneratedEvaluatorRef`
instances for generated rubric evaluators.
evaluators: Evaluator names.
model: Model deployment for the LLM judge.
include_data_mapping: Whether to include field-level data mapping
(required for the JSONL data source, not needed for response-based).
@@ -235,38 +183,7 @@ def _build_testing_criteria(
definitions.
"""
criteria: list[dict[str, Any]] = []
for entry_spec in evaluators:
if isinstance(entry_spec, GeneratedEvaluatorRef):
short = entry_spec.display_name or entry_spec.name
ref_entry: dict[str, Any] = {
"type": "azure_ai_evaluator",
"name": short,
"evaluator_name": entry_spec.name,
"initialization_parameters": {"deployment_name": model},
}
if entry_spec.version is not None:
ref_entry["evaluator_version"] = entry_spec.version
else:
logger.warning(
"GeneratedEvaluatorRef '%s' has no pinned version; the eval run "
"will resolve to whichever version is current at execution time. "
"Pin the version for reproducible runs.",
entry_spec.name,
)
if include_data_mapping:
# Rubric evaluators accept conversation arrays like agent
# evaluators, plus tool_definitions when items are tool-aware.
ref_mapping: dict[str, str] = {
"query": "{{item.query_messages}}",
"response": "{{item.response_messages}}",
}
if include_tool_definitions:
ref_mapping["tool_definitions"] = "{{item.tool_definitions}}"
ref_entry["data_mapping"] = ref_mapping
criteria.append(ref_entry)
continue
name = entry_spec
for name in evaluators:
qualified = _resolve_evaluator(name)
short = name if not name.startswith("builtin.") else name.split(".")[-1]
@@ -330,9 +247,9 @@ def _build_item_schema(
def _resolve_default_evaluators(
evaluators: Sequence[str | GeneratedEvaluatorRef] | None,
evaluators: Sequence[str] | None,
items: Sequence[EvalItem | dict[str, Any]] | None = None,
) -> list[str | GeneratedEvaluatorRef]:
) -> list[str]:
"""Resolve evaluators, applying defaults when ``None``.
Defaults to relevance + coherence + task_adherence. Automatically adds
@@ -341,7 +258,7 @@ def _resolve_default_evaluators(
if evaluators is not None:
return list(evaluators)
result: list[str | GeneratedEvaluatorRef] = list(_DEFAULT_EVALUATORS)
result = list(_DEFAULT_EVALUATORS)
if items is not None:
has_tools = any((item.tools if isinstance(item, EvalItem) else item.get("tool_definitions")) for item in items)
if has_tools:
@@ -350,24 +267,14 @@ def _resolve_default_evaluators(
def _filter_tool_evaluators(
evaluators: list[str | GeneratedEvaluatorRef],
evaluators: list[str],
items: Sequence[EvalItem | dict[str, Any]],
) -> list[str | GeneratedEvaluatorRef]:
"""Remove tool evaluators if no items have tool definitions.
Generated rubric evaluators are tool-aware but not tool-required; they
are preserved regardless of whether items carry tool definitions.
"""
) -> list[str]:
"""Remove tool evaluators if no items have tool definitions."""
has_tools = any((item.tools if isinstance(item, EvalItem) else item.get("tool_definitions")) for item in items)
if has_tools:
return evaluators
def _is_tool_only(spec: str | GeneratedEvaluatorRef) -> bool:
if isinstance(spec, GeneratedEvaluatorRef):
return False
return _resolve_evaluator(spec) in _TOOL_EVALUATORS
filtered = [e for e in evaluators if not _is_tool_only(e)]
filtered = [e for e in evaluators if _resolve_evaluator(e) not in _TOOL_EVALUATORS]
if not filtered:
raise ValueError(
f"All requested evaluators {evaluators} require tool definitions, "
@@ -375,7 +282,7 @@ def _filter_tool_evaluators(
"or choose evaluators that do not require tools."
)
if len(filtered) < len(evaluators):
removed = [e for e in evaluators if _is_tool_only(e)]
removed = [e for e in evaluators if _resolve_evaluator(e) in _TOOL_EVALUATORS]
logger.info("Removed tool evaluators %s (no items have tools)", removed)
return filtered
@@ -447,114 +354,6 @@ def _extract_per_evaluator(run: RunRetrieveResponse) -> dict[str, dict[str, int]
return per_eval
_RUBRIC_DIMENSION_KEYS: tuple[str, ...] = ("dimension_scores", "rubric_scores")
"""Property keys that may carry per-dimension rubric breakdowns.
The published Foundry rubric-evaluator output format uses
``properties.dimension_scores`` (see the Microsoft Learn "Rubric
evaluators" reference). Earlier preview builds and some SDK shapes
used ``rubric_scores``; we accept both for defensive forward/backward
compatibility.
"""
def _parse_dimension_entries(raw: Any) -> list[RubricScore]:
"""Parse a raw list-like payload into ``RubricScore`` instances.
Returns an empty list when ``raw`` is falsy, not iterable, or
contains no well-formed entries.
"""
if not raw:
return []
try:
raw_iter: Iterable[Any] = iter(raw)
except TypeError:
return []
parsed: list[RubricScore] = []
for raw_entry in raw_iter:
entry: Any = raw_entry
try:
rid: Any
score_val: Any
applicable: Any
weight: Any
reason: Any
if isinstance(entry, dict):
entry_any = cast("dict[str, Any]", entry)
rid = entry_any.get("id")
score_val = entry_any.get("score")
applicable = entry_any.get("applicable")
weight = entry_any.get("weight")
reason = entry_any.get("reason", "")
else:
rid = getattr(entry, "id", None)
score_val = getattr(entry, "score", None)
applicable = getattr(entry, "applicable", None)
weight = getattr(entry, "weight", None)
reason = getattr(entry, "reason", "") or ""
if rid is None or weight is None or applicable is None:
continue
parsed.append(
RubricScore(
id=str(rid),
score=int(score_val) if isinstance(score_val, (int, float)) else None,
applicable=bool(applicable),
weight=int(weight),
reason=str(reason) if reason is not None else "",
)
)
except (TypeError, ValueError):
logger.debug("Skipping malformed rubric dimension entry: %s", cast("Any", entry), exc_info=True)
return parsed
def _extract_rubric_scores(sample: Any) -> list[RubricScore] | None:
"""Extract typed ``RubricScore`` instances from an evaluator's raw sample payload.
Foundry rubric evaluators include a per-dimension breakdown under
``properties.dimension_scores`` on each result (preview builds used
``rubric_scores``; both keys are accepted, with the canonical
``dimension_scores`` taking priority). The exact location may
vary across SDK versions, so this helper accepts a few shapes:
* The SDK ``sample`` object exposes
``properties.dimension_scores`` / ``properties.rubric_scores``.
* The ``sample`` is a dict containing the same under
``properties.<key>``.
* The ``sample`` is a dict with ``dimension_scores`` /
``rubric_scores`` at the top level.
Returns ``None`` when no rubric scores are present (i.e. the
evaluator was not a rubric evaluator).
"""
if sample is None:
return None
containers: list[Any] = []
properties: Any = getattr(sample, "properties", None)
if properties is not None:
containers.append(properties)
if isinstance(sample, dict):
sample_any = cast("dict[str, Any]", sample)
props_dict: Any = sample_any.get("properties")
if props_dict is not None and props_dict is not properties:
containers.append(props_dict)
containers.append(sample_any)
for container in containers:
for key in _RUBRIC_DIMENSION_KEYS:
raw: Any = None
if isinstance(container, dict):
raw = cast("dict[str, Any]", container).get(key)
elif hasattr(container, key):
raw = getattr(container, key, None)
parsed = _parse_dimension_entries(raw)
if parsed:
return parsed
return None
async def _fetch_output_items(
client: AsyncOpenAI,
eval_id: str,
@@ -578,15 +377,12 @@ async def _fetch_output_items(
# Extract per-evaluator scores
scores: list[EvalScoreResult] = []
for r in oi.results or []:
sample = r.sample
dimensions = _extract_rubric_scores(sample)
scores.append(
EvalScoreResult(
name=r.name,
score=r.score,
passed=r.passed,
sample=sample,
dimensions=dimensions,
sample=r.sample,
)
)
@@ -598,18 +394,15 @@ async def _fetch_output_items(
output_text: str | None = None
response_id: str | None = None
# mypy infers oi.sample as dict[str, object] | None, but the
# OpenAI SDK actually returns a typed Sample model. Cast to Any so
# both type checkers accept the attribute access pattern.
oi_sample: Any = oi.sample
if oi_sample is not None:
err = oi_sample.error
if err is not None and (err.code or err.message):
sample = oi.sample
if sample is not None: # pyright: ignore[reportUnnecessaryComparison]
err = sample.error
if err is not None and (err.code or err.message): # pyright: ignore[reportUnnecessaryComparison]
error_code = err.code or None
error_message = err.message or None
usage = oi_sample.usage
if usage is not None and usage.total_tokens:
usage = sample.usage
if usage is not None and usage.total_tokens: # pyright: ignore[reportUnnecessaryComparison]
token_usage = {
"prompt_tokens": usage.prompt_tokens,
"completion_tokens": usage.completion_tokens,
@@ -618,13 +411,13 @@ async def _fetch_output_items(
}
# Extract input/output text
if oi_sample.input:
parts = [si.content for si in oi_sample.input if si.role == "user"]
if sample.input:
parts = [si.content for si in sample.input if si.role == "user"]
if parts:
input_text = " ".join(parts)
if oi_sample.output:
parts = [so.content or "" for so in oi_sample.output if so.role == "assistant"]
if sample.output:
parts = [so.content or "" for so in sample.output if so.role == "assistant"]
if parts:
output_text = " ".join(parts)
@@ -679,7 +472,7 @@ async def _evaluate_via_responses_impl(
*,
client: AsyncOpenAI,
response_ids: Sequence[str],
evaluators: list[str | GeneratedEvaluatorRef],
evaluators: list[str],
model: str,
eval_name: str,
poll_interval: float,
@@ -780,11 +573,8 @@ class FoundryEvals:
(from ``azure.ai.projects.aio``). Provide this or *client*.
model: Model deployment name for the evaluator LLM judge.
Resolved from ``client.model`` when omitted.
evaluators: Evaluator specifications. Entries may be built-in
short names (e.g. ``"relevance"``), fully-qualified
``"builtin.*"`` names, or :class:`GeneratedEvaluatorRef`
instances for previously generated rubric evaluators. When
``None`` (default), uses smart defaults based on item data.
evaluators: Evaluator names (e.g. ``["relevance", "tool_call_accuracy"]``).
When ``None`` (default), uses smart defaults based on item data.
conversation_split: How to split multi-turn conversations into
query/response halves. Defaults to ``LAST_TURN``. Pass a
``ConversationSplit`` enum value or a custom callable — see
@@ -833,7 +623,7 @@ class FoundryEvals:
client: FoundryChatClient | None = None,
project_client: AIProjectClient | None = None,
model: str | None = None,
evaluators: Sequence[str | GeneratedEvaluatorRef] | None = None,
evaluators: Sequence[str] | None = None,
conversation_split: ConversationSplitter = ConversationSplit.LAST_TURN,
poll_interval: float = 5.0,
timeout: float = 180.0,
@@ -852,9 +642,7 @@ class FoundryEvals:
"Model is required. Pass model= explicitly or use a FoundryChatClient that has a model configured."
)
self._model = resolved_model
self._evaluators: list[str | GeneratedEvaluatorRef] | None = (
list(evaluators) if evaluators is not None else None
)
self._evaluators = list(evaluators) if evaluators is not None else None
self._conversation_split = conversation_split
self._poll_interval = poll_interval
self._timeout = timeout
@@ -890,7 +678,7 @@ class FoundryEvals:
async def _evaluate_via_dataset(
self,
items: Sequence[EvalItem],
evaluators: list[str | GeneratedEvaluatorRef],
evaluators: list[str],
eval_name: str,
) -> EvalResults:
"""Evaluate using JSONL dataset upload path."""
+1 -1
View File
@@ -26,7 +26,7 @@ dependencies = [
"agent-framework-core>=1.7.0,<2",
"agent-framework-openai>=1.7.0,<2",
"azure-ai-inference>=1.0.0b9,<1.0.0b10",
"azure-ai-projects>=2.2.0,<3.0",
"azure-ai-projects>=2.1.0,<3.0",
]
[tool.uv]
@@ -203,7 +203,7 @@ async def test_raw_foundry_agent_chat_client_prepare_options_accepts_function_to
async def test_raw_foundry_agent_chat_client_prepare_options_strips_client_side_fields() -> None:
"""Test that _prepare_options strips tool-loop fields but preserves model for non-session requests."""
"""Test that _prepare_options strips model and tool-loop fields from run_options."""
mock_project = MagicMock()
mock_openai = MagicMock()
@@ -235,49 +235,16 @@ async def test_raw_foundry_agent_chat_client_prepare_options_strips_client_side_
options={"tools": [my_func]},
)
# model is preserved for non-session (PromptAgent) requests
assert result["model"] == "gpt-4.1"
assert "model" not in result
assert "tools" not in result
assert "tool_choice" not in result
assert "parallel_tool_calls" not in result
# agent_reference is required so the Responses API can resolve model server-side; see #5582.
assert result == {
"model": "gpt-4.1",
"extra_body": {"agent_reference": {"name": "test-agent", "type": "agent_reference"}},
}
async def test_raw_foundry_agent_chat_client_prepare_options_strips_model_for_hosted_session() -> None:
"""Test that model is stripped when using a hosted agent session (not a PromptAgent)."""
mock_project = MagicMock()
mock_openai = MagicMock()
mock_project.get_openai_client.return_value = mock_openai
client = RawFoundryAgentChatClient(
project_client=mock_project,
agent_name="test-agent",
)
with patch(
"agent_framework_openai._chat_client.RawOpenAIChatClient._prepare_options",
new_callable=AsyncMock,
return_value={
"model": "gpt-4.1",
"previous_response_id": "resp_abc",
},
):
result = await client._prepare_options(
messages=[Message(role="user", contents="hi")],
options={"conversation_id": "agent-session-123"},
)
assert "model" not in result
assert "previous_response_id" not in result
assert result["extra_body"]["agent_session_id"] == "agent-session-123"
assert result["extra_body"]["agent_reference"] == {"name": "test-agent", "type": "agent_reference"}
async def test_raw_foundry_agent_chat_client_prepare_options_injects_agent_reference_first_turn() -> None:
"""First-turn (no conversation_id) Prompt Agent calls must carry agent_reference in extra_body.
@@ -305,6 +272,7 @@ async def test_raw_foundry_agent_chat_client_prepare_options_injects_agent_refer
options={},
)
assert "model" not in result
assert result["extra_body"] == {
"agent_reference": {"name": "test-agent", "type": "agent_reference", "version": "2"},
}
@@ -365,8 +333,7 @@ async def test_raw_foundry_agent_chat_client_prepare_options_skips_agent_referen
options={},
)
# model is preserved for non-session requests (platform tolerates it for hosted agents)
assert result["model"] == "gpt-4.1"
assert "model" not in result
# No extra_body at all is the cleanest signal — agent_reference must not be injected here.
assert "extra_body" not in result
@@ -396,39 +363,6 @@ async def test_raw_foundry_agent_chat_client_prepare_options_respects_caller_age
assert result["extra_body"]["agent_reference"] == caller_reference
async def test_raw_foundry_agent_chat_client_prepare_options_preserves_model_for_resp_continuation() -> None:
"""Test that model is preserved when conversation_id is a resp_* continuation (HostedAgent v1 / v2-no-session)."""
mock_project = MagicMock()
mock_openai = MagicMock()
mock_project.get_openai_client.return_value = mock_openai
client = RawFoundryAgentChatClient(
project_client=mock_project,
agent_name="test-agent",
)
with patch(
"agent_framework_openai._chat_client.RawOpenAIChatClient._prepare_options",
new_callable=AsyncMock,
return_value={
"model": "gpt-4.1",
"previous_response_id": "resp_abc123",
},
):
result = await client._prepare_options(
messages=[Message(role="user", contents="hi")],
options={"conversation_id": "resp_abc123"},
)
# model preserved — resp_* is standard Responses API continuity, not a hosted session
assert result["model"] == "gpt-4.1"
# previous_response_id preserved — not stripped outside hosted session path
assert result["previous_response_id"] == "resp_abc123"
# no agent_session_id injected
assert "extra_body" not in result or "agent_session_id" not in result.get("extra_body", {})
async def test_raw_foundry_agent_chat_client_prepare_options_maps_agent_session_id_to_extra_body() -> None:
"""Test that service_session_id is forwarded as agent_session_id for hosted sessions."""
@@ -25,25 +25,16 @@ from agent_framework._evaluation import (
from agent_framework._workflows._workflow import WorkflowRunResult
from openai import AsyncOpenAI
from agent_framework_foundry import GeneratedEvaluatorRef
from agent_framework_foundry._foundry_evals import (
_AGENT_EVALUATORS,
_BUILTIN_EVALUATORS,
_TOOL_EVALUATORS,
FoundryEvals,
_build_item_schema,
_build_testing_criteria,
_extract_per_evaluator,
_extract_result_counts,
_extract_rubric_scores,
_fetch_output_items,
_filter_tool_evaluators,
_poll_eval_run,
_resolve_default_evaluators,
_resolve_evaluator,
_resolve_openai_client,
evaluate_foundry_target,
evaluate_traces,
)
@@ -815,67 +806,6 @@ class TestBuildTestingCriteria:
for c in criteria:
assert "tool_definitions" in c["data_mapping"], f"{c['name']} missing tool_definitions"
def test_generated_evaluator_ref_pinned_version(self) -> None:
ref = GeneratedEvaluatorRef(name="my-rubric", version="1")
criteria = _build_testing_criteria([ref], "gpt-4o", include_data_mapping=True)
assert len(criteria) == 1
c = criteria[0]
assert c["type"] == "azure_ai_evaluator"
assert c["evaluator_name"] == "my-rubric"
assert c["evaluator_version"] == "1"
assert c["name"] == "my-rubric"
assert c["initialization_parameters"] == {"deployment_name": "gpt-4o"}
assert c["data_mapping"] == {
"query": "{{item.query_messages}}",
"response": "{{item.response_messages}}",
}
def test_generated_evaluator_ref_display_name_used_as_short(self) -> None:
ref = GeneratedEvaluatorRef(name="my-rubric", version="2", display_name="My Rubric")
criteria = _build_testing_criteria([ref], "gpt-4o")
assert criteria[0]["name"] == "My Rubric"
assert criteria[0]["evaluator_name"] == "my-rubric"
def test_generated_evaluator_ref_tool_definitions_added(self) -> None:
ref = GeneratedEvaluatorRef(name="my-rubric", version="1")
criteria = _build_testing_criteria(
[ref],
"gpt-4o",
include_data_mapping=True,
include_tool_definitions=True,
)
assert criteria[0]["data_mapping"]["tool_definitions"] == "{{item.tool_definitions}}"
def test_generated_evaluator_ref_unpinned_warns(self, caplog: pytest.LogCaptureFixture) -> None:
import logging
ref = GeneratedEvaluatorRef.latest("my-rubric")
with caplog.at_level(logging.WARNING, logger="agent_framework_foundry._foundry_evals"):
criteria = _build_testing_criteria([ref], "gpt-4o")
assert "evaluator_version" not in criteria[0]
assert any("no pinned version" in r.message for r in caplog.records)
def test_generated_evaluator_ref_mixed_with_builtins(self) -> None:
ref = GeneratedEvaluatorRef(name="my-rubric", version="1")
criteria = _build_testing_criteria(
["relevance", ref, "task_adherence"],
"gpt-4o",
include_data_mapping=True,
)
assert [c["name"] for c in criteria] == ["relevance", "my-rubric", "task_adherence"]
assert criteria[0]["evaluator_name"] == "builtin.relevance"
assert criteria[1]["evaluator_name"] == "my-rubric"
assert criteria[2]["evaluator_name"] == "builtin.task_adherence"
# ---------------------------------------------------------------------------
# _build_item_schema
@@ -1333,29 +1263,6 @@ class TestFilterToolEvaluators:
items,
)
def test_preserves_generated_ref_when_no_tools(self) -> None:
ref = GeneratedEvaluatorRef(name="rubric", version="1")
items = [
EvalItem(conversation=[Message("user", ["q"]), Message("assistant", ["r"])]),
]
result = _filter_tool_evaluators(
["relevance", ref, "tool_call_accuracy"],
items,
)
assert "relevance" in result
assert ref in result
assert "tool_call_accuracy" not in result
def test_generated_ref_alone_does_not_raise(self) -> None:
ref = GeneratedEvaluatorRef(name="rubric", version="1")
items = [
EvalItem(conversation=[Message("user", ["q"]), Message("assistant", ["r"])]),
]
result = _filter_tool_evaluators([ref], items)
assert result == [ref]
# ---------------------------------------------------------------------------
# EvalResults
@@ -2360,6 +2267,7 @@ class TestEvalResultsWithItems:
class TestFetchOutputItems:
async def test_fetches_and_converts_output_items(self) -> None:
from agent_framework_foundry._foundry_evals import _fetch_output_items
# Build mock output items matching the OpenAI SDK schema
mock_result = MagicMock()
@@ -2421,6 +2329,7 @@ class TestFetchOutputItems:
assert item.error_code is None
async def test_handles_errored_item(self) -> None:
from agent_framework_foundry._foundry_evals import _fetch_output_items
mock_error = MagicMock()
mock_error.code = "QueryExtractionError"
@@ -2452,6 +2361,7 @@ class TestFetchOutputItems:
assert len(item.scores) == 0
async def test_handles_api_failure_gracefully(self) -> None:
from agent_framework_foundry._foundry_evals import _fetch_output_items
mock_client = MagicMock()
mock_client.evals.runs.output_items.list = AsyncMock(side_effect=TypeError("API error"))
@@ -2459,166 +2369,6 @@ class TestFetchOutputItems:
items = await _fetch_output_items(mock_client, "eval_1", "run_1")
assert items == []
async def test_extracts_rubric_scores_from_dict_sample(self) -> None:
mock_result = MagicMock()
mock_result.name = "my-rubric"
mock_result.score = 0.85
mock_result.passed = True
mock_result.sample = {
"properties": {
"rubric_scores": [
{"id": "policy", "score": 4, "applicable": True, "weight": 1, "reason": "ok"},
{"id": "safety", "score": None, "applicable": False, "weight": 1, "reason": "n/a"},
]
}
}
mock_oi = MagicMock()
mock_oi.id = "oi_1"
mock_oi.status = "pass"
mock_oi.results = [mock_result]
mock_oi.sample = None
mock_oi.datasource_item = {}
mock_client = MagicMock()
mock_client.evals.runs.output_items.list = AsyncMock(return_value=_AsyncPage([mock_oi]))
items = await _fetch_output_items(mock_client, "eval_1", "run_1")
assert len(items) == 1
scores = items[0].scores
assert len(scores) == 1
assert scores[0].dimensions is not None
assert len(scores[0].dimensions) == 2
policy = next(d for d in scores[0].dimensions if d.id == "policy")
assert policy.score == 4
assert policy.applicable is True
assert policy.weight == 1
assert policy.reason == "ok"
safety = next(d for d in scores[0].dimensions if d.id == "safety")
assert safety.score is None
assert safety.applicable is False
async def test_no_rubric_scores_when_absent(self) -> None:
mock_result = MagicMock()
mock_result.name = "relevance"
mock_result.score = 0.85
mock_result.passed = True
mock_result.sample = None
mock_oi = MagicMock()
mock_oi.id = "oi_2"
mock_oi.status = "pass"
mock_oi.results = [mock_result]
mock_oi.sample = None
mock_oi.datasource_item = {}
mock_client = MagicMock()
mock_client.evals.runs.output_items.list = AsyncMock(return_value=_AsyncPage([mock_oi]))
items = await _fetch_output_items(mock_client, "eval_1", "run_1")
assert items[0].scores[0].dimensions is None
class TestExtractRubricScores:
def test_handles_attribute_style_properties(self) -> None:
rs = MagicMock()
rs.id = "policy"
rs.score = 5
rs.applicable = True
rs.weight = 2
rs.reason = "ok"
sample = MagicMock()
sample.properties = MagicMock()
sample.properties.rubric_scores = [rs]
result = _extract_rubric_scores(sample)
assert result is not None
assert result[0].id == "policy"
assert result[0].score == 5
assert result[0].weight == 2
def test_top_level_rubric_scores_in_dict(self) -> None:
sample = {"rubric_scores": [{"id": "a", "score": 3, "applicable": True, "weight": 1, "reason": "r"}]}
result = _extract_rubric_scores(sample)
assert result is not None
assert result[0].id == "a"
def test_returns_none_when_missing(self) -> None:
assert _extract_rubric_scores(None) is None
assert _extract_rubric_scores({}) is None
assert _extract_rubric_scores({"properties": {}}) is None
def test_skips_malformed_entries(self) -> None:
sample = {
"properties": {
"rubric_scores": [
{"id": "good", "score": 3, "applicable": True, "weight": 1, "reason": "ok"},
{"id": "bad-no-weight", "score": 2, "applicable": True, "reason": "x"},
]
}
}
result = _extract_rubric_scores(sample)
assert result is not None
assert len(result) == 1
assert result[0].id == "good"
def test_canonical_dimension_scores_key_from_docs(self) -> None:
"""Per the Microsoft Learn docs, runtime output uses ``properties.dimension_scores``."""
sample = {
"properties": {
"dimension_scores": [
{
"id": "intent_recognition",
"score": 5,
"applicable": True,
"weight": 9,
"reason": "Identified correctly.",
},
{
"id": "general_quality",
"score": 4,
"applicable": True,
"weight": 5,
"reason": "Strong overall.",
},
]
}
}
result = _extract_rubric_scores(sample)
assert result is not None
assert [r.id for r in result] == ["intent_recognition", "general_quality"]
assert [r.score for r in result] == [5, 4]
assert [r.weight for r in result] == [9, 5]
def test_dimension_scores_via_attribute(self) -> None:
"""Canonical key also resolves when properties exposes ``dimension_scores`` as an attr."""
rs = MagicMock()
rs.id = "policy_enforcement"
rs.score = 1
rs.applicable = True
rs.weight = 5
rs.reason = "violated"
sample = MagicMock()
sample.properties = MagicMock(spec=["dimension_scores"])
sample.properties.dimension_scores = [rs]
result = _extract_rubric_scores(sample)
assert result is not None
assert result[0].id == "policy_enforcement"
assert result[0].score == 1
# ---------------------------------------------------------------------------
# _poll_eval_run — timeout / failed / canceled paths
@@ -2628,6 +2378,7 @@ class TestExtractRubricScores:
class TestPollEvalRun:
async def test_timeout_returns_timeout_status(self) -> None:
"""Poll timeout returns EvalResults with status='timeout'."""
from agent_framework_foundry._foundry_evals import _poll_eval_run
mock_client = MagicMock()
mock_pending = MagicMock()
@@ -2641,6 +2392,7 @@ class TestPollEvalRun:
async def test_failed_run_returns_error(self) -> None:
"""Failed run returns EvalResults with error message."""
from agent_framework_foundry._foundry_evals import _poll_eval_run
mock_client = MagicMock()
mock_failed = MagicMock()
@@ -2658,6 +2410,7 @@ class TestPollEvalRun:
async def test_canceled_run_returns_canceled_status(self) -> None:
"""Canceled run returns EvalResults with status='canceled'."""
from agent_framework_foundry._foundry_evals import _poll_eval_run
mock_client = MagicMock()
mock_canceled = MagicMock()
@@ -2682,6 +2435,7 @@ class TestPollEvalRun:
class TestEvaluateTraces:
async def test_raises_without_required_args(self) -> None:
"""Raises ValueError when no response_ids, trace_ids, or agent_id given."""
from agent_framework_foundry._foundry_evals import evaluate_traces
mock_client = MagicMock()
with pytest.raises(ValueError, match="Provide at least one of"):
@@ -2692,6 +2446,7 @@ class TestEvaluateTraces:
async def test_response_ids_path(self) -> None:
"""evaluate_traces with response_ids uses the responses API path."""
from agent_framework_foundry._foundry_evals import evaluate_traces
mock_client = MagicMock()
@@ -2739,6 +2494,7 @@ class TestEvaluateTraces:
async def test_trace_ids_path(self) -> None:
"""evaluate_traces with trace_ids builds azure_ai_traces data source."""
from agent_framework_foundry._foundry_evals import evaluate_traces
mock_client = MagicMock()
@@ -2778,6 +2534,7 @@ class TestEvaluateTraces:
class TestEvaluateFoundryTarget:
async def test_happy_path(self) -> None:
"""evaluate_foundry_target creates eval + run and polls to completion."""
from agent_framework_foundry._foundry_evals import evaluate_foundry_target
mock_client = MagicMock()
@@ -2913,11 +2670,13 @@ class TestEvaluatorSetConsistency:
"""Verify that _AGENT_EVALUATORS and _TOOL_EVALUATORS are subsets of _BUILTIN_EVALUATORS."""
def test_agent_evaluators_subset(self):
from agent_framework_foundry._foundry_evals import _AGENT_EVALUATORS, _BUILTIN_EVALUATORS
diff = _AGENT_EVALUATORS - set(_BUILTIN_EVALUATORS.values())
assert not diff, f"_AGENT_EVALUATORS has names not in _BUILTIN_EVALUATORS: {diff}"
def test_tool_evaluators_subset(self):
from agent_framework_foundry._foundry_evals import _BUILTIN_EVALUATORS, _TOOL_EVALUATORS
diff = _TOOL_EVALUATORS - set(_BUILTIN_EVALUATORS.values())
assert not diff, f"_TOOL_EVALUATORS has names not in _BUILTIN_EVALUATORS: {diff}"
@@ -2931,6 +2690,7 @@ class TestEvaluatorSetConsistency:
class TestEvaluateTracesAgentId:
async def test_agent_id_only_path(self) -> None:
"""evaluate_traces with agent_id only builds azure_ai_traces data source."""
from agent_framework_foundry._foundry_evals import evaluate_traces
mock_client = MagicMock()
@@ -2988,6 +2748,7 @@ class TestFilterToolEvaluatorsRaises:
class TestEvaluateFoundryTargetValidation:
async def test_target_without_type_raises(self) -> None:
"""target dict without 'type' key raises ValueError."""
from agent_framework_foundry._foundry_evals import evaluate_foundry_target
mock_client = MagicMock()
with pytest.raises(ValueError, match="'type' key"):
@@ -9,9 +9,9 @@ import logging
import os
import tempfile
import threading
from collections.abc import AsyncIterable, AsyncIterator, Generator, Mapping, Sequence
from collections.abc import AsyncIterable, AsyncIterator, Generator, Sequence
from contextlib import AbstractAsyncContextManager, AsyncExitStack, suppress
from dataclasses import asdict, is_dataclass
from dataclasses import asdict, dataclass, is_dataclass
from pathlib import Path
from typing import Protocol, cast
@@ -264,28 +264,73 @@ def _checkpoint_storage_for_context(root: str, context_id: str) -> FileCheckpoin
# Foundry Toolbox Auth integration
# Consent-URL error code returned by the Foundry MCP gateway when calling `/list`
CONSENT_ERROR_CODE = -32007
CONSENT_ERROR_CODE = -32006
def consent_url_from_error(exc: BaseException) -> str | None:
"""Return the consent URL when ``exc`` wraps a Foundry MCP gateway consent error.
@dataclass
class ConsentError:
name: str
consent_url: str
The Agent Framework MCP layer surfaces gateway consent failures by wrapping the underlying
``McpError`` inside an :class:`AgentFrameworkException` (typically a ``ToolExecutionException``
raised from ``MCPStreamableHTTPTool.__aenter__``). This helper inspects ``exc.args`` for a
wrapped ``McpError`` whose ``error.code`` is :data:`CONSENT_ERROR_CODE`; when found, the
consent link the gateway returned in ``error.message`` is returned. Returns ``None`` for
anything else, so callers can do ``if (url := consent_url_from_error(ex)) is None: raise``.
def consent_url_from_error(exc: BaseException) -> list[ConsentError] | None:
"""Return the consent URLs when ``exc`` wraps Foundry MCP gateway consent errors.
Args:
exc: The exception to inspect.
Returns:
The consent URL if ``exc`` wraps a consent ``McpError``, otherwise ``None``.
The consent URL(s) extracted from the error, or ``None`` if no consent error was found.
"""
inner_exception = next((arg for arg in exc.args if isinstance(arg, McpError)), None)
if inner_exception is not None and inner_exception.error.code == CONSENT_ERROR_CODE:
return inner_exception.error.message
# Parse the error message
# The error message is structured with the following format:
# "tools/list failed for 1 tool source(s), succeeded for 0 tool source(s) {"errors":[{"name": ..."
# where the second part is a JSON string that can be deserialized into an object with the following shape:
# ruff: disable[ERA001]
# {
# "errors" : [
# {
# "name": "Name of the MCP tool that requires consent",
# "type" : "mcp",
# "error": {
# "code": "CONSENT_REQUIRED",
# "message": consent_url,
# }
# }
# ]
# }
# ruff: enable[ERA001]
try:
consent_errors: list[ConsentError] = []
error_message_start = inner_exception.error.message.find("{")
if error_message_start == -1:
logger.warning("Consent error message does not contain JSON: %s", inner_exception.error.message)
return None
consent_details_json = inner_exception.error.message[error_message_start:]
consent_details = json.loads(consent_details_json)
if "errors" not in consent_details or not isinstance(consent_details["errors"], list):
logger.warning("Consent error message JSON does not contain 'errors' list: %s", consent_details_json)
return None
for error in consent_details["errors"]:
if (
isinstance(error, dict)
and error.get("type") == "mcp" # type: ignore
and "error" in error
and isinstance(error["error"], dict)
and error["error"].get("code") == "CONSENT_REQUIRED" # type: ignore
and "message" in error["error"]
):
consent_url = error["error"]["message"] # type: ignore
if isinstance(consent_url, str):
consent_errors.append(ConsentError(name=error.get("name", "Unknown"), consent_url=consent_url)) # type: ignore
else:
logger.warning("Consent URL in error message is not a valid URL: %s", consent_url) # type: ignore
if consent_errors:
return consent_errors
except json.JSONDecodeError:
logger.warning("Failed to parse consent details JSON: %s", inner_exception.error.message)
return None
@@ -448,18 +493,19 @@ class ResponsesHostServer(ResponsesAgentServerHost):
try:
await self._ensure_agent_ready()
except AgentFrameworkException as ex:
consent_url = consent_url_from_error(ex)
if consent_url is None:
consent_errors = consent_url_from_error(ex)
if consent_errors is None:
raise
logger.warning("OAuth consent required for Foundry MCP gateway.")
oauth_item = OAuthConsentRequestOutputItem(
id=IdGenerator.new_id("oacr"),
consent_link=consent_url,
server_label="Foundry Toolbox",
)
builder = response_event_stream.add_output_item(oauth_item.id)
yield builder.emit_added(oauth_item)
yield builder.emit_done(oauth_item)
for consent_error in consent_errors:
logger.warning("Consent URL for tool '%s': %s", consent_error.name, consent_error.consent_url)
oauth_item = OAuthConsentRequestOutputItem(
id=IdGenerator.new_id("oacr"),
consent_link=consent_error.consent_url,
server_label=consent_error.name,
)
builder = response_event_stream.add_output_item(oauth_item.id)
yield builder.emit_added(oauth_item)
yield builder.emit_done(oauth_item)
yield response_event_stream.emit_completed()
return
@@ -472,12 +518,14 @@ class ResponsesHostServer(ResponsesAgentServerHost):
# Run the agent in non-streaming mode
response = await self._agent.run(stream=False, **run_kwargs) # type: ignore[reportUnknownMemberType]
async for item in _to_outputs_for_messages(
response_event_stream,
response.messages,
approval_storage=self._approval_storage,
):
yield item
for message in response.messages:
for content in message.contents:
async for item in _to_outputs(
response_event_stream,
content,
approval_storage=self._approval_storage,
):
yield item
yield response_event_stream.emit_completed()
else:
if tracker is None: # pragma: no cover - defensive, set above
@@ -521,7 +569,7 @@ class ResponsesHostServer(ResponsesAgentServerHost):
by the hosting infrastructure or files will be preserved upon deactivation.
"""
input_items = await context.get_input_items()
input_messages = await _items_to_messages(input_items, approval_storage=self._approval_storage)
input_messages = await _items_to_messages(input_items)
is_streaming_request = request.stream is not None and request.stream is True
_, are_options_set = _to_chat_options(request)
@@ -618,12 +666,10 @@ class ResponsesHostServer(ResponsesAgentServerHost):
checkpoint_storage=write_storage,
)
async for item in _to_outputs_for_messages(
response_event_stream,
response.messages,
approval_storage=self._approval_storage,
):
yield item
for message in response.messages:
for content in message.contents:
async for item in _to_outputs(response_event_stream, content):
yield item
await self._delete_not_latest_checkpoints(write_storage, self._agent.workflow.name)
yield response_event_stream.emit_completed()
@@ -643,9 +689,7 @@ class ResponsesHostServer(ResponsesAgentServerHost):
for event in tracker.handle(content):
yield event
if tracker.needs_async:
async for item in _to_outputs(
response_event_stream, content, approval_storage=self._approval_storage
):
async for item in _to_outputs(response_event_stream, content):
yield item
tracker.needs_async = False
@@ -731,7 +775,7 @@ class _OutputItemTracker:
yield self._fc_builder.emit_arguments_delta(args_str)
elif content.type == "mcp_server_tool_call" and content.tool_name:
key = content.call_id or f"{content.server_name or 'default'}::{content.tool_name}"
key = f"{content.server_name or 'default'}::{content.tool_name}"
if self._active_type != "mcp_server_tool_call" or self._active_id != key:
yield from self._close()
yield from self._open_mcp_call(content)
@@ -740,24 +784,6 @@ class _OutputItemTracker:
if self._mcp_builder is not None:
yield self._mcp_builder.emit_arguments_delta(args_str)
elif (
content.type == "mcp_server_tool_result"
and self._active_type == "mcp_server_tool_call"
and self._mcp_builder is not None
and content.call_id is not None
and content.call_id == self._mcp_builder.item_id
):
accumulated = "".join(self._accumulated)
yield self._mcp_builder.emit_arguments_done(accumulated)
yield self._mcp_builder.emit_completed()
yield self._mcp_builder.emit_done(output=_stringify_mcp_output(content.output))
self._mcp_builder = None
self._active_type = None
self._active_id = None
self._accumulated.clear()
self.needs_async = False
return
else:
yield from self._close()
self.needs_async = True
@@ -797,10 +823,9 @@ class _OutputItemTracker:
self._mcp_builder = self._stream.add_output_item_mcp_call(
server_label=content.server_name or "default",
name=content.tool_name or "",
item_id=content.call_id,
)
self._active_type = "mcp_server_tool_call"
self._active_id = content.call_id or f"{content.server_name or 'default'}::{content.tool_name}"
self._active_id = f"{content.server_name or 'default'}::{content.tool_name}"
yield self._mcp_builder.emit_added()
def _close(self) -> Generator[ResponseStreamEvent]:
@@ -948,19 +973,16 @@ async def _item_to_message(item: Item, *, approval_storage: ApprovalStorage | No
if item.type == "mcp_call":
mcp = cast(ItemMcpToolCall, item)
contents = [
Content.from_mcp_server_tool_call(
mcp.id,
mcp.name,
server_name=mcp.server_label,
arguments=mcp.arguments,
)
]
if getattr(mcp, "output", None) is not None:
contents.append(Content.from_mcp_server_tool_result(call_id=mcp.id, output=mcp.output))
return Message(
role="assistant",
contents=contents,
contents=[
Content.from_mcp_server_tool_call(
mcp.id,
mcp.name,
server_name=mcp.server_label,
arguments=mcp.arguments,
)
],
)
if item.type == "mcp_approval_request":
@@ -1221,19 +1243,16 @@ async def _output_item_to_message(item: OutputItem, *, approval_storage: Approva
if item.type == "mcp_call":
mcp = cast(OutputItemMcpToolCall, item)
contents = [
Content.from_mcp_server_tool_call(
mcp.id,
mcp.name,
server_name=mcp.server_label,
arguments=mcp.arguments,
)
]
if getattr(mcp, "output", None) is not None:
contents.append(Content.from_mcp_server_tool_result(call_id=mcp.id, output=mcp.output))
return Message(
role="assistant",
contents=contents,
contents=[
Content.from_mcp_server_tool_call(
mcp.id,
mcp.name,
server_name=mcp.server_label,
arguments=mcp.arguments,
)
],
)
if item.type == "mcp_approval_request":
@@ -1610,7 +1629,6 @@ async def _to_outputs(
mcp_call = stream.add_output_item_mcp_call(
server_label=content.server_name or "default",
name=content.tool_name or "",
item_id=content.call_id,
)
yield mcp_call.emit_added()
async for event in mcp_call.aarguments(_arguments_to_str(content.arguments)):
@@ -1685,91 +1703,4 @@ async def _to_outputs(
logger.warning(f"Content type '{content.type}' is not supported yet. This is usually safe to ignore.")
def _stringify_mcp_output(output: Any) -> str:
"""Convert hosted MCP output payloads into the string shape expected by mcp_call.output."""
if output is None:
return ""
if isinstance(output, str):
return output
if isinstance(output, Mapping):
text = cast(Any, output).get("text")
if isinstance(text, str):
return text
return json.dumps(output, default=str)
if isinstance(output, Sequence) and not isinstance(output, (str, bytes, bytearray)):
parts: list[str] = []
entries = cast(Sequence[object], output)
for entry in entries:
if isinstance(entry, Content) and entry.type == "text":
parts.append(entry.text or "")
continue
parts.append(_stringify_mcp_output(entry))
return "".join(parts)
return str(output)
def _emit_completed_mcp_call(
stream: ResponseEventStream,
call_content: Content,
*,
arguments: str,
output: str,
) -> Generator[ResponseStreamEvent]:
"""Emit a single completed MCP call item carrying both arguments and output."""
mcp_call = stream.add_output_item_mcp_call(
server_label=call_content.server_name or "default",
name=call_content.tool_name or "",
item_id=call_content.call_id,
)
yield mcp_call.emit_added()
yield mcp_call.emit_arguments_done(arguments)
yield mcp_call.emit_completed()
yield mcp_call.emit_done(output=output)
async def _to_outputs_for_messages(
stream: ResponseEventStream,
messages: Sequence[Message],
*,
approval_storage: ApprovalStorage | None = None,
) -> AsyncIterator[ResponseStreamEvent]:
"""Convert messages to output events with hosted-MCP call/result coalescing.
Parse once in message/content order and emit either:
- a single canonical completed ``mcp_call`` when adjacent hosted MCP
call/result content are encountered, or
- standard output items for all other content types.
"""
pending_mcp_call: Content | None = None
for message in messages:
for content in message.contents:
if pending_mcp_call is not None:
if content.type == "mcp_server_tool_result" and content.call_id == pending_mcp_call.call_id:
for event in _emit_completed_mcp_call(
stream,
pending_mcp_call,
arguments=_arguments_to_str(pending_mcp_call.arguments),
output=_stringify_mcp_output(content.output),
):
yield event
pending_mcp_call = None
continue
async for event in _to_outputs(stream, pending_mcp_call, approval_storage=approval_storage):
yield event
pending_mcp_call = None
if content.type == "mcp_server_tool_call" and content.call_id:
pending_mcp_call = content
continue
async for event in _to_outputs(stream, content, approval_storage=approval_storage):
yield event
if pending_mcp_call is not None:
async for event in _to_outputs(stream, pending_mcp_call, approval_storage=approval_storage):
yield event
# endregion
@@ -25,7 +25,7 @@ classifiers = [
dependencies = [
"agent-framework-core>=1.7.0,<2",
"azure-ai-agentserver-core>=2.0.0b3,<3",
"azure-ai-agentserver-responses>=1.0.0b7,<2",
"azure-ai-agentserver-responses>=1.0.0b5,<2",
"azure-ai-agentserver-invocations>=1.0.0b3,<2",
]
@@ -11,33 +11,24 @@ the registered _handle_create handler.
from __future__ import annotations
import json
import uuid
from collections.abc import AsyncIterator, Awaitable, Callable, Sequence
from collections.abc import AsyncIterator, Callable
from dataclasses import dataclass
from typing import Literal, overload
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import pytest
from agent_framework import (
AgentExecutorRequest,
AgentResponse,
AgentResponseUpdate,
AgentSession,
Content,
FileCheckpointStorage,
HistoryProvider,
Message,
RawAgent,
ResponseStream,
SupportsAgentRun,
WorkflowAgent,
WorkflowBuilder,
WorkflowCheckpoint,
WorkflowCheckpointException,
WorkflowContext,
WorkflowMessage,
executor,
)
from azure.ai.agentserver.responses import InMemoryResponseProvider
from mcp import McpError
@@ -48,6 +39,7 @@ from agent_framework_foundry_hosting import ResponsesHostServer
from agent_framework_foundry_hosting._responses import (
_AZURE_RESPONSES_MESSAGE_ROLE_TYPE, # pyright: ignore[reportPrivateUsage]
CONSENT_ERROR_CODE,
ConsentError,
FileBasedFunctionApprovalStorage, # pyright: ignore[reportPrivateUsage]
InMemoryFunctionApprovalStorage, # pyright: ignore[reportPrivateUsage]
_item_to_message, # pyright: ignore[reportPrivateUsage]
@@ -110,7 +102,7 @@ def _make_agent(
return agent
def _make_server(agent: Any, **kwargs: Any) -> ResponsesHostServer:
def _make_server(agent: MagicMock, **kwargs: Any) -> ResponsesHostServer:
"""Create a ResponsesHostServer with an in-memory store."""
return ResponsesHostServer(agent, store=InMemoryResponseProvider(), **kwargs)
@@ -269,50 +261,6 @@ class TestNonStreaming:
assert "function_call_output" in types
assert "message" in types
async def test_hosted_mcp_call_and_result_persist_as_single_mcp_call(self) -> None:
agent = _make_agent(
response=AgentResponse(
messages=[
Message(
role="assistant",
contents=[
Content.from_mcp_server_tool_call(
call_id="mcp_abc123",
tool_name="search",
server_name="api_specs",
arguments='{"q": "cats"}',
)
],
),
Message(
role="tool",
contents=[
Content.from_mcp_server_tool_result(
call_id="mcp_abc123",
output=[Content.from_text(text="found 10 cats")],
)
],
),
Message(role="assistant", contents=[Content.from_text("I found 10 cats!")]),
]
)
)
server = _make_server(agent)
resp = await _post(server, stream=False)
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "completed"
types = [item["type"] for item in body["output"]]
assert "mcp_call" in types
assert "custom_tool_call_output" not in types
mcp_items = [item for item in body["output"] if item["type"] == "mcp_call"]
assert len(mcp_items) == 1
assert mcp_items[0]["id"] == "mcp_abc123"
assert mcp_items[0]["output"] == "found 10 cats"
async def test_reasoning_content(self) -> None:
agent = _make_agent(
response=AgentResponse(
@@ -670,53 +618,6 @@ class TestStreaming:
assert "response.output_item.added" in types
assert "response.output_item.done" in types
async def test_mcp_tool_call_and_result_streaming_emit_single_completed_mcp_call(self) -> None:
agent = _make_agent(
stream_updates=[
AgentResponseUpdate(
contents=[
Content.from_mcp_server_tool_call(
call_id="mcp_abc123",
tool_name="search",
server_name="api_specs",
arguments='{"q":',
)
],
role="assistant",
),
AgentResponseUpdate(
contents=[
Content.from_mcp_server_tool_call(
call_id="mcp_abc123",
tool_name="search",
server_name="api_specs",
arguments=' "cats"}',
)
],
role="assistant",
),
AgentResponseUpdate(
contents=[
Content.from_mcp_server_tool_result(
call_id="mcp_abc123",
output=[Content.from_text(text="found 10 cats")],
)
],
role="tool",
),
]
)
server = _make_server(agent)
resp = await _post(server, stream=True)
assert resp.status_code == 200
events = _parse_sse_events(resp.text)
done_events = [e for e in events if e["event"] == "response.output_item.done"]
assert len(done_events) == 1
assert done_events[0]["data"]["item"]["type"] == "mcp_call"
assert done_events[0]["data"]["item"]["id"] == "mcp_abc123"
assert done_events[0]["data"]["item"]["output"] == "found 10 cats"
# endregion
@@ -820,24 +721,6 @@ class TestOutputItemToMessage:
assert msg.contents[0].server_name == "my_server"
assert msg.contents[0].tool_name == "search"
async def test_mcp_call_with_output_reconstructs_mcp_result_content(self) -> None:
from azure.ai.agentserver.responses.models import OutputItemMcpToolCall
item = OutputItemMcpToolCall({
"type": "mcp_call",
"id": "mcp-1",
"server_label": "my_server",
"name": "search",
"arguments": '{"q": "test"}',
"output": "found 10 cats",
})
msg = await _output_item_to_message(item)
assert msg.role == "assistant"
assert len(msg.contents) == 2
assert msg.contents[0].type == "mcp_server_tool_call"
assert msg.contents[1].type == "mcp_server_tool_result"
assert msg.contents[1].output == "found 10 cats"
async def test_mcp_approval_request(self) -> None:
from azure.ai.agentserver.responses.models import OutputItemMcpApprovalRequest
@@ -1307,25 +1190,6 @@ class TestItemToMessage:
assert msg.contents[0].server_name == "my_server"
assert msg.contents[0].tool_name == "search"
async def test_mcp_call_with_output_reconstructs_mcp_result_content(self) -> None:
from azure.ai.agentserver.responses.models import ItemMcpToolCall
item = ItemMcpToolCall({
"type": "mcp_call",
"id": "mcp-1",
"server_label": "my_server",
"name": "search",
"arguments": '{"q": "test"}',
"output": "found 10 cats",
})
msg = await _item_to_message(item)
assert msg is not None
assert msg.role == "assistant"
assert len(msg.contents) == 2
assert msg.contents[0].type == "mcp_server_tool_call"
assert msg.contents[1].type == "mcp_server_tool_result"
assert msg.contents[1].output == "found 10 cats"
async def test_mcp_approval_request(self) -> None:
from azure.ai.agentserver.responses.models import ItemMcpApprovalRequest
@@ -2074,71 +1938,6 @@ class TestMultiTurnMixedContent:
assert len(fc_contents) >= 1
assert fc_contents[0].name == "search"
async def test_hosted_mcp_call_round_trip_does_not_orphan_function_call_output(self) -> None:
"""Turn 1 produces hosted MCP call + result, turn 2 must replay both without orphaning output."""
agent = _make_multi_response_agent([
AgentResponse(
messages=[
Message(
role="assistant",
contents=[
Content.from_mcp_server_tool_call(
call_id="mcp_abc123",
tool_name="search",
server_name="api_specs",
arguments='{"q": "cats"}',
)
],
),
Message(
role="tool",
contents=[
Content.from_mcp_server_tool_result(
call_id="mcp_abc123",
output=[Content.from_text(text="found 10 cats")],
)
],
),
Message(role="assistant", contents=[Content.from_text("I found 10 cats!")]),
]
),
AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("Here are more details")])]),
])
server = _make_server(agent)
resp1 = await _post(server, input_text="Search for cats", stream=False)
assert resp1.status_code == 200
response_id = resp1.json()["id"]
types1 = [item["type"] for item in resp1.json()["output"]]
assert "mcp_call" in types1
assert "custom_tool_call_output" not in types1
resp2 = await _post_json(
server,
{
"model": "test-model",
"input": "Tell me more",
"stream": False,
"previous_response_id": response_id,
},
)
assert resp2.status_code == 200
assert resp2.json()["status"] == "completed"
second_call_messages = agent.run.call_args_list[1].kwargs["messages"]
mcp_call_contents = [c for m in second_call_messages for c in m.contents if c.type == "mcp_server_tool_call"]
mcp_result_contents = [
c for m in second_call_messages for c in m.contents if c.type == "mcp_server_tool_result"
]
function_result_contents = [c for m in second_call_messages for c in m.contents if c.type == "function_result"]
assert len(mcp_call_contents) >= 1
assert len(mcp_result_contents) >= 1
assert all((c.call_id or "") != "mcp_abc123" for c in function_result_contents)
assert any((c.call_id or "") == "mcp_abc123" for c in mcp_call_contents)
assert any((c.call_id or "") == "mcp_abc123" for c in mcp_result_contents)
async def test_multi_turn_reasoning_in_history(self) -> None:
"""Turn 1 produces reasoning + text, turn 2 sees them in history."""
agent = _make_multi_response_agent([
@@ -3269,7 +3068,10 @@ class TestCheckpointContextPathValidation:
# region Agent lifecycle (lazy entry & OAuth consent surfacing)
def _make_consent_error(url: str = "https://consent.example.com/auth") -> Exception:
def _make_consent_error(
url: str = "https://consent.example.com/auth",
name: str = "Foundry Toolbox",
) -> Exception:
"""Build an exception wrapping a Foundry MCP gateway consent error.
Mirrors the real-world wrapping produced by ``MCPStreamableHTTPTool.__aenter__``,
@@ -3277,17 +3079,34 @@ def _make_consent_error(url: str = "https://consent.example.com/auth") -> Except
``ToolExecutionException`` (an ``AgentFrameworkException`` subclass) with the
original error attached via ``inner_exception``. ``consent_url_from_error``
then finds the wrapped ``McpError`` in ``exc.args``.
The McpError message uses the structured Foundry MCP gateway format:
a human-readable prefix followed by a JSON document describing each
failed tool source and its consent URL.
"""
from agent_framework.exceptions import ToolExecutionException
inner = McpError(ErrorData(code=CONSENT_ERROR_CODE, message=url))
payload = json.dumps({
"errors": [
{
"name": name,
"type": "mcp",
"error": {
"code": "CONSENT_REQUIRED",
"message": url,
},
}
]
})
message = f"tools/list failed for 1 tool source(s), succeeded for 0 tool source(s) {payload}"
inner = McpError(ErrorData(code=CONSENT_ERROR_CODE, message=message))
return ToolExecutionException("MCP consent required", inner_exception=inner)
class TestConsentUrlFromError:
def test_returns_consent_url_when_inner_arg_is_consent_mcp_error(self) -> None:
exc = _make_consent_error("https://example.com/consent")
assert consent_url_from_error(exc) == "https://example.com/consent"
exc = _make_consent_error("https://example.com/consent", name="my-tool")
assert consent_url_from_error(exc) == [ConsentError(name="my-tool", consent_url="https://example.com/consent")]
def test_returns_none_when_no_mcp_error_in_args(self) -> None:
assert consent_url_from_error(Exception("boom")) is None
@@ -3304,6 +3123,13 @@ class TestConsentUrlFromError:
bare = McpError(ErrorData(code=CONSENT_ERROR_CODE, message="https://x"))
assert consent_url_from_error(bare) is None
def test_returns_none_when_message_has_no_json(self) -> None:
from agent_framework.exceptions import ToolExecutionException
inner = McpError(ErrorData(code=CONSENT_ERROR_CODE, message="no json here"))
exc = ToolExecutionException("MCP consent required", inner_exception=inner)
assert consent_url_from_error(exc) is None
class TestAgentLifecycle:
async def test_agent_entered_lazily_on_first_request(self) -> None:
@@ -3450,498 +3276,3 @@ class TestOAuthConsentSurfacing:
# endregion
# region Workflow agent hosting (end-to-end)
class _ToolApprovalWorkflowAgentMock(SupportsAgentRun):
"""Inner agent for a hosted ``WorkflowAgent`` whose first run emits a
``FunctionApprovalRequestContent`` and whose follow-up run (after
receiving a ``FunctionApprovalResponseContent`` in its inputs) returns a
final assistant text response.
Mirrors a real agent whose tool invocation requires user approval. Used
here to exercise the full HTTP pipeline through ``ResponsesHostServer``
when the hosted agent is a ``WorkflowAgent`` containing a tool-approval
flow.
"""
def __init__(
self,
name: str,
*,
tool_name: str = "delete_file",
tool_arguments: dict[str, Any] | None = None,
approval_request_ids: Sequence[str] | None = None,
final_text: str = "done",
) -> None:
self.id = str(uuid.uuid4())
self.name = name
self.description: str | None = None
self._tool_name = tool_name
self._tool_arguments = tool_arguments or {"path": "/tmp/example"}
self._approval_request_ids: list[str] = list(approval_request_ids) if approval_request_ids else []
self._final_text = final_text
self.run_count = 0
self.last_run_messages: list[Message] = []
def create_session(self, **kwargs: Any) -> AgentSession:
return AgentSession()
def get_session(self, *, service_session_id: str, **kwargs: Any) -> AgentSession:
return AgentSession()
def _next_request_id(self) -> str:
# Stable across calls: when the workflow checkpoint round-trips through
# restore, ``AgentExecutor`` re-invokes the inner agent during replay.
# We must surface the *same* approval request id on each invocation so
# the workflow's pending-request id matches the id the test echoes
# back as ``mcp_approval_response``.
if self._approval_request_ids:
return self._approval_request_ids[0]
return str(uuid.uuid4())
def _build_approval_request(self) -> Content:
request_id = self._next_request_id()
function_call = Content.from_function_call(
call_id=request_id,
name=self._tool_name,
arguments=self._tool_arguments,
additional_properties={"server_label": "test_server"},
)
return Content.from_function_approval_request(id=request_id, function_call=function_call)
@overload
def run(
self,
messages: str | Content | Message | Sequence[str | Content | Message] | None = ...,
*,
stream: Literal[False] = ...,
session: AgentSession | None = ...,
**kwargs: Any,
) -> Awaitable[AgentResponse[Any]]: ...
@overload
def run(
self,
messages: str | Content | Message | Sequence[str | Content | Message] | None = ...,
*,
stream: Literal[True],
session: AgentSession | None = ...,
**kwargs: Any,
) -> ResponseStream[AgentResponseUpdate, AgentResponse[Any]]: ...
def run(
self,
messages: str | Content | Message | Sequence[str | Content | Message] | None = None,
*,
stream: bool = False,
session: AgentSession | None = None,
**kwargs: Any,
) -> Awaitable[AgentResponse] | ResponseStream[AgentResponseUpdate, AgentResponse]:
if stream:
return self._run_stream(messages=messages, **kwargs)
return self._run(messages=messages, **kwargs)
@staticmethod
def _normalize(
messages: str | Content | Message | Sequence[str | Content | Message] | None,
) -> list[Message]:
if messages is None:
return []
if isinstance(messages, str):
return [Message(role="user", contents=[Content.from_text(text=messages)])]
if isinstance(messages, Message):
return [messages]
if isinstance(messages, Content):
return [Message(role="user", contents=[messages])]
result: list[Message] = []
for item in messages:
if isinstance(item, Message):
result.append(item)
elif isinstance(item, Content):
result.append(Message(role="user", contents=[item]))
else:
result.append(Message(role="user", contents=[Content.from_text(text=item)]))
return result
@staticmethod
def _approval_responses_in(messages: list[Message]) -> list[Content]:
return [c for m in messages for c in m.contents if c.type == "function_approval_response"]
async def _run(
self,
messages: str | Content | Message | Sequence[str | Content | Message] | None = None,
**kwargs: Any,
) -> AgentResponse:
normalized = self._normalize(messages)
self.last_run_messages = normalized
self.run_count += 1
if self._approval_responses_in(normalized):
return AgentResponse(messages=[Message("assistant", [Content.from_text(text=self._final_text)])])
approval = self._build_approval_request()
return AgentResponse(messages=[Message("assistant", [approval])])
def _run_stream(
self,
messages: str | Content | Message | Sequence[str | Content | Message] | None = None,
**kwargs: Any,
) -> ResponseStream[AgentResponseUpdate, AgentResponse]:
normalized = self._normalize(messages)
self.last_run_messages = normalized
self.run_count += 1
approvals = self._approval_responses_in(normalized)
async def _iter() -> AsyncIterator[AgentResponseUpdate]:
if approvals:
yield AgentResponseUpdate(
contents=[Content.from_text(text=self._final_text)],
role="assistant",
author_name=self.name,
)
return
yield AgentResponseUpdate(
contents=[self._build_approval_request()],
role="assistant",
author_name=self.name,
)
return ResponseStream(_iter(), finalizer=AgentResponse.from_updates)
def _build_text_workflow_agent(text: str) -> WorkflowAgent:
"""Build a minimal ``WorkflowAgent`` whose inner agent emits a fixed text."""
class _TextAgent(SupportsAgentRun):
def __init__(self, name: str, text: str) -> None:
self.id = str(uuid.uuid4())
self.name = name
self.description: str | None = None
self._text = text
def create_session(self, **kwargs: Any) -> AgentSession:
return AgentSession()
def get_session(self, *, service_session_id: str, **kwargs: Any) -> AgentSession:
return AgentSession()
@overload
def run(
self,
messages: Any = ...,
*,
stream: Literal[False] = ...,
session: AgentSession | None = ...,
**kwargs: Any,
) -> Awaitable[AgentResponse[Any]]: ...
@overload
def run(
self,
messages: Any = ...,
*,
stream: Literal[True],
session: AgentSession | None = ...,
**kwargs: Any,
) -> ResponseStream[AgentResponseUpdate, AgentResponse[Any]]: ...
def run(
self,
messages: Any = None,
*,
stream: bool = False,
session: AgentSession | None = None,
**kwargs: Any,
) -> Awaitable[AgentResponse] | ResponseStream[AgentResponseUpdate, AgentResponse]:
text = self._text
name = self.name
async def _aresult() -> AgentResponse:
return AgentResponse(messages=[Message("assistant", [Content.from_text(text=text)])])
async def _aiter() -> AsyncIterator[AgentResponseUpdate]:
yield AgentResponseUpdate(
contents=[Content.from_text(text=text)],
role="assistant",
author_name=name,
)
if stream:
return ResponseStream(_aiter(), finalizer=AgentResponse.from_updates)
return _aresult()
inner = _TextAgent("text-agent", text)
@executor
async def start(messages: list[Message], ctx: WorkflowContext[AgentExecutorRequest]) -> None:
await ctx.send_message(AgentExecutorRequest(messages=messages, should_respond=True))
workflow = WorkflowBuilder(start_executor=start).add_edge(start, inner).build()
return WorkflowAgent(workflow=workflow, name="Text Workflow Agent")
def _build_approval_workflow_agent(
*,
approval_request_id: str,
tool_name: str = "delete_file",
tool_arguments: dict[str, Any] | None = None,
final_text: str = "done",
) -> tuple[WorkflowAgent, _ToolApprovalWorkflowAgentMock]:
"""Build a ``WorkflowAgent`` whose inner agent emits a tool approval request."""
mock_agent = _ToolApprovalWorkflowAgentMock(
name="approval-agent",
tool_name=tool_name,
tool_arguments=tool_arguments or {"path": "/tmp/secret.txt"},
approval_request_ids=[approval_request_id],
final_text=final_text,
)
@executor
async def start(messages: list[Message], ctx: WorkflowContext[AgentExecutorRequest]) -> None:
await ctx.send_message(AgentExecutorRequest(messages=messages, should_respond=True))
workflow = WorkflowBuilder(start_executor=start).add_edge(start, mock_agent).build()
workflow_agent = WorkflowAgent(workflow=workflow, name="Approval Workflow Agent")
return workflow_agent, mock_agent
class TestWorkflowAgentHosting:
"""End-to-end HTTP tests for ``ResponsesHostServer`` hosting a ``WorkflowAgent``.
These tests drive ``_handle_inner_workflow`` through the ASGI stack:
they exercise checkpoint write/restore (multi-turn) and the
tool-approval round-trip path, which is the primary differentiator
relative to the regular agent path.
"""
async def test_basic_text_response(self) -> None:
workflow_agent = _build_text_workflow_agent("hello from workflow")
server = _make_server(workflow_agent)
resp = await _post(server, input_text="hi", stream=False)
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "completed"
text_found = any(
part.get("type") == "output_text" and part.get("text") == "hello from workflow"
for item in body["output"]
if item["type"] == "message"
for part in item.get("content", [])
)
assert text_found, f"Expected workflow output text in {body['output']}"
async def test_basic_text_response_streaming(self) -> None:
workflow_agent = _build_text_workflow_agent("hello stream")
server = _make_server(workflow_agent)
resp = await _post(server, input_text="hi", stream=True)
assert resp.status_code == 200
events = _parse_sse_events(resp.text)
types = _sse_event_types(events)
assert types[0] == "response.created"
assert types[-1] == "response.completed"
assert "response.output_text.delta" in types
text_done = [e for e in events if e["event"] == "response.output_text.done"]
assert any(e["data"]["text"] == "hello stream" for e in text_done)
async def test_non_streaming_emits_mcp_approval_request_and_persists_to_storage(self) -> None:
workflow_agent, mock_agent = _build_approval_workflow_agent(approval_request_id="apr_wf_ns")
server = _make_server(workflow_agent)
resp = await _post(server, stream=False)
assert resp.status_code == 200
body = resp.json()
assert body["status"] == "completed"
approval_items = [it for it in body["output"] if it["type"] == "mcp_approval_request"]
assert len(approval_items) == 1
assert approval_items[0]["name"] == "delete_file"
assert approval_items[0]["server_label"] == "test_server"
approval_request_id = approval_items[0]["id"]
# The id surfaced over the wire is generated by the response stream
# builder; the original approval ``Content`` (carrying the inner
# ``function_call``) must be persisted under that id so the next
# turn can reconstruct it.
loaded = await server._approval_storage.load_approval_request( # pyright: ignore[reportPrivateUsage]
approval_request_id
)
assert loaded.type == "function_approval_request"
assert loaded.function_call.name == "delete_file" # type: ignore[attr-defined]
assert mock_agent.run_count == 1
async def test_streaming_emits_mcp_approval_request_and_persists_to_storage(self) -> None:
workflow_agent, mock_agent = _build_approval_workflow_agent(approval_request_id="apr_wf_st")
server = _make_server(workflow_agent)
resp = await _post(server, stream=True)
assert resp.status_code == 200
events = _parse_sse_events(resp.text)
types = _sse_event_types(events)
assert types[0] == "response.created"
assert types[-1] == "response.completed"
approval_request_id: str | None = None
for e in events:
if e["event"] != "response.output_item.added":
continue
item = e["data"].get("item") or {}
if item.get("type") == "mcp_approval_request":
approval_request_id = item.get("id")
break
assert approval_request_id is not None
loaded = await server._approval_storage.load_approval_request( # pyright: ignore[reportPrivateUsage]
approval_request_id
)
assert loaded.type == "function_approval_request"
assert mock_agent.run_count == 1
async def test_round_trip_approval_response_resumes_workflow_agent(self) -> None:
"""Two-turn HTTP round-trip:
Turn 1 emits ``mcp_approval_request`` and writes a workflow
checkpoint under the response id. Turn 2 sends the
``mcp_approval_response`` with ``previous_response_id`` set, so the
host restores the checkpoint, the WorkflowAgent routes the
approval response back to the paused inner agent, and the inner
agent emits the final assistant text.
"""
workflow_agent, mock_agent = _build_approval_workflow_agent(
approval_request_id="apr_wf_rt",
final_text="done with approval",
)
server = _make_server(workflow_agent)
first = await _post(server, stream=False)
assert first.status_code == 200
first_body = first.json()
first_response_id = first_body["id"]
approval_items = [it for it in first_body["output"] if it["type"] == "mcp_approval_request"]
assert len(approval_items) == 1
approval_request_id = approval_items[0]["id"]
assert mock_agent.run_count == 1
second_payload: dict[str, Any] = {
"model": "test-model",
"input": [
{
"type": "mcp_approval_response",
"approval_request_id": approval_request_id,
"approve": True,
}
],
"stream": False,
"previous_response_id": first_response_id,
}
second = await _post_json(server, second_payload)
assert second.status_code == 200
second_body = second.json()
assert second_body["status"] == "completed"
# The inner agent must have been resumed (restore replay + new turn).
# Restore call is a no-op for the mock (no input); the new-turn call
# delivers the approval response, so run_count grows by at least 1.
assert mock_agent.run_count >= 2
# The final assistant text from the resumed inner agent surfaces in
# the HTTP output.
text_pieces = [
part.get("text", "")
for item in second_body["output"]
if item["type"] == "message"
for part in item.get("content", [])
if part.get("type") == "output_text"
]
assert any("done with approval" in t for t in text_pieces), (
f"expected resumed workflow output, got {second_body['output']}"
)
# The new-turn invocation of the inner agent must have received the
# approval response routed back through WorkflowAgent.
approval_responses = [
c for m in mock_agent.last_run_messages for c in m.contents if c.type == "function_approval_response"
]
assert len(approval_responses) == 1
assert approval_responses[0].approved is True # type: ignore[attr-defined]
async def test_round_trip_approval_response_streaming(self) -> None:
"""Streaming variant of the round-trip: turn 2 is requested with
``stream=true`` and surfaces the resumed text as SSE events."""
workflow_agent, mock_agent = _build_approval_workflow_agent(
approval_request_id="apr_wf_rt_st",
final_text="streamed-done",
)
server = _make_server(workflow_agent)
first = await _post(server, stream=False)
first_body = first.json()
first_response_id = first_body["id"]
approval_request_id = next(it["id"] for it in first_body["output"] if it["type"] == "mcp_approval_request")
second = await _post_json(
server,
{
"model": "test-model",
"input": [
{
"type": "mcp_approval_response",
"approval_request_id": approval_request_id,
"approve": True,
}
],
"stream": True,
"previous_response_id": first_response_id,
},
)
assert second.status_code == 200
events = _parse_sse_events(second.text)
types = _sse_event_types(events)
assert types[0] == "response.created"
assert types[-1] == "response.completed"
text_done = [e for e in events if e["event"] == "response.output_text.done"]
assert any("streamed-done" in e["data"]["text"] for e in text_done)
assert mock_agent.run_count >= 2
async def test_round_trip_approval_response_rejected(self) -> None:
"""Sending ``approve=False`` must surface as ``approved=False`` to the
inner agent on resume."""
workflow_agent, mock_agent = _build_approval_workflow_agent(
approval_request_id="apr_wf_reject",
final_text="acknowledged",
)
server = _make_server(workflow_agent)
first = await _post(server, stream=False)
first_body = first.json()
first_response_id = first_body["id"]
approval_request_id = next(it["id"] for it in first_body["output"] if it["type"] == "mcp_approval_request")
second = await _post_json(
server,
{
"model": "test-model",
"input": [
{
"type": "mcp_approval_response",
"approval_request_id": approval_request_id,
"approve": False,
}
],
"stream": False,
"previous_response_id": first_response_id,
},
)
assert second.status_code == 200
approval_responses = [
c for m in mock_agent.last_run_messages for c in m.contents if c.type == "function_approval_response"
]
assert len(approval_responses) == 1
assert approval_responses[0].approved is False # type: ignore[attr-defined]
# endregion
+4 -4
View File
@@ -57,19 +57,19 @@ math = [
[dependency-groups]
dev = [
"uv==0.11.17",
"ruff==0.15.15",
"uv==0.11.6",
"ruff==0.15.8",
"pytest==9.0.3",
"mypy==1.20.0",
"pyright==1.1.408",
#tasks
"poethepoet==0.46.0",
"poethepoet==0.42.1",
"rich>=13.7.1,<15.0.0",
"tomli==2.4.1",
"tomli-w==1.2.0",
# tau2 from source (not available on PyPI)
"tau2@ git+https://github.com/sierra-research/tau2-bench@5ba9e3e56db57c5e4114bf7f901291f09b2c5619",
"prek==0.4.3",
"prek==0.3.9",
]
[project.scripts]
+7 -7
View File
@@ -28,25 +28,25 @@ dependencies = [
[dependency-groups]
dev = [
"uv==0.11.17",
"uv==0.11.6",
"flit==3.12.0",
"ruff==0.15.15",
"ruff==0.15.8",
"pytest==9.0.3",
"pytest-asyncio==1.4.0",
"pytest-asyncio==1.3.0",
"pytest-cov==7.1.0",
"pytest-xdist[psutil]==3.8.0",
"pytest-timeout==2.4.0",
"pytest-retry==1.7.0",
"mypy==1.20.0",
"pyright==1.1.408",
"mcp[ws]==1.27.2",
"mcp[ws]==1.27.0",
"opentelemetry-sdk==1.40.0",
"azure-monitor-opentelemetry==1.8.8",
"azure-monitor-opentelemetry==1.8.7",
#tasks
"poethepoet==0.46.0",
"poethepoet==0.42.1",
"rich>=13.7.1,<16.0.0",
"tomli==2.4.1",
"prek==0.4.3",
"prek==0.3.9",
]
[tool.uv]
@@ -109,10 +109,7 @@ async def main() -> None:
print(f"\n [calling tool: {content.name}]", flush=True)
print(" ", end="", flush=True)
# Show web search activity when the result arrives with action details.
elif (
content.type in ("search_tool_call", "search_tool_result")
and getattr(content, "tool_name", None) == "web_search"
):
elif content.type in ("search_tool_call", "search_tool_result") and getattr(content, "tool_name", None) == "web_search":
action = None
if content.type == "search_tool_result" and isinstance(content.result, dict):
action = content.result.get("action", {})
@@ -134,11 +134,15 @@ def handle_response_and_requests(response: AgentResponse) -> dict[str, HandoffAg
if message.text:
print(f"- {message.author_name or message.role}: {message.text}")
for content in message.contents:
if content.type == "function_call" and content.name == WorkflowAgent.REQUEST_INFO_FUNCTION_NAME:
request_function_args = WorkflowAgent.RequestInfoFunctionArgs.from_dict(content.arguments) # type: ignore
request_id = request_function_args.request_id
request_event = request_function_args.request_event
pending_requests[request_id] = request_event.data
if content.type == "function_call":
if isinstance(content.arguments, dict):
request = WorkflowAgent.RequestInfoFunctionArgs.from_dict(content.arguments)
elif isinstance(content.arguments, str):
request = WorkflowAgent.RequestInfoFunctionArgs.from_json(content.arguments)
else:
raise ValueError("Invalid arguments type. Expecting a request info structure for this sample.")
if isinstance(request.data, HandoffAgentUserRequest):
pending_requests[request.request_id] = request.data
return pending_requests
@@ -3,8 +3,10 @@
import asyncio
import os
import sys
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential
@@ -139,14 +141,28 @@ async def main() -> None:
# Handle the human review if required.
if human_review_function_call:
# Parse the human review request arguments.
human_request_args = WorkflowAgent.RequestInfoFunctionArgs.from_dict(human_review_function_call.arguments) # type: ignore
request_payload = human_request_args.request_event.data
human_request_args = human_review_function_call.arguments
if isinstance(human_request_args, str):
request: WorkflowAgent.RequestInfoFunctionArgs = WorkflowAgent.RequestInfoFunctionArgs.from_json(
human_request_args
)
elif isinstance(human_request_args, Mapping):
request = WorkflowAgent.RequestInfoFunctionArgs.from_dict(dict(human_request_args))
else:
raise TypeError("Unexpected argument type for human review function call.")
request_payload: Any = request.data
if not isinstance(request_payload, HumanReviewRequest):
raise ValueError("Human review request payload must be a HumanReviewRequest.")
if not request_payload.agent_request:
raise ValueError("Human review request must contain an agent_request.")
agent_request = request_payload.agent_request
if agent_request is None:
raise ValueError("Human review request must include agent_request.")
request_id = agent_request.request_id
# Mock a human response approval for demonstration purposes.
human_response = ReviewResponse(request_id=request_payload.agent_request.request_id, feedback="", approved=True)
human_response = ReviewResponse(request_id=request_id, feedback="", approved=True)
# Create the function call result object to send back to the agent.
human_review_function_result = Content(
"function_result",
@@ -64,6 +64,7 @@ actions:
### Agent Invocation
- `InvokeAzureAgent` - Call an Azure AI agent
- `InvokePromptAgent` - Call a local prompt agent
### Tool Invocation
- `InvokeFunctionTool` - Call a registered Python function
@@ -20,7 +20,7 @@ You can connect to MCP servers in Foundry Toolbox that use different authenticat
- **Agent identity authentication**: The tool requires an agent identity token to authenticate. Sample MCP server: `https://{foundry-resource-name}.cognitiveservices.azure.com/language/mcp?api-version=2025-11-15-preview` (Azure Language MCP server) with agent identity for authentication.
- **Entra Pass-through authentication**: The tool requires an Entra pass-through token to authenticate. Sample MCP server: Microsoft Outlook MCP server with Entra pass-through for authentication.
> Definitions of these authentication methods can be found in the [agent.manifest.yaml](agent.manifest.yaml) file in this sample.
> Definitions of these authentication methods can be found in the [agent.manifest.yaml](agent.manifest.yaml) file in this sample. The GitHub MCP connection defaults to using a PAT for authentication in this sample, but you can switch to OAuth2 by changing the `project_connection_id` field in the `agent.manifest.yaml` file and following the instructions in the comments.
There are also Non-MCP tools in the toolbox that support different authentication methods. Learn more at the [Foundry sample repository](https://github.com/microsoft-foundry/foundry-samples/blob/main/samples/python/hosted-agents/SUPPORTED_TOOLBOX_SCENARIOS.md).
@@ -18,92 +18,92 @@ template:
- name: AZURE_AI_MODEL_DEPLOYMENT_NAME
value: "{{AZURE_AI_MODEL_DEPLOYMENT_NAME}}"
- name: TOOLBOX_NAME
value: "agent-tools-2"
# parameters:
# properties:
# - name: mcp_endpoint
# # `azd ai agent init -m` will prompt for this value when initializing the agent manifest
# secret: false
# description: URL of the public MCP server (e.g. https://gitmcp.io/Azure/azure-rest-api-specs) that does not require authentication
# - name: github_pat
# # `azd ai agent init -m` will prompt for this value when initializing the agent manifest.
# # Only needed when the GitHub MCP connection is configured to use the `github-mcp-pat-conn`
# # PAT-based connection below; if you use the `github-mcp-oauth-conn` OAuth2 connection
# # instead, you can leave this empty.
# secret: true
# description: GitHub Personal Access Token used to authenticate with the GitHub MCP server (only needed when using the PAT connection; press Enter if using OAuth2 instead)
# - name: language_mcp_entra_audience
# secret: false
# description: Entra ID audience for the Azure Language MCP server (e.g. https://cognitiveservices.azure.com/)
# - name: language_mcp_target_url
# secret: false
# description: URL of the Azure Language MCP server that accepts agent identity tokens (e.g. https://{foundry-resource-name}.cognitiveservices.azure.com/language/mcp?api-version=2025-11-15-preview)
# - name: outlook_mail_entra_audience
# secret: false
# description: Entra ID audience for the Outlook Mail MCP server
# - name: outlook_mail_entra_mcp_target
# secret: false
# description: URL of the Outlook Mail MCP server that accepts user Entra tokens
value: "agent-tools"
parameters:
properties:
- name: mcp_endpoint
# `azd ai agent init -m` will prompt for this value when initializing the agent manifest
secret: false
description: URL of the public MCP server (e.g. https://gitmcp.io/Azure/azure-rest-api-specs) that does not require authentication
- name: github_pat
# `azd ai agent init -m` will prompt for this value when initializing the agent manifest.
# Only needed when the GitHub MCP connection is configured to use the `github-mcp-pat-conn`
# PAT-based connection below; if you use the `github-mcp-oauth-conn` OAuth2 connection
# instead, you can leave this empty.
secret: true
description: GitHub Personal Access Token used to authenticate with the GitHub MCP server (only needed when using the PAT connection; press Enter if using OAuth2 instead)
# - name: language_mcp_entra_audience
# secret: false
# description: Entra ID audience for the Azure Language MCP server (e.g. https://cognitiveservices.azure.com/)
# - name: language_mcp_target_url
# secret: false
# description: URL of the Azure Language MCP server that accepts agent identity tokens (e.g. https://{foundry-resource-name}.cognitiveservices.azure.com/language/mcp?api-version=2025-11-15-preview)
# - name: outlook_mail_entra_audience
# secret: false
# description: Entra ID audience for the Outlook Mail MCP server
# - name: outlook_mail_entra_mcp_target
# secret: false
# description: URL of the Outlook Mail MCP server that accepts user Entra tokens
resources:
- kind: model
id: gpt-4.1-mini
name: AZURE_AI_MODEL_DEPLOYMENT_NAME
# - kind: connection
# # A connection that uses a GitHub Personal Access Token (PAT) to authenticate with the GitHub MCP server
# name: github-mcp-pat-conn
# category: RemoteTool
# authType: CustomKeys
# target: https://api.githubcopilot.com/mcp
# credentials:
# type: CustomKeys
# keys:
# Authorization: "Bearer {{ github_pat }}"
# - kind: connection
# # A connection that uses OAuth2 to authenticate with the GitHub MCP server
# name: github-mcp-oauth-conn
# category: RemoteTool
# authType: OAuth2
# target: https://api.githubcopilot.com/mcp
# connectorName: foundrygithubmcp
# credentials:
# type: OAuth2
# clientId: managed
# clientSecret: managed
- kind: connection
# A connection that uses a GitHub Personal Access Token (PAT) to authenticate with the GitHub MCP server
name: github-mcp-pat-conn
category: RemoteTool
authType: CustomKeys
target: https://api.githubcopilot.com/mcp
credentials:
type: CustomKeys
keys:
Authorization: "Bearer {{ github_pat }}"
- kind: connection
# A connection that uses OAuth2 to authenticate with the GitHub MCP server
name: github-mcp-oauth-conn
category: RemoteTool
authType: OAuth2
target: https://api.githubcopilot.com/mcp
connectorName: foundrygithubmcp
credentials:
type: OAuth2
clientId: managed
clientSecret: managed
# - kind: connection
# name: language-mcp-conn
# category: RemoteTool
# authType: AgenticIdentity
# audience: "{{ language_mcp_entra_audience }}"
# target: "{{ language_mcp_target_url }}"
# # - kind: connection
# # name: outlook-mail-conn
# # category: RemoteTool
# # authType: UserEntraToken
# # audience: "{{ outlook_mail_entra_audience }}"
# # target: "{{ outlook_mail_entra_mcp_target }}"
# - kind: toolbox
# name: agent-tools
# tools:
# - type: web_search
# name: web_search
# - type: code_interpreter
# name: code_interpreter
# # - type: mcp
# # # This MCP tool doesn't require authentication
# # server_label: noauth_mcp
# # server_url: "{{ mcp_endpoint }}"
# # require_approval: "never"
# - type: mcp
# # This MCP tool uses the GitHub MCP server with a PAT for authentication or OAuth2
# server_label: github
# project_connection_id: github-mcp-pat-conn # use `github-mcp-oauth-conn` for OAuth2 authentication
# require_approval: "never"
# - type: mcp
# # This MCP tool uses the Azure Language MCP server with agent identity for authentication
# server_label: language-mcp
# project_connection_id: language-mcp-conn
# require_approval: "never"
# # - type: mcp
# # server_label: outlook-mail
# # project_connection_id: outlook-mail-conn
# # require_approval: "never"
# - kind: connection
# name: outlook-mail-conn
# category: RemoteTool
# authType: UserEntraToken
# audience: "{{ outlook_mail_entra_audience }}"
# target: "{{ outlook_mail_entra_mcp_target }}"
- kind: toolbox
name: agent-tools
tools:
- type: web_search
name: web_search
- type: code_interpreter
name: code_interpreter
- type: mcp
# This MCP tool doesn't require authentication
server_label: noauth_mcp
server_url: "{{ mcp_endpoint }}"
require_approval: "never"
- type: mcp
# This MCP tool uses the GitHub MCP server with a PAT for authentication or OAuth2
server_label: github
project_connection_id: github-mcp-pat-conn # use `github-mcp-oauth-conn` for OAuth2 authentication
require_approval: "never"
# - type: mcp
# # This MCP tool uses the Azure Language MCP server with agent identity for authentication
# server_label: language-mcp
# project_connection_id: language-mcp-conn
# require_approval: "never"
# - type: mcp
# server_label: outlook-mail
# project_connection_id: outlook-mail-conn
# require_approval: "never"
@@ -1,4 +1,3 @@
# agent-framework
# agent-framework-foundry-hosting
agent-framework
agent-framework-foundry-hosting
mcp>=1.24.0,<2
@@ -28,7 +28,6 @@ import zipfile
from pathlib import Path
from azure.ai.projects.aio import AIProjectClient
from azure.ai.projects.models import CreateSkillVersionFromFilesBody
from azure.core.exceptions import ResourceNotFoundError
from azure.identity.aio import DefaultAzureCredential
from dotenv import load_dotenv
@@ -69,13 +68,8 @@ async def main() -> None:
name = skill_md.parent.name
print(f"Provisioning skill '{name}' from {skill_md.relative_to(SKILLS_DIR.parent)}...")
await _delete_skill_if_exists(project, name)
imported = await project.beta.skills.create_from_files(
name,
content=CreateSkillVersionFromFilesBody(
files=[(f"{name}.zip", _zip_skill_md(skill_md), "application/zip")]
),
)
print(f" Imported skill '{imported.name}' (id={imported.skill_id}, version={imported.version}).")
imported = await project.beta.skills.create_from_package(_zip_skill_md(skill_md))
print(f" Imported skill '{imported.name}' (id={imported.skill_id}, has_blob={imported.has_blob}).")
print("Verifying skills via project.beta.skills.list()...")
listed = {skill.name: skill async for skill in project.beta.skills.list()}
@@ -85,8 +79,8 @@ async def main() -> None:
if skill is None:
raise RuntimeError(f"Skill '{name}' was imported but is not present in the project listing.")
print(
f" OK '{skill.name}': id={skill.id}, "
f"description={skill.description!r}, default_version={skill.default_version}"
f" OK '{skill.name}': id={skill.skill_id}, "
f"description={skill.description!r}, has_blob={skill.has_blob}"
)
print("Done.")
@@ -1,12 +1,3 @@
FOUNDRY_PROJECT_ENDPOINT="<your-project-endpoint>"
FOUNDRY_MODEL="<your-model-deployment>"
# Only needed for evaluate_with_rubric_sample.py — connects to the
# pre-existing Foundry agent that the rubric evaluator was created against.
FOUNDRY_AGENT_NAME="<your-agent-name>"
FOUNDRY_AGENT_VERSION="<your-agent-version>"
# Only needed for evaluate_with_rubric_sample.py — references a rubric
# evaluator you created in Foundry. Pin the version for reproducible runs.
FOUNDRY_RUBRIC_NAME="<your-rubric-name>"
FOUNDRY_RUBRIC_VERSION="<your-rubric-version>"
@@ -35,34 +35,6 @@ Evaluate what already happened — zero changes to agent code:
uv run samples/05-end-to-end/evaluation/foundry_evals/evaluate_traces_sample.py
```
### Referencing a rubric evaluator created in Foundry
Foundry users can create rubric evaluators in the Foundry portal (or
through the dedicated SDK / REST surface). Once an evaluator exists,
agent-framework consumes it like any other evaluator: pass a
`GeneratedEvaluatorRef(name=..., version=...)` in the `evaluators=`
list and pin the version for reproducible runs.
```python
from agent_framework.foundry import FoundryEvals, GeneratedEvaluatorRef
evals = FoundryEvals(
evaluators=[
GeneratedEvaluatorRef(name="reservation-policy-rubric", version="3"),
"relevance",
"coherence",
],
)
```
Quality gates on rubric output use the standard `EvalResults` helpers,
including `assert_dimension_score_at_least(...)` for per-dimension
thresholds.
See [`evaluate_with_rubric_sample.py`](./evaluate_with_rubric_sample.py)
for a runnable end-to-end example that combines a rubric evaluator with
built-in evaluators and gates a per-dimension threshold.
## Setup
Create a `.env` file with configuration as in the `.env.example` file in this folder.
@@ -72,4 +44,3 @@ Create a `.env` file with configuration as in the `.env.example` file in this fo
- **"I want to test my agent during development"** → `evaluate_agent_sample.py`, Pattern 1
- **"I want to evaluate past agent runs"** → `evaluate_traces_sample.py`
- **"I want to inspect/modify eval data before submitting"** → `evaluate_agent_sample.py`, Pattern 2
- **"I want to score against a custom rubric I created in Foundry"** → `evaluate_with_rubric_sample.py`
@@ -1,138 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
"""Evaluate a Foundry agent against a rubric evaluator that was created in Foundry.
Rubric evaluators are LLM-as-judge evaluators with custom scoring dimensions
that you define for your domain. agent-framework consumes pre-existing rubric
evaluators — they are authored in the Foundry portal (or via the dedicated
SDK / REST surface) and referenced here by name and version.
See: https://learn.microsoft.com/azure/ai-foundry/concepts/evaluation-evaluators/rubric-evaluators
This sample demonstrates:
1. Connecting to a pre-existing Foundry agent (PromptAgent or HostedAgent).
2. Referencing a pre-existing rubric evaluator by ``name`` and ``version``.
3. Mixing the rubric with built-in Foundry evaluators in one run.
4. Asserting per-dimension thresholds with
``EvalResults.assert_dimension_score_at_least(...)`` for CI quality gates.
Starting condition / prerequisites:
- An Azure AI Foundry project with a deployed model.
- A registered Foundry agent (PromptAgent or HostedAgent) in that project.
This is the agent the rubric is meant to evaluate.
- A rubric evaluator already created in the Foundry portal against that
agent. Creating rubrics through the portal currently requires picking a
Foundry agent as the generation context, so this prerequisite is implied
by having a rubric at all.
- Set the following in .env (see ``.env.example``):
- ``FOUNDRY_PROJECT_ENDPOINT``
- ``FOUNDRY_AGENT_NAME`` and ``FOUNDRY_AGENT_VERSION`` for the agent
- ``FOUNDRY_RUBRIC_NAME`` and ``FOUNDRY_RUBRIC_VERSION`` for the rubric
- ``FOUNDRY_MODEL`` for the rubric judge model
"""
import asyncio
import os
from agent_framework import EvalNotPassedError, evaluate_agent
from agent_framework.foundry import FoundryAgent, FoundryChatClient, FoundryEvals, GeneratedEvaluatorRef
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
load_dotenv(override=True)
async def main() -> None:
# 1. Connect to the existing Foundry agent that the rubric was created
# against. PromptAgents and HostedAgents are both supported.
credential = AzureCliCredential()
project_endpoint = os.environ["FOUNDRY_PROJECT_ENDPOINT"]
agent = FoundryAgent(
project_endpoint=project_endpoint,
agent_name=os.environ["FOUNDRY_AGENT_NAME"],
agent_version=os.environ.get("FOUNDRY_AGENT_VERSION"),
credential=credential,
)
# 2. Reference the pre-existing rubric evaluator by name + version.
# Always pin a version for reproducible CI runs; versionless refs
# resolve to "latest" and emit a warning at evaluation time.
rubric_name = os.environ["FOUNDRY_RUBRIC_NAME"]
rubric_version = os.environ["FOUNDRY_RUBRIC_VERSION"]
rubric = GeneratedEvaluatorRef(name=rubric_name, version=rubric_version)
# 3. Mix the rubric with built-in evaluators in a single FoundryEvals
# config. FoundryEvals talks to Foundry over the project endpoint, so
# we hand it a FoundryChatClient configured with the same credential.
eval_client = FoundryChatClient(
project_endpoint=project_endpoint,
model=os.environ["FOUNDRY_MODEL"],
credential=credential,
)
evals = FoundryEvals(
client=eval_client,
evaluators=[
rubric,
FoundryEvals.RELEVANCE,
FoundryEvals.COHERENCE,
],
)
# =========================================================================
# Run evaluation
# =========================================================================
print("=" * 60)
print(f"Evaluating '{agent.name}' with rubric '{rubric_name}' (version {rubric_version})")
print("=" * 60)
results = await evaluate_agent(
agent=agent,
queries=[
"What's the weather like in Seattle?",
"Should I bring an umbrella to London tomorrow?",
],
evaluators=evals,
)
for r in results:
print(f"Status: {r.status}")
print(f"Results: {r.passed}/{r.total} passed")
print(f"Portal: {r.report_url}")
if r.all_passed:
print("[PASS] All passed")
else:
print(f"[FAIL] {r.failed} failed")
# =========================================================================
# Per-dimension quality gate
# =========================================================================
# Rubric evaluators emit per-dimension scores (1–5) on top of the overall
# weighted score. Use assert_dimension_score_at_least to gate CI on a
# specific dimension — e.g., never ship if a critical dimension drops
# below 3.
#
# The dimension_id must match an id defined on your rubric in Foundry.
# ``general_quality`` is used here because it's the conventional
# ``always_applicable: true`` dimension in the Foundry docs' example
# rubric — swap it for whatever dimension id(s) your rubric actually
# defines.
print()
print("=" * 60)
print("Per-dimension quality gate")
print("=" * 60)
for r in results:
try:
r.assert_dimension_score_at_least(
"general_quality",
min_score=3.0,
evaluator=rubric_name,
)
print(f"[PASS] {r.provider}: general_quality >= 3 on every item")
except EvalNotPassedError as exc:
print(f"[FAIL] {r.provider}: dimension gate tripped: {exc}")
if __name__ == "__main__":
asyncio.run(main())
@@ -74,22 +74,19 @@ async def run_agent_framework() -> None:
client = OpenAIChatClient(model="gpt-4.1-mini")
# Create specialized agents
python_expert = Agent(
client=client,
python_expert = Agent(client=client,
name="python_expert",
instructions="You are a Python programming expert. Answer Python-related questions.",
description="Expert in Python programming",
)
javascript_expert = Agent(
client=client,
javascript_expert = Agent(client=client,
name="javascript_expert",
instructions="You are a JavaScript programming expert. Answer JavaScript-related questions.",
description="Expert in JavaScript programming",
)
database_expert = Agent(
client=client,
database_expert = Agent(client=client,
name="database_expert",
instructions="You are a database expert. Answer SQL and database-related questions.",
description="Expert in databases and SQL",
@@ -98,8 +95,7 @@ async def run_agent_framework() -> None:
workflow = GroupChatBuilder(
participants=[python_expert, javascript_expert, database_expert],
max_rounds=1,
orchestrator_agent=Agent(
client=client,
orchestrator_agent=Agent(client=client,
name="selector_manager",
instructions="Based on the conversation, select the most appropriate expert to respond next.",
),
@@ -113,8 +113,7 @@ async def run_agent_framework() -> None:
client = OpenAIChatClient(model="gpt-4.1-mini")
# Create triage agent
triage_agent = Agent(
client=client,
triage_agent = Agent(client=client,
name="triage",
instructions=(
"You are a triage agent. Analyze the user's request and route to the appropriate specialist:\n"
@@ -126,8 +125,7 @@ async def run_agent_framework() -> None:
)
# Create billing specialist
billing_agent = Agent(
client=client,
billing_agent = Agent(client=client,
name="billing_agent",
instructions="You are a billing specialist. Help with payment and billing questions. Provide clear assistance.",
description="Handles billing and payment questions",
@@ -135,8 +133,7 @@ async def run_agent_framework() -> None:
)
# Create technical support specialist
tech_support = Agent(
client=client,
tech_support = Agent(client=client,
name="technical_support",
instructions="You are technical support. Help with technical issues. Provide clear assistance.",
description="Handles technical support questions",
@@ -73,8 +73,7 @@ async def run_agent_framework() -> None:
# Create agent with tool
client = OpenAIChatClient(model="gpt-4.1-mini")
agent = Agent(
client=client,
agent = Agent(client=client,
name="assistant",
instructions="You are a helpful assistant. Use available tools to answer questions.",
tools=[get_weather],
@@ -61,8 +61,7 @@ async def run_agent_framework() -> None:
client = OpenAIChatClient(model="gpt-4.1-mini")
# Create specialized writer agent
writer = Agent(
client=client,
writer = Agent(client=client,
name="writer",
instructions="You are a creative writer. Write short, engaging content.",
)
@@ -76,8 +75,7 @@ async def run_agent_framework() -> None:
)
# Create coordinator agent with writer tool
coordinator = Agent(
client=client,
coordinator = Agent(client=client,
name="coordinator",
instructions="You coordinate with specialized agents. Delegate writing tasks to the writer agent.",
tools=[writer_tool],

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