Removed unnecessary export command

* Add declarative HttpRequestAction support to workflows

* Clean up response body for diagnostics  and fix tests.

* Fix merge with main.

* Remove redundant fallback for request content headers.

* Add declarative InvokeHttpRequest sample

* Fix solution file and update sample yaml comments

* Add final newline to sample class to fix formatting failure

dotnet/agent-framework-dotnet.slnx

@@ -163,10 +163,10 @@
     <Project Path="samples/02-agents/AgentsWithFoundry/Agent_Step25_ToolboxServerSideTools/Agent_Step25_ToolboxServerSideTools.csproj" />
   </Folder>
   <Folder Name="/Samples/02-agents/Evaluation/">
-    <Project Path="samples/02-agents/Evaluation/Evaluation_SimpleEval/Evaluation_SimpleEval.csproj" />
     <Project Path="samples/02-agents/Evaluation/Evaluation_CustomEvals/Evaluation_CustomEvals.csproj" />
     <Project Path="samples/02-agents/Evaluation/Evaluation_ExpectedOutputs/Evaluation_ExpectedOutputs.csproj" />
     <Project Path="samples/02-agents/Evaluation/Evaluation_Multimodal/Evaluation_Multimodal.csproj" />
+    <Project Path="samples/02-agents/Evaluation/Evaluation_SimpleEval/Evaluation_SimpleEval.csproj" />
   </Folder>
   <Folder Name="/Samples/02-agents/AgentWithMemory/">
     <File Path="samples/02-agents/AgentWithMemory/README.md" />
@@ -226,6 +226,7 @@
     <Project Path="samples/03-workflows/Declarative/HostedWorkflow/HostedWorkflow.csproj" />
     <Project Path="samples/03-workflows/Declarative/InputArguments/InputArguments.csproj" />
     <Project Path="samples/03-workflows/Declarative/InvokeFunctionTool/InvokeFunctionTool.csproj" />
+    <Project Path="samples/03-workflows/Declarative/InvokeHttpRequest/InvokeHttpRequest.csproj" />
     <Project Path="samples/03-workflows/Declarative/InvokeMcpTool/InvokeMcpTool.csproj" />
     <Project Path="samples/03-workflows/Declarative/Marketing/Marketing.csproj" />
     <Project Path="samples/03-workflows/Declarative/StudentTeacher/StudentTeacher.csproj" />
@@ -347,17 +348,17 @@
     <File Path="samples/02-agents/A2A/README.md" />
     <Project Path="samples/02-agents/A2A/A2AAgent_AsFunctionTools/A2AAgent_AsFunctionTools.csproj" />
     <Project Path="samples/02-agents/A2A/A2AAgent_PollingForTaskCompletion/A2AAgent_PollingForTaskCompletion.csproj" />
-    <Project Path="samples/02-agents/A2A/A2AAgent_StreamReconnection/A2AAgent_StreamReconnection.csproj" />
     <Project Path="samples/02-agents/A2A/A2AAgent_ProtocolSelection/A2AAgent_ProtocolSelection.csproj" />
+    <Project Path="samples/02-agents/A2A/A2AAgent_StreamReconnection/A2AAgent_StreamReconnection.csproj" />
   </Folder>  
   <Folder Name="/Samples/05-end-to-end/">
     <Project Path="samples/05-end-to-end/AgentWithPurview/AgentWithPurview.csproj" />
     <Project Path="samples/05-end-to-end/M365Agent/M365Agent.csproj" />
   </Folder>
   <Folder Name="/Samples/05-end-to-end/Evaluation/">
+    <Project Path="samples/05-end-to-end/Evaluation/Evaluation_ConversationSplits/Evaluation_ConversationSplits.csproj" />
     <Project Path="samples/05-end-to-end/Evaluation/Evaluation_FoundryQuality/Evaluation_FoundryQuality.csproj" />
     <Project Path="samples/05-end-to-end/Evaluation/Evaluation_MixedProviders/Evaluation_MixedProviders.csproj" />
-    <Project Path="samples/05-end-to-end/Evaluation/Evaluation_ConversationSplits/Evaluation_ConversationSplits.csproj" />
   </Folder>
   <Folder Name="/Samples/05-end-to-end/A2AClientServer/">
     <File Path="samples/05-end-to-end/A2AClientServer/README.md" />
@@ -543,8 +544,8 @@
     <Project Path="src/Microsoft.Agents.AI.Declarative/Microsoft.Agents.AI.Declarative.csproj" />
     <Project Path="src/Microsoft.Agents.AI.DevUI/Microsoft.Agents.AI.DevUI.csproj" />
     <Project Path="src/Microsoft.Agents.AI.DurableTask/Microsoft.Agents.AI.DurableTask.csproj" />
-    <Project Path="src/Microsoft.Agents.AI.Foundry/Microsoft.Agents.AI.Foundry.csproj" />
     <Project Path="src/Microsoft.Agents.AI.Foundry.Hosting/Microsoft.Agents.AI.Foundry.Hosting.csproj" />
+    <Project Path="src/Microsoft.Agents.AI.Foundry/Microsoft.Agents.AI.Foundry.csproj" />
     <Project Path="src/Microsoft.Agents.AI.GitHub.Copilot/Microsoft.Agents.AI.GitHub.Copilot.csproj" />
     <Project Path="src/Microsoft.Agents.AI.Hosting.A2A.AspNetCore/Microsoft.Agents.AI.Hosting.A2A.AspNetCore.csproj" />
     <Project Path="src/Microsoft.Agents.AI.Hosting.A2A/Microsoft.Agents.AI.Hosting.A2A.csproj" />

dotnet/samples/03-workflows/Declarative/InvokeHttpRequest/InvokeHttpRequest.csproj

@@ -0,0 +1,38 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFrameworks>net10.0</TargetFrameworks>
+    <Nullable>enable</Nullable>
+    <ImplicitUsings>enable</ImplicitUsings>
+  </PropertyGroup>
+
+  <PropertyGroup>
+    <InjectIsExternalInitOnLegacy>true</InjectIsExternalInitOnLegacy>
+    <InjectSharedFoundryAgents>true</InjectSharedFoundryAgents>
+    <InjectSharedWorkflowsExecution>true</InjectSharedWorkflowsExecution>
+    <InjectSharedWorkflowsSettings>true</InjectSharedWorkflowsSettings>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Configuration" />
+    <PackageReference Include="Microsoft.Extensions.Configuration.Binder" />
+    <PackageReference Include="Microsoft.Extensions.Configuration.EnvironmentVariables" />
+    <PackageReference Include="Microsoft.Extensions.Configuration.Json" />
+    <PackageReference Include="Microsoft.Extensions.Configuration.UserSecrets" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection" />
+    <PackageReference Include="Microsoft.Extensions.Logging" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\..\..\src\Microsoft.Agents.AI.Workflows.Declarative\Microsoft.Agents.AI.Workflows.Declarative.csproj" />
+    <ProjectReference Include="..\..\..\..\src\Microsoft.Agents.AI.Workflows.Declarative.Foundry\Microsoft.Agents.AI.Workflows.Declarative.Foundry.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <None Include="InvokeHttpRequest.yaml">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+  </ItemGroup>
+
+</Project>

dotnet/samples/03-workflows/Declarative/InvokeHttpRequest/InvokeHttpRequest.yaml

@@ -0,0 +1,76 @@
+#
+# This workflow demonstrates using HttpRequestAction to call a REST API directly
+# from the workflow without going through an AI agent first.
+#
+# HttpRequestAction allows workflows to:
+# - Fetch data from external HTTP endpoints
+# - Store the parsed response in workflow variables for later use
+# - Add the response body to the conversation so a downstream agent can
+#   answer questions based on it
+#
+# This sample fetches public metadata for the dotnet/runtime repository from
+# the GitHub REST API (no authentication required) and uses an agent to
+# answer follow-up questions about it.
+#
+# Example input:
+# How many subscribers does the repository have?
+#
+kind: Workflow
+trigger:
+
+  kind: OnConversationStart
+  id: workflow_invoke_http_request_demo
+  actions:
+
+    # Capture the original user message for input to the follow-up agent.
+    - kind: SetVariable
+      id: set_user_message
+      variable: Local.InputMessage
+      value: =System.LastMessage
+
+    # Set the repository org/name used to form the request URL.
+    - kind: SetVariable
+      id: set_repo_name
+      variable: Local.RepoName
+      value: microsoft/agent-framework
+
+    # Invoke the GitHub repo API. The response body is parsed into Local.RepoInfo
+    # and also added to the conversation (via conversationId) so the agent below
+    # can answer questions based on it.
+    - kind: HttpRequestAction
+      id: fetch_repo_info
+      conversationId: =System.ConversationId
+      method: GET
+      url: =Concatenate("https://api.github.com/repos/", Local.RepoName)
+      headers:
+        Accept: application/vnd.github+json
+        User-Agent: agent-framework-sample
+      response: Local.RepoInfo
+
+    # Display a confirmation message showing key fields from the parsed response.
+    - kind: SendMessage
+      id: show_repo_summary
+      message: "Fetched repo: visibility={Local.RepoInfo.visibility}, description={Local.RepoInfo.description}"
+
+    # Use the agent to summarize the repo using the conversation context.
+    - kind: InvokeAzureAgent
+      id: summarize_repo
+      conversationId: =System.ConversationId
+      agent:
+        name: GitHubRepoInfoAgent
+      input:
+        messages: =UserMessage("Please provide a brief summary of this GitHub repository based on the data already in the conversation.")
+      output:
+        autoSend: true
+        messages: Local.AgentResponse
+
+    # Allow the user to ask follow-up questions about the repo in a loop.
+    - kind: InvokeAzureAgent
+      id: invoke_followup
+      conversationId: =System.ConversationId
+      agent:
+        name: GitHubRepoInfoAgent
+      input:
+        messages: =Local.InputMessage
+        externalLoop:
+          when: =Upper(System.LastMessage.Text) <> "EXIT"

dotnet/samples/03-workflows/Declarative/InvokeHttpRequest/Program.cs

@@ -0,0 +1,95 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using Azure.AI.Projects;
+using Azure.AI.Projects.Agents;
+using Azure.Identity;
+using Microsoft.Agents.AI.Workflows.Declarative;
+using Microsoft.Extensions.Configuration;
+using Shared.Foundry;
+using Shared.Workflows;
+
+namespace Demo.Workflows.Declarative.InvokeHttpRequest;
+
+/// <summary>
+/// Demonstrates a workflow that uses HttpRequestAction to call a REST API
+/// directly from the workflow.
+/// </summary>
+/// <remarks>
+/// <para>
+/// The HttpRequestAction allows workflows to issue HTTP requests and:
+/// </para>
+/// <list type="bullet">
+/// <item>Fetch data from external REST endpoints</item>
+/// <item>Store the parsed response in workflow variables</item>
+/// <item>Add the response body to the conversation so an agent can answer
+///       questions based on it</item>
+/// </list>
+/// <para>
+/// This sample fetches public metadata for the dotnet/runtime repository from
+/// the GitHub REST API (no authentication required) and uses a Foundry agent
+/// to answer follow-up questions about it. Type "EXIT" to end the conversation.
+/// </para>
+/// <para>
+/// See the README.md file in the parent folder (../README.md) for detailed
+/// information about the configuration required to run this sample.
+/// </para>
+/// </remarks>
+internal sealed class Program
+{
+    public static async Task Main(string[] args)
+    {
+        // Initialize configuration
+        IConfiguration configuration = Application.InitializeConfig();
+        Uri foundryEndpoint = new(configuration.GetValue(Application.Settings.FoundryEndpoint));
+
+        // Ensure sample agent exists in Foundry. The agent has no tools - it answers
+        // questions about the GitHub repository using only the JSON data that the
+        // HttpRequestAction adds to the conversation.
+        await CreateAgentAsync(foundryEndpoint, configuration);
+
+        // Get input from command line or console
+        string workflowInput = Application.GetInput(args);
+
+        // The default HttpRequestHandler is sufficient for this sample because the
+        // GitHub REST endpoint used here does not require authentication. For
+        // authenticated endpoints, supply a custom Func<HttpRequestInfo, ..., HttpClient?>
+        // to DefaultHttpRequestHandler so each request can be routed through a
+        // pre-configured (cached) HttpClient with the appropriate credentials.
+        await using DefaultHttpRequestHandler httpRequestHandler = new();
+
+        // Create the workflow factory with the HTTP request handler
+        WorkflowFactory workflowFactory = new("InvokeHttpRequest.yaml", foundryEndpoint)
+        {
+            HttpRequestHandler = httpRequestHandler
+        };
+
+        // Execute the workflow
+        WorkflowRunner runner = new() { UseJsonCheckpoints = true };
+        await runner.ExecuteAsync(workflowFactory.CreateWorkflow, workflowInput);
+    }
+
+    private static async Task CreateAgentAsync(Uri foundryEndpoint, IConfiguration configuration)
+    {
+        // WARNING: DefaultAzureCredential is convenient for development but requires careful consideration in production.
+        AIProjectClient aiProjectClient = new(foundryEndpoint, new DefaultAzureCredential());
+
+        await aiProjectClient.CreateAgentAsync(
+            agentName: "GitHubRepoInfoAgent",
+            agentDefinition: DefineAgent(configuration),
+            agentDescription: "Answers questions about a GitHub repository using HTTP response data in the conversation");
+    }
+
+    private static DeclarativeAgentDefinition DefineAgent(IConfiguration configuration)
+    {
+        return new DeclarativeAgentDefinition(configuration.GetValue(Application.Settings.FoundryModel))
+        {
+            Instructions =
+                """
+                Answer the user's questions about the GitHub repository using only the
+                JSON data already present in the conversation history.
+                If the answer is not contained in the conversation, say so plainly
+                rather than guessing. Be concise and helpful.
+                """
+        };
+    }
+}

dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/ChatMessageExtensions.cs

@@ -16,6 +16,60 @@ internal static class ChatMessageExtensions
     public static RecordValue ToRecord(this ChatMessage message) =>
         FormulaValue.NewRecordFromFields(message.GetMessageFields());
 
+    /// <summary>
+    /// Merges the user-authored <paramref name="input"/> with the round-tripped
+    /// <paramref name="inputMessage"/> returned by <c>AgentProvider.CreateMessageAsync</c>
+    /// to produce the value stored in <c>System.LastMessage</c>.
+    /// </summary>
+    /// <remarks>
+    /// The agent service often strips or alters <see cref="TextContent"/> on round-trip,
+    /// while replacing inline media (<see cref="DataContent"/>, <see cref="UriContent"/>)
+    /// with server-side references (typically <see cref="HostedFileContent"/>).
+    /// We want both: the original text (so <c>=System.LastMessage.Text</c> works) and
+    /// the server's media references (so subsequent actions don't re-upload large blobs).
+    /// <para>
+    /// Strategy: keep <paramref name="inputMessage"/> as the base — it has the server-generated
+    /// <see cref="ChatMessage.MessageId"/> and any provider-augmented metadata, and is forward-
+    /// compatible with new properties added on <see cref="ChatMessage"/> in the abstractions
+    /// layer. Only the <see cref="ChatMessage.Contents"/> list is mutated to substitute
+    /// original <see cref="TextContent"/> items in place (and append any extras the round-trip
+    /// dropped). Non-text content items returned by the service are left untouched so
+    /// server-side references survive.
+    /// </para>
+    /// </remarks>
+    public static ChatMessage MergeForLastMessage(this ChatMessage input, ChatMessage? inputMessage)
+    {
+        if (inputMessage is null)
+        {
+            return input;
+        }
+
+        // Build a queue of the original text items, in order. Fall back to ChatMessage.Text
+        // if the input has no explicit TextContent entries.
+        Queue<TextContent> originalTexts = new(input.Contents.OfType<TextContent>());
+        if (originalTexts.Count == 0 && !string.IsNullOrEmpty(input.Text))
+        {
+            originalTexts.Enqueue(new TextContent(input.Text));
+        }
+
+        // Replace TextContent items in inputMessage.Contents with the originals, in order.
+        for (int i = 0; i < inputMessage.Contents.Count && originalTexts.Count > 0; i++)
+        {
+            if (inputMessage.Contents[i] is TextContent)
+            {
+                inputMessage.Contents[i] = originalTexts.Dequeue();
+            }
+        }
+
+        // Append any remaining original text items that the round-trip dropped entirely.
+        while (originalTexts.Count > 0)
+        {
+            inputMessage.Contents.Add(originalTexts.Dequeue());
+        }
+
+        return inputMessage;
+    }
+
     public static TableValue ToTable(this IEnumerable<ChatMessage> messages) =>
         FormulaValue.NewTable(TypeSchema.Message.RecordType, messages.Select(message => message.ToRecord()));
 

dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeWorkflowExecutor.cs

@@ -43,7 +43,11 @@ internal sealed class DeclarativeWorkflowExecutor<TInput>(
         await declarativeContext.QueueConversationUpdateAsync(conversationId, isExternal: true, cancellationToken).ConfigureAwait(false);
 
         ChatMessage inputMessage = await options.AgentProvider.CreateMessageAsync(conversationId, input, cancellationToken).ConfigureAwait(false);
-        await declarativeContext.SetLastMessageAsync(inputMessage).ConfigureAwait(false);
+
+        // Use the original input for System.LastMessage to ensure Text is preserved (the
+        // service may strip text on round-trip), but substitute server-side media references
+        // (e.g., HostedFileContent) so subsequent actions don't re-upload large blobs.
+        await declarativeContext.SetLastMessageAsync(input.MergeForLastMessage(inputMessage)).ConfigureAwait(false);
 
         await context.SendResultMessageAsync(this.Id, cancellationToken).ConfigureAwait(false);
     }

dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Kit/RootExecutor.cs

@@ -58,7 +58,6 @@ public abstract class RootExecutor<TInput> : Executor<TInput>, IResettableExecut
     public override async ValueTask HandleAsync(TInput message, IWorkflowContext context, CancellationToken cancellationToken)
     {
         DeclarativeWorkflowContext declarativeContext = new(context, this._state);
-        await this.ExecuteAsync(message, declarativeContext, cancellationToken).ConfigureAwait(false);
 
         ChatMessage input = (this._inputTransform ?? DefaultInputTransform).Invoke(message);
 
@@ -69,7 +68,13 @@ public abstract class RootExecutor<TInput> : Executor<TInput>, IResettableExecut
         await declarativeContext.QueueConversationUpdateAsync(this._conversationId, isExternal: true, cancellationToken).ConfigureAwait(false);
 
         ChatMessage inputMessage = await this._agentProvider.CreateMessageAsync(this._conversationId, input, cancellationToken).ConfigureAwait(false);
-        await declarativeContext.SetLastMessageAsync(inputMessage).ConfigureAwait(false);
+
+        // Use the original input for System.LastMessage to ensure Text is preserved (the
+        // service may strip text on round-trip), but substitute server-side media references
+        // (e.g., HostedFileContent) so subsequent actions don't re-upload large blobs.
+        await declarativeContext.SetLastMessageAsync(input.MergeForLastMessage(inputMessage)).ConfigureAwait(false);
+
+        await this.ExecuteAsync(message, declarativeContext, cancellationToken).ConfigureAwait(false);
 
         await declarativeContext.SendResultMessageAsync(this.Id, cancellationToken).ConfigureAwait(false);
     }

dotnet/src/Shared/Workflows/Execution/WorkflowFactory.cs

@@ -25,6 +25,9 @@ internal sealed class WorkflowFactory(string workflowFile, Uri foundryEndpoint)
     // Assign to provide MCP tool capabilities
     public IMcpToolHandler? McpToolHandler { get; init; }
 
+    // Assign to enable HttpRequestAction support
+    public IHttpRequestHandler? HttpRequestHandler { get; init; }
+
     /// <summary>
     /// Create the workflow from the declarative YAML.  Includes definition of the
     /// <see cref="DeclarativeWorkflowOptions" /> and the associated <see cref="ResponseAgentProvider"/>.
@@ -46,6 +49,7 @@ internal sealed class WorkflowFactory(string workflowFile, Uri foundryEndpoint)
                 ConversationId = this.ConversationId,
                 LoggerFactory = this.LoggerFactory,
                 McpToolHandler = this.McpToolHandler,
+                HttpRequestHandler = this.HttpRequestHandler,
             };
 
         string workflowPath = Path.Combine(AppContext.BaseDirectory, workflowFile);

dotnet/src/Shared/Workflows/Execution/WorkflowRunner.cs

@@ -162,7 +162,10 @@ internal sealed class WorkflowRunner
 
                 case RequestInfoEvent requestInfo:
                     Debug.WriteLine($"REQUEST #{requestInfo.Request.RequestId}");
-                    externalResponse = requestInfo.Request;
+                    if (response is null || !string.Equals(requestInfo.Request.RequestId, response.RequestId, StringComparison.Ordinal))
+                    {
+                        externalResponse = requestInfo.Request;
+                    }
                     break;
 
                 case ConversationUpdateEvent invokeEvent:

dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/Extensions/ChatMessageExtensionsTests.cs

@@ -769,4 +769,165 @@ public sealed class ChatMessageExtensionsTests
                 break;
         }
     }
+
+    [Fact]
+    public void MergeForLastMessageReturnsInputWhenInputMessageIsNull()
+    {
+        // Arrange
+        ChatMessage input = new(ChatRole.User, "hello") { MessageId = "local" };
+
+        // Act
+        ChatMessage result = input.MergeForLastMessage(null);
+
+        // Assert
+        Assert.Same(input, result);
+    }
+
+    [Fact]
+    public void MergeForLastMessageReturnsSameInstanceAsRoundTripped()
+    {
+        // Arrange: returning the round-tripped instance keeps the merge forward-compatible
+        // with future ChatMessage properties (e.g., new metadata fields) without explicit copies.
+        ChatMessage input = new(ChatRole.User, "original");
+        ChatMessage roundTripped = new(ChatRole.User, "stripped") { MessageId = "server" };
+
+        // Act
+        ChatMessage result = input.MergeForLastMessage(roundTripped);
+
+        // Assert
+        Assert.Same(roundTripped, result);
+    }
+
+    [Fact]
+    public void MergeForLastMessagePrefersOriginalTextOverRoundTrippedText()
+    {
+        // Arrange
+        ChatMessage input = new(ChatRole.User, "original text");
+        ChatMessage roundTripped = new(ChatRole.User, "stripped") { MessageId = "server-id" };
+
+        // Act
+        ChatMessage result = input.MergeForLastMessage(roundTripped);
+
+        // Assert
+        Assert.Equal("server-id", result.MessageId);
+        Assert.Equal("original text", result.Text);
+        TextContent text = Assert.IsType<TextContent>(Assert.Single(result.Contents));
+        Assert.Equal("original text", text.Text);
+    }
+
+    [Fact]
+    public void MergeForLastMessageReplacesTextInPlaceAndKeepsServerMedia()
+    {
+        // Arrange
+        HostedFileContent serverRef = new("file-abc");
+        ChatMessage input = new(ChatRole.User, [new TextContent("look at this:"), new DataContent("data:image/jpeg;base64,QUJD", "image/jpeg")]);
+        ChatMessage roundTripped = new(ChatRole.User, [new TextContent("stripped"), serverRef]) { MessageId = "server-id" };
+
+        // Act
+        ChatMessage result = input.MergeForLastMessage(roundTripped);
+
+        // Assert: server's text slot is replaced with original text; server's media reference is preserved.
+        Assert.Equal("server-id", result.MessageId);
+        Assert.Collection(result.Contents,
+            c => Assert.Equal("look at this:", Assert.IsType<TextContent>(c).Text),
+            c => Assert.Same(serverRef, c));
+    }
+
+    [Fact]
+    public void MergeForLastMessageAppendsOriginalTextWhenRoundTripHasNoTextSlot()
+    {
+        // Arrange: round-tripped message has only media (no text slot to replace).
+        HostedFileContent serverRef = new("file-1");
+        ChatMessage input = new(ChatRole.User, [new TextContent("middle"), new DataContent("data:image/jpeg;base64,QUE=", "image/jpeg")]);
+        ChatMessage roundTripped = new(ChatRole.User, [serverRef]) { MessageId = "id" };
+
+        // Act
+        ChatMessage result = input.MergeForLastMessage(roundTripped);
+
+        // Assert: media kept; original text appended at end.
+        Assert.Collection(result.Contents,
+            c => Assert.Same(serverRef, c),
+            c => Assert.Equal("middle", Assert.IsType<TextContent>(c).Text));
+    }
+
+    [Fact]
+    public void MergeForLastMessageReplacesMultipleTextSlotsInOrder()
+    {
+        // Arrange: input has two text items; round-tripped has two text slots interleaved with media.
+        HostedFileContent firstRef = new("file-1");
+        HostedFileContent secondRef = new("file-2");
+        ChatMessage input = new(ChatRole.User, [new TextContent("first"), new TextContent("second")]);
+        ChatMessage roundTripped = new(ChatRole.User, [firstRef, new TextContent("a"), secondRef, new TextContent("b")]) { MessageId = "id" };
+
+        // Act
+        ChatMessage result = input.MergeForLastMessage(roundTripped);
+
+        // Assert
+        Assert.Collection(result.Contents,
+            c => Assert.Same(firstRef, c),
+            c => Assert.Equal("first", Assert.IsType<TextContent>(c).Text),
+            c => Assert.Same(secondRef, c),
+            c => Assert.Equal("second", Assert.IsType<TextContent>(c).Text));
+    }
+
+    [Fact]
+    public void MergeForLastMessageFallsBackToInputTextWhenInputHasNoTextContent()
+    {
+        // Arrange: ChatMessage(role, "string") populates Text but no explicit TextContent
+        // when Contents is initially empty in some construction paths. Verify we still
+        // recover the original Text via input.Text.
+        ChatMessage input = new(ChatRole.User, "fallback text");
+        ChatMessage roundTripped = new(ChatRole.User, [new TextContent("stripped")]) { MessageId = "id" };
+
+        // Act
+        ChatMessage result = input.MergeForLastMessage(roundTripped);
+
+        // Assert
+        Assert.Equal("fallback text", Assert.IsType<TextContent>(Assert.Single(result.Contents)).Text);
+    }
+
+    [Fact]
+    public void MergeForLastMessagePreservesServerAuthoredProperties()
+    {
+        // Arrange: server (round-trip) is authoritative for metadata. Returning the
+        // round-tripped instance means any future ChatMessage property is automatically
+        // preserved without code changes here.
+        ChatMessage input = new(ChatRole.User, "hi")
+        {
+            AuthorName = "client-side",
+            AdditionalProperties = new AdditionalPropertiesDictionary { ["client"] = "value" },
+        };
+        ChatMessage roundTripped = new(ChatRole.User, [new TextContent("stripped")])
+        {
+            MessageId = "server",
+            AuthorName = "server-side",
+            AdditionalProperties = new AdditionalPropertiesDictionary { ["server"] = "value" },
+        };
+
+        // Act
+        ChatMessage result = input.MergeForLastMessage(roundTripped);
+
+        // Assert
+        Assert.Equal("server", result.MessageId);
+        Assert.Equal("server-side", result.AuthorName);
+        Assert.NotNull(result.AdditionalProperties);
+        Assert.True(result.AdditionalProperties.ContainsKey("server"));
+        Assert.False(result.AdditionalProperties.ContainsKey("client"));
+    }
+
+    [Fact]
+    public void MergeForLastMessageHandlesEmptyInputContents()
+    {
+        // Arrange
+        ChatMessage input = new(ChatRole.User, new List<AIContent>());
+        HostedFileContent serverRef = new("file-only");
+        ChatMessage roundTripped = new(ChatRole.User, [serverRef]) { MessageId = "id" };
+
+        // Act
+        ChatMessage result = input.MergeForLastMessage(roundTripped);
+
+        // Assert: nothing to splice; round-tripped returned unchanged.
+        Assert.Same(roundTripped, result);
+        Assert.Equal("file-only", Assert.IsType<HostedFileContent>(Assert.Single(result.Contents)).FileId);
+    }
 }

python/CHANGELOG.md

@@ -7,6 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [1.2.2] - 2026-04-29
+
+### Added
+- **agent-framework-azure-contentunderstanding**: New alpha package — Azure AI Content Understanding context provider that auto-analyzes file attachments (documents, images, audio, video) and injects structured results into the LLM context, with multi-document session state, configurable timeout, output filtering via `AnalysisSection`, and auto-registered `list_documents` / `get_analyzed_document` tools ([#4829](https://github.com/microsoft/agent-framework/pull/4829))
+- **agent-framework-foundry-hosting**: Add hosted Durable Workflow support — propagate full conversation history to workflow agents and wire `Workflow.as_agent()` end-to-end via the foundry hosting layer ([#5531](https://github.com/microsoft/agent-framework/pull/5531))
+
+### Changed
+- **agent-framework-orchestrations**: [BREAKING] Standardize orchestration terminal outputs as `AgentResponse` so `Workflow.as_agent()` returns the final answer only; aligns sequential-approval (`with_request_info`) and concurrent (`intermediate_outputs=True`) flows on the same output contract ([#5301](https://github.com/microsoft/agent-framework/pull/5301))
+- **agent-framework-core**, **agent-framework-declarative**: Preserve `Workflow.run()` shared state across calls so multi-turn `WorkflowAgent` invocations retain context, accept `list[Message]` input in the declarative start executor, and coerce `Enum` values when serializing PowerFx symbols ([#5531](https://github.com/microsoft/agent-framework/pull/5531))
+- **dependencies**: Update workspace package dependencies and preserve `mcp[ws]` / `uvicorn[standard]` extras through override-dependencies in `/python` ([#5555](https://github.com/microsoft/agent-framework/pull/5555))
+
+### Fixed
+- **agent-framework-core**: Fix observability spans not being correctly nested when using streaming ([#5552](https://github.com/microsoft/agent-framework/pull/5552))
+- **agent-framework-openai**: Fix `file_search` citations breaking the assistant-message history roundtrip — skip `hosted_file` content in the assistant role so the Responses API no longer rejects `input_file` ([#5557](https://github.com/microsoft/agent-framework/pull/5557))
+
 ## [1.2.1] - 2026-04-28
 
 ### Added
@@ -1003,7 +1018,8 @@ Release candidate for **agent-framework-core** and **agent-framework-azure-ai**
 
 For more information, see the [announcement blog post](https://devblogs.microsoft.com/foundry/introducing-microsoft-agent-framework-the-open-source-engine-for-agentic-ai-apps/).
 
-[Unreleased]: https://github.com/microsoft/agent-framework/compare/python-1.2.1...HEAD
+[Unreleased]: https://github.com/microsoft/agent-framework/compare/python-1.2.2...HEAD
+[1.2.2]: https://github.com/microsoft/agent-framework/compare/python-1.2.1...python-1.2.2
 [1.2.1]: https://github.com/microsoft/agent-framework/compare/python-1.2.0...python-1.2.1
 [1.2.0]: https://github.com/microsoft/agent-framework/compare/python-1.1.1...python-1.2.0
 [1.1.1]: https://github.com/microsoft/agent-framework/compare/python-1.1.0...python-1.1.1

python/packages/a2a/pyproject.toml

@@ -4,7 +4,7 @@ description = "A2A integration for Microsoft Agent Framework."
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -23,7 +23,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
     "a2a-sdk>=0.3.5,<0.3.24",
 ]
 

python/packages/ag-ui/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "agent-framework-ag-ui"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 description = "AG-UI protocol integration for Agent Framework"
 readme = "README.md"
 license-files = ["LICENSE"]
@@ -22,7 +22,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
     "ag-ui-protocol>=0.1.16,<0.2",
     "fastapi>=0.115.0,<0.133.1",
     "uvicorn[standard]>=0.30.0,<0.42.0"

python/packages/anthropic/agent_framework_anthropic/_chat_client.py

@@ -872,6 +872,8 @@ class RawAnthropicClient(
         tool_mode = validate_tool_mode(options.get("tool_choice"))
         if tool_mode is None:
             return result or None
+        if "allowed_tools" in tool_mode:
+            logger.warning("allowed_tools is not supported by Anthropic; the setting will be ignored")
         allow_multiple = options.get("allow_multiple_tool_calls")
         match tool_mode.get("mode"):
             case "auto":

python/packages/anthropic/pyproject.toml

@@ -4,7 +4,7 @@ description = "Anthropic integration for Microsoft Agent Framework."
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -23,7 +23,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
     "anthropic>=0.80.0,<0.80.1",
 ]
 

python/packages/azure-ai-search/pyproject.toml

@@ -4,7 +4,7 @@ description = "Azure AI Search integration for Microsoft Agent Framework."
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -23,7 +23,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
     "azure-search-documents>=11.7.0b2,<11.7.0b3",
 ]
 

python/packages/azure-contentunderstanding/pyproject.toml

@@ -4,7 +4,7 @@ description = "Azure Content Understanding integration for Microsoft Agent Frame
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com" }]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0a260401"
+version = "1.0.0a260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -23,8 +23,9 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.0.0,<2",
-    "azure-ai-contentunderstanding>=1.0.0,<1.1",
+    "agent-framework-core>=1.2.2,<2",
+    "agent-framework-foundry>=1.2.2,<2",
+    "azure-ai-contentunderstanding>=1.0.1,<1.1",
     "aiohttp>=3.9,<4",
     "filetype>=1.2,<2",
 ]

python/packages/azure-contentunderstanding/samples/01-get-started/01_document_qa.py

@@ -15,12 +15,10 @@ import os
 from pathlib import Path
 
 from agent_framework import Agent, Content, Message
-from agent_framework.foundry import FoundryChatClient
+from agent_framework.foundry import ContentUnderstandingContextProvider, FoundryChatClient
 from azure.identity import AzureCliCredential
 from dotenv import load_dotenv
 
-from agent_framework.foundry import ContentUnderstandingContextProvider
-
 load_dotenv()
 
 """

python/packages/azure-contentunderstanding/samples/01-get-started/02_multi_turn_session.py

@@ -15,12 +15,10 @@ import os
 from pathlib import Path
 
 from agent_framework import Agent, AgentSession, Content, Message
-from agent_framework.foundry import FoundryChatClient
+from agent_framework.foundry import ContentUnderstandingContextProvider, FoundryChatClient
 from azure.identity import AzureCliCredential
 from dotenv import load_dotenv
 
-from agent_framework.foundry import ContentUnderstandingContextProvider
-
 load_dotenv()
 
 """

python/packages/azure-contentunderstanding/samples/01-get-started/03_multimodal_chat.py

@@ -16,12 +16,10 @@ import time
 from pathlib import Path
 
 from agent_framework import Agent, AgentSession, Content, Message
-from agent_framework.foundry import FoundryChatClient
+from agent_framework.foundry import ContentUnderstandingContextProvider, FoundryChatClient
 from azure.identity import AzureCliCredential
 from dotenv import load_dotenv
 
-from agent_framework.foundry import ContentUnderstandingContextProvider
-
 load_dotenv()
 
 """

python/packages/azure-contentunderstanding/samples/01-get-started/04_invoice_processing.py

@@ -16,13 +16,11 @@ import os
 from pathlib import Path
 
 from agent_framework import Agent, AgentSession, Content, Message
-from agent_framework.foundry import FoundryChatClient
+from agent_framework.foundry import ContentUnderstandingContextProvider, FoundryChatClient
 from azure.identity import AzureCliCredential
 from dotenv import load_dotenv
 from pydantic import BaseModel, Field
 
-from agent_framework.foundry import ContentUnderstandingContextProvider
-
 load_dotenv()
 
 """

python/packages/azure-contentunderstanding/samples/02-devui/01-multimodal_agent/agent.py

@@ -21,13 +21,11 @@ Run with DevUI:
 import os
 
 from agent_framework import Agent
-from agent_framework.foundry import FoundryChatClient
+from agent_framework.foundry import ContentUnderstandingContextProvider, FoundryChatClient
 from azure.core.credentials import AzureKeyCredential
 from azure.identity import AzureCliCredential
 from dotenv import load_dotenv
 
-from agent_framework.foundry import ContentUnderstandingContextProvider
-
 load_dotenv()
 
 # --- Auth ---

python/packages/azure-contentunderstanding/samples/02-devui/02-file_search_agent/foundry_backend/agent.py

@@ -32,17 +32,16 @@ Run with DevUI:
 import os
 
 from agent_framework import Agent
-from agent_framework.foundry import FoundryChatClient
+from agent_framework.foundry import (
+    ContentUnderstandingContextProvider,
+    FileSearchConfig,
+    FoundryChatClient,
+)
 from azure.core.credentials import AzureKeyCredential
 from azure.identity import AzureCliCredential
 from dotenv import load_dotenv
 from openai import AzureOpenAI
 
-from agent_framework.foundry import (
-    ContentUnderstandingContextProvider,
-    FileSearchConfig,
-)
-
 load_dotenv()
 
 # --- Auth ---

python/packages/azure-cosmos/pyproject.toml

@@ -4,7 +4,7 @@ description = "Azure Cosmos DB history provider integration for Microsoft Agent
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -23,7 +23,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
     "azure-cosmos>=4.3.0,<5",
 ]
 

python/packages/azurefunctions/pyproject.toml

@@ -4,7 +4,7 @@ description = "Azure Functions integration for Microsoft Agent Framework."
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -22,7 +22,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
     "agent-framework-durabletask",
     "azure-functions>=1.24.0,<2",
     "azure-functions-durable>=1.3.1,<2",

python/packages/bedrock/agent_framework_bedrock/_chat_client.py

@@ -405,6 +405,8 @@ class BedrockChatClient(
 
         tool_config = self._prepare_tools(options.get("tools"))
         if tool_mode := validate_tool_mode(options.get("tool_choice")):
+            if "allowed_tools" in tool_mode:
+                logger.warning("allowed_tools is not supported by Bedrock; the setting will be ignored")
             match tool_mode.get("mode"):
                 case "none":
                     # Bedrock doesn't support toolChoice "none".

python/packages/bedrock/pyproject.toml

@@ -4,7 +4,7 @@ description = "Amazon Bedrock integration for Microsoft Agent Framework."
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -23,7 +23,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
     "boto3>=1.35.0,<2.0.0",
     "botocore>=1.35.0,<2.0.0",
 ]

python/packages/chatkit/pyproject.toml

@@ -4,7 +4,7 @@ description = "OpenAI ChatKit integration for Microsoft Agent Framework."
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -22,7 +22,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
     "openai-chatkit>=1.4.1,<2.0.0",
 ]
 

python/packages/claude/pyproject.toml

@@ -4,7 +4,7 @@ description = "Claude Agent SDK integration for Microsoft Agent Framework."
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -23,7 +23,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
     "claude-agent-sdk>=0.1.36,<0.1.49",
 ]
 

python/packages/copilotstudio/pyproject.toml

@@ -4,7 +4,7 @@ description = "Copilot Studio integration for Microsoft Agent Framework."
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -23,7 +23,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
     "microsoft-agents-copilotstudio-client>=0.3.1,<0.3.2",
 ]
 

python/packages/core/agent_framework/_types.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import base64
+import contextlib
 import json
 import logging
 import re
@@ -2890,6 +2891,7 @@ class ResponseStream(AsyncIterable[UpdateT], Generic[UpdateT, FinalT]):
         self._inner_stream_source: ResponseStream[Any, Any] | Awaitable[ResponseStream[Any, Any]] | None = None
         self._wrap_inner: bool = False
         self._map_update: Callable[[Any], UpdateT | Awaitable[UpdateT]] | None = None
+        self._pull_context_manager_factories: list[Callable[[], contextlib.AbstractContextManager[Any]]] = []
 
     def map(
         self,
@@ -3008,11 +3010,18 @@ class ResponseStream(AsyncIterable[UpdateT], Generic[UpdateT, FinalT]):
         return self
 
     async def __anext__(self) -> UpdateT:
-        if self._iterator is None:
-            stream = await self._get_stream()
-            self._iterator = stream.__aiter__()
         try:
-            update: UpdateT = await self._iterator.__anext__()
+            with contextlib.ExitStack() as stack:
+                for factory in self._pull_context_manager_factories:
+                    stack.enter_context(factory())
+                # Resolve the underlying stream inside the pull contexts so that any
+                # spans/contexts created during stream resolution (e.g. inner chat
+                # completion spans created on the first pull of a wrapped agent stream)
+                # inherit the active context (e.g. an outer agent invoke span).
+                if self._iterator is None:
+                    stream = await self._get_stream()
+                    self._iterator = stream.__aiter__()
+                update: UpdateT = await self._iterator.__anext__()
         except StopAsyncIteration:
             self._consumed = True
             await self._run_cleanup_hooks()
@@ -3038,9 +3047,25 @@ class ResponseStream(AsyncIterable[UpdateT], Generic[UpdateT, FinalT]):
                 update = hooked
         return update
 
+    async def _resolve_stream_with_pull_contexts(self) -> AsyncIterable[UpdateT]:
+        """Resolve the underlying stream while activating any registered pull context managers.
+
+        Used by ``__await__`` and ``get_final_response`` so that any spans/contexts created
+        during stream resolution (e.g. when the source is an Awaitable that internally
+        creates child telemetry spans) inherit the same active context as iterator pulls.
+        ``__anext__`` resolves the stream inside its own ExitStack and so calls ``_get_stream``
+        directly.
+        """
+        if self._stream is not None:
+            return await self._get_stream()
+        with contextlib.ExitStack() as stack:
+            for factory in self._pull_context_manager_factories:
+                stack.enter_context(factory())
+            return await self._get_stream()
+
     def __await__(self) -> Any:
         async def _wrap() -> ResponseStream[UpdateT, FinalT]:
-            await self._get_stream()
+            await self._resolve_stream_with_pull_contexts()
             return self
 
         return _wrap().__await__()
@@ -3064,10 +3089,12 @@ class ResponseStream(AsyncIterable[UpdateT], Generic[UpdateT, FinalT]):
         """
         if self._wrap_inner:
             if self._inner_stream is None:
-                # Use _get_stream() to resolve the awaitable - this properly handles
+                # Use _resolve_stream_with_pull_contexts() so that any spans/contexts
+                # created while resolving the awaitable (e.g. inner telemetry spans)
+                # inherit the same active context as iterator pulls. This also handles
                 # the case where _stream_source and _inner_stream_source are the same
                 # coroutine (e.g., from from_awaitable), avoiding double-await errors.
-                await self._get_stream()
+                await self._resolve_stream_with_pull_contexts()
             if self._inner_stream is None:
                 raise RuntimeError("Inner stream not available")
             if not self._finalized and not self._consumed:
@@ -3177,6 +3204,25 @@ class ResponseStream(AsyncIterable[UpdateT], Generic[UpdateT, FinalT]):
         self._cleanup_hooks.append(hook)
         return self
 
+    def with_pull_context_manager(
+        self,
+        cm_factory: Callable[[], contextlib.AbstractContextManager[Any]],
+    ) -> ResponseStream[UpdateT, FinalT]:
+        """Register a context manager factory invoked around each underlying iterator pull.
+
+        The factory is called once per ``__anext__`` and the returned context manager wraps
+        the await of the underlying iterator. This is useful for state that needs to be
+        active while the inner async work runs - for example, attaching an OpenTelemetry
+        span to the current context so child spans created by inner code (HTTP clients,
+        tool execution) are correctly parented.
+
+        Because the context manager is entered and exited within the same ``__anext__``
+        invocation, attach/detach style operations remain symmetric in the same async
+        context regardless of where the stream is iterated.
+        """
+        self._pull_context_manager_factories.append(cm_factory)
+        return self
+
     async def _run_cleanup_hooks(self) -> None:
         if self._cleanup_run:
             return
@@ -3200,10 +3246,12 @@ class ToolMode(TypedDict, total=False):
     Fields:
         mode: One of "auto", "required", or "none".
         required_function_name: Optional function name when `mode == "required"`.
+        allowed_tools: Optional list of tool names when `mode` is `"auto"` or `"required"`.
     """
 
     mode: Literal["auto", "required", "none"]
     required_function_name: str
+    allowed_tools: list[str]
 
 
 # region TypedDict-based Chat Options
@@ -3436,7 +3484,7 @@ def validate_tool_mode(
 
     Returns:
         A ToolMode dict (contains keys: "mode", and optionally
-        "required_function_name"), or ``None`` when not provided.
+        "required_function_name" or "allowed_tools"), or ``None`` when not provided.
 
     Raises:
         ContentError: If the tool_choice string is invalid.
@@ -3453,6 +3501,17 @@ def validate_tool_mode(
         raise ContentError(f"Invalid tool choice: {tool_choice['mode']}")
     if tool_choice["mode"] != "required" and "required_function_name" in tool_choice:
         raise ContentError("tool_choice with mode other than 'required' cannot have 'required_function_name'")
+    if tool_choice["mode"] not in ("auto", "required") and "allowed_tools" in tool_choice:
+        raise ContentError("tool_choice 'allowed_tools' is only valid when mode is 'auto' or 'required'")
+    if "allowed_tools" in tool_choice:
+        allowed_tools = tool_choice["allowed_tools"]
+        if isinstance(allowed_tools, str) or not isinstance(allowed_tools, Sequence):
+            raise ContentError("tool_choice 'allowed_tools' must be a non-string sequence of strings")
+        if not all(isinstance(tool_name, str) for tool_name in allowed_tools):
+            raise ContentError("tool_choice 'allowed_tools' must contain only strings")
+        normalized_tool_choice = dict(tool_choice)
+        normalized_tool_choice["allowed_tools"] = list(allowed_tools)
+        return cast(ToolMode, normalized_tool_choice)
     return tool_choice
 
 

python/packages/core/agent_framework/_workflows/_agent.py

@@ -437,6 +437,13 @@ class WorkflowAgent(BaseAgent):
                     yield event
 
         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(
                     stream=True,

python/packages/core/agent_framework/_workflows/_runner.py

@@ -278,7 +278,12 @@ class Runner:
                     "Please rebuild the original workflow before resuming."
                 )
 
-            # Restore state
+            # Restore state. Clear first so import_state (which merges) does
+            # not leak stale keys from a prior run on this Workflow instance.
+            # This matters more now that Workflow.run() no longer wipes state
+            # per call - the only reset point for shared state on a reused
+            # instance is at restore time.
+            self._state.clear()
             self._state.import_state(checkpoint.state)
             # Restore executor states using the restored state
             await self._restore_executor_states()

python/packages/core/agent_framework/_workflows/_workflow.py

@@ -299,7 +299,7 @@ class Workflow(DictConvertible):
     async def _run_workflow_with_tracing(
         self,
         initial_executor_fn: Callable[[], Awaitable[None]] | None = None,
-        reset_context: bool = True,
+        is_continuation: bool = False,
         streaming: bool = False,
         function_invocation_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None,
         client_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None,
@@ -310,13 +310,19 @@ class Workflow(DictConvertible):
         of external callers to maintain context across different workflow runs.
 
         Args:
-            initial_executor_fn: Optional function to execute initial executor
-            reset_context: Whether to reset the context for a new run
-            streaming: Whether to enable streaming mode for agents
+            initial_executor_fn: Optional function to execute initial executor.
+            is_continuation: True when this run is a continuation of prior
+                work (a checkpoint restore or a responses-only replay) rather
+                than a fresh new turn delivered via the start executor with
+                ``message=...``. Continuations preserve per-run accounting
+                (iteration counter and run kwargs) from the prior turn;
+                fresh-message runs reset them. Shared workflow state is
+                preserved in both cases.
+            streaming: Whether to enable streaming mode for agents.
             function_invocation_kwargs: Optional kwargs to store in State for function
-                invocations in subagents
+                invocations in subagents.
             client_kwargs: Optional kwargs to store in State for chat client
-                invocations in subagents
+                invocations in subagents.
 
         Yields:
             WorkflowEvent: The events generated during the workflow execution.
@@ -345,16 +351,26 @@ class Workflow(DictConvertible):
                     in_progress = WorkflowEvent.status(WorkflowRunState.IN_PROGRESS)
                 yield in_progress  # noqa: RUF070
 
-                # Reset context for a new run if supported
-                if reset_context:
+                # Per-run reset for fresh-message runs only. We deliberately
+                # do NOT clear shared workflow state (`_state.clear()`) or the
+                # runner context's in-flight messages (`reset_for_new_run()`)
+                # here - state and pending work persist across `run()` calls
+                # so that a `WorkflowAgent` can deliver multi-turn input on
+                # the same instance and have prior turns' context survive.
+                # Iteration counting and per-run kwargs ARE per-run though,
+                # so they're reset here.
+                if not is_continuation:
                     self._runner.reset_iteration_count()
-                    self._runner.context.reset_for_new_run()
-                    self._state.clear()
 
                 # Store run kwargs in State so executors can access them.
-                # Only overwrite when new kwargs are explicitly provided or state was
-                # just cleared (fresh run). On continuation (reset_context=False) with
-                # no new kwargs, preserve the kwargs from the original run.
+                # Per-run kwargs semantics:
+                # - On a fresh message run, prior kwargs go away (set to {}
+                #   by default, or to the new kwargs if provided). This
+                #   prevents stale kwargs from a prior turn leaking into the
+                #   current turn.
+                # - On a continuation (checkpoint restore or responses), the
+                #   prior run's kwargs are preserved unless the caller
+                #   explicitly provides new kwargs.
                 if function_invocation_kwargs is not None or client_kwargs is not None:
                     combined_kwargs: dict[str, Any] = {}
                     if function_invocation_kwargs is not None:
@@ -366,11 +382,12 @@ class Workflow(DictConvertible):
                             client_kwargs, "client_kwargs"
                         )
                     self._state.set(WORKFLOW_RUN_KWARGS_KEY, combined_kwargs)
-                elif reset_context:
+                elif not is_continuation:
                     self._state.set(WORKFLOW_RUN_KWARGS_KEY, {})
                 self._state.commit()  # Commit immediately so kwargs are available
 
-                # Set streaming mode after reset
+                # Set streaming mode (always set explicitly per run since
+                # reset_for_new_run() no longer runs to clear it).
                 self._runner_context.set_streaming(streaming)
 
                 # Execute initial setup if provided
@@ -585,13 +602,31 @@ class Workflow(DictConvertible):
         if checkpoint_storage is not None:
             self._runner.context.set_runtime_checkpoint_storage(checkpoint_storage)
 
-        initial_executor_fn, reset_context = self._resolve_execution_mode(
-            message, responses, checkpoint_id, checkpoint_storage
-        )
+        # Async validation: a fresh-message run is only allowed when the
+        # runner context has fully drained from any prior run. If it still
+        # has in-flight executor messages, the prior run didn't complete -
+        # the caller must either resume from a checkpoint or wait for the
+        # prior run to drain. (Pending request_info events are intentionally
+        # NOT blocked here: a follow-up run with message=... is the normal
+        # way to deliver a response to those pending requests, e.g. via
+        # WorkflowAgent._process_pending_requests.)
+        # NOTE: _validate_run_params already enforces that ``message`` is
+        # mutually exclusive with both ``checkpoint_id`` and ``responses``,
+        # so we don't need to re-check those here.
+        if message is not None and await self._runner.context.has_messages():
+            raise RuntimeError(
+                "Cannot start a new run with 'message' while in-flight executor "
+                "messages remain from a prior run. Resume from a checkpoint "
+                "(checkpoint_id=...) or wait for the prior run to complete. "
+                "Workflows that need to recover from a mid-run failure must use "
+                "checkpointing; there is no in-process recovery path."
+            )
+
+        initial_executor_fn = self._resolve_execution_mode(message, responses, checkpoint_id, checkpoint_storage)
 
         async for event in self._run_workflow_with_tracing(
             initial_executor_fn=initial_executor_fn,
-            reset_context=reset_context,
+            is_continuation=(message is None),
             streaming=streaming,
             function_invocation_kwargs=function_invocation_kwargs,
             client_kwargs=client_kwargs,
@@ -674,12 +709,8 @@ class Workflow(DictConvertible):
         responses: Mapping[str, Any] | None,
         checkpoint_id: str | None,
         checkpoint_storage: CheckpointStorage | None,
-    ) -> tuple[Callable[[], Awaitable[None]], bool]:
-        """Determine the initial executor function and reset_context flag based on parameters.
-
-        Returns:
-            A tuple of (initial_executor_fn, reset_context).
-        """
+    ) -> Callable[[], Awaitable[None]]:
+        """Determine the initial executor function based on parameters."""
         if responses is not None:
             if checkpoint_id is not None:
                 # Combined: restore checkpoint then send responses
@@ -689,13 +720,9 @@ class Workflow(DictConvertible):
             else:
                 # Send responses only (requires pending requests in workflow state)
                 initial_executor_fn = functools.partial(self._send_responses_internal, responses)
-            return initial_executor_fn, False
+            return initial_executor_fn
         # Regular run or checkpoint restoration
-        initial_executor_fn = functools.partial(
-            self._execute_with_message_or_checkpoint, message, checkpoint_id, checkpoint_storage
-        )
-        reset_context = message is not None and checkpoint_id is None
-        return initial_executor_fn, reset_context
+        return functools.partial(self._execute_with_message_or_checkpoint, message, checkpoint_id, checkpoint_storage)
 
     async def _restore_and_send_responses(
         self,

python/packages/core/agent_framework/declarative/__init__.py

@@ -21,10 +21,15 @@ _IMPORTS = [
     "AgentFactory",
     "AgentExternalInputRequest",
     "AgentExternalInputResponse",
+    "DeclarativeActionError",
     "DeclarativeLoaderError",
     "DeclarativeWorkflowError",
+    "DefaultHttpRequestHandler",
     "ExternalInputRequest",
     "ExternalInputResponse",
+    "HttpRequestHandler",
+    "HttpRequestInfo",
+    "HttpRequestResult",
     "ProviderLookupError",
     "ProviderTypeMapping",
     "WorkflowFactory",

python/packages/core/agent_framework/declarative/__init__.pyi

@@ -4,10 +4,15 @@ from agent_framework_declarative import (
     AgentExternalInputRequest,
     AgentExternalInputResponse,
     AgentFactory,
+    DeclarativeActionError,
     DeclarativeLoaderError,
     DeclarativeWorkflowError,
+    DefaultHttpRequestHandler,
     ExternalInputRequest,
     ExternalInputResponse,
+    HttpRequestHandler,
+    HttpRequestInfo,
+    HttpRequestResult,
     ProviderLookupError,
     ProviderTypeMapping,
     WorkflowFactory,
@@ -18,10 +23,15 @@ __all__ = [
     "AgentExternalInputRequest",
     "AgentExternalInputResponse",
     "AgentFactory",
+    "DeclarativeActionError",
     "DeclarativeLoaderError",
     "DeclarativeWorkflowError",
+    "DefaultHttpRequestHandler",
     "ExternalInputRequest",
     "ExternalInputResponse",
+    "HttpRequestHandler",
+    "HttpRequestInfo",
+    "HttpRequestResult",
     "ProviderLookupError",
     "ProviderTypeMapping",
     "WorkflowFactory",

python/packages/core/agent_framework/foundry/__init__.py

@@ -15,7 +15,10 @@ from typing import Any
 _IMPORTS: dict[str, tuple[str, str]] = {
     "AnalysisSection": ("agent_framework_azure_contentunderstanding", "agent-framework-azure-contentunderstanding"),
     "AnthropicFoundryClient": ("agent_framework_anthropic", "agent-framework-anthropic"),
-    "ContentUnderstandingContextProvider": ("agent_framework_azure_contentunderstanding", "agent-framework-azure-contentunderstanding"),
+    "ContentUnderstandingContextProvider": (
+        "agent_framework_azure_contentunderstanding",
+        "agent-framework-azure-contentunderstanding",
+    ),
     "DocumentStatus": ("agent_framework_azure_contentunderstanding", "agent-framework-azure-contentunderstanding"),
     "FileSearchBackend": ("agent_framework_azure_contentunderstanding", "agent-framework-azure-contentunderstanding"),
     "FileSearchConfig": ("agent_framework_azure_contentunderstanding", "agent-framework-azure-contentunderstanding"),

python/packages/core/agent_framework/foundry/__init__.pyi

@@ -4,12 +4,12 @@
 # Install the relevant packages for full type support.
 
 from agent_framework_anthropic import AnthropicFoundryClient, RawAnthropicFoundryClient
-from agent_framework_azure_contentunderstanding import (
-    AnalysisSection,
-    ContentUnderstandingContextProvider,
-    DocumentStatus,
-    FileSearchBackend,
-    FileSearchConfig,
+from agent_framework_azure_contentunderstanding import (  # pyright: ignore[reportMissingImports]
+    AnalysisSection,  # pyright: ignore[reportUnknownVariableType]
+    ContentUnderstandingContextProvider,  # pyright: ignore[reportUnknownVariableType]
+    DocumentStatus,  # pyright: ignore[reportUnknownVariableType]
+    FileSearchBackend,  # pyright: ignore[reportUnknownVariableType]
+    FileSearchConfig,  # pyright: ignore[reportUnknownVariableType]
 )
 from agent_framework_foundry import (
     FoundryAgent,

python/packages/core/agent_framework/observability.py

@@ -26,6 +26,7 @@ from time import perf_counter, time_ns
 from typing import TYPE_CHECKING, Any, ClassVar, Final, Generic, Literal, TypedDict, cast, overload
 
 from dotenv import load_dotenv
+from opentelemetry import context as otel_context
 from opentelemetry import metrics, trace
 
 from . import __version__ as version_info
@@ -1277,27 +1278,8 @@ class ChatTelemetryLayer(Generic[OptionsCoT]):
         )
 
         if stream:
-            result_stream = cast(
-                ResponseStream[ChatResponseUpdate, ChatResponse[Any]],
-                super_get_response(
-                    messages=messages,
-                    stream=True,
-                    options=opts,
-                    compaction_strategy=compaction_strategy,
-                    tokenizer=tokenizer,
-                    function_invocation_kwargs=function_invocation_kwargs,
-                    client_kwargs=merged_client_kwargs,
-                ),
-            )
+            span = _start_streaming_span(attributes, OtelAttr.REQUEST_MODEL)
 
-            # Create span directly without trace.use_span() context attachment.
-            # Streaming spans are closed asynchronously in cleanup hooks, which run
-            # in a different async context than creation — using use_span() would
-            # cause "Failed to detach context" errors from OpenTelemetry.
-            operation = attributes.get(OtelAttr.OPERATION, "operation")
-            span_name = attributes.get(OtelAttr.REQUEST_MODEL, "unknown")
-            span = get_tracer().start_span(f"{operation} {span_name}")
-            span.set_attributes(attributes)
             if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED and messages:
                 _capture_messages(
                     span=span,
@@ -1319,6 +1301,24 @@ class ChatTelemetryLayer(Generic[OptionsCoT]):
             def _record_duration() -> None:
                 duration_state["duration"] = perf_counter() - start_time
 
+            try:
+                result_stream = cast(
+                    ResponseStream[ChatResponseUpdate, ChatResponse[Any]],
+                    super_get_response(
+                        messages=messages,
+                        stream=True,
+                        options=opts,
+                        compaction_strategy=compaction_strategy,
+                        tokenizer=tokenizer,
+                        function_invocation_kwargs=function_invocation_kwargs,
+                        client_kwargs=merged_client_kwargs,
+                    ),
+                )
+            except Exception as exception:
+                capture_exception(span=span, exception=exception, timestamp=time_ns())
+                _close_span()
+                raise
+
             async def _finalize_stream() -> None:
                 from ._types import ChatResponse
 
@@ -1357,11 +1357,18 @@ class ChatTelemetryLayer(Generic[OptionsCoT]):
                 finally:
                     _close_span()
 
-            # Register a weak reference callback to close the span if stream is garbage collected
-            # without being consumed. This ensures spans don't leak if users don't consume streams.
-            wrapped_stream: ResponseStream[ChatResponseUpdate, ChatResponse[Any]] = result_stream.with_cleanup_hook(
-                _record_duration
-            ).with_cleanup_hook(_finalize_stream)
+            # The pull context manager attaches the span around each underlying iterator pull so
+            # that child spans created during the pull (e.g. HTTP requests, inner tool execution)
+            # are parented under this chat span. Attach and detach happen in the same async
+            # context as the pull, avoiding cross-context cleanup issues. The weakref finalizer
+            # ensures the span is closed even if the stream is garbage collected without being
+            # consumed.
+            wrapped_stream: ResponseStream[ChatResponseUpdate, ChatResponse[Any]] = (
+                result_stream
+                .with_cleanup_hook(_record_duration)
+                .with_cleanup_hook(_finalize_stream)
+                .with_pull_context_manager(lambda: _activate_span(span))
+            )
             weakref.finalize(wrapped_stream, _close_span)
             return wrapped_stream
 
@@ -1543,23 +1550,8 @@ class AgentTelemetryLayer:
         inner_accumulated_usage_token = INNER_ACCUMULATED_USAGE.set({})
 
         if stream:
-            try:
-                run_result: object = execute()
-                if isinstance(run_result, ResponseStream):
-                    result_stream: ResponseStream[AgentResponseUpdate, AgentResponse[Any]] = run_result  # pyright: ignore[reportUnknownVariableType]
-                elif isinstance(run_result, Awaitable):
-                    result_stream = ResponseStream.from_awaitable(run_result)  # type: ignore[arg-type]  # pyright: ignore[reportArgumentType]
-                else:
-                    raise RuntimeError("Streaming telemetry requires a ResponseStream result.")
-            except Exception:
-                INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.reset(inner_response_telemetry_captured_fields_token)
-                INNER_ACCUMULATED_USAGE.reset(inner_accumulated_usage_token)
-                raise
+            span = _start_streaming_span(attributes, OtelAttr.AGENT_NAME)
 
-            operation = attributes.get(OtelAttr.OPERATION, "operation")
-            span_name = attributes.get(OtelAttr.AGENT_NAME, "unknown")
-            span = get_tracer().start_span(f"{operation} {span_name}")
-            span.set_attributes(attributes)
             if OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED and messages:
                 _capture_messages(
                     span=span,
@@ -1581,6 +1573,21 @@ class AgentTelemetryLayer:
             def _record_duration() -> None:
                 duration_state["duration"] = perf_counter() - start_time
 
+            try:
+                run_result: object = execute()
+                if isinstance(run_result, ResponseStream):
+                    result_stream: ResponseStream[AgentResponseUpdate, AgentResponse[Any]] = run_result  # pyright: ignore[reportUnknownVariableType]
+                elif isinstance(run_result, Awaitable):
+                    result_stream = ResponseStream.from_awaitable(run_result)  # type: ignore[arg-type]  # pyright: ignore[reportArgumentType]
+                else:
+                    raise RuntimeError("Streaming telemetry requires a ResponseStream result.")
+            except Exception as exception:
+                capture_exception(span=span, exception=exception, timestamp=time_ns())
+                INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.reset(inner_response_telemetry_captured_fields_token)
+                INNER_ACCUMULATED_USAGE.reset(inner_accumulated_usage_token)
+                _close_span()
+                raise
+
             async def _finalize_stream() -> None:
                 from ._types import AgentResponse
 
@@ -1620,9 +1627,18 @@ class AgentTelemetryLayer:
                     INNER_ACCUMULATED_USAGE.reset(inner_accumulated_usage_token)
                     _close_span()
 
-            wrapped_stream: ResponseStream[AgentResponseUpdate, AgentResponse[Any]] = result_stream.with_cleanup_hook(
-                _record_duration
-            ).with_cleanup_hook(_finalize_stream)
+            # The pull context manager attaches the span around each underlying iterator pull so
+            # that child spans created during the pull (e.g. inner chat completion spans from the
+            # underlying ChatTelemetryLayer) are parented under this agent invoke span. Attach and
+            # detach happen in the same async context as the pull, avoiding cross-context cleanup
+            # issues. The weakref finalizer ensures the span is closed even if the stream is
+            # garbage collected without being consumed.
+            wrapped_stream: ResponseStream[AgentResponseUpdate, AgentResponse[Any]] = (
+                result_stream
+                .with_cleanup_hook(_record_duration)
+                .with_cleanup_hook(_finalize_stream)
+                .with_pull_context_manager(lambda: _activate_span(span))
+            )
             weakref.finalize(wrapped_stream, _close_span)
             return wrapped_stream
 
@@ -1809,6 +1825,27 @@ def get_function_span(
     )
 
 
+@contextlib.contextmanager
+def _activate_span(span: trace.Span) -> Generator[None]:
+    """Attach ``span`` as the current span in the OpenTelemetry context.
+
+    Designed to be used as a per-pull context manager registered on a
+    ``ResponseStream`` via ``with_pull_context_manager``: it attaches the span
+    before each underlying iterator pull and detaches immediately after, so
+    child spans created during the pull (HTTP clients, inner chat completions,
+    tool execution) are correctly parented under ``span``.
+
+    Because attach and detach happen within the same ``__anext__`` invocation
+    (and therefore the same async task / contextvars context), there is no risk
+    of "Failed to detach context" warnings from cross-context cleanup.
+    """
+    token = otel_context.attach(trace.set_span_in_context(span))
+    try:
+        yield
+    finally:
+        otel_context.detach(token)
+
+
 @contextlib.contextmanager
 def _get_span(
     attributes: dict[str, Any],
@@ -1831,6 +1868,29 @@ def _get_span(
         yield current_span
 
 
+def _start_streaming_span(attributes: dict[str, Any], span_name_attribute: str) -> trace.Span:
+    """Start a non-current span for a streaming operation.
+
+    Unlike :func:`_get_span`, the returned span is not attached to the current
+    OpenTelemetry context. The caller is responsible for:
+
+    - Ending the span via cleanup hooks on the wrapped
+      :class:`~agent_framework._types.ResponseStream`.
+    - Activating the span around each iterator pull via
+      :func:`_activate_span` registered with ``with_pull_context_manager`` so
+      that child spans created during stream production inherit it as parent.
+
+    Streaming spans are closed asynchronously in cleanup hooks that run in a
+    different async context than creation, so attaching the span at creation
+    time would cause "Failed to detach context" errors from OpenTelemetry.
+    """
+    operation = attributes.get(OtelAttr.OPERATION, "operation")
+    span_name = attributes.get(span_name_attribute, "unknown")
+    span = get_tracer().start_span(f"{operation} {span_name}")
+    span.set_attributes(attributes)
+    return span
+
+
 def _get_instructions_from_options(options: Any) -> str | list[str] | None:
     """Extract instructions from options dict."""
     if options is None:

python/packages/core/pyproject.toml

@@ -4,7 +4,7 @@ description = "Microsoft Agent Framework for building AI Agents with Python. Thi
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.2.1"
+version = "1.2.2"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"

python/packages/core/tests/core/test_observability.py

@@ -3313,3 +3313,487 @@ async def test_agent_invoke_span_aggregates_usage_on_max_iterations_exhaustion(s
     # The invoke_agent span must aggregate usage from the in-loop call and the final exhaustion call
     assert agent_span.attributes.get(OtelAttr.INPUT_TOKENS) == 500
     assert agent_span.attributes.get(OtelAttr.OUTPUT_TOKENS) == 100
+
+
+# region Test span nesting (parent-child relationships)
+
+
+@pytest.mark.parametrize("stream", [False, True])
+async def test_chat_span_nested_under_agent_span(span_exporter: InMemorySpanExporter, stream: bool):
+    """The inner chat span must be a child of the outer agent invoke span."""
+
+    class NestedChatClient(ChatTelemetryLayer, BaseChatClient[Any]):
+        def service_url(self):
+            return "https://test.example.com"
+
+        def _inner_get_response(
+            self, *, messages: MutableSequence[Message], stream: bool, options: dict[str, Any], **kwargs: Any
+        ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]:
+            if stream:
+
+                async def _stream() -> AsyncIterable[ChatResponseUpdate]:
+                    yield ChatResponseUpdate(contents=[Content.from_text("Hello")], role="assistant")
+                    yield ChatResponseUpdate(
+                        contents=[Content.from_text(" world")], role="assistant", finish_reason="stop"
+                    )
+
+                def _finalize(updates: Sequence[ChatResponseUpdate]) -> ChatResponse:
+                    return ChatResponse(
+                        messages=[Message(role="assistant", contents=["Hello world"])],
+                        response_id="resp_1",
+                        usage_details=UsageDetails(input_token_count=3, output_token_count=4),
+                        finish_reason="stop",
+                    )
+
+                return ResponseStream(_stream(), finalizer=_finalize)
+
+            async def _get() -> ChatResponse:
+                return ChatResponse(
+                    messages=[Message(role="assistant", contents=["Hello world"])],
+                    response_id="resp_1",
+                    usage_details=UsageDetails(input_token_count=3, output_token_count=4),
+                    finish_reason="stop",
+                )
+
+            return _get()
+
+    agent = Agent(
+        client=NestedChatClient(),
+        id="nested_agent_id",
+        name="nested_agent",
+        default_options={"model": "NestedModel"},
+    )
+
+    span_exporter.clear()
+    if stream:
+        result_stream = agent.run("Test message", stream=True)
+        async for _ in result_stream:
+            pass
+        await result_stream.get_final_response()
+    else:
+        await agent.run("Test message")
+
+    spans = span_exporter.get_finished_spans()
+    assert len(spans) == 2
+
+    span_by_op = {s.attributes[OtelAttr.OPERATION.value]: s for s in spans}
+    agent_span = span_by_op[OtelAttr.AGENT_INVOKE_OPERATION]
+    chat_span = span_by_op[OtelAttr.CHAT_COMPLETION_OPERATION]
+
+    # Agent span has no parent (it is the root)
+    assert agent_span.parent is None
+
+    # Chat span's parent must be the agent span
+    assert chat_span.parent is not None
+    assert chat_span.parent.span_id == agent_span.context.span_id
+    assert chat_span.parent.trace_id == agent_span.context.trace_id
+
+    # Both spans must share the same trace
+    assert chat_span.context.trace_id == agent_span.context.trace_id
+
+
+@pytest.mark.parametrize("stream", [False, True])
+async def test_function_call_spans_nested_under_agent_span(span_exporter: InMemorySpanExporter, stream: bool):
+    """All inner spans (chat completions and execute_tool) must be children of the agent span."""
+    from agent_framework import Content
+    from agent_framework._tools import FunctionInvocationLayer
+
+    @tool(name="get_weather", description="Get the weather for a location")
+    def get_weather(location: str) -> str:
+        return f"The weather in {location} is sunny."
+
+    class NestedToolChatClient(FunctionInvocationLayer, ChatTelemetryLayer, BaseChatClient[Any]):
+        def __init__(self) -> None:
+            super().__init__()
+            self.call_count = 0
+
+        def service_url(self):
+            return "https://test.example.com"
+
+        def _inner_get_response(
+            self, *, messages: MutableSequence[Message], stream: bool, options: dict[str, Any], **kwargs: Any
+        ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]:
+            self.call_count += 1
+            is_first = self.call_count == 1
+
+            if stream:
+
+                async def _stream() -> AsyncIterable[ChatResponseUpdate]:
+                    if is_first:
+                        yield ChatResponseUpdate(
+                            contents=[
+                                Content.from_function_call(
+                                    call_id="call_123",
+                                    name="get_weather",
+                                    arguments='{"location": "Seattle"}',
+                                )
+                            ],
+                            role="assistant",
+                        )
+                    else:
+                        yield ChatResponseUpdate(
+                            contents=[Content.from_text("The weather in Seattle is sunny!")],
+                            role="assistant",
+                            finish_reason="stop",
+                        )
+
+                def _finalize(updates: Sequence[ChatResponseUpdate]) -> ChatResponse:
+                    return ChatResponse.from_updates(updates)
+
+                return ResponseStream(_stream(), finalizer=_finalize)
+
+            async def _get() -> ChatResponse:
+                if is_first:
+                    return ChatResponse(
+                        messages=[
+                            Message(
+                                role="assistant",
+                                contents=[
+                                    Content.from_function_call(
+                                        call_id="call_123",
+                                        name="get_weather",
+                                        arguments='{"location": "Seattle"}',
+                                    )
+                                ],
+                            )
+                        ],
+                    )
+                return ChatResponse(
+                    messages=[Message(role="assistant", contents=["The weather in Seattle is sunny!"])],
+                    finish_reason="stop",
+                )
+
+            return _get()
+
+    agent = Agent(
+        client=NestedToolChatClient(),
+        id="tool_agent_id",
+        name="tool_agent",
+        default_options={"model": "ToolModel", "tools": [get_weather], "tool_choice": "auto"},
+    )
+
+    span_exporter.clear()
+    if stream:
+        result_stream = agent.run("What's the weather in Seattle?", stream=True)
+        async for _ in result_stream:
+            pass
+        await result_stream.get_final_response()
+    else:
+        await agent.run("What's the weather in Seattle?")
+
+    spans = span_exporter.get_finished_spans()
+
+    invoke_spans = [s for s in spans if s.attributes.get(OtelAttr.OPERATION.value) == OtelAttr.AGENT_INVOKE_OPERATION]
+    chat_spans = [s for s in spans if s.attributes.get(OtelAttr.OPERATION.value) == OtelAttr.CHAT_COMPLETION_OPERATION]
+    tool_spans = [s for s in spans if s.attributes.get(OtelAttr.OPERATION.value) == OtelAttr.TOOL_EXECUTION_OPERATION]
+
+    assert len(invoke_spans) == 1, f"Expected 1 invoke_agent span, got {len(invoke_spans)}"
+    assert len(chat_spans) == 2, f"Expected 2 chat spans, got {len(chat_spans)}"
+    assert len(tool_spans) == 1, f"Expected 1 execute_tool span, got {len(tool_spans)}"
+
+    agent_span = invoke_spans[0]
+    assert agent_span.parent is None
+
+    # All inner spans must be parented under the agent invoke span
+    for inner in (*chat_spans, *tool_spans):
+        assert inner.parent is not None, f"Span {inner.name} has no parent"
+        assert inner.parent.span_id == agent_span.context.span_id, (
+            f"Span {inner.name} parent={inner.parent.span_id} != agent={agent_span.context.span_id}"
+        )
+        assert inner.context.trace_id == agent_span.context.trace_id
+
+
+@pytest.mark.parametrize("stream", [False, True])
+async def test_chat_span_nested_under_explicit_outer_span(
+    span_exporter: InMemorySpanExporter, mock_chat_client, stream: bool
+):
+    """Chat telemetry spans (including streaming) must inherit a user-provided outer span as parent."""
+    from agent_framework.observability import get_tracer
+
+    client = mock_chat_client()
+    span_exporter.clear()
+
+    tracer = get_tracer()
+    with tracer.start_as_current_span("outer") as outer_span:
+        outer_ctx = outer_span.get_span_context()
+        if stream:
+            stream_obj = client.get_response(
+                stream=True, messages=[Message(role="user", contents=["Test"])], options={"model": "Test"}
+            )
+            async for _ in stream_obj:
+                pass
+            await stream_obj.get_final_response()
+        else:
+            await client.get_response(messages=[Message(role="user", contents=["Test"])], options={"model": "Test"})
+
+    spans = span_exporter.get_finished_spans()
+    chat_spans = [s for s in spans if s.attributes.get(OtelAttr.OPERATION.value) == OtelAttr.CHAT_COMPLETION_OPERATION]
+    assert len(chat_spans) == 1
+    chat_span = chat_spans[0]
+
+    assert chat_span.parent is not None
+    assert chat_span.parent.span_id == outer_ctx.span_id
+    assert chat_span.context.trace_id == outer_ctx.trace_id
+
+
+@pytest.mark.parametrize("stream", [False, True])
+async def test_http_span_nested_under_chat_span(span_exporter: InMemorySpanExporter, stream: bool):
+    """A span created inside ``_inner_get_response`` (e.g. an HTTP client call to the LLM provider)
+    must be parented under the chat completion span.
+
+    This validates that the chat span context is active while the inner client implementation
+    runs, both for non-streaming responses and while streaming updates are being pulled.
+    """
+    from agent_framework.observability import get_tracer
+
+    tracer = get_tracer()
+
+    class HttpEmittingClient(ChatTelemetryLayer, BaseChatClient[Any]):
+        def service_url(self):
+            return "https://test.example.com"
+
+        def _inner_get_response(
+            self, *, messages: MutableSequence[Message], stream: bool, options: dict[str, Any], **kwargs: Any
+        ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]:
+            if stream:
+
+                async def _stream() -> AsyncIterable[ChatResponseUpdate]:
+                    # Simulate an HTTP request to the model provider while producing the stream.
+                    with tracer.start_as_current_span("HTTP POST"):
+                        pass
+                    yield ChatResponseUpdate(contents=[Content.from_text("hi")], role="assistant", finish_reason="stop")
+
+                def _finalize(updates: Sequence[ChatResponseUpdate]) -> ChatResponse:
+                    return ChatResponse.from_updates(updates)
+
+                return ResponseStream(_stream(), finalizer=_finalize)
+
+            async def _get() -> ChatResponse:
+                # Simulate an HTTP request to the model provider during the call.
+                with tracer.start_as_current_span("HTTP POST"):
+                    pass
+                return ChatResponse(
+                    messages=[Message(role="assistant", contents=["done"])],
+                    usage_details=UsageDetails(input_token_count=1, output_token_count=1),
+                )
+
+            return _get()
+
+    span_exporter.clear()
+    client = HttpEmittingClient()
+    if stream:
+        result_stream = client.get_response(
+            stream=True, messages=[Message(role="user", contents=["Test"])], options={"model": "Test"}
+        )
+        async for _ in result_stream:
+            pass
+        await result_stream.get_final_response()
+    else:
+        await client.get_response(messages=[Message(role="user", contents=["Test"])], options={"model": "Test"})
+
+    spans = span_exporter.get_finished_spans()
+    chat_spans = [s for s in spans if s.attributes.get(OtelAttr.OPERATION.value) == OtelAttr.CHAT_COMPLETION_OPERATION]
+    http_spans = [s for s in spans if s.name == "HTTP POST"]
+    assert len(chat_spans) == 1
+    assert len(http_spans) == 1
+
+    chat_span = chat_spans[0]
+    http_span = http_spans[0]
+
+    assert http_span.parent is not None
+    assert http_span.parent.span_id == chat_span.context.span_id
+    assert http_span.context.trace_id == chat_span.context.trace_id
+
+
+# region Test ResponseStream.with_pull_context_manager
+
+
+async def test_with_pull_context_manager_enters_and_exits_per_pull():
+    """The registered factory is entered and exited symmetrically around each iterator pull."""
+    import contextlib
+
+    events: list[str] = []
+
+    @contextlib.contextmanager
+    def cm():
+        events.append("enter")
+        try:
+            yield
+        finally:
+            events.append("exit")
+
+    async def src() -> AsyncIterable[int]:
+        yield 1
+        yield 2
+
+    stream: ResponseStream[int, list[int]] = ResponseStream(src(), finalizer=lambda updates: list(updates))
+    stream.with_pull_context_manager(cm)
+
+    pulled = [u async for u in stream]
+
+    assert pulled == [1, 2]
+    # Enter/exit must be balanced and there must be at least one pair per yielded update.
+    assert events.count("enter") == events.count("exit")
+    assert events.count("enter") >= 2
+    # Verify symmetric ordering (no overlapping pairs).
+    for i in range(0, len(events), 2):
+        assert events[i] == "enter"
+        assert events[i + 1] == "exit"
+
+
+async def test_with_pull_context_manager_exits_on_iteration_error():
+    """The pull context is exited even when the underlying stream raises mid-iteration."""
+    import contextlib
+
+    events: list[str] = []
+
+    @contextlib.contextmanager
+    def cm():
+        events.append("enter")
+        try:
+            yield
+        finally:
+            events.append("exit")
+
+    async def src() -> AsyncIterable[int]:
+        yield 1
+        raise RuntimeError("boom")
+
+    stream: ResponseStream[int, list[int]] = ResponseStream(src(), finalizer=lambda updates: list(updates))
+    stream.with_pull_context_manager(cm)
+
+    with pytest.raises(RuntimeError, match="boom"):
+        async for _ in stream:
+            pass
+
+    # Enter/exit balanced even on the failing pull.
+    assert events.count("enter") == events.count("exit")
+    assert events.count("enter") >= 2
+
+
+async def test_with_pull_context_manager_wraps_stream_resolution_via_await():
+    """Awaiting a ``from_awaitable`` stream resolves the inner stream under the pull contexts."""
+    import contextlib
+
+    events: list[str] = []
+
+    @contextlib.contextmanager
+    def cm():
+        events.append("enter")
+        try:
+            yield
+        finally:
+            events.append("exit")
+
+    async def inner() -> AsyncIterable[int]:
+        yield 1
+
+    async def make_stream() -> ResponseStream[int, list[int]]:
+        # Record that we resolve while a pull context is active.
+        events.append("resolving")
+        return ResponseStream(inner(), finalizer=lambda updates: list(updates))
+
+    stream: ResponseStream[int, list[int]] = ResponseStream.from_awaitable(make_stream())
+    stream.with_pull_context_manager(cm)
+
+    await stream  # Triggers _resolve_stream_with_pull_contexts via __await__
+
+    assert "resolving" in events
+    resolve_index = events.index("resolving")
+    assert events[resolve_index - 1] == "enter"  # Pull context active during resolution
+
+
+# region Test streaming telemetry error paths
+
+
+@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True)
+async def test_chat_streaming_super_failure_closes_span(span_exporter: InMemorySpanExporter, enable_sensitive_data):
+    """If the underlying client raises synchronously when constructing the stream, the chat
+    span is ended and the exception is recorded (no span leak)."""
+
+    class FailingClient(ChatTelemetryLayer, BaseChatClient[Any]):
+        def service_url(self):
+            return "https://test.example.com"
+
+        def _inner_get_response(
+            self, *, messages: MutableSequence[Message], stream: bool, options: dict[str, Any], **kwargs: Any
+        ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]:
+            raise RuntimeError("inner failed")
+
+    span_exporter.clear()
+    client = FailingClient()
+    with pytest.raises(RuntimeError, match="inner failed"):
+        client.get_response(stream=True, messages=[Message(role="user", contents=["Test"])], options={"model": "Test"})
+
+    spans = span_exporter.get_finished_spans()
+    chat_spans = [s for s in spans if s.attributes.get(OtelAttr.OPERATION.value) == OtelAttr.CHAT_COMPLETION_OPERATION]
+    assert len(chat_spans) == 1
+    assert chat_spans[0].status.status_code == StatusCode.ERROR
+
+
+@pytest.mark.parametrize("enable_sensitive_data", [True], indirect=True)
+async def test_agent_streaming_execute_failure_closes_span_and_resets_contextvars(
+    span_exporter: InMemorySpanExporter, enable_sensitive_data
+):
+    """If ``execute()`` raises synchronously during streaming agent invocation, the agent span is
+    ended, the exception is recorded, and the telemetry contextvars are reset."""
+    from agent_framework.observability import (
+        INNER_ACCUMULATED_USAGE,
+        INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS,
+    )
+
+    class _FailingExecuteAgent:
+        AGENT_PROVIDER_NAME = "test_provider"
+
+        def __init__(self):
+            self._id = "failing_execute"
+            self._name = "Failing Execute"
+            self._description = "Agent whose stream call raises synchronously"
+            self._default_options: dict[str, Any] = {}
+
+        @property
+        def id(self):
+            return self._id
+
+        @property
+        def name(self):
+            return self._name
+
+        @property
+        def description(self):
+            return self._description
+
+        @property
+        def default_options(self):
+            return self._default_options
+
+        def run(self, messages=None, *, stream: bool = False, session=None, **kwargs):
+            if stream:
+                raise RuntimeError("execute failed")
+            raise NotImplementedError
+
+    class FailingExecuteAgent(AgentTelemetryLayer, _FailingExecuteAgent):
+        pass
+
+    # Sentinel values to detect that contextvars were reset to their pre-call state.
+    sentinel_fields: set[str] = set()
+    sentinel_usage: dict[str, Any] = {}
+    fields_token = INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.set(sentinel_fields)
+    usage_token = INNER_ACCUMULATED_USAGE.set(sentinel_usage)
+    try:
+        agent = FailingExecuteAgent()
+        span_exporter.clear()
+        with pytest.raises(RuntimeError, match="execute failed"):
+            agent.run(messages="Hello", stream=True)
+
+        # Contextvars must be back to the sentinel values registered before the call.
+        assert INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.get() is sentinel_fields
+        assert INNER_ACCUMULATED_USAGE.get() is sentinel_usage
+    finally:
+        INNER_ACCUMULATED_USAGE.reset(usage_token)
+        INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.reset(fields_token)
+
+    spans = span_exporter.get_finished_spans()
+    agent_spans = [s for s in spans if s.attributes.get(OtelAttr.OPERATION.value) == OtelAttr.AGENT_INVOKE_OPERATION]
+    assert len(agent_spans) == 1
+    assert agent_spans[0].status.status_code == StatusCode.ERROR

python/packages/core/tests/core/test_types.py

@@ -1087,16 +1087,20 @@ def test_chat_tool_mode():
     required_any: ToolMode = {"mode": "required"}
     required_mode: ToolMode = {"mode": "required", "required_function_name": "example_function"}
     none_mode: ToolMode = {"mode": "none"}
+    allowed_mode: ToolMode = {"mode": "auto", "allowed_tools": ["get_weather", "search_docs"]}
 
     # Check the type and content
     assert auto_mode["mode"] == "auto"
     assert "required_function_name" not in auto_mode
+    assert "allowed_tools" not in auto_mode
     assert required_any["mode"] == "required"
     assert "required_function_name" not in required_any
     assert required_mode["mode"] == "required"
     assert required_mode["required_function_name"] == "example_function"
     assert none_mode["mode"] == "none"
     assert "required_function_name" not in none_mode
+    assert allowed_mode["mode"] == "auto"
+    assert allowed_mode["allowed_tools"] == ["get_weather", "search_docs"]
 
     # equality of dicts
     assert {"mode": "required", "required_function_name": "example_function"} == {
@@ -1154,6 +1158,45 @@ def test_chat_options_tool_choice_validation():
     with raises(ContentError):
         validate_tool_mode({"mode": "auto", "required_function_name": "should_not_be_here"})
 
+    # Valid allowed_tools
+    assert validate_tool_mode({"mode": "auto", "allowed_tools": ["get_weather"]}) == {
+        "mode": "auto",
+        "allowed_tools": ["get_weather"],
+    }
+    assert validate_tool_mode({"mode": "auto", "allowed_tools": ["get_weather", "search_docs"]}) == {
+        "mode": "auto",
+        "allowed_tools": ["get_weather", "search_docs"],
+    }
+
+    # allowed_tools valid with required mode
+    assert validate_tool_mode({"mode": "required", "allowed_tools": ["get_weather"]}) == {
+        "mode": "required",
+        "allowed_tools": ["get_weather"],
+    }
+
+    # allowed_tools invalid with none mode
+    with raises(ContentError):
+        validate_tool_mode({"mode": "none", "allowed_tools": ["get_weather"]})
+
+    # allowed_tools must be a non-string sequence of strings
+    with raises(ContentError):
+        validate_tool_mode({"mode": "auto", "allowed_tools": "get_weather"})
+    with raises(ContentError):
+        validate_tool_mode({"mode": "auto", "allowed_tools": 123})
+    with raises(ContentError):
+        validate_tool_mode({"mode": "auto", "allowed_tools": ["get_weather", 123]})
+
+    # Empty list is valid (caller explicitly allows no tools)
+    assert validate_tool_mode({"mode": "auto", "allowed_tools": []}) == {
+        "mode": "auto",
+        "allowed_tools": [],
+    }
+
+    # Tuple is normalized to list
+    result = validate_tool_mode({"mode": "auto", "allowed_tools": ("get_weather",)})
+    assert result is not None
+    assert result["allowed_tools"] == ["get_weather"]
+
 
 def test_chat_options_merge(tool_tool, ai_tool) -> None:
     """Test merge_chat_options utility function."""

python/packages/core/tests/workflow/test_workflow.py

@@ -488,8 +488,13 @@ class StateTrackingExecutor(Executor):
         await ctx.yield_output(existing_messages.copy())  # type: ignore
 
 
-async def test_workflow_multiple_runs_no_state_collision():
-    """Test that running the same workflow instance multiple times doesn't have state collision."""
+async def test_workflow_multiple_runs_preserve_state():
+    """Test that running the same workflow instance multiple times preserves shared state.
+
+    State preservation is the new default - calling ``Workflow.run`` repeatedly
+    on the same instance behaves like a chat agent maintaining memory across
+    turns. Callers that want fresh state should rebuild the Workflow.
+    """
     with tempfile.TemporaryDirectory() as temp_dir:
         storage = FileCheckpointStorage(temp_dir)
 
@@ -503,29 +508,45 @@ async def test_workflow_multiple_runs_no_state_collision():
             .build()
         )
 
-        # Run 1: Should only see messages from run 1
+        # Run 1: Single record from run 1
         result1 = await workflow.run(StateTrackingMessage(data="message1", run_id="run1"))
         assert result1.get_final_state() == WorkflowRunState.IDLE
         outputs1 = result1.get_outputs()
         assert outputs1[0] == ["run1:message1"]
 
-        # Run 2: Should only see messages from run 2, not run 1
+        # Run 2: State from run 1 persists; run 2's record appends.
         result2 = await workflow.run(StateTrackingMessage(data="message2", run_id="run2"))
         assert result2.get_final_state() == WorkflowRunState.IDLE
         outputs2 = result2.get_outputs()
-        assert outputs2[0] == ["run2:message2"]  # Should NOT contain run1 data
+        assert outputs2[0] == ["run1:message1", "run2:message2"]
 
-        # Run 3: Should only see messages from run 3
+        # Run 3: Same - all three accumulate.
         result3 = await workflow.run(StateTrackingMessage(data="message3", run_id="run3"))
         assert result3.get_final_state() == WorkflowRunState.IDLE
         outputs3 = result3.get_outputs()
-        assert outputs3[0] == ["run3:message3"]  # Should NOT contain run1 or run2 data
+        assert outputs3[0] == ["run1:message1", "run2:message2", "run3:message3"]
 
-        # Verify that each run only processed its own message
-        # This confirms that the checkpointable context properly resets between runs
-        assert outputs1[0] != outputs2[0]
-        assert outputs2[0] != outputs3[0]
-        assert outputs1[0] != outputs3[0]
+
+async def test_workflow_multiple_runs_no_state_collision_after_rebuild():
+    """Rebuilding the Workflow gives a fresh shared-state slate."""
+    with tempfile.TemporaryDirectory() as temp_dir:
+        storage = FileCheckpointStorage(temp_dir)
+
+        def _build():
+            executor = StateTrackingExecutor(id="state_executor")
+            return (
+                WorkflowBuilder(start_executor=executor, checkpoint_storage=storage)
+                .add_edge(executor, executor)
+                .build()
+            )
+
+        wf1 = _build()
+        result1 = await wf1.run(StateTrackingMessage(data="message1", run_id="run1"))
+        assert result1.get_outputs()[0] == ["run1:message1"]
+
+        wf2 = _build()
+        result2 = await wf2.run(StateTrackingMessage(data="message2", run_id="run2"))
+        assert result2.get_outputs()[0] == ["run2:message2"]
 
 
 async def test_workflow_checkpoint_runtime_only_configuration(
@@ -932,6 +953,31 @@ async def test_agent_streaming_vs_non_streaming() -> None:
     assert accumulated_text == "Hello World", f"Expected 'Hello World', got '{accumulated_text}'"
 
 
+async def test_workflow_run_inflight_messages_guard(simple_executor: Executor) -> None:
+    """``run(message=...)`` must reject in-flight executor messages from a prior run.
+
+    Workflows preserve state and pending messages across :meth:`Workflow.run`
+    calls. If a prior run aborted before the runner drained those pending
+    messages (e.g. it raised :class:`WorkflowConvergenceException`), the next
+    fresh-message call should fail loudly instead of silently mixing the
+    leftover messages with the new turn. The supported recovery path is to
+    resume from a checkpoint; there is no in-process recovery hatch.
+    """
+    workflow = WorkflowBuilder(start_executor=simple_executor).add_edge(simple_executor, simple_executor).build()
+    test_message = WorkflowMessage(data="test", source_id="test", target_id=None)
+
+    # Simulate an aborted prior run by leaving a message in the runner context.
+    workflow._runner.context._messages["test"] = [test_message]
+    assert await workflow._runner.context.has_messages()
+
+    with pytest.raises(RuntimeError, match="in-flight executor messages"):
+        await workflow.run(test_message)
+
+    with pytest.raises(RuntimeError, match="in-flight executor messages"):
+        async for _ in workflow.run(test_message, stream=True):
+            pass
+
+
 async def test_workflow_run_parameter_validation(simple_executor: Executor) -> None:
     """Test that stream properly validate parameter combinations."""
     workflow = WorkflowBuilder(start_executor=simple_executor).add_edge(simple_executor, simple_executor).build()
@@ -942,13 +988,15 @@ async def test_workflow_run_parameter_validation(simple_executor: Executor) -> N
     result = await workflow.run(test_message)
     assert result.get_final_state() == WorkflowRunState.IDLE
 
-    # Invalid: both message and checkpoint_id
+    # Invalid: message + checkpoint_id (mutually exclusive). Multi-turn
+    # state preservation is handled by Workflow.run preserving state across
+    # calls, so the host pattern is two separate calls (restore-then-run),
+    # not a single combined call.
     with pytest.raises(ValueError, match="Cannot provide both 'message' and 'checkpoint_id'"):
-        await workflow.run(test_message, checkpoint_id="fake_id")
+        await workflow.run(test_message, checkpoint_id="some-checkpoint")
 
-    # Invalid: both message and checkpoint_id (streaming)
     with pytest.raises(ValueError, match="Cannot provide both 'message' and 'checkpoint_id'"):
-        async for _ in workflow.run(test_message, checkpoint_id="fake_id", stream=True):
+        async for _ in workflow.run(test_message, checkpoint_id="some-checkpoint", stream=True):
             pass
 
     # Invalid: none of message or checkpoint_id

python/packages/declarative/AGENTS.md

@@ -8,7 +8,8 @@ YAML/JSON-based declarative agent and workflow definitions.
 - **`WorkflowFactory`** - Creates workflows from declarative definitions
 - **`WorkflowState`** - State management for declarative workflows
 - **`ProviderTypeMapping`** - Maps provider types to implementations
-- **`DeclarativeLoaderError`** / **`ProviderLookupError`** - Error types
+- **`HttpRequestHandler`** / **`DefaultHttpRequestHandler`** - Pluggable HTTP transport for the `HttpRequestAction` declarative action (configured via `WorkflowFactory(http_request_handler=...)`)
+- **`DeclarativeLoaderError`** / **`ProviderLookupError`** / **`DeclarativeWorkflowError`** / **`DeclarativeActionError`** - Error types
 
 ## External Input Handling
 

python/packages/declarative/agent_framework_declarative/__init__.py

@@ -6,9 +6,14 @@ from ._loader import AgentFactory, DeclarativeLoaderError, ProviderLookupError,
 from ._workflows import (
     AgentExternalInputRequest,
     AgentExternalInputResponse,
+    DeclarativeActionError,
     DeclarativeWorkflowError,
+    DefaultHttpRequestHandler,
     ExternalInputRequest,
     ExternalInputResponse,
+    HttpRequestHandler,
+    HttpRequestInfo,
+    HttpRequestResult,
     WorkflowFactory,
     WorkflowState,
 )
@@ -22,10 +27,15 @@ __all__ = [
     "AgentExternalInputRequest",
     "AgentExternalInputResponse",
     "AgentFactory",
+    "DeclarativeActionError",
     "DeclarativeLoaderError",
     "DeclarativeWorkflowError",
+    "DefaultHttpRequestHandler",
     "ExternalInputRequest",
     "ExternalInputResponse",
+    "HttpRequestHandler",
+    "HttpRequestInfo",
+    "HttpRequestResult",
     "ProviderLookupError",
     "ProviderTypeMapping",
     "WorkflowFactory",

python/packages/declarative/agent_framework_declarative/_workflows/__init__.py

@@ -25,6 +25,7 @@ from ._declarative_base import (
     LoopIterationResult,
 )
 from ._declarative_builder import ALL_ACTION_EXECUTORS, DeclarativeWorkflowBuilder
+from ._errors import DeclarativeActionError, DeclarativeWorkflowError
 from ._executors_agents import (
     AGENT_ACTION_EXECUTORS,
     AGENT_REGISTRY_KEY,
@@ -67,6 +68,10 @@ from ._executors_external_input import (
     RequestExternalInputExecutor,
     WaitForInputExecutor,
 )
+from ._executors_http import (
+    HTTP_ACTION_EXECUTORS,
+    HttpRequestActionExecutor,
+)
 from ._executors_tools import (
     FUNCTION_TOOL_REGISTRY_KEY,
     TOOL_ACTION_EXECUTORS,
@@ -78,7 +83,13 @@ from ._executors_tools import (
     ToolApprovalState,
     ToolInvocationResult,
 )
-from ._factory import DeclarativeWorkflowError, WorkflowFactory
+from ._factory import WorkflowFactory
+from ._http_handler import (
+    DefaultHttpRequestHandler,
+    HttpRequestHandler,
+    HttpRequestInfo,
+    HttpRequestResult,
+)
 from ._state import WorkflowState
 
 __all__ = [
@@ -90,6 +101,7 @@ __all__ = [
     "DECLARATIVE_STATE_KEY",
     "EXTERNAL_INPUT_EXECUTORS",
     "FUNCTION_TOOL_REGISTRY_KEY",
+    "HTTP_ACTION_EXECUTORS",
     "TOOL_ACTION_EXECUTORS",
     "TOOL_APPROVAL_STATE_KEY",
     "TOOL_REGISTRY_KEY",
@@ -106,12 +118,14 @@ __all__ = [
     "ContinueLoopExecutor",
     "ConversationData",
     "CreateConversationExecutor",
+    "DeclarativeActionError",
     "DeclarativeActionExecutor",
     "DeclarativeMessage",
     "DeclarativeStateData",
     "DeclarativeWorkflowBuilder",
     "DeclarativeWorkflowError",
     "DeclarativeWorkflowState",
+    "DefaultHttpRequestHandler",
     "EmitEventExecutor",
     "EndConversationExecutor",
     "EndWorkflowExecutor",
@@ -120,6 +134,10 @@ __all__ = [
     "ExternalLoopState",
     "ForeachInitExecutor",
     "ForeachNextExecutor",
+    "HttpRequestActionExecutor",
+    "HttpRequestHandler",
+    "HttpRequestInfo",
+    "HttpRequestResult",
     "InvokeAzureAgentExecutor",
     "InvokeFunctionToolExecutor",
     "JoinExecutor",

python/packages/declarative/agent_framework_declarative/_workflows/_declarative_base.py

@@ -32,10 +32,12 @@ import uuid
 from collections.abc import Mapping
 from dataclasses import dataclass
 from decimal import Decimal as _Decimal
+from enum import Enum
 from typing import Any, Literal, cast
 
 from agent_framework import (
     Executor,
+    Message,
     WorkflowContext,
 )
 from agent_framework._workflows._state import State
@@ -120,7 +122,20 @@ def _make_powerfx_safe(value: Any) -> Any:
     Returns:
         A PowerFx-safe representation of the value
     """
-    if value is None or isinstance(value, _POWERFX_SAFE_TYPES):
+    if value is None:
+        return value
+
+    # Enum coercion must run BEFORE the primitive type check: many MAF
+    # enums (e.g. MessageRole) are ``str``-subclass enums, so they pass
+    # ``isinstance(v, str)`` but pythonnet refuses to convert them to
+    # ``System.String`` and raises ``'MessageRole' value cannot be
+    # converted to System.<X>'`` for every PowerFx primitive type. Reduce
+    # to the underlying value (or its string form) so PowerFx sees a
+    # plain ``str``/``int``.
+    if isinstance(value, Enum):
+        return _make_powerfx_safe(value.value)
+
+    if isinstance(value, _POWERFX_SAFE_TYPES):
         return value
 
     if isinstance(value, dict):
@@ -197,6 +212,16 @@ class DeclarativeWorkflowState:
             result = self._state.get(DECLARATIVE_STATE_KEY)
         return cast(DeclarativeStateData, result)
 
+    def is_initialized(self) -> bool:
+        """Return True when declarative state has been initialized.
+
+        Useful for distinguishing a fresh start from a continuation: when
+        Workflow state preserves data across run() calls (multi-turn
+        scenarios), the start executor needs to avoid calling initialize()
+        and clobbering the prior turn's Conversation/Local/System data.
+        """
+        return self._state.get(DECLARATIVE_STATE_KEY) is not None
+
     def set_state_data(self, data: DeclarativeStateData) -> None:
         """Set the full state data dict in state."""
         self._state.set(DECLARATIVE_STATE_KEY, data)
@@ -873,6 +898,20 @@ class DeclarativeActionExecutor(Executor):
         Follows .NET's DefaultTransform pattern - accepts any input type:
         - dict/Mapping: Used directly as workflow.inputs
         - str: Converted to {"input": value}
+        - list[Message]: Treated as the agent-facing message contract
+          (e.g. from WorkflowAgent / as_agent()). The prior conversation
+          history is stored in ``Conversation.messages``/
+          ``Conversation.history`` and mirrored to
+          ``System.conversations.{id}.messages`` so workflows that
+          reference ``=Conversation.messages`` (e.g. InvokeAzureAgent) see
+          assistant turns and other earlier messages, including non-text
+          content. At the start of a turn this history excludes the current
+          user message; that message's text is instead used as the string
+          input (``Inputs.input``) and surfaced via ``System.LastMessage*``
+          for backward compatibility with simple text-only workflows. Agent
+          executors are responsible for appending the current user message
+          to ``Conversation.messages`` immediately before invoking the
+          inner agent.
         - DeclarativeMessage: Internal message, no initialization needed
         - Any other type: Converted via str() to {"input": str(value)}
 
@@ -888,6 +927,100 @@ class DeclarativeActionExecutor(Executor):
         if isinstance(trigger, dict):
             # Structured inputs - use directly
             state.initialize(trigger)  # type: ignore
+        elif isinstance(trigger, list) and all(isinstance(m, Message) for m in trigger):  # pyright: ignore[reportUnknownVariableType]
+            # list[Message] (e.g. from WorkflowAgent / as_agent()).
+            messages_list = cast(list[Message], trigger)
+
+            # Detect continuation: if the workflow's shared state already
+            # carries declarative data from a prior turn (because the host
+            # restored a checkpoint and dispatched this run with
+            # reset_context=False), we MUST NOT call state.initialize() -
+            # that would wipe Conversation.messages, Local.*, System.* etc.
+            # Instead, treat the trigger as the new turn's user input only:
+            # update Inputs.input, append the new user message to existing
+            # Conversation history, and refresh System.LastMessage*.
+            #
+            # Continuation = declarative state already exists in the workflow's
+            # shared state (either left over in-memory from a prior turn on
+            # the same instance, or restored from a checkpoint just before
+            # this run). In that case state.initialize() would wipe Local.*,
+            # System.*, Conversation.* etc., destroying the cross-turn
+            # context we're trying to preserve.
+            is_continuation = state.is_initialized()
+
+            # Locate the trailing user message in the trigger.
+            last_user_index = -1
+            for idx in range(len(messages_list) - 1, -1, -1):
+                if str(messages_list[idx].role).lower() == "user":
+                    last_user_index = idx
+                    break
+
+            if last_user_index >= 0:
+                last_user_msg = messages_list[last_user_index]
+                last_user_text = last_user_msg.text or ""
+                last_user_id = getattr(last_user_msg, "message_id", "") or ""
+                history_messages = messages_list[:last_user_index] + messages_list[last_user_index + 1 :]
+            else:
+                history_messages = list(messages_list)
+                tail = messages_list[-1] if messages_list else None
+                last_user_text = (tail.text or "") if tail is not None else ""
+                last_user_id = getattr(tail, "message_id", "") or "" if tail is not None else ""
+
+            if is_continuation:
+                # Continuation turn: keep prior Conversation.messages intact.
+                # Refresh inputs and surface the new user message via the
+                # System.LastMessage* fields. We deliberately do NOT append
+                # the new user message to Conversation.messages here: agent
+                # executors append the live user input themselves before
+                # invoking the inner agent (matching the first-turn
+                # contract where Conversation.messages holds prior turns
+                # only).
+                #
+                # Note: ``state.set("Inputs.input", ...)`` would route to
+                # the Custom namespace (Inputs is not a recognized top-level
+                # writable namespace - see DeclarativeWorkflowState.set).
+                # PowerFx expressions like ``=Workflow.Inputs.input`` /
+                # ``=inputs.input`` read state_data["Inputs"] directly, so
+                # we update that dict in place via get_state_data /
+                # set_state_data.
+                state_data = state.get_state_data()
+                inputs_dict = state_data.get("Inputs")
+                if not isinstance(inputs_dict, dict):
+                    inputs_dict = {}
+                    state_data["Inputs"] = inputs_dict
+                inputs_dict["input"] = last_user_text
+                state.set_state_data(state_data)
+                # Trailing non-user messages (e.g. tool results) sandwiched
+                # before the new user message in the trigger are still
+                # appended so later actions see them.
+                for msg in history_messages:
+                    state.append("Conversation.messages", msg)
+                    state.append("Conversation.history", msg)
+                conversation_id = state.get("System.ConversationId")
+                if conversation_id:
+                    conv_path = f"System.conversations.{conversation_id}.messages"
+                    for msg in history_messages:
+                        state.append(conv_path, msg)
+                state.set("System.LastMessage", {"Text": last_user_text, "Id": last_user_id})
+                state.set("System.LastMessageText", last_user_text)
+                state.set("System.LastMessageId", last_user_id)
+            else:
+                # First turn: full initialization.
+                state.initialize({"input": last_user_text})
+
+                for msg in history_messages:
+                    state.append("Conversation.messages", msg)
+                    state.append("Conversation.history", msg)
+
+                conversation_id = state.get("System.ConversationId")
+                if conversation_id:
+                    conv_path = f"System.conversations.{conversation_id}.messages"
+                    for msg in history_messages:
+                        state.append(conv_path, msg)
+
+                state.set("System.LastMessage", {"Text": last_user_text, "Id": last_user_id})
+                state.set("System.LastMessageText", last_user_text)
+                state.set("System.LastMessageId", last_user_id)
         elif isinstance(trigger, str):
             # String input - wrap in dict and populate System.LastMessage.Text
             # so YAML expressions like =System.LastMessage.Text see the user input
@@ -895,10 +1028,11 @@ class DeclarativeActionExecutor(Executor):
             state.set("System.LastMessage", {"Text": trigger, "Id": ""})
             state.set("System.LastMessageText", trigger)
         elif not isinstance(
-            trigger, (ActionTrigger, ActionComplete, ConditionResult, LoopIterationResult, LoopControl)
+            trigger,
+            (ActionTrigger, ActionComplete, ConditionResult, LoopIterationResult, LoopControl),  # pyright: ignore[reportUnknownArgumentType]
         ):
             # Any other type - convert to string like .NET's DefaultTransform
-            input_str = str(trigger)
+            input_str = str(cast(Any, trigger))
             state.initialize({"input": input_str})
             state.set("System.LastMessage", {"Text": input_str, "Id": ""})
             state.set("System.LastMessageText", input_str)

python/packages/declarative/agent_framework_declarative/_workflows/_declarative_builder.py

@@ -26,6 +26,7 @@ from ._declarative_base import (
     DeclarativeActionExecutor,
     LoopIterationResult,
 )
+from ._errors import DeclarativeWorkflowError
 from ._executors_agents import AGENT_ACTION_EXECUTORS, InvokeAzureAgentExecutor
 from ._executors_basic import BASIC_ACTION_EXECUTORS
 from ._executors_control_flow import (
@@ -39,7 +40,9 @@ from ._executors_control_flow import (
     SwitchEvaluatorExecutor,
 )
 from ._executors_external_input import EXTERNAL_INPUT_EXECUTORS
+from ._executors_http import HTTP_ACTION_EXECUTORS, HttpRequestActionExecutor
 from ._executors_tools import TOOL_ACTION_EXECUTORS, InvokeFunctionToolExecutor
+from ._http_handler import HttpRequestHandler
 
 logger = logging.getLogger(__name__)
 
@@ -51,6 +54,7 @@ ALL_ACTION_EXECUTORS = {
     **AGENT_ACTION_EXECUTORS,
     **EXTERNAL_INPUT_EXECUTORS,
     **TOOL_ACTION_EXECUTORS,
+    **HTTP_ACTION_EXECUTORS,
 }
 
 # Action kinds that terminate control flow (no fall-through to successor)
@@ -85,6 +89,7 @@ ACTION_REQUIRED_FIELDS: dict[str, list[str]] = {
     "WaitForHumanInput": ["variable"],
     "EmitEvent": ["event"],
     "InvokeFunctionTool": ["functionName"],
+    "HttpRequestAction": ["url"],
 }
 
 # Alternate field names that satisfy required field requirements
@@ -129,6 +134,7 @@ class DeclarativeWorkflowBuilder:
         checkpoint_storage: Any | None = None,
         validate: bool = True,
         max_iterations: int | None = None,
+        http_request_handler: HttpRequestHandler | None = None,
     ):
         """Initialize the builder.
 
@@ -141,6 +147,9 @@ class DeclarativeWorkflowBuilder:
             validate: Whether to validate the workflow definition before building (default: True)
             max_iterations: Maximum runner supersteps. Falls back to the YAML ``maxTurns``
                 field, then to the core default (100).
+            http_request_handler: Handler used to dispatch HttpRequestAction requests.
+                Must be supplied when the workflow contains any HttpRequestAction;
+                otherwise build raises ``DeclarativeWorkflowError``.
         """
         self._yaml_def = yaml_definition
         self._workflow_id = workflow_id or yaml_definition.get("name", "declarative_workflow")
@@ -152,6 +161,7 @@ class DeclarativeWorkflowBuilder:
         self._pending_gotos: list[tuple[Any, str]] = []  # (goto_executor, target_id)
         self._validate = validate
         self._seen_explicit_ids: set[str] = set()  # Track explicit IDs for duplicate detection
+        self._http_request_handler = http_request_handler
         # Resolve max_iterations: explicit arg > YAML maxTurns > core default
         resolved = max_iterations if max_iterations is not None else yaml_definition.get("maxTurns")
         if resolved is not None and (not isinstance(resolved, int) or resolved <= 0):
@@ -458,6 +468,19 @@ class DeclarativeWorkflowBuilder:
             executor = InvokeAzureAgentExecutor(action_def, id=action_id, agents=self._agents)
         elif kind == "InvokeFunctionTool":
             executor = InvokeFunctionToolExecutor(action_def, id=action_id, tools=self._tools)
+        elif kind == "HttpRequestAction":
+            if self._http_request_handler is None:
+                raise DeclarativeWorkflowError(
+                    f"Workflow defines HttpRequestAction '{action_id}' but no "
+                    "http_request_handler was supplied to WorkflowFactory. Pass "
+                    "http_request_handler=DefaultHttpRequestHandler() (or a custom "
+                    "implementation) to enable HTTP requests."
+                )
+            executor = HttpRequestActionExecutor(
+                action_def,
+                id=action_id,
+                http_request_handler=self._http_request_handler,
+            )
         else:
             executor = executor_class(action_def, id=action_id)
         self._executors[action_id] = executor

python/packages/declarative/agent_framework_declarative/_workflows/_errors.py

@@ -0,0 +1,38 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Error types for declarative workflow executor modules.
+
+This module exists so that executor modules and the builder (e.g.
+``_executors_http``, ``_declarative_builder``) can raise declarative-specific
+exceptions without importing from ``_factory``. ``_factory`` imports
+``_declarative_builder`` which imports the executor modules; pulling
+:class:`DeclarativeWorkflowError` from ``_factory`` into an executor or
+builder module would therefore introduce a circular import.
+"""
+
+from __future__ import annotations
+
+from agent_framework.exceptions import WorkflowException
+
+
+class DeclarativeWorkflowError(WorkflowException):
+    """Raised for build-time / factory-level declarative workflow errors.
+
+    Used for YAML parsing/validation issues, missing configuration (e.g. an
+    HTTP request handler not supplied for a workflow that contains an
+    ``HttpRequestAction``), and other errors detected before workflow
+    execution begins.
+    """
+
+    pass
+
+
+class DeclarativeActionError(WorkflowException):
+    """Raised when a declarative action fails at run time.
+
+    Used by executor modules for runtime failures (e.g. transport errors,
+    non-2xx responses from :class:`HttpRequestActionExecutor`). Build-time and
+    factory-level errors continue to use :class:`DeclarativeWorkflowError`.
+    """
+
+    pass

python/packages/declarative/agent_framework_declarative/_workflows/_executors_control_flow.py

@@ -17,6 +17,7 @@ The key insight is that control flow becomes GRAPH STRUCTURE, not executor logic
 from typing import Any, cast
 
 from agent_framework import (
+    Message,
     WorkflowContext,
     handler,
 )
@@ -492,7 +493,13 @@ class JoinExecutor(DeclarativeActionExecutor):
     @handler
     async def handle_action(
         self,
-        trigger: dict[str, Any] | str | ActionTrigger | ActionComplete | ConditionResult | LoopIterationResult,
+        trigger: dict[str, Any]
+        | str
+        | list[Message]
+        | ActionTrigger
+        | ActionComplete
+        | ConditionResult
+        | LoopIterationResult,
         ctx: WorkflowContext[ActionComplete],
     ) -> None:
         """Simply pass through to continue the workflow."""

python/packages/declarative/agent_framework_declarative/_workflows/_executors_http.py

@@ -0,0 +1,417 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Executor for the ``HttpRequestAction`` declarative action.
+
+Mirrors the .NET ``HttpRequestExecutor``: dispatches an HTTP request through the
+configured :class:`HttpRequestHandler`, parses the response body, and assigns
+the parsed body and response headers to the declared state paths.
+
+Security note: response bodies can echo secrets and may be very large. Diagnostic
+messages produced for non-2xx responses truncate the body to 256 characters and
+collapse CR/LF/TAB to spaces (parity with .NET ``FormatBodyForDiagnostics``).
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from collections.abc import Mapping
+from typing import Any
+
+import httpx
+from agent_framework import (
+    Message,
+    WorkflowContext,
+    handler,
+)
+
+from ._declarative_base import (
+    ActionComplete,
+    DeclarativeActionExecutor,
+    DeclarativeWorkflowState,
+)
+from ._errors import DeclarativeActionError
+from ._http_handler import HttpRequestHandler, HttpRequestInfo, HttpRequestResult
+
+__all__ = [
+    "HTTP_ACTION_EXECUTORS",
+    "HttpRequestActionExecutor",
+]
+
+logger = logging.getLogger(__name__)
+
+_MAX_BODY_DIAGNOSTIC_LENGTH = 256
+_BODY_TRUNCATION_SUFFIX = " \u2026 [truncated]"
+
+
+# Body discriminator aliases. Long forms match the .NET object-model type
+# names so YAML produced by .NET round-trips. Short forms are the .NET YAML
+# convention used in test fixtures.
+_BODY_KIND_JSON = {"json", "JsonRequestContent"}
+_BODY_KIND_RAW = {"raw", "RawRequestContent"}
+_BODY_KIND_NONE = {"none", "NoRequestContent"}
+
+
+def _get_path(action_def: Mapping[str, Any], key: str) -> str | None:
+    """Extract a state path from ``response``/``responseHeaders`` field.
+
+    Supports two YAML shapes (matches .NET serialization round-trips):
+
+    - ``response: Local.MyVar`` (plain string).
+    - ``response: { path: Local.MyVar }`` (object form).
+    """
+    value = action_def.get(key)
+    if isinstance(value, str):
+        return value or None
+    if isinstance(value, Mapping):
+        path = value.get("path")  # type: ignore[reportUnknownMemberType, reportUnknownVariableType]
+        return path if isinstance(path, str) and path else None
+    return None
+
+
+def _format_body_for_diagnostics(body: str | None) -> str:
+    """Truncate and sanitise a response body for inclusion in error messages.
+
+    Mirrors the .NET ``FormatBodyForDiagnostics`` helper:
+
+    - Empty/None -> empty string.
+    - Replaces CR/LF/TAB with spaces.
+    - Truncates to 256 chars with a unicode-ellipsis ``[truncated]`` suffix.
+    """
+    if not body:
+        return ""
+
+    truncated = len(body) > _MAX_BODY_DIAGNOSTIC_LENGTH
+    head = body[:_MAX_BODY_DIAGNOSTIC_LENGTH] if truncated else body
+    sanitized = head.replace("\r", " ").replace("\n", " ").replace("\t", " ")
+    return sanitized + _BODY_TRUNCATION_SUFFIX if truncated else sanitized
+
+
+def _parse_response_body(body: str | None) -> Any:
+    """Parse an HTTP response body the same way the .NET executor does.
+
+    JSON-first: if the body parses as JSON, the parsed value is returned. Other
+    bodies are returned as the raw string. Empty/None bodies return ``None``.
+    """
+    if body is None or body == "":
+        return None
+    try:
+        return json.loads(body)
+    except json.JSONDecodeError:
+        return body
+
+
+def _format_query_value(value: Any) -> str | None:
+    """Format a query-parameter value for URL inclusion.
+
+    Mirrors .NET ``FormatQueryValue``: ``None`` is dropped, ``bool`` becomes
+    lower-case ``"true"``/``"false"``, numerics use invariant ``str()``, and
+    other values fall through to ``str()``.
+    """
+    if value is None:
+        return None
+    if isinstance(value, bool):
+        return "true" if value else "false"
+    if isinstance(value, str):
+        return value
+    return str(value)
+
+
+def _get_messages_path(state: DeclarativeWorkflowState, conversation_id_expr: str | None) -> str | None:
+    """Return the configured conversation messages path, if any.
+
+    Returns ``System.conversations.{evaluated_id}.messages`` when a
+    ``conversation_id_expr`` is configured and evaluates to a non-empty value.
+    Returns ``None`` when no conversation id expression is configured or when
+    the expression evaluates to ``None`` or an empty string (matches .NET
+    ``GetConversationId`` behaviour where empty becomes ``null`` and the
+    response is not appended).
+    """
+    if not conversation_id_expr:
+        return None
+    evaluated = state.eval_if_expression(conversation_id_expr)
+    if evaluated is None or (isinstance(evaluated, str) and not evaluated):
+        return None
+    return f"System.conversations.{evaluated}.messages"
+
+
+class HttpRequestActionExecutor(DeclarativeActionExecutor):
+    """Executor for the ``HttpRequestAction`` declarative action.
+
+    Dispatches through the supplied :class:`HttpRequestHandler` and:
+
+    - Parses the response body (JSON-first, raw string fall-back).
+    - Assigns the parsed body to ``response`` path (if configured).
+    - Folds multi-value response headers (comma-joined) and assigns them to
+      ``responseHeaders`` path (if configured).
+    - On 2xx with non-empty body and a configured ``conversationId``, appends
+      an Assistant :class:`agent_framework.Message` to
+      ``System.conversations.{id}.messages``.
+    - On non-2xx, still publishes ``responseHeaders`` (diagnostic) and raises
+      :class:`DeclarativeActionError` with a status-coded message containing a
+      truncated/sanitised body preview.
+
+    Transport errors (``httpx.TimeoutException``, ``TimeoutError``,
+    ``httpx.HTTPError``) become :class:`DeclarativeActionError`. ``CancelledError``
+    is intentionally NOT caught so that workflow cancellation propagates.
+    """
+
+    def __init__(
+        self,
+        action_def: dict[str, Any],
+        *,
+        id: str | None = None,
+        http_request_handler: HttpRequestHandler,
+    ) -> None:
+        """Create an HTTP request action executor.
+
+        Args:
+            action_def: Parsed ``HttpRequestAction`` YAML dict.
+            id: Optional executor id (defaults to action id or generated).
+            http_request_handler: Handler used to dispatch HTTP requests.
+                Required: the builder enforces presence at workflow-build time.
+        """
+        super().__init__(action_def, id=id)
+        self._http_request_handler = http_request_handler
+
+    @handler
+    async def handle_action(
+        self,
+        trigger: Any,
+        ctx: WorkflowContext[ActionComplete],
+    ) -> None:
+        """Execute the HTTP request action."""
+        state = await self._ensure_state_initialized(ctx, trigger)
+
+        method = self._get_method(state)
+        url = self._get_url(state)
+        headers = self._get_headers(state)
+        query_parameters = self._get_query_parameters(state)
+        body, body_content_type = self._get_body(state)
+        timeout_ms = self._get_timeout_ms(state)
+        conversation_id_expr = self._action_def.get("conversationId")
+        connection_name = self._get_connection_name(state)
+
+        info = HttpRequestInfo(
+            method=method,
+            url=url,
+            headers=headers or {},
+            query_parameters=query_parameters or {},
+            body=body,
+            body_content_type=body_content_type,
+            timeout_ms=timeout_ms,
+            connection_name=connection_name,
+        )
+
+        try:
+            result = await self._http_request_handler.send(info)
+        except (httpx.TimeoutException, TimeoutError) as exc:
+            raise DeclarativeActionError(f"HTTP request to '{url}' timed out.") from exc
+        except DeclarativeActionError:
+            raise
+        except httpx.HTTPError as exc:
+            raise DeclarativeActionError(f"HTTP request to '{url}' failed: {type(exc).__name__}") from exc
+        except Exception as exc:
+            # Custom HttpRequestHandler implementations may raise arbitrary
+            # exception types. Wrap them in DeclarativeActionError so workflow
+            # error handling stays uniform regardless of transport. Note that
+            # ``asyncio.CancelledError`` is a ``BaseException`` (not
+            # ``Exception``) and so still propagates unmodified, preserving
+            # workflow-cancellation semantics.
+            raise DeclarativeActionError(f"HTTP request to '{url}' failed: {type(exc).__name__}") from exc
+
+        if result.is_success_status_code:
+            self._assign_response(state, result)
+            self._assign_response_headers(state, result)
+            self._append_response_to_conversation(state, conversation_id_expr, result.body)
+            await ctx.send_message(ActionComplete())
+            return
+
+        # Non-success path: still publish headers diagnostically, then raise.
+        self._assign_response_headers(state, result)
+        body_preview = _format_body_for_diagnostics(result.body)
+        if body_preview:
+            message = f"HTTP request to '{url}' failed with status code {result.status_code}. Body: '{body_preview}'"
+        else:
+            message = f"HTTP request to '{url}' failed with status code {result.status_code}."
+        raise DeclarativeActionError(message)
+
+    # ----- Field resolution ----------------------------------------------------
+
+    def _get_method(self, state: DeclarativeWorkflowState) -> str:
+        method = self._action_def.get("method")
+        evaluated = state.eval_if_expression(method) if method is not None else None
+        if not evaluated:
+            return "GET"
+        return str(evaluated).upper()
+
+    def _get_url(self, state: DeclarativeWorkflowState) -> str:
+        raw = self._action_def.get("url")
+        if raw is None:
+            raise ValueError("HttpRequestAction requires a 'url' field.")
+        evaluated = state.eval_if_expression(raw)
+        if not isinstance(evaluated, str) or not evaluated:
+            raise ValueError("HttpRequestAction 'url' evaluated to an empty value.")
+        return evaluated
+
+    def _get_headers(self, state: DeclarativeWorkflowState) -> dict[str, str] | None:
+        raw_headers = self._action_def.get("headers")
+        if not isinstance(raw_headers, Mapping) or not raw_headers:
+            return None
+        result: dict[str, str] = {}
+        for key, value in raw_headers.items():  # type: ignore[reportUnknownVariableType]
+            if not isinstance(key, str) or not key:
+                continue
+            evaluated = state.eval_if_expression(value)
+            if evaluated is None:
+                continue
+            text = str(evaluated)
+            if not text:
+                continue
+            result[key] = text
+        return result or None
+
+    def _get_query_parameters(self, state: DeclarativeWorkflowState) -> dict[str, str] | None:
+        raw_params = self._action_def.get("queryParameters")
+        if not isinstance(raw_params, Mapping) or not raw_params:
+            return None
+        result: dict[str, str] = {}
+        for key, value in raw_params.items():  # type: ignore[reportUnknownVariableType]
+            if not isinstance(key, str) or not key or value is None:
+                continue
+            evaluated = state.eval_if_expression(value)
+            formatted = _format_query_value(evaluated)
+            if formatted is not None:
+                result[key] = formatted
+        return result or None
+
+    def _get_body(self, state: DeclarativeWorkflowState) -> tuple[str | None, str | None]:
+        raw_body = self._action_def.get("body")
+        if raw_body is None:
+            return None, None
+        if not isinstance(raw_body, Mapping):
+            raise ValueError(
+                "HttpRequestAction 'body' must be a mapping with a 'kind' field (json, raw) or omitted entirely."
+            )
+
+        kind_value: Any = raw_body.get("kind") or raw_body.get("$kind")  # type: ignore[reportUnknownMemberType]
+        if kind_value is None:
+            raise ValueError(
+                "HttpRequestAction 'body' is missing 'kind'. Use 'json', 'raw', or omit 'body' for no request body."
+            )
+        if not isinstance(kind_value, str):
+            raise ValueError(f"HttpRequestAction 'body.kind' must be a string, got {kind_value!r}.")
+
+        if kind_value in _BODY_KIND_NONE:
+            return None, None
+
+        if kind_value in _BODY_KIND_JSON:
+            content_expr: Any = raw_body.get("content")  # type: ignore[reportUnknownMemberType]
+            if content_expr is None:
+                return None, None
+            evaluated = state.eval_if_expression(content_expr)
+            try:
+                body_text = json.dumps(evaluated, default=str)
+            except (TypeError, ValueError) as exc:
+                raise ValueError(f"HttpRequestAction 'body.content' could not be serialised as JSON: {exc}") from exc
+            return body_text, "application/json"
+
+        if kind_value in _BODY_KIND_RAW:
+            content_expr = raw_body.get("content")  # type: ignore[reportUnknownMemberType]
+            content_type_expr: Any = raw_body.get("contentType")  # type: ignore[reportUnknownMemberType]
+            content: str | None = None
+            if content_expr is not None:
+                evaluated = state.eval_if_expression(content_expr)
+                content = None if evaluated is None else str(evaluated)
+            content_type: str | None = None
+            if content_type_expr is not None:
+                ct_eval = state.eval_if_expression(content_type_expr)
+                ct_text = None if ct_eval is None else str(ct_eval)
+                content_type = ct_text or None
+            # Match .NET RawRequestContent semantics: when a raw body is sent
+            # without an explicit content type, default to text/plain so the
+            # request is interpretable by servers.
+            if content is not None and not content_type:
+                content_type = "text/plain"
+            return content, content_type
+
+        raise ValueError(
+            f"HttpRequestAction 'body.kind' has unsupported value '{kind_value}'. "
+            "Expected one of: json, raw, JsonRequestContent, RawRequestContent, "
+            "NoRequestContent."
+        )
+
+    def _get_timeout_ms(self, state: DeclarativeWorkflowState) -> int | None:
+        raw = self._action_def.get("requestTimeoutInMilliseconds")
+        if raw is None:
+            return None
+        evaluated = state.eval_if_expression(raw)
+        if evaluated is None:
+            return None
+        try:
+            value = int(evaluated)
+        except (TypeError, ValueError):
+            logger.debug(
+                "HttpRequestAction: ignoring non-numeric requestTimeoutInMilliseconds=%r",
+                evaluated,
+            )
+            return None
+        return value if value > 0 else None
+
+    def _get_connection_name(self, state: DeclarativeWorkflowState) -> str | None:
+        connection = self._action_def.get("connection")
+        if not isinstance(connection, Mapping):
+            return None
+        name_expr: Any = connection.get("name")  # type: ignore[reportUnknownMemberType]
+        if name_expr is None:
+            return None
+        evaluated = state.eval_if_expression(name_expr)
+        if evaluated is None:
+            return None
+        text = str(evaluated)
+        return text or None
+
+    # ----- Result handling -----------------------------------------------------
+
+    def _assign_response(self, state: DeclarativeWorkflowState, result: HttpRequestResult) -> None:
+        path = _get_path(self._action_def, "response")
+        if path is None:
+            return
+        state.set(path, _parse_response_body(result.body))
+
+    def _assign_response_headers(self, state: DeclarativeWorkflowState, result: HttpRequestResult) -> None:
+        path = _get_path(self._action_def, "responseHeaders")
+        if path is None:
+            return
+        if not result.headers:
+            state.set(path, None)
+            return
+        # Fold multi-value headers with commas (standard HTTP folding) only at
+        # assignment time. The raw multi-value dict on HttpRequestResult.headers
+        # is left untouched so callers/tests can inspect duplicates.
+        flattened: dict[str, str] = {}
+        for key, values in result.headers.items():
+            flattened[key] = ",".join(values)
+        state.set(path, flattened)
+
+    def _append_response_to_conversation(
+        self,
+        state: DeclarativeWorkflowState,
+        conversation_id_expr: str | None,
+        body: str,
+    ) -> None:
+        if not body:
+            return
+        messages_path = _get_messages_path(state, conversation_id_expr)
+        if messages_path is None:
+            return
+        # Mirrors InvokeAzureAgentExecutor: rely on state.append to lazily
+        # create the conversation entry. Avoids re-parsing the id back out
+        # of the dotted path string.
+        message = Message(role="assistant", contents=[body])
+        state.append(messages_path, message)
+
+
+HTTP_ACTION_EXECUTORS: dict[str, type[DeclarativeActionExecutor]] = {
+    "HttpRequestAction": HttpRequestActionExecutor,
+}

python/packages/declarative/agent_framework_declarative/_workflows/_factory.py

@@ -24,18 +24,16 @@ from agent_framework import (
     SupportsAgentRun,
     Workflow,
 )
-from agent_framework.exceptions import WorkflowException
 
 from .._loader import AgentFactory
 from ._declarative_builder import DeclarativeWorkflowBuilder
+from ._errors import DeclarativeWorkflowError
+from ._http_handler import HttpRequestHandler
 
 logger = logging.getLogger("agent_framework.declarative")
 
 
-class DeclarativeWorkflowError(WorkflowException):
-    """Exception raised for errors in declarative workflow processing."""
-
-    pass
+__all__ = ["WorkflowFactory"]
 
 
 class WorkflowFactory:
@@ -92,6 +90,7 @@ class WorkflowFactory:
         env_file: str | None = None,
         checkpoint_storage: CheckpointStorage | None = None,
         max_iterations: int | None = None,
+        http_request_handler: HttpRequestHandler | None = None,
     ) -> None:
         """Initialize the workflow factory.
 
@@ -105,6 +104,12 @@ class WorkflowFactory:
             max_iterations: Optional maximum runner supersteps.  Overrides the YAML ``maxTurns``
                 field and the core default (100).  Workflows with ``GotoAction`` loops (e.g.
                 DeepResearch) typically need a higher value.
+            http_request_handler: Optional handler used to dispatch HTTP requests for
+                ``HttpRequestAction``. Required if the workflow contains any
+                ``HttpRequestAction``; build will fail with :class:`DeclarativeWorkflowError`
+                otherwise. Use :class:`agent_framework.declarative.DefaultHttpRequestHandler`
+                for a no-policy ``httpx``-based default, or supply your own implementation
+                to enforce SSRF guards, allowlisting, or auth resolution.
 
         Examples:
             .. code-block:: python
@@ -144,6 +149,7 @@ class WorkflowFactory:
         self._tools: dict[str, Any] = {}  # Tool registry for InvokeFunctionTool actions
         self._checkpoint_storage = checkpoint_storage
         self._max_iterations = max_iterations
+        self._http_request_handler = http_request_handler
 
     def create_workflow_from_yaml_path(
         self,
@@ -387,6 +393,7 @@ class WorkflowFactory:
                 tools=self._tools,
                 checkpoint_storage=self._checkpoint_storage,
                 max_iterations=self._max_iterations,
+                http_request_handler=self._http_request_handler,
             )
             workflow = graph_builder.build()
         except ValueError as e:

python/packages/declarative/agent_framework_declarative/_workflows/_http_handler.py

@@ -0,0 +1,237 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""HTTP request handler abstraction for declarative workflows.
+
+Mirrors the .NET ``IHttpRequestHandler`` / ``DefaultHttpRequestHandler`` pair from
+``Microsoft.Agents.AI.Workflows.Declarative``. Provides:
+
+- :class:`HttpRequestInfo` — request input data passed from the executor.
+- :class:`HttpRequestResult` — response data returned to the executor.
+- :class:`HttpRequestHandler` — :class:`typing.Protocol` callers implement to plug
+  in custom transports (e.g. with allowlisting, mTLS, retries, etc.).
+- :class:`DefaultHttpRequestHandler` — production-grade default backed by
+  ``httpx.AsyncClient``.
+
+Security note: :class:`DefaultHttpRequestHandler` performs **no** URL filtering
+or SSRF protection. Production deployments should supply a custom handler that
+enforces an allowlist or DNS-rebinding-resistant policy. This split mirrors the
+.NET design.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable, Callable, Mapping
+from dataclasses import dataclass, field
+from typing import Any, Protocol, runtime_checkable
+
+import httpx
+
+__all__ = [
+    "DefaultHttpRequestHandler",
+    "HttpRequestHandler",
+    "HttpRequestInfo",
+    "HttpRequestResult",
+]
+
+
+@dataclass
+class HttpRequestInfo:
+    """Description of an HTTP request to be dispatched by a :class:`HttpRequestHandler`.
+
+    Mirrors the .NET ``HttpRequestInfo`` record. Field semantics:
+
+    - ``method``: HTTP method (``GET``, ``POST``, etc.). Already upper-cased by the executor.
+    - ``url``: Absolute URL. Already evaluated from the YAML expression.
+    - ``headers``: Single-value header map (case-insensitive keys per HTTP semantics
+      but stored as authored). Empty values are skipped by the executor.
+    - ``query_parameters``: String key/value pairs appended to the URL.
+    - ``body``: Request body bytes/text, or ``None`` for no body.
+    - ``body_content_type``: Content type to send (e.g. ``application/json``).
+      Ignored when ``body`` is ``None``.
+    - ``timeout_ms``: Per-request timeout in milliseconds. ``None`` => use the
+      handler's default.
+    - ``connection_name``: Optional Foundry connection name for handlers that
+      resolve auth/credentials by connection.
+    """
+
+    method: str
+    url: str
+    headers: dict[str, str] = field(default_factory=dict)  # type: ignore[reportUnknownVariableType]
+    query_parameters: dict[str, str] = field(default_factory=dict)  # type: ignore[reportUnknownVariableType]
+    body: str | None = None
+    body_content_type: str | None = None
+    timeout_ms: int | None = None
+    connection_name: str | None = None
+
+
+@dataclass
+class HttpRequestResult:
+    """Response returned by a :class:`HttpRequestHandler`.
+
+    Mirrors the .NET ``HttpRequestResult`` record. ``headers`` preserves
+    multi-value response headers (e.g. multiple ``Set-Cookie`` headers) as a
+    ``dict[str, list[str]]``. The executor folds duplicates into a single
+    comma-joined string only at the point it assigns ``responseHeaders`` to
+    workflow state.
+
+    Header keys are normalized to lowercase so that lookups are consistent
+    regardless of the server's transmitted casing (HTTP headers are
+    case-insensitive per RFC 7230 §3.2). Custom :class:`HttpRequestHandler`
+    implementations should follow the same convention.
+    """
+
+    status_code: int
+    is_success_status_code: bool
+    body: str
+    headers: dict[str, list[str]] = field(default_factory=dict)  # type: ignore[reportUnknownVariableType]
+
+
+@runtime_checkable
+class HttpRequestHandler(Protocol):
+    """Protocol for HTTP request handlers used by ``HttpRequestAction``.
+
+    Implementations must be safe to call concurrently from multiple workflow
+    runs. Implementations are responsible for any URL allowlisting, SSRF
+    guards, retry policies, auth resolution, and other policies that the
+    workflow author wants applied.
+    """
+
+    async def send(self, info: HttpRequestInfo) -> HttpRequestResult:
+        """Dispatch ``info`` and return the response result.
+
+        Args:
+            info: Description of the request to send.
+
+        Returns:
+            The response. Implementations should NOT raise on non-2xx status
+            codes; instead, set ``is_success_status_code`` accordingly. They
+            SHOULD raise on transport-level failures (connection refused,
+            DNS errors, timeouts).
+        """
+        ...
+
+
+ClientProvider = Callable[[HttpRequestInfo], Awaitable["httpx.AsyncClient | None"]]
+
+
+class DefaultHttpRequestHandler:
+    """Default :class:`HttpRequestHandler` backed by :class:`httpx.AsyncClient`.
+
+    Construction modes:
+
+    1. ``DefaultHttpRequestHandler()`` — owns an internal client created lazily
+       on first ``send()``. Closed by :meth:`aclose`.
+    2. ``DefaultHttpRequestHandler(client=existing)`` — caller-owned client.
+       Not closed by :meth:`aclose`.
+    3. ``DefaultHttpRequestHandler(client_provider=cb)`` — per-request client
+       lookup (parity with .NET's ``httpClientProvider`` callback). The
+       provider may return ``None`` to fall back to the owned/default client.
+
+    .. warning::
+
+       This handler performs **no** URL filtering or SSRF protection. Wrap or
+       replace it with a custom handler in production.
+    """
+
+    def __init__(
+        self,
+        *,
+        client: httpx.AsyncClient | None = None,
+        client_provider: ClientProvider | None = None,
+    ) -> None:
+        self._owned_client: httpx.AsyncClient | None = None
+        self._caller_client = client
+        self._client_provider = client_provider
+        # Guards lazy creation of ``_owned_client`` against concurrent first
+        # ``send()`` calls leaking duplicate clients.
+        self._owned_client_lock = asyncio.Lock()
+
+    async def send(self, info: HttpRequestInfo) -> HttpRequestResult:
+        """Dispatch the request and return the parsed result."""
+        if not info.url:
+            raise ValueError("HttpRequestInfo.url must be a non-empty string.")
+        if not info.method:
+            raise ValueError("HttpRequestInfo.method must be a non-empty string.")
+
+        client = await self._resolve_client(info)
+
+        timeout: httpx.Timeout | object
+        if info.timeout_ms is not None and info.timeout_ms > 0:
+            timeout = httpx.Timeout(info.timeout_ms / 1000.0)
+        else:
+            timeout = httpx.USE_CLIENT_DEFAULT
+
+        headers = dict(info.headers)
+        content: bytes | str | None = None
+        if info.body is not None:
+            content = info.body
+            if not _has_header(headers, "content-type"):
+                # Match .NET DefaultHttpRequestHandler: when a body is sent
+                # without an explicit content type, default to ``text/plain``
+                # so the request is interpretable by servers and direct
+                # callers (not just the YAML executor) get sensible defaults.
+                headers["Content-Type"] = info.body_content_type or "text/plain"
+
+        params: Mapping[str, str] | None = info.query_parameters or None
+
+        response = await client.request(
+            method=info.method,
+            url=info.url,
+            params=params,
+            headers=headers or None,
+            content=content,
+            timeout=timeout,  # type: ignore[arg-type]
+        )
+
+        # Preserve multi-value headers (e.g. multiple Set-Cookie) as list[str].
+        # Normalize names to lowercase so lookups are consistent and case
+        # variations from the transport do not create duplicate logical keys
+        # (HTTP headers are case-insensitive per RFC 7230 §3.2).
+        result_headers: dict[str, list[str]] = {}
+        for key, value in response.headers.multi_items():
+            result_headers.setdefault(key.lower(), []).append(value)
+
+        body_text = response.text
+
+        return HttpRequestResult(
+            status_code=response.status_code,
+            is_success_status_code=200 <= response.status_code < 300,
+            body=body_text,
+            headers=result_headers,
+        )
+
+    async def aclose(self) -> None:
+        """Release the owned client, if any. Caller-owned clients are NOT closed."""
+        if self._owned_client is not None:
+            await self._owned_client.aclose()
+            self._owned_client = None
+
+    async def _resolve_client(self, info: HttpRequestInfo) -> httpx.AsyncClient:
+        """Pick a client for this request: provider → caller → lazily-owned."""
+        if self._client_provider is not None:
+            provided = await self._client_provider(info)
+            if provided is not None:
+                return provided
+        if self._caller_client is not None:
+            return self._caller_client
+        if self._owned_client is None:
+            # Double-checked locking under asyncio.Lock so concurrent first
+            # callers don't each create a fresh httpx.AsyncClient and orphan
+            # one of them.
+            async with self._owned_client_lock:
+                if self._owned_client is None:
+                    self._owned_client = httpx.AsyncClient()
+        return self._owned_client
+
+    async def __aenter__(self) -> DefaultHttpRequestHandler:
+        return self
+
+    async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
+        await self.aclose()
+
+
+def _has_header(headers: Mapping[str, str], name: str) -> bool:
+    """Case-insensitive header presence check."""
+    needle = name.lower()
+    return any(key.lower() == needle for key in headers)

python/packages/declarative/pyproject.toml

@@ -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.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -22,7 +22,8 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
+    "httpx>=0.27,<1",
     "powerfx>=0.0.32,<0.0.35; python_version < '3.14'",
     "pyyaml>=6.0,<7.0",
 ]

python/packages/declarative/tests/test_default_http_request_handler.py

@@ -0,0 +1,329 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for ``DefaultHttpRequestHandler``.
+
+These tests exercise the real handler against ``httpx.MockTransport`` (no real
+network) to cover the parts of the handler not exercisable through the executor
+stub: query-param URL composition, content-type forwarding, per-request
+timeout overrides, multi-value response header preservation, and client
+ownership semantics.
+"""
+
+from __future__ import annotations
+
+import sys
+
+import httpx
+import pytest
+
+try:
+    import powerfx  # noqa: F401
+
+    _powerfx_available = True
+except (ImportError, RuntimeError):
+    _powerfx_available = False
+
+# These tests don't actually need PowerFx, but the rest of the suite gates on
+# Python versions and we keep behaviour consistent.
+pytestmark = pytest.mark.skipif(
+    sys.version_info >= (3, 14),
+    reason="Skipped on Python 3.14+ to keep parity with rest of declarative suite",
+)
+
+from agent_framework_declarative._workflows._http_handler import (  # noqa: E402
+    DefaultHttpRequestHandler,
+    HttpRequestInfo,
+)
+
+
+def _make_handler(transport: httpx.MockTransport) -> DefaultHttpRequestHandler:
+    """Return a handler with a MockTransport-backed caller-owned client."""
+    client = httpx.AsyncClient(transport=transport)
+    return DefaultHttpRequestHandler(client=client)
+
+
+class TestRequestComposition:
+    @pytest.mark.asyncio
+    async def test_query_parameters_merged_into_url(self) -> None:
+        captured: dict[str, httpx.Request] = {}
+
+        def respond(request: httpx.Request) -> httpx.Response:
+            captured["req"] = request
+            return httpx.Response(200, text="ok")
+
+        handler = _make_handler(httpx.MockTransport(respond))
+        try:
+            await handler.send(
+                HttpRequestInfo(
+                    method="GET",
+                    url="https://api.example.test/items",
+                    query_parameters={"q": "alpha", "limit": "5"},
+                )
+            )
+        finally:
+            await handler.aclose()
+
+        req = captured["req"]
+        # httpx exposes the merged URL with QS appended
+        assert req.url.params.get("q") == "alpha"
+        assert req.url.params.get("limit") == "5"
+
+    @pytest.mark.asyncio
+    async def test_body_content_type_forwarded(self) -> None:
+        captured: dict[str, httpx.Request] = {}
+
+        def respond(request: httpx.Request) -> httpx.Response:
+            captured["req"] = request
+            return httpx.Response(204)
+
+        handler = _make_handler(httpx.MockTransport(respond))
+        try:
+            await handler.send(
+                HttpRequestInfo(
+                    method="POST",
+                    url="https://api.example.test/items",
+                    body='{"k":"v"}',
+                    body_content_type="application/json",
+                )
+            )
+        finally:
+            await handler.aclose()
+
+        req = captured["req"]
+        assert req.headers.get("content-type") == "application/json"
+        assert req.content == b'{"k":"v"}'
+
+    @pytest.mark.asyncio
+    async def test_existing_content_type_header_not_overwritten(self) -> None:
+        captured: dict[str, httpx.Request] = {}
+
+        def respond(request: httpx.Request) -> httpx.Response:
+            captured["req"] = request
+            return httpx.Response(200, text="ok")
+
+        handler = _make_handler(httpx.MockTransport(respond))
+        try:
+            await handler.send(
+                HttpRequestInfo(
+                    method="POST",
+                    url="https://api.example.test/items",
+                    headers={"Content-Type": "application/xml"},  # caller wins
+                    body="<x/>",
+                    body_content_type="application/json",
+                )
+            )
+        finally:
+            await handler.aclose()
+
+        req = captured["req"]
+        assert req.headers.get("content-type") == "application/xml"
+
+    @pytest.mark.asyncio
+    async def test_body_without_content_type_defaults_to_text_plain(self) -> None:
+        """Match .NET DefaultHttpRequestHandler: body without explicit content type → ``text/plain``."""
+        captured: dict[str, httpx.Request] = {}
+
+        def respond(request: httpx.Request) -> httpx.Response:
+            captured["req"] = request
+            return httpx.Response(204)
+
+        handler = _make_handler(httpx.MockTransport(respond))
+        try:
+            await handler.send(
+                HttpRequestInfo(
+                    method="POST",
+                    url="https://api.example.test/items",
+                    body="hello",
+                    # No body_content_type and no Content-Type header.
+                )
+            )
+        finally:
+            await handler.aclose()
+
+        req = captured["req"]
+        assert req.headers.get("content-type") == "text/plain"
+        assert req.content == b"hello"
+
+
+class TestTimeout:
+    @pytest.mark.asyncio
+    async def test_per_request_timeout_surfaces_as_timeout_exception(self) -> None:
+        def respond(request: httpx.Request) -> httpx.Response:
+            raise httpx.TimeoutException("simulated timeout", request=request)
+
+        handler = _make_handler(httpx.MockTransport(respond))
+        try:
+            with pytest.raises(httpx.TimeoutException):
+                await handler.send(
+                    HttpRequestInfo(
+                        method="GET",
+                        url="https://api.example.test/slow",
+                        timeout_ms=50,
+                    )
+                )
+        finally:
+            await handler.aclose()
+
+
+class TestResponseHeaders:
+    @pytest.mark.asyncio
+    async def test_multi_value_headers_preserved(self) -> None:
+        def respond(request: httpx.Request) -> httpx.Response:
+            return httpx.Response(
+                200,
+                text="ok",
+                headers=[
+                    ("Content-Type", "application/json"),
+                    ("Set-Cookie", "a=1"),
+                    ("Set-Cookie", "b=2"),
+                ],
+            )
+
+        handler = _make_handler(httpx.MockTransport(respond))
+        try:
+            result = await handler.send(HttpRequestInfo(method="GET", url="https://api.example.test/x"))
+        finally:
+            await handler.aclose()
+
+        assert result.is_success_status_code
+        # The handler keeps multi-value headers as list[str].
+        assert result.headers.get("set-cookie") == ["a=1", "b=2"]
+        assert result.headers.get("content-type") == ["application/json"]
+
+
+class TestClientOwnership:
+    @pytest.mark.asyncio
+    async def test_owned_client_is_closed_on_aclose(self) -> None:
+        handler = DefaultHttpRequestHandler()
+        # Inject a MockTransport-backed client into the owned slot and verify
+        # aclose() releases it. Avoids real network access.
+        owned = httpx.AsyncClient(transport=httpx.MockTransport(lambda r: httpx.Response(200, text="ok")))
+        handler._owned_client = owned
+        assert not owned.is_closed
+        await handler.aclose()
+        assert owned.is_closed
+
+    @pytest.mark.asyncio
+    async def test_caller_owned_client_is_not_closed(self) -> None:
+        client = httpx.AsyncClient(transport=httpx.MockTransport(lambda r: httpx.Response(200, text="ok")))
+        handler = DefaultHttpRequestHandler(client=client)
+        await handler.send(HttpRequestInfo(method="GET", url="https://api.example.test/x"))
+        await handler.aclose()
+        assert not client.is_closed
+        await client.aclose()  # cleanup
+
+    @pytest.mark.asyncio
+    async def test_concurrent_first_send_creates_single_owned_client(self) -> None:
+        """Concurrent first-send calls must not race-leak duplicate clients.
+
+        Without the lock, two concurrent calls on a fresh handler would each
+        observe ``_owned_client is None`` and create their own
+        ``httpx.AsyncClient``, orphaning one. Verify that lazy initialization
+        is serialized: all concurrent sends end up using the same client and
+        ``aclose()`` cleanly closes it.
+        """
+        import asyncio
+
+        # Patch httpx.AsyncClient to count constructions, but only when called
+        # from inside _resolve_client (no transport=) so we don't break the
+        # MockTransport-backed clients used elsewhere.
+        original_ctor = httpx.AsyncClient
+        construction_count = 0
+
+        def counting_ctor(*args, **kwargs):  # type: ignore[no-untyped-def]
+            nonlocal construction_count
+            if not args and not kwargs:
+                construction_count += 1
+                return original_ctor(transport=httpx.MockTransport(lambda r: httpx.Response(200, text="ok")))
+            return original_ctor(*args, **kwargs)
+
+        import agent_framework_declarative._workflows._http_handler as hh
+
+        hh.httpx.AsyncClient = counting_ctor  # type: ignore[assignment]
+        try:
+            handler = DefaultHttpRequestHandler()
+            try:
+                await asyncio.gather(*[
+                    handler.send(HttpRequestInfo(method="GET", url="https://api.example.test/x")) for _ in range(8)
+                ])
+            finally:
+                await handler.aclose()
+        finally:
+            hh.httpx.AsyncClient = original_ctor  # type: ignore[assignment]
+
+        assert construction_count == 1, (
+            f"Expected exactly 1 owned client to be lazily created but got {construction_count}"
+        )
+
+
+class TestClientProvider:
+    @pytest.mark.asyncio
+    async def test_client_provider_overrides_default(self) -> None:
+        captured: dict[str, str] = {}
+
+        def primary(request: httpx.Request) -> httpx.Response:
+            captured["transport"] = "primary"
+            return httpx.Response(200, text="primary")
+
+        def provided(request: httpx.Request) -> httpx.Response:
+            captured["transport"] = "provided"
+            return httpx.Response(200, text="provided")
+
+        primary_client = httpx.AsyncClient(transport=httpx.MockTransport(primary))
+        provided_client = httpx.AsyncClient(transport=httpx.MockTransport(provided))
+
+        async def provider(info: HttpRequestInfo) -> httpx.AsyncClient:
+            return provided_client
+
+        handler = DefaultHttpRequestHandler(client=primary_client, client_provider=provider)
+        try:
+            result = await handler.send(HttpRequestInfo(method="GET", url="https://api.example.test/x"))
+            assert result.body == "provided"
+            assert captured["transport"] == "provided"
+        finally:
+            await handler.aclose()
+            await primary_client.aclose()
+            await provided_client.aclose()
+
+    @pytest.mark.asyncio
+    async def test_client_provider_returning_none_falls_back(self) -> None:
+        captured: dict[str, str] = {}
+
+        def primary(request: httpx.Request) -> httpx.Response:
+            captured["transport"] = "primary"
+            return httpx.Response(200, text="primary")
+
+        async def provider(info: HttpRequestInfo) -> httpx.AsyncClient | None:
+            return None
+
+        primary_client = httpx.AsyncClient(transport=httpx.MockTransport(primary))
+        handler = DefaultHttpRequestHandler(client=primary_client, client_provider=provider)
+        try:
+            result = await handler.send(HttpRequestInfo(method="GET", url="https://api.example.test/x"))
+            assert result.body == "primary"
+        finally:
+            await handler.aclose()
+            await primary_client.aclose()
+
+
+class TestValidation:
+    @pytest.mark.asyncio
+    async def test_empty_url_raises(self) -> None:
+        handler = DefaultHttpRequestHandler()
+        with pytest.raises(ValueError):
+            await handler.send(HttpRequestInfo(method="GET", url=""))
+
+    @pytest.mark.asyncio
+    async def test_empty_method_raises(self) -> None:
+        handler = DefaultHttpRequestHandler()
+        with pytest.raises(ValueError):
+            await handler.send(HttpRequestInfo(method="", url="https://x.test/"))
+
+
+class TestAsyncContextManager:
+    @pytest.mark.asyncio
+    async def test_context_manager_closes_owned_client(self) -> None:
+        async with DefaultHttpRequestHandler() as handler:
+            owned = httpx.AsyncClient(transport=httpx.MockTransport(lambda r: httpx.Response(200, text="ok")))
+            handler._owned_client = owned
+        assert owned.is_closed

python/packages/declarative/tests/test_http_request_executor.py

@@ -0,0 +1,645 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for HttpRequestActionExecutor.
+
+These tests use a stub HttpRequestHandler that returns canned HttpRequestResults.
+No real network or httpx transports are exercised. See
+test_default_http_request_handler.py for tests that exercise the real
+DefaultHttpRequestHandler against httpx.MockTransport.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+from typing import Any
+
+import httpx
+import pytest
+
+try:
+    import powerfx  # noqa: F401
+
+    _powerfx_available = True
+except (ImportError, RuntimeError):
+    _powerfx_available = False
+
+pytestmark = pytest.mark.skipif(
+    not _powerfx_available or sys.version_info >= (3, 14),
+    reason="PowerFx engine not available (requires dotnet runtime)",
+)
+
+from agent_framework_declarative._workflows import (  # noqa: E402
+    DECLARATIVE_STATE_KEY,
+    DeclarativeActionError,
+    DeclarativeWorkflowError,
+    HttpRequestHandler,
+    HttpRequestInfo,
+    HttpRequestResult,
+    WorkflowFactory,
+)
+
+
+class StubHandler:
+    """Test stub that records the last call and returns a canned result."""
+
+    def __init__(
+        self,
+        result: HttpRequestResult | None = None,
+        *,
+        raise_exc: BaseException | None = None,
+    ) -> None:
+        self.result = result
+        self.raise_exc = raise_exc
+        self.last_info: HttpRequestInfo | None = None
+        self.call_count = 0
+
+    async def send(self, info: HttpRequestInfo) -> HttpRequestResult:
+        self.call_count += 1
+        self.last_info = info
+        if self.raise_exc is not None:
+            raise self.raise_exc
+        assert self.result is not None
+        return self.result
+
+
+def _ok(body: str = "", headers: dict[str, list[str]] | None = None) -> HttpRequestResult:
+    return HttpRequestResult(
+        status_code=200,
+        is_success_status_code=True,
+        body=body,
+        headers=headers or {},
+    )
+
+
+def _err(status: int = 500, body: str = "", headers: dict[str, list[str]] | None = None) -> HttpRequestResult:
+    return HttpRequestResult(
+        status_code=status,
+        is_success_status_code=False,
+        body=body,
+        headers=headers or {},
+    )
+
+
+async def _run(yaml_def: dict[str, Any], handler: HttpRequestHandler) -> Any:
+    """Build & run a workflow, returning final WorkflowState."""
+    factory = WorkflowFactory(http_request_handler=handler)
+    workflow = factory.create_workflow_from_definition(yaml_def)
+    return await workflow.run({})
+
+
+def _state(workflow: Any, events: Any) -> dict[str, Any]:
+    """Read declarative state out of the workflow after run completes."""
+    return workflow._state.get(DECLARATIVE_STATE_KEY) or {}
+
+
+# Helper used by parametrised path tests
+_TEST_URL = "https://api.example.test/items"
+
+
+def _action(
+    *,
+    method: str | None = None,
+    url: str = _TEST_URL,
+    headers: dict[str, Any] | None = None,
+    query_parameters: dict[str, Any] | None = None,
+    body: dict[str, Any] | None = None,
+    response: Any = None,
+    response_headers: Any = None,
+    conversation_id: str | None = None,
+    request_timeout_ms: int | None = None,
+    connection: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    action: dict[str, Any] = {
+        "kind": "HttpRequestAction",
+        "id": "http_action",
+        "url": url,
+    }
+    if method is not None:
+        action["method"] = method
+    if headers is not None:
+        action["headers"] = headers
+    if query_parameters is not None:
+        action["queryParameters"] = query_parameters
+    if body is not None:
+        action["body"] = body
+    if response is not None:
+        action["response"] = response
+    if response_headers is not None:
+        action["responseHeaders"] = response_headers
+    if conversation_id is not None:
+        action["conversationId"] = conversation_id
+    if request_timeout_ms is not None:
+        action["requestTimeoutInMilliseconds"] = request_timeout_ms
+    if connection is not None:
+        action["connection"] = connection
+    return action
+
+
+def _yaml(action: dict[str, Any]) -> dict[str, Any]:
+    return {"name": "http_test", "actions": [action]}
+
+
+# ---------- Success path: response parsing ----------------------------------
+
+
+class TestSuccessPath:
+    @pytest.mark.asyncio
+    async def test_get_parses_json_object(self) -> None:
+        handler = StubHandler(_ok('{"key":"value","number":42}'))
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action(method="GET", response="Local.Result")))
+        await workflow.run({})
+
+        decl = workflow._state.get(DECLARATIVE_STATE_KEY)
+        assert decl["Local"]["Result"] == {"key": "value", "number": 42}
+        assert handler.last_info is not None
+        assert handler.last_info.method == "GET"
+        assert handler.last_info.url == _TEST_URL
+
+    @pytest.mark.asyncio
+    async def test_get_parses_plain_string(self) -> None:
+        handler = StubHandler(_ok("not-json content"))
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action(response="Local.Result")))
+        await workflow.run({})
+
+        decl = workflow._state.get(DECLARATIVE_STATE_KEY)
+        assert decl["Local"]["Result"] == "not-json content"
+
+    @pytest.mark.asyncio
+    async def test_get_empty_body_yields_none(self) -> None:
+        handler = StubHandler(_ok(""))
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action(response="Local.Result")))
+        await workflow.run({})
+
+        decl = workflow._state.get(DECLARATIVE_STATE_KEY)
+        assert decl["Local"]["Result"] is None
+
+    @pytest.mark.asyncio
+    async def test_response_object_form_path(self) -> None:
+        handler = StubHandler(_ok('{"x":1}'))
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action(response={"path": "Local.Result"})))
+        await workflow.run({})
+
+        decl = workflow._state.get(DECLARATIVE_STATE_KEY)
+        assert decl["Local"]["Result"] == {"x": 1}
+
+    @pytest.mark.asyncio
+    async def test_no_response_path_does_not_assign(self) -> None:
+        handler = StubHandler(_ok('{"x":1}'))
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action()))
+        # Should complete without error and without writing anything
+        await workflow.run({})
+
+
+# ---------- Method / headers / query params --------------------------------
+
+
+class TestRequestComposition:
+    @pytest.mark.asyncio
+    async def test_default_method_is_get(self) -> None:
+        handler = StubHandler(_ok())
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action()))
+        await workflow.run({})
+
+        assert handler.last_info is not None
+        assert handler.last_info.method == "GET"
+
+    @pytest.mark.asyncio
+    async def test_method_uppercased(self) -> None:
+        handler = StubHandler(_ok())
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action(method="post")))
+        await workflow.run({})
+
+        assert handler.last_info is not None
+        assert handler.last_info.method == "POST"
+
+    @pytest.mark.asyncio
+    async def test_headers_are_forwarded_and_empty_skipped(self) -> None:
+        handler = StubHandler(_ok())
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(
+            _yaml(
+                _action(
+                    headers={
+                        "Accept": "application/json",
+                        "X-Empty": "",
+                        "Authorization": "Bearer token",
+                    }
+                )
+            )
+        )
+        await workflow.run({})
+
+        assert handler.last_info is not None
+        assert handler.last_info.headers == {
+            "Accept": "application/json",
+            "Authorization": "Bearer token",
+        }
+
+    @pytest.mark.asyncio
+    async def test_query_parameters_stringified(self) -> None:
+        handler = StubHandler(_ok())
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(
+            _yaml(
+                _action(
+                    query_parameters={
+                        "name": "alpha",
+                        "limit": 10,
+                        "active": True,
+                        "ratio": 0.5,
+                        "missing": None,  # dropped
+                    }
+                )
+            )
+        )
+        await workflow.run({})
+
+        assert handler.last_info is not None
+        assert handler.last_info.query_parameters == {
+            "name": "alpha",
+            "limit": "10",
+            "active": "true",
+            "ratio": "0.5",
+        }
+
+
+# ---------- Body composition ------------------------------------------------
+
+
+class TestBody:
+    @pytest.mark.asyncio
+    async def test_post_json_body_sets_content_type_and_serialises(self) -> None:
+        handler = StubHandler(_ok())
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(
+            _yaml(
+                _action(
+                    method="POST",
+                    body={"kind": "json", "content": {"k": "v", "n": 1}},
+                )
+            )
+        )
+        await workflow.run({})
+
+        info = handler.last_info
+        assert info is not None
+        assert info.body_content_type == "application/json"
+        assert info.body is not None
+        # JSON serialized, key order may vary
+        import json
+
+        assert json.loads(info.body) == {"k": "v", "n": 1}
+
+    @pytest.mark.asyncio
+    async def test_post_raw_body_uses_declared_content_type(self) -> None:
+        handler = StubHandler(_ok())
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(
+            _yaml(
+                _action(
+                    method="POST",
+                    body={
+                        "kind": "raw",
+                        "content": "raw body text",
+                        "contentType": "text/plain",
+                    },
+                )
+            )
+        )
+        await workflow.run({})
+
+        info = handler.last_info
+        assert info is not None
+        assert info.body == "raw body text"
+        assert info.body_content_type == "text/plain"
+
+    @pytest.mark.asyncio
+    async def test_post_raw_body_without_content_type_defaults_to_text_plain(self) -> None:
+        """Match .NET RawRequestContent: no contentType => default text/plain.
+
+        Otherwise the request is sent without a Content-Type header which most
+        servers will treat as application/octet-stream and fail to parse.
+        """
+        handler = StubHandler(_ok())
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(
+            _yaml(
+                _action(
+                    method="POST",
+                    body={"kind": "raw", "content": "plain body"},
+                )
+            )
+        )
+        await workflow.run({})
+
+        info = handler.last_info
+        assert info is not None
+        assert info.body == "plain body"
+        assert info.body_content_type == "text/plain"
+
+    @pytest.mark.asyncio
+    async def test_long_form_body_kinds_accepted(self) -> None:
+        handler = StubHandler(_ok())
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(
+            _yaml(
+                _action(
+                    method="POST",
+                    body={"kind": "JsonRequestContent", "content": {"k": 1}},
+                )
+            )
+        )
+        await workflow.run({})
+        info = handler.last_info
+        assert info is not None
+        assert info.body_content_type == "application/json"
+
+    @pytest.mark.asyncio
+    async def test_unknown_body_kind_raises(self) -> None:
+        handler = StubHandler(_ok())
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action(body={"kind": "weirdform", "content": "x"})))
+        with pytest.raises(Exception) as excinfo:
+            await workflow.run({})
+        # Should surface as ValueError (potentially wrapped by runner)
+        msg = str(excinfo.value)
+        assert "weirdform" in msg or "unsupported value" in msg
+
+    @pytest.mark.asyncio
+    async def test_no_body_omitted(self) -> None:
+        handler = StubHandler(_ok())
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action()))
+        await workflow.run({})
+        info = handler.last_info
+        assert info is not None
+        assert info.body is None
+        assert info.body_content_type is None
+
+
+# ---------- Non-2xx and error handling -------------------------------------
+
+
+class TestErrorHandling:
+    @pytest.mark.asyncio
+    async def test_non_2xx_raises_declarative_action_error(self) -> None:
+        handler = StubHandler(_err(status=500, body="server exploded"))
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action()))
+        with pytest.raises(DeclarativeActionError) as excinfo:
+            await workflow.run({})
+        msg = str(excinfo.value)
+        assert "500" in msg
+        assert "server exploded" in msg
+
+    @pytest.mark.asyncio
+    async def test_non_2xx_long_body_truncated(self) -> None:
+        big_body = "A" * 1000
+        handler = StubHandler(_err(status=500, body=big_body))
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action()))
+        with pytest.raises(DeclarativeActionError) as excinfo:
+            await workflow.run({})
+        msg = str(excinfo.value)
+        assert "[truncated]" in msg
+        assert len(msg) < 512
+        # Should NOT contain the full 1000-char body
+        assert big_body not in msg
+
+    @pytest.mark.asyncio
+    async def test_non_2xx_empty_body_omits_body_section(self) -> None:
+        handler = StubHandler(_err(status=404, body=""))
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action()))
+        with pytest.raises(DeclarativeActionError) as excinfo:
+            await workflow.run({})
+        msg = str(excinfo.value)
+        assert "404" in msg
+        assert "Body:" not in msg
+
+    @pytest.mark.asyncio
+    async def test_non_2xx_control_chars_collapsed(self) -> None:
+        handler = StubHandler(_err(status=500, body="line1\r\nline2\tlong"))
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action()))
+        with pytest.raises(DeclarativeActionError) as excinfo:
+            await workflow.run({})
+        msg = str(excinfo.value)
+        assert "\r" not in msg
+        assert "\n" not in msg
+        assert "\t" not in msg
+        assert "line1  line2 long" in msg
+
+    @pytest.mark.asyncio
+    async def test_timeout_exception_becomes_declarative_action_error(self) -> None:
+        handler = StubHandler(raise_exc=httpx.TimeoutException("timeout"))
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action()))
+        with pytest.raises(DeclarativeActionError) as excinfo:
+            await workflow.run({})
+        assert "timed out" in str(excinfo.value)
+
+    @pytest.mark.asyncio
+    async def test_stdlib_timeout_error_becomes_declarative_action_error(self) -> None:
+        handler = StubHandler(raise_exc=TimeoutError("clock"))
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action()))
+        with pytest.raises(DeclarativeActionError) as excinfo:
+            await workflow.run({})
+        assert "timed out" in str(excinfo.value)
+
+    @pytest.mark.asyncio
+    async def test_transport_error_becomes_declarative_action_error(self) -> None:
+        handler = StubHandler(raise_exc=httpx.ConnectError("dns failure"))
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action()))
+        with pytest.raises(DeclarativeActionError) as excinfo:
+            await workflow.run({})
+        msg = str(excinfo.value)
+        assert "failed" in msg
+        assert _TEST_URL in msg
+
+    @pytest.mark.asyncio
+    async def test_cancelled_error_propagates_unchanged(self) -> None:
+        """CancelledError from the handler must propagate so cancellation works."""
+        handler = StubHandler(raise_exc=asyncio.CancelledError())
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action()))
+        # CancelledError is allowed to surface as either CancelledError or as
+        # the runner's wrapped form, but it MUST NOT be DeclarativeActionError.
+        with pytest.raises(BaseException) as excinfo:
+            await workflow.run({})
+        assert not isinstance(excinfo.value, DeclarativeActionError)
+
+    @pytest.mark.asyncio
+    async def test_generic_exception_from_custom_handler_wrapped(self) -> None:
+        """A custom handler raising a non-httpx Exception must be wrapped.
+
+        Authors can plug in custom HttpRequestHandler implementations that use
+        any transport (requests-like clients, gRPC bridges, mock test doubles,
+        etc.). The executor must wrap arbitrary Exception subclasses uniformly
+        so that workflow error handling stays consistent across transports.
+        """
+        handler = StubHandler(raise_exc=RuntimeError("custom transport blew up"))
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action()))
+        with pytest.raises(DeclarativeActionError) as excinfo:
+            await workflow.run({})
+        msg = str(excinfo.value)
+        assert "failed" in msg
+        assert "RuntimeError" in msg
+        assert _TEST_URL in msg
+
+
+# ---------- Response headers ------------------------------------------------
+
+
+class TestResponseHeaders:
+    @pytest.mark.asyncio
+    async def test_response_headers_folded_with_commas(self) -> None:
+        handler = StubHandler(
+            _ok(
+                "ok",
+                headers={
+                    "Content-Type": ["application/json"],
+                    "Set-Cookie": ["a=1", "b=2"],
+                },
+            )
+        )
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action(response_headers="Local.H")))
+        await workflow.run({})
+        decl = workflow._state.get(DECLARATIVE_STATE_KEY)
+        h = decl["Local"]["H"]
+        assert h["Content-Type"] == "application/json"
+        assert h["Set-Cookie"] == "a=1,b=2"
+
+    @pytest.mark.asyncio
+    async def test_response_headers_empty_assigned_none(self) -> None:
+        handler = StubHandler(_ok("ok", headers={}))
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action(response_headers="Local.H")))
+        await workflow.run({})
+        decl = workflow._state.get(DECLARATIVE_STATE_KEY)
+        assert decl["Local"]["H"] is None
+
+    @pytest.mark.asyncio
+    async def test_non_2xx_still_publishes_headers(self) -> None:
+        handler = StubHandler(_err(status=500, body="boom", headers={"X-Trace": ["abc"]}))
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action(response_headers="Local.H")))
+        with pytest.raises(DeclarativeActionError):
+            await workflow.run({})
+        decl = workflow._state.get(DECLARATIVE_STATE_KEY)
+        assert decl["Local"]["H"] == {"X-Trace": "abc"}
+
+
+# ---------- ConversationId append -------------------------------------------
+
+
+class TestConversationAppend:
+    @pytest.mark.asyncio
+    async def test_conversation_id_appends_message(self) -> None:
+        handler = StubHandler(_ok('{"answer":"hello"}'))
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(
+            _yaml(
+                _action(
+                    response="Local.Result",
+                    conversation_id="conv-test-1",
+                )
+            )
+        )
+        await workflow.run({})
+        decl = workflow._state.get(DECLARATIVE_STATE_KEY)
+        conv = decl["System"]["conversations"].get("conv-test-1")
+        assert conv is not None
+        assert len(conv["messages"]) == 1
+
+    @pytest.mark.asyncio
+    async def test_empty_conversation_id_does_not_append(self) -> None:
+        handler = StubHandler(_ok('{"answer":"hello"}'))
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action(response="Local.Result", conversation_id="")))
+        await workflow.run({})
+        decl = workflow._state.get(DECLARATIVE_STATE_KEY)
+        # Auto-init creates an entry for the System.ConversationId conversation,
+        # but it should NOT have HTTP-appended messages from us.
+        for _cid, conv in decl["System"]["conversations"].items():
+            assert conv["messages"] == []
+
+    @pytest.mark.asyncio
+    async def test_empty_body_skips_conversation_append(self) -> None:
+        handler = StubHandler(_ok(""))
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action(conversation_id="conv-test-1")))
+        await workflow.run({})
+        decl = workflow._state.get(DECLARATIVE_STATE_KEY)
+        # No conversation entry should have been created either.
+        assert "conv-test-1" not in decl["System"]["conversations"]
+
+
+# ---------- Connection name -------------------------------------------------
+
+
+class TestConnection:
+    @pytest.mark.asyncio
+    async def test_connection_name_forwarded(self) -> None:
+        handler = StubHandler(_ok())
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action(connection={"name": "my-connection"})))
+        await workflow.run({})
+        assert handler.last_info is not None
+        assert handler.last_info.connection_name == "my-connection"
+
+
+# ---------- Build-time validation -------------------------------------------
+
+
+class TestBuildTimeValidation:
+    def test_missing_url_fails_validation(self) -> None:
+        handler = StubHandler(_ok())
+        factory = WorkflowFactory(http_request_handler=handler)
+        bad = {
+            "name": "no_url",
+            "actions": [{"kind": "HttpRequestAction", "id": "x"}],
+        }
+        with pytest.raises(DeclarativeWorkflowError):
+            factory.create_workflow_from_definition(bad)
+
+    def test_missing_handler_fails_at_build(self) -> None:
+        factory = WorkflowFactory()  # no handler
+        with pytest.raises(DeclarativeWorkflowError) as excinfo:
+            factory.create_workflow_from_definition(_yaml(_action()))
+        assert "http_request_handler" in str(excinfo.value)
+
+
+# ---------- Timeout forwarding ----------------------------------------------
+
+
+class TestTimeout:
+    @pytest.mark.asyncio
+    async def test_timeout_ms_forwarded(self) -> None:
+        handler = StubHandler(_ok())
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action(request_timeout_ms=2500)))
+        await workflow.run({})
+        assert handler.last_info is not None
+        assert handler.last_info.timeout_ms == 2500
+
+    @pytest.mark.asyncio
+    async def test_timeout_ms_zero_treated_as_unset(self) -> None:
+        handler = StubHandler(_ok())
+        factory = WorkflowFactory(http_request_handler=handler)
+        workflow = factory.create_workflow_from_definition(_yaml(_action(request_timeout_ms=0)))
+        await workflow.run({})
+        assert handler.last_info is not None
+        assert handler.last_info.timeout_ms is None

python/packages/declarative/tests/test_http_request_yaml_integration.py

@@ -0,0 +1,111 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""End-to-end YAML integration test for ``HttpRequestAction``.
+
+Loads the ``tests/workflows/http_request.yaml`` fixture (parity with the .NET
+integration fixture) through ``WorkflowFactory.create_workflow_from_yaml_path``
+with a stub :class:`HttpRequestHandler` and asserts state is populated.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+try:
+    import powerfx  # noqa: F401
+
+    _powerfx_available = True
+except (ImportError, RuntimeError):
+    _powerfx_available = False
+
+pytestmark = [
+    pytest.mark.skipif(
+        not _powerfx_available,
+        reason="powerfx not available — declarative workflows require it.",
+    ),
+    pytest.mark.skipif(
+        sys.version_info >= (3, 14),
+        reason="Skipped on Python 3.14+ to keep parity with declarative suite.",
+    ),
+]
+
+from agent_framework_declarative import WorkflowFactory  # noqa: E402
+from agent_framework_declarative._workflows import DECLARATIVE_STATE_KEY  # noqa: E402
+from agent_framework_declarative._workflows._http_handler import (  # noqa: E402
+    HttpRequestInfo,
+    HttpRequestResult,
+)
+
+FIXTURE_PATH = Path(__file__).parent / "workflows" / "http_request.yaml"
+
+
+class _StubHandler:
+    """Test double that records requests and returns a canned response."""
+
+    def __init__(self, result: HttpRequestResult) -> None:
+        self._result = result
+        self.received: list[HttpRequestInfo] = []
+
+    async def send(self, info: HttpRequestInfo) -> HttpRequestResult:
+        self.received.append(info)
+        return self._result
+
+
+@pytest.mark.asyncio
+async def test_http_request_yaml_roundtrip() -> None:
+    handler = _StubHandler(
+        HttpRequestResult(
+            status_code=200,
+            is_success_status_code=True,
+            body='{"name": "runtime", "visibility": "public", "stars": 12345}',
+            headers={
+                "content-type": ["application/json"],
+                "x-ratelimit-remaining": ["59"],
+            },
+        )
+    )
+
+    factory = WorkflowFactory(http_request_handler=handler)
+    workflow = factory.create_workflow_from_yaml_path(FIXTURE_PATH)
+    await workflow.run({})
+
+    decl: dict[str, Any] = workflow._state.get(DECLARATIVE_STATE_KEY) or {}
+    local = decl.get("Local") or {}
+
+    assert local.get("RepoOwner") == "dotnet"
+    repo_info = local.get("RepoInfo")
+    assert isinstance(repo_info, dict), f"Expected dict body, got {type(repo_info)!r}"
+    assert repo_info["name"] == "runtime"
+    assert repo_info["visibility"] == "public"
+    assert repo_info["stars"] == 12345
+
+    repo_headers = local.get("RepoHeaders")
+    assert isinstance(repo_headers, dict)
+    # Single-value header surfaces as plain string.
+    assert repo_headers.get("content-type") == "application/json"
+    assert repo_headers.get("x-ratelimit-remaining") == "59"
+
+    # Stub got the right call.
+    assert len(handler.received) == 1
+    sent = handler.received[0]
+    assert sent.method == "GET"
+    assert sent.url == "https://api.github.com/repos/dotnet/runtime"
+    assert sent.headers["Accept"] == "application/vnd.github+json"
+    assert sent.headers["User-Agent"] == "agent-framework-integration-test"
+
+
+@pytest.mark.asyncio
+async def test_http_request_yaml_missing_handler_fails_at_build_time() -> None:
+    """Without an http_request_handler, building the workflow must raise."""
+    from agent_framework_declarative._workflows._errors import DeclarativeWorkflowError
+
+    factory = WorkflowFactory()  # no handler configured
+    with pytest.raises(DeclarativeWorkflowError) as excinfo:
+        factory.create_workflow_from_yaml_path(FIXTURE_PATH)
+    msg = str(excinfo.value)
+    assert "HttpRequestAction" in msg
+    assert "http_request_handler" in msg

python/packages/declarative/tests/test_powerfx_safe.py

@@ -0,0 +1,59 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Regression tests for ``_make_powerfx_safe``.
+
+PowerFx (via pythonnet) only accepts plain primitives, dicts, and lists.
+``Enum`` instances - especially ``str``- and ``int``-subclass enums like
+MAF's ``MessageRole`` - silently pass ``isinstance(v, str)`` /
+``isinstance(v, int)`` checks but blow up later inside pythonnet with
+``'<EnumName>' value cannot be converted to System.<X>``. These tests
+pin down the Enum coercion branch so we don't regress that interop fix.
+"""
+
+from enum import Enum, IntEnum
+
+from agent_framework_declarative._workflows._declarative_base import _make_powerfx_safe
+
+
+class _StrRole(str, Enum):
+    USER = "user"
+    SYSTEM = "system"
+
+
+class _IntCode(IntEnum):
+    ONE = 1
+    TWO = 2
+
+
+class _PlainEnum(Enum):
+    X = "x"
+    Y = 42
+
+
+def test_str_subclass_enum_reduces_to_str():
+    assert _make_powerfx_safe(_StrRole.USER) == "user"
+    assert type(_make_powerfx_safe(_StrRole.USER)) is str
+
+
+def test_int_subclass_enum_reduces_to_int():
+    assert _make_powerfx_safe(_IntCode.ONE) == 1
+    assert type(_make_powerfx_safe(_IntCode.ONE)) is int
+
+
+def test_plain_enum_reduces_to_underlying_value():
+    assert _make_powerfx_safe(_PlainEnum.X) == "x"
+    assert _make_powerfx_safe(_PlainEnum.Y) == 42
+
+
+def test_enum_inside_dict_is_coerced():
+    safe = _make_powerfx_safe({"role": _StrRole.USER, "code": _IntCode.TWO})
+    assert safe == {"role": "user", "code": 2}
+    assert type(safe["role"]) is str
+    assert type(safe["code"]) is int
+
+
+def test_enum_inside_list_is_coerced():
+    safe = _make_powerfx_safe([_StrRole.USER, _IntCode.ONE])
+    assert safe == ["user", 1]
+    assert type(safe[0]) is str
+    assert type(safe[1]) is int

python/packages/declarative/tests/test_workflow_factory.py

@@ -4,10 +4,8 @@
 
 import pytest
 
-from agent_framework_declarative._workflows._factory import (
-    DeclarativeWorkflowError,
-    WorkflowFactory,
-)
+from agent_framework_declarative._workflows._errors import DeclarativeWorkflowError
+from agent_framework_declarative._workflows._factory import WorkflowFactory
 
 try:
     import powerfx  # noqa: F401
@@ -228,6 +226,94 @@ actions:
         outputs = result.get_outputs()
         assert any("hello-world" in str(o) for o in outputs), f"Expected 'hello-world' in outputs but got: {outputs}"
 
+    async def test_as_agent_round_trip_with_last_message_text(self):
+        """Regression test: a declarative workflow built via WorkflowFactory must be
+        consumable as an AIAgent via Workflow.as_agent().
+
+        Specifically, the declarative start executor must accept list[Message]
+        (the input passed by WorkflowAgent) and populate System.LastMessageText
+        so =System.LastMessageText is resolvable in the YAML.
+        """
+        factory = WorkflowFactory()
+        workflow = factory.create_workflow_from_yaml("""
+name: as-agent-roundtrip-test
+actions:
+  - kind: SetVariable
+    variable: Local.echo
+    value: =System.LastMessageText
+  - kind: SendActivity
+    activity:
+      text: =Local.echo
+""")
+
+        agent = workflow.as_agent(name="echo-agent")
+        response = await agent.run("Hello there")
+
+        assert "Hello there" in response.text, (
+            f"Expected 'Hello there' in agent response text but got: {response.text!r}"
+        )
+
+    async def test_as_agent_continuation_preserves_prior_state(self):
+        """Regression test for the ``is_continuation`` branch in
+        ``DeclarativeWorkflowExecutor._ensure_state_initialized``.
+
+        Verifies, end-to-end via ``Workflow.as_agent()``:
+          * Turn 1 initializes the declarative state via ``state.initialize``.
+          * Turn 2 takes the *continuation* branch (skips ``state.initialize``),
+            so any non-Inputs/non-System state stamped on turn 1 survives.
+          * Turn 2 still refreshes ``Inputs.input`` and
+            ``System.LastMessage*`` to the new user message.
+
+        Without state preservation, ``Workflow.run`` would clear shared state
+        on entry and ``state.initialize`` would re-run on every turn,
+        wiping the marker we stamped between calls.
+        """
+        from agent_framework_declarative._workflows._declarative_base import DECLARATIVE_STATE_KEY
+
+        factory = WorkflowFactory()
+        workflow = factory.create_workflow_from_yaml("""
+name: as-agent-continuation-test
+actions:
+  - kind: SendActivity
+    activity:
+      text: =System.LastMessageText
+""")
+
+        agent = workflow.as_agent(name="continuation-agent")
+
+        first = await agent.run("turn-1-msg")
+        assert first.text == "turn-1-msg", f"Expected turn-1 echo 'turn-1-msg', got: {first.text!r}"
+
+        # Stamp a marker into the declarative state between turns. The
+        # continuation branch must preserve it; a state-clearing run would
+        # wipe ``DECLARATIVE_STATE_KEY`` and force re-initialization.
+        state_data = workflow._state.get(DECLARATIVE_STATE_KEY)
+        assert isinstance(state_data, dict), "Expected declarative state to be initialized after turn 1"
+        state_data["Local"] = {"persisted_marker": "kept-from-turn-1"}
+        workflow._state.set(DECLARATIVE_STATE_KEY, state_data)
+        workflow._state.commit()
+
+        second = await agent.run("turn-2-msg")
+        assert second.text == "turn-2-msg", (
+            f"Expected System.LastMessageText to refresh to 'turn-2-msg', got: {second.text!r}"
+        )
+
+        # The continuation branch in ``_ensure_state_initialized`` must:
+        # 1. preserve the cross-turn marker we stamped above
+        # 2. refresh Inputs.input and System.LastMessage* to the new turn
+        post_state = workflow._state.get(DECLARATIVE_STATE_KEY)
+        assert isinstance(post_state, dict), "declarative state vanished between turns"
+        local = post_state.get("Local", {})
+        assert local.get("persisted_marker") == "kept-from-turn-1", (
+            f"Cross-turn marker was wiped (state was reset). post_state Local={local!r}"
+        )
+        assert post_state.get("Inputs", {}).get("input") == "turn-2-msg", (
+            f"Inputs.input not refreshed on turn 2: {post_state.get('Inputs')!r}"
+        )
+        assert post_state.get("System", {}).get("LastMessageText") == "turn-2-msg", (
+            f"System.LastMessageText not refreshed on turn 2: {post_state.get('System')!r}"
+        )
+
 
 class TestWorkflowFactoryAgentRegistration:
     """Tests for agent registration."""

python/packages/declarative/tests/workflows/http_request.yaml

@@ -0,0 +1,29 @@
+#
+# Integration fixture: end-to-end HttpRequestAction round-trip using a
+# stub HttpRequestHandler. Mirrors the .NET integration fixture in
+# dotnet/tests/.../Workflows/HttpRequest.yaml.
+#
+kind: Workflow
+trigger:
+
+  kind: OnConversationStart
+  id: workflow_http_request_test
+  actions:
+
+    # Set the repo owner used to form the request URL.
+    - kind: SetVariable
+      id: set_repo_owner
+      variable: Local.RepoOwner
+      value: dotnet
+
+    # Invoke the (stubbed) GitHub repo API.
+    - kind: HttpRequestAction
+      id: fetch_repo_info
+      conversationId: =System.ConversationId
+      method: GET
+      url: =Concatenate("https://api.github.com/repos/", Local.RepoOwner, "/runtime")
+      headers:
+        Accept: application/vnd.github+json
+        User-Agent: agent-framework-integration-test
+      response: Local.RepoInfo
+      responseHeaders: Local.RepoHeaders

python/packages/devui/pyproject.toml

@@ -4,7 +4,7 @@ description = "Debug UI for Microsoft Agent Framework with OpenAI-compatible API
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://github.com/microsoft/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -23,7 +23,7 @@ classifiers = [
   "Typing :: Typed",
 ]
  dependencies = [
-     "agent-framework-core>=1.2.1,<2",
+     "agent-framework-core>=1.2.2,<2",
      "openai>=1.99.0,<3",
      "opentelemetry-sdk>=1.39.0,<2",
      "fastapi>=0.115.0,<0.133.1",

python/packages/durabletask/pyproject.toml

@@ -4,7 +4,7 @@ description = "Durable Task integration for Microsoft Agent Framework."
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -22,7 +22,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
     "durabletask>=1.3.0,<2",
     "durabletask-azuremanaged>=1.3.0,<2",
     "python-dateutil>=2.8.0,<3",

python/packages/foundry/pyproject.toml

@@ -4,7 +4,7 @@ description = "Microsoft Foundry integrations for Microsoft Agent Framework."
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.2.1"
+version = "1.2.2"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -23,8 +23,8 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
-    "agent-framework-openai>=1.1.0,<2",
+    "agent-framework-core>=1.2.2,<2",
+    "agent-framework-openai>=1.2.2,<2",
     "azure-ai-inference>=1.0.0b9,<1.0.0b10",
     "azure-ai-projects>=2.1.0,<3.0",
 ]

python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py

@@ -272,50 +272,86 @@ class ResponsesHostServer(ResponsesAgentServerHost):
         if not isinstance(self._agent, WorkflowAgent):
             raise RuntimeError("Agent is not a workflow agent.")
 
-        # Restore from the latest checkpoint if available, otherwise start with an empty history
+        # Determine the latest checkpoint (if any) so we can resume the
+        # workflow's prior state for this turn. The directory is keyed by
+        # the inbound context id (conversation_id when set, otherwise
+        # previous_response_id). Multi-turn declarative workflows need the
+        # workflow's internal state (e.g. Conversation.messages,
+        # intermediate Local.* variables) to survive across user turns;
+        # the only place that state lives is the workflow checkpoint, so
+        # on every turn we restore the latest checkpoint and feed the new
+        # input back into the start executor as a continuation rather than
+        # a fresh run.
+        latest_checkpoint_id: str | None = None
+        restore_storage: FileCheckpointStorage | None = None
         if context_id is not None:
-            checkpoint_storage = FileCheckpointStorage(os.path.join(self._checkpoint_storage_path, context_id))
-            latest_checkpoint = await checkpoint_storage.get_latest(workflow_name=self._agent.workflow.name)
+            restore_storage = FileCheckpointStorage(os.path.join(self._checkpoint_storage_path, context_id))
+            latest_checkpoint = await restore_storage.get_latest(workflow_name=self._agent.workflow.name)
             if latest_checkpoint is not None:
-                if not is_streaming_request:
-                    _ = await self._agent.run(
-                        stream=False,
-                        checkpoint_id=latest_checkpoint.checkpoint_id,
-                        checkpoint_storage=checkpoint_storage,
-                    )
-                else:
-                    # Consume the streaming or the invocation will result in a no-op
-                    async for _ in self._agent.run(
-                        stream=True,
-                        checkpoint_id=latest_checkpoint.checkpoint_id,
-                        checkpoint_storage=checkpoint_storage,
-                    ):
-                        pass
+                latest_checkpoint_id = latest_checkpoint.checkpoint_id
+
+        # Storage that will receive checkpoints written during this turn.
+        # When the caller chains with previous_response_id, the next turn
+        # will reference the current response_id as its previous_response_id,
+        # so new checkpoints must land under the current response_id (or the
+        # conversation_id when set). When conversation_id is set, this
+        # matches restore_storage; when only previous_response_id was
+        # supplied, restore_storage points at the *prior* response's
+        # directory and write_storage points at the *current* response's.
+        write_context_id = context.conversation_id or context.response_id
+        write_storage = FileCheckpointStorage(os.path.join(self._checkpoint_storage_path, write_context_id))
+
+        # Multi-turn pattern: when we have a prior checkpoint, restore it
+        # first (drive the workflow back to idle with prior state intact),
+        # then make a separate call that delivers the new user input. This
+        # depends on Workflow.run preserving shared state across calls. The
+        # restore-only call may yield events from any pending in-flight
+        # work in the checkpoint; we consume those internally here so they
+        # don't surface to the response stream as duplicates.
+        #
+        # If the restored checkpoint had pending request_info events, the
+        # restore-only call replays them through
+        # ``WorkflowAgent._convert_workflow_event_to_agent_response_updates``
+        # and populates ``self._agent.pending_requests``. That is the correct
+        # state: those requests are genuinely outstanding, and the next
+        # ``run(input_messages, ...)`` call may contain ``function_call_output``
+        # items (carried as FunctionResult/FunctionApprovalResponse content)
+        # that fulfill them via :meth:`WorkflowAgent._process_pending_requests`.
+        if latest_checkpoint_id is not None:
+            if is_streaming_request:
+                async for _ in self._agent.run(
+                    stream=True,
+                    checkpoint_id=latest_checkpoint_id,
+                    checkpoint_storage=restore_storage,
+                ):
+                    pass
+            else:
+                await self._agent.run(
+                    stream=False,
+                    checkpoint_id=latest_checkpoint_id,
+                    checkpoint_storage=restore_storage,
+                )
 
         # Now run the agent with the latest input
         response_event_stream = ResponseEventStream(response_id=context.response_id, model=request.model)
 
-        # Create a new checkpoint storage for this response based on the following rules:
-        # - If no previous response ID or conversation ID is provided,
-        #   create a new checkpoint storage for this response
-        # - If a previous response ID is provided, create a new checkpoint storage for this response
-        # - If a conversation ID is provided, reuse the existing checkpoint storage for the conversation
-        context_id = context.conversation_id or context.response_id
-        checkpoint_storage = FileCheckpointStorage(os.path.join(self._checkpoint_storage_path, context_id))
-
         yield response_event_stream.emit_created()
         yield response_event_stream.emit_in_progress()
 
         if not is_streaming_request:
-            # Run the agent in non-streaming mode
-            response = await self._agent.run(input_messages, stream=False, checkpoint_storage=checkpoint_storage)
+            # Run the agent in non-streaming mode with the new user input.
+            response = await self._agent.run(
+                input_messages,
+                stream=False,
+                checkpoint_storage=write_storage,
+            )
 
             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(checkpoint_storage, self._agent.workflow.name)
+            await self._delete_not_latest_checkpoints(write_storage, self._agent.workflow.name)
             yield response_event_stream.emit_completed()
             return
 
@@ -323,8 +359,12 @@ class ResponsesHostServer(ResponsesAgentServerHost):
         # lazily created on matching content, closed when a different type arrives.
         tracker = _OutputItemTracker(response_event_stream)
 
-        # Run the workflow agent in streaming mode
-        async for update in self._agent.run(input_messages, stream=True, checkpoint_storage=checkpoint_storage):
+        # Run the workflow agent in streaming mode with the new user input.
+        async for update in self._agent.run(
+            input_messages,
+            stream=True,
+            checkpoint_storage=write_storage,
+        ):
             for content in update.contents:
                 for event in tracker.handle(content):
                     yield event
@@ -337,7 +377,7 @@ class ResponsesHostServer(ResponsesAgentServerHost):
         for event in tracker.close():
             yield event
 
-        await self._delete_not_latest_checkpoints(checkpoint_storage, self._agent.workflow.name)
+        await self._delete_not_latest_checkpoints(write_storage, self._agent.workflow.name)
         yield response_event_stream.emit_completed()
 
     @staticmethod

python/packages/foundry_hosting/pyproject.toml

@@ -4,7 +4,7 @@ description = "Foundry Hosting integration for Microsoft Agent Framework."
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0a260428"
+version = "1.0.0a260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -23,10 +23,10 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
-    "azure-ai-agentserver-core==2.0.0b3",
-    "azure-ai-agentserver-responses==1.0.0b5",
-    "azure-ai-agentserver-invocations==1.0.0b3",
+    "agent-framework-core>=1.2.2,<2",
+    "azure-ai-agentserver-core>=2.0.0b3,<3",
+    "azure-ai-agentserver-responses>=1.0.0b5,<2",
+    "azure-ai-agentserver-invocations>=1.0.0b3,<2",
 ]
 
 [tool.uv]

python/packages/foundry_local/pyproject.toml

@@ -4,7 +4,7 @@ description = "Foundry Local integration for Microsoft Agent Framework."
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -23,7 +23,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
     "agent-framework-openai>=1.1.0,<2",
     "foundry-local-sdk>=0.5.1,<0.5.2",
 ]

python/packages/gemini/agent_framework_gemini/_chat_client.py

@@ -823,19 +823,28 @@ class RawGeminiChatClient(
 
         match tool_mode.get("mode"):
             case "auto":
-                function_calling_mode, allowed_names = types.FunctionCallingConfigMode.AUTO, None
+                if "allowed_tools" in tool_mode:
+                    function_calling_mode = types.FunctionCallingConfigMode.VALIDATED
+                    allowed_names = list(tool_mode["allowed_tools"])
+                else:
+                    function_calling_mode, allowed_names = types.FunctionCallingConfigMode.AUTO, None
             case "none":
                 function_calling_mode, allowed_names = types.FunctionCallingConfigMode.NONE, None
             case "required":
                 function_calling_mode = types.FunctionCallingConfigMode.ANY
                 name = tool_mode.get("required_function_name")
-                allowed_names = [name] if name else None
+                if name:
+                    allowed_names = [name]
+                elif "allowed_tools" in tool_mode:
+                    allowed_names = list(tool_mode["allowed_tools"])
+                else:
+                    allowed_names = None
             case unknown_mode:
                 logger.warning("Unsupported tool_choice mode for Gemini: %s", unknown_mode)
                 return None
 
         function_calling_kwargs: dict[str, Any] = {"mode": function_calling_mode}
-        if allowed_names:
+        if allowed_names is not None:
             function_calling_kwargs["allowed_function_names"] = allowed_names
 
         return types.ToolConfig(function_calling_config=types.FunctionCallingConfig(**function_calling_kwargs))

python/packages/gemini/pyproject.toml

@@ -4,7 +4,7 @@ description = "Google Gemini integration for Microsoft Agent Framework."
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0a260428"
+version = "1.0.0a260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -24,7 +24,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2.0",
+    "agent-framework-core>=1.2.2,<2.0",
     "google-genai>=1.65.0,<2.0.0",
 ]
 

python/packages/gemini/tests/test_gemini_client.py

@@ -1157,6 +1157,86 @@ async def test_unknown_tool_choice_mode_is_ignored() -> None:
     assert not hasattr(config, "tool_config") or config.tool_config is None
 
 
+async def test_tool_choice_auto_with_allowed_tools_uses_VALIDATED() -> None:
+    """Maps auto + allowed_tools to FunctionCallingConfigMode.VALIDATED with allowed_function_names."""
+    tool = _make_dummy_tool()
+    client, mock = _make_gemini_client()
+    mock.aio.models.generate_content = AsyncMock(return_value=_make_response([_make_part(text="Hi")]))
+
+    await client.get_response(
+        messages=[Message(role="user", contents=[Content.from_text("Hi")])],
+        options={
+            "tools": [tool],
+            "tool_choice": {"mode": "auto", "allowed_tools": ["dummy", "other"]},
+        },
+    )
+
+    config: types.GenerateContentConfig = mock.aio.models.generate_content.call_args.kwargs["config"]
+    function_calling_config = config.tool_config.function_calling_config
+    assert function_calling_config.mode == "VALIDATED"
+    assert function_calling_config.allowed_function_names == ["dummy", "other"]
+
+
+async def test_tool_choice_auto_with_empty_allowed_tools_uses_VALIDATED() -> None:
+    """Maps auto + empty allowed_tools to VALIDATED with empty allowed_function_names."""
+    tool = _make_dummy_tool()
+    client, mock = _make_gemini_client()
+    mock.aio.models.generate_content = AsyncMock(return_value=_make_response([_make_part(text="Hi")]))
+
+    await client.get_response(
+        messages=[Message(role="user", contents=[Content.from_text("Hi")])],
+        options={
+            "tools": [tool],
+            "tool_choice": {"mode": "auto", "allowed_tools": []},
+        },
+    )
+
+    config: types.GenerateContentConfig = mock.aio.models.generate_content.call_args.kwargs["config"]
+    function_calling_config = config.tool_config.function_calling_config
+    assert function_calling_config.mode == "VALIDATED"
+    assert function_calling_config.allowed_function_names == []
+
+
+async def test_tool_choice_required_with_allowed_tools_uses_ANY() -> None:
+    """Maps required + allowed_tools to ANY with allowed_function_names."""
+    tool = _make_dummy_tool()
+    client, mock = _make_gemini_client()
+    mock.aio.models.generate_content = AsyncMock(return_value=_make_response([_make_part(text="Hi")]))
+
+    await client.get_response(
+        messages=[Message(role="user", contents=[Content.from_text("Hi")])],
+        options={
+            "tools": [tool],
+            "tool_choice": {"mode": "required", "allowed_tools": ["dummy"]},
+        },
+    )
+
+    config: types.GenerateContentConfig = mock.aio.models.generate_content.call_args.kwargs["config"]
+    function_calling_config = config.tool_config.function_calling_config
+    assert function_calling_config.mode == "ANY"
+    assert function_calling_config.allowed_function_names == ["dummy"]
+
+
+async def test_tool_choice_required_function_name_takes_precedence_over_allowed_tools() -> None:
+    """When both required_function_name and allowed_tools are present, required_function_name wins."""
+    tool = _make_dummy_tool()
+    client, mock = _make_gemini_client()
+    mock.aio.models.generate_content = AsyncMock(return_value=_make_response([_make_part(text="Hi")]))
+
+    await client.get_response(
+        messages=[Message(role="user", contents=[Content.from_text("Hi")])],
+        options={
+            "tools": [tool],
+            "tool_choice": {"mode": "required", "required_function_name": "dummy", "allowed_tools": ["other"]},
+        },
+    )
+
+    config: types.GenerateContentConfig = mock.aio.models.generate_content.call_args.kwargs["config"]
+    function_calling_config = config.tool_config.function_calling_config
+    assert function_calling_config.mode == "ANY"
+    assert function_calling_config.allowed_function_names == ["dummy"]
+
+
 # built-in tool factories
 
 

python/packages/github_copilot/pyproject.toml

@@ -4,7 +4,7 @@ description = "GitHub Copilot integration for Microsoft Agent Framework."
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -23,7 +23,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
     "github-copilot-sdk>=0.2.1,<=0.2.1; python_version >= '3.11'",
 ]
 

python/packages/hyperlight/pyproject.toml

@@ -4,7 +4,7 @@ description = "Hyperlight CodeAct integrations for Microsoft Agent Framework."
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0a260428"
+version = "1.0.0a260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -22,7 +22,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
     "hyperlight-sandbox>=0.3.0,<0.4",
     "hyperlight-sandbox-backend-wasm>=0.3.0,<0.4 ; ((sys_platform == 'linux' and platform_machine == 'x86_64') or (sys_platform == 'win32' and platform_machine == 'AMD64')) and python_version < '3.14'",
     "hyperlight-sandbox-python-guest>=0.3.0,<0.4",

python/packages/lab/pyproject.toml

@@ -4,7 +4,7 @@ description = "Experimental modules for Microsoft Agent Framework"
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -22,7 +22,7 @@ classifiers = [
   "Programming Language :: Python :: 3.14",
 ]
 dependencies = [
-  "agent-framework-core>=1.2.1,<2",
+  "agent-framework-core>=1.2.2,<2",
 ]
 
 [project.optional-dependencies]

python/packages/mem0/pyproject.toml

@@ -4,7 +4,7 @@ description = "Mem0 integration for Microsoft Agent Framework."
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -23,7 +23,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
     "mem0ai>=1.0.0,<2",
 ]
 

python/packages/ollama/pyproject.toml

@@ -4,7 +4,7 @@ description = "Ollama integration for Microsoft Agent Framework."
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://learn.microsoft.com/en-us/agent-framework/"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -23,7 +23,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
     "ollama>=0.5.3,<0.5.4",
 ]
 

python/packages/openai/agent_framework_openai/_chat_client.py

@@ -241,6 +241,85 @@ OpenAIChatOptionsT = TypeVar(
 # endregion
 
 
+# region Helpers
+
+
+def _annotations_to_output_text(annotations: Sequence[Annotation] | None) -> list[dict[str, Any]]:
+    """Convert framework `Annotation` objects to Responses API `output_text` annotation dicts.
+
+    Citations from `file_search`, `code_interpreter` file paths, and url citations all collapse
+    to `Annotation(type="citation", ...)` in the framework. The original API form is recovered
+    here so assistant messages roundtrip cleanly through history forwarding.
+
+    Each Responses API annotation dict carries at most one `start_index`/`end_index` pair, so an
+    `Annotation` with multiple `annotated_regions` is fanned out into one entry per region.
+    Regions missing valid integer span bounds are skipped.
+    """
+    if not annotations:
+        return []
+    out: list[dict[str, Any]] = []
+    for annotation in annotations:
+        if annotation.get("type") != "citation":
+            continue
+        props = annotation.get("additional_properties") or {}
+        regions = annotation.get("annotated_regions") or []
+        file_id = annotation.get("file_id")
+        url = annotation.get("url")
+        title = annotation.get("title")
+        container_id = props.get("container_id")
+
+        if container_id and file_id:
+            for region in regions:
+                start = region.get("start_index")
+                end = region.get("end_index")
+                if not (isinstance(start, int) and isinstance(end, int)):
+                    continue
+                entry: dict[str, Any] = {
+                    "type": "container_file_citation",
+                    "container_id": container_id,
+                    "file_id": file_id,
+                    "start_index": start,
+                    "end_index": end,
+                }
+                if url:
+                    entry["filename"] = url
+                out.append(entry)
+        elif url and not file_id and regions:
+            for region in regions:
+                start = region.get("start_index")
+                end = region.get("end_index")
+                if not (isinstance(start, int) and isinstance(end, int)):
+                    continue
+                out.append({
+                    "type": "url_citation",
+                    "url": url,
+                    "title": title or "",
+                    "start_index": start,
+                    "end_index": end,
+                })
+        elif file_id and url:
+            entry = {
+                "type": "file_citation",
+                "file_id": file_id,
+                "filename": url,
+            }
+            if (idx := props.get("index")) is not None:
+                entry["index"] = idx
+            out.append(entry)
+        elif file_id:
+            entry = {
+                "type": "file_path",
+                "file_id": file_id,
+            }
+            if (idx := props.get("index")) is not None:
+                entry["index"] = idx
+            out.append(entry)
+    return out
+
+
+# endregion
+
+
 # region ResponsesClient
 
 
@@ -1217,6 +1296,12 @@ class RawOpenAIChatClient(  # type: ignore[misc]
                             "type": "function",
                             "name": func_name,
                         }
+                    elif mode == "auto" and (allowed := tool_mode.get("allowed_tools")) is not None:
+                        run_options["tool_choice"] = {
+                            "type": "allowed_tools",
+                            "mode": "auto",
+                            "tools": [{"type": "function", "name": name} for name in allowed],
+                        }
                     else:
                         run_options["tool_choice"] = mode
         else:
@@ -1374,7 +1459,7 @@ class RawOpenAIChatClient(  # type: ignore[misc]
                     return {
                         "type": "output_text",
                         "text": content.text,
-                        "annotations": [],
+                        "annotations": _annotations_to_output_text(getattr(content, "annotations", None)),
                     }
                 return {
                     "type": "input_text",
@@ -1522,6 +1607,13 @@ class RawOpenAIChatClient(  # type: ignore[misc]
                     "approve": content.approved,
                 }
             case "hosted_file":
+                # `input_file` is an input-only content type in the Responses API and is rejected
+                # inside an assistant message. Hosted-file content on an assistant message
+                # represents a citation produced by a hosted tool (e.g., file_search) and cannot be
+                # meaningfully replayed as input — drop it. The accompanying text annotations carry
+                # the citation context for round-tripping.
+                if role == "assistant":
+                    return {}
                 return {
                     "type": "input_file",
                     "file_id": content.file_id,
@@ -2502,45 +2594,63 @@ class RawOpenAIChatClient(  # type: ignore[misc]
 
                 ann_type = _get_ann_value("type")
                 ann_file_id = _get_ann_value("file_id")
+                # Hosted-file citations attach as text annotations (matching the non-streaming path)
+                # so they don't roundtrip as standalone `input_file` items in assistant history.
                 if ann_type == "file_path":
                     if ann_file_id:
+                        annotation_obj = Annotation(
+                            type="citation",
+                            file_id=str(ann_file_id),
+                            additional_properties={
+                                "annotation_index": event.annotation_index,
+                                "index": _get_ann_value("index"),
+                            },
+                            raw_representation=annotation,
+                        )
                         contents.append(
-                            Content.from_hosted_file(
-                                file_id=str(ann_file_id),
-                                additional_properties={
-                                    "annotation_index": event.annotation_index,
-                                    "index": _get_ann_value("index"),
-                                },
-                                raw_representation=event,
-                            )
+                            Content.from_text(text="", annotations=[annotation_obj], raw_representation=event)
                         )
                 elif ann_type == "file_citation":
                     if ann_file_id:
+                        ann_filename = _get_ann_value("filename")
+                        annotation_obj = Annotation(
+                            type="citation",
+                            file_id=str(ann_file_id),
+                            url=ann_filename,
+                            additional_properties={
+                                "annotation_index": event.annotation_index,
+                                "index": _get_ann_value("index"),
+                            },
+                            raw_representation=annotation,
+                        )
                         contents.append(
-                            Content.from_hosted_file(
-                                file_id=str(ann_file_id),
-                                additional_properties={
-                                    "annotation_index": event.annotation_index,
-                                    "filename": _get_ann_value("filename"),
-                                    "index": _get_ann_value("index"),
-                                },
-                                raw_representation=event,
-                            )
+                            Content.from_text(text="", annotations=[annotation_obj], raw_representation=event)
                         )
                 elif ann_type == "container_file_citation":
                     if ann_file_id:
+                        ann_filename = _get_ann_value("filename")
+                        ann_start = _get_ann_value("start_index")
+                        ann_end = _get_ann_value("end_index")
+                        annotation_obj = Annotation(
+                            type="citation",
+                            file_id=str(ann_file_id),
+                            url=ann_filename,
+                            additional_properties={
+                                "annotation_index": event.annotation_index,
+                                "container_id": _get_ann_value("container_id"),
+                            },
+                            raw_representation=annotation,
+                        )
+                        if ann_start is not None and ann_end is not None:
+                            annotation_obj["annotated_regions"] = [
+                                TextSpanRegion(
+                                    type="text_span",
+                                    start_index=ann_start,
+                                    end_index=ann_end,
+                                )
+                            ]
                         contents.append(
-                            Content.from_hosted_file(
-                                file_id=str(ann_file_id),
-                                additional_properties={
-                                    "annotation_index": event.annotation_index,
-                                    "container_id": _get_ann_value("container_id"),
-                                    "filename": _get_ann_value("filename"),
-                                    "start_index": _get_ann_value("start_index"),
-                                    "end_index": _get_ann_value("end_index"),
-                                },
-                                raw_representation=event,
-                            )
+                            Content.from_text(text="", annotations=[annotation_obj], raw_representation=event)
                         )
                 elif ann_type == "url_citation":
                     ann_url = _get_ann_value("url")

python/packages/openai/agent_framework_openai/_chat_completion_client.py

@@ -662,6 +662,12 @@ class RawOpenAIChatCompletionClient(  # type: ignore[misc]
                         "type": "function",
                         "function": {"name": func_name},
                     }
+                elif mode in ("auto", "required") and tool_mode.get("allowed_tools") is not None:
+                    logger.warning(
+                        "allowed_tools is not supported by the Chat Completions API; "
+                        "the setting will be ignored. Use OpenAIChatClient (Responses API) instead."
+                    )
+                    run_options["tool_choice"] = mode
                 else:
                     run_options["tool_choice"] = mode
 

python/packages/openai/pyproject.toml

@@ -4,7 +4,7 @@ description = "OpenAI integrations for Microsoft Agent Framework."
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.2.1"
+version = "1.2.2"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -23,7 +23,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
     "openai>=1.99.0,<3",
 ]
 

python/packages/openai/tests/openai/test_openai_chat_client.py

@@ -1914,6 +1914,285 @@ def test_hosted_file_content_preparation() -> None:
     assert result["file_id"] == "file_abc123"
 
 
+def test_assistant_text_preserves_citation_annotations_on_roundtrip() -> None:
+    """Citation annotations on assistant text should survive serialization back to the Responses API.
+
+    Previously `output_text.annotations` was hardcoded to `[]`, silently dropping `file_search`
+    citation context on every roundtrip. Preserving them keeps citations intact across
+    multi-agent forwarding.
+    """
+    from agent_framework._types import Annotation, TextSpanRegion
+
+    client = OpenAIChatClient(model="test-model", api_key="test-key")
+
+    text_content = Content.from_text(
+        "Per the docs, the answer is X. See also the report.",
+        annotations=[
+            Annotation(
+                type="citation",
+                file_id="file-abc123",
+                url="guidelines.md",
+                additional_properties={"index": 12},
+            ),
+            Annotation(
+                type="citation",
+                title="Quarterly Report",
+                url="https://example.com/report",
+                annotated_regions=[TextSpanRegion(type="text_span", start_index=40, end_index=46)],
+            ),
+            Annotation(
+                type="citation",
+                file_id="file-container456",
+                url="data.csv",
+                additional_properties={"container_id": "container-789"},
+                annotated_regions=[TextSpanRegion(type="text_span", start_index=0, end_index=3)],
+            ),
+        ],
+    )
+
+    result = client._prepare_content_for_openai("assistant", text_content)
+
+    assert result["type"] == "output_text"
+    annotations = result["annotations"]
+    assert len(annotations) == 3
+
+    file_citation = next(a for a in annotations if a["type"] == "file_citation")
+    assert file_citation["file_id"] == "file-abc123"
+    assert file_citation["filename"] == "guidelines.md"
+    assert file_citation["index"] == 12
+
+    url_citation = next(a for a in annotations if a["type"] == "url_citation")
+    assert url_citation["url"] == "https://example.com/report"
+    assert url_citation["title"] == "Quarterly Report"
+    assert url_citation["start_index"] == 40
+    assert url_citation["end_index"] == 46
+
+    container = next(a for a in annotations if a["type"] == "container_file_citation")
+    assert container["file_id"] == "file-container456"
+    assert container["container_id"] == "container-789"
+    assert container["filename"] == "data.csv"
+    assert container["start_index"] == 0
+    assert container["end_index"] == 3
+
+
+def test_assistant_text_preserves_file_path_annotation() -> None:
+    """A `file_path`-style citation (file_id only, no url) should serialize as `file_path`."""
+    from agent_framework._types import Annotation
+
+    client = OpenAIChatClient(model="test-model", api_key="test-key")
+
+    text_content = Content.from_text(
+        "See attached.",
+        annotations=[
+            Annotation(
+                type="citation",
+                file_id="file-only",
+                additional_properties={"index": 42},
+            ),
+        ],
+    )
+
+    result = client._prepare_content_for_openai("assistant", text_content)
+
+    assert result["type"] == "output_text"
+    annotations = result["annotations"]
+    assert annotations == [{"type": "file_path", "file_id": "file-only", "index": 42}]
+
+
+def test_assistant_text_fans_out_multiple_annotated_regions() -> None:
+    """A url_citation with multiple `annotated_regions` should emit one entry per region.
+
+    The Responses API annotation dict carries one start/end pair, so a framework Annotation
+    with N regions must produce N output annotation entries.
+    """
+    from agent_framework._types import Annotation, TextSpanRegion
+
+    client = OpenAIChatClient(model="test-model", api_key="test-key")
+
+    text_content = Content.from_text(
+        "See report. The report says X. Also report.",
+        annotations=[
+            Annotation(
+                type="citation",
+                title="Report",
+                url="https://example.com/report",
+                annotated_regions=[
+                    TextSpanRegion(type="text_span", start_index=4, end_index=10),
+                    TextSpanRegion(type="text_span", start_index=16, end_index=22),
+                    TextSpanRegion(type="text_span", start_index=36, end_index=42),
+                ],
+            ),
+        ],
+    )
+
+    result = client._prepare_content_for_openai("assistant", text_content)
+    annotations = result["annotations"]
+    assert len(annotations) == 3
+    assert all(a["type"] == "url_citation" for a in annotations)
+    spans = [(a["start_index"], a["end_index"]) for a in annotations]
+    assert spans == [(4, 10), (16, 22), (36, 42)]
+
+
+def test_assistant_text_skips_regions_with_invalid_span() -> None:
+    """Regions missing integer start/end bounds are skipped rather than emitted with `None`."""
+    from agent_framework._types import Annotation, TextSpanRegion
+
+    client = OpenAIChatClient(model="test-model", api_key="test-key")
+
+    text_content = Content.from_text(
+        "See report.",
+        annotations=[
+            Annotation(
+                type="citation",
+                title="Report",
+                url="https://example.com/report",
+                annotated_regions=[
+                    TextSpanRegion(type="text_span"),  # type: ignore[typeddict-item]
+                    TextSpanRegion(type="text_span", start_index=4, end_index=10),
+                ],
+            ),
+        ],
+    )
+
+    result = client._prepare_content_for_openai("assistant", text_content)
+    annotations = result["annotations"]
+    assert len(annotations) == 1
+    assert annotations[0]["start_index"] == 4
+    assert annotations[0]["end_index"] == 10
+
+
+def test_assistant_text_without_annotations_emits_empty_list() -> None:
+    """Plain assistant text should still emit `annotations: []` (Azure validation requires the field)."""
+    client = OpenAIChatClient(model="test-model", api_key="test-key")
+
+    result = client._prepare_content_for_openai("assistant", Content.from_text("hello"))
+
+    assert result["type"] == "output_text"
+    assert result["text"] == "hello"
+    assert result["annotations"] == []
+
+
+def test_streamed_file_citation_coalesces_onto_surrounding_text() -> None:
+    """Streamed citation events emit empty-text Content with annotations; `_finalize_response`
+    coalesces consecutive text contents and unions their annotations, so the citation lands on
+    the merged assistant text content (not a stray empty-text entry).
+
+    Without this, span indices in the annotation would reference `text == ""` after roundtrip.
+    """
+    text_event = MagicMock()
+    text_event.type = "response.output_text.delta"
+    text_event.delta = "Hello world."
+    text_event.item_id = "item_1"
+    text_event.output_index = 0
+    text_event.content_index = 0
+
+    citation_event = MagicMock()
+    citation_event.type = "response.output_text.annotation.added"
+    citation_event.annotation_index = 0
+    citation_event.annotation = {
+        "type": "file_citation",
+        "file_id": "file-abc",
+        "filename": "guidelines.md",
+        "index": 5,
+    }
+
+    client = OpenAIChatClient(model="test-model", api_key="test-key")
+    chat_options = ChatOptions()
+    function_call_ids: dict[int, tuple[str, str]] = {}
+
+    update1 = client._parse_chunk_from_openai(text_event, chat_options, function_call_ids)
+    update2 = client._parse_chunk_from_openai(citation_event, chat_options, function_call_ids)
+
+    response = ChatResponse.from_updates([update1, update2])
+
+    assert len(response.messages) == 1
+    contents = response.messages[0].contents
+    assert len(contents) == 1
+    merged = contents[0]
+    assert merged.type == "text"
+    assert merged.text == "Hello world."
+    assert merged.annotations is not None
+    assert len(merged.annotations) == 1
+    assert merged.annotations[0]["file_id"] == "file-abc"
+
+
+def test_streamed_file_citation_roundtrips_as_assistant_history() -> None:
+    """End-to-end: file_citation arrives via streaming, then gets forwarded as assistant history.
+
+    Reproduces the user-reported sequential/group-chat workflow bug where one agent's
+    `file_search` citations became `input_file` items in the next agent's request and were
+    rejected by the Responses API.
+    """
+    client = OpenAIChatClient(model="test-model", api_key="test-key")
+    chat_options = ChatOptions()
+    function_call_ids: dict[int, tuple[str, str]] = {}
+
+    text_event = MagicMock()
+    text_event.type = "response.output_text.delta"
+    text_event.delta = "According to the docs, the answer is X."
+    text_event.item_id = "item_1"
+    text_event.output_index = 0
+    text_event.content_index = 0
+
+    citation_event = MagicMock()
+    citation_event.type = "response.output_text.annotation.added"
+    citation_event.annotation_index = 0
+    citation_event.annotation = {
+        "type": "file_citation",
+        "file_id": "file-xyz789",
+        "filename": "guidelines.md",
+        "index": 12,
+    }
+
+    update1 = client._parse_chunk_from_openai(text_event, chat_options, function_call_ids)
+    update2 = client._parse_chunk_from_openai(citation_event, chat_options, function_call_ids)
+
+    assistant_history = Message(
+        role="assistant",
+        contents=[*update1.contents, *update2.contents],
+    )
+    prepared = client._prepare_message_for_openai(assistant_history)
+
+    assert len(prepared) == 1
+    content_items = prepared[0].get("content", [])
+    types = [c.get("type") for c in content_items]
+    assert "input_file" not in types, f"input_file leaked into assistant history: {types}"
+    output_text_items = [c for c in content_items if c.get("type") == "output_text"]
+    assert any(
+        any(a.get("type") == "file_citation" and a.get("file_id") == "file-xyz789" for a in c.get("annotations", []))
+        for c in output_text_items
+    ), "file_citation annotation should survive the streaming → history roundtrip"
+
+
+def test_hosted_file_in_assistant_message_does_not_emit_input_file() -> None:
+    """Hosted file citations attached to an assistant message must not roundtrip as `input_file`.
+
+    The Responses API rejects `input_file` items inside an assistant role's content array;
+    `input_file` is an input-only content type. This guards the multi-agent / sequential workflow
+    case where one agent's `file_search` citations get forwarded as history to the next call.
+    """
+    client = OpenAIChatClient(model="test-model", api_key="test-key")
+
+    assistant_msg = Message(
+        role="assistant",
+        contents=[
+            Content.from_text("According to the docs, the answer is X."),
+            Content.from_hosted_file(file_id="file_abc123"),
+        ],
+    )
+
+    prepared = client._prepare_message_for_openai(assistant_msg)
+
+    assert len(prepared) == 1
+    assistant_item = prepared[0]
+    assert assistant_item["role"] == "assistant"
+    content_types = [c.get("type") for c in assistant_item.get("content", [])]
+    assert "input_file" not in content_types, (
+        f"`input_file` is not valid inside an assistant message; got {content_types}"
+    )
+    assert "output_text" in content_types
+
+
 def test_function_approval_response_with_mcp_tool_call() -> None:
     """Test function approval response content with MCP server tool call content."""
     client = OpenAIChatClient(model="test-model", api_key="test-key")
@@ -2682,7 +2961,7 @@ def test_streaming_response_in_progress_type() -> None:
 
 
 def test_streaming_annotation_added_with_file_path() -> None:
-    """Test streaming annotation added event with file_path type extracts HostedFileContent."""
+    """Streaming `file_path` should attach as a text annotation, matching non-streaming."""
     client = OpenAIChatClient(model="test-model", api_key="test-key")
     chat_options = ChatOptions()
     function_call_ids: dict[int, tuple[str, str]] = {}
@@ -2700,15 +2979,23 @@ def test_streaming_annotation_added_with_file_path() -> None:
 
     assert len(response.contents) == 1
     content = response.contents[0]
-    assert content.type == "hosted_file"
-    assert content.file_id == "file-abc123"
-    assert content.additional_properties is not None
-    assert content.additional_properties.get("annotation_index") == 0
-    assert content.additional_properties.get("index") == 42
+    assert content.type == "text"
+    assert content.annotations is not None
+    assert len(content.annotations) == 1
+    annotation = content.annotations[0]
+    assert annotation["type"] == "citation"
+    assert annotation["file_id"] == "file-abc123"
+    assert annotation["additional_properties"]["annotation_index"] == 0
+    assert annotation["additional_properties"]["index"] == 42
 
 
 def test_streaming_annotation_added_with_file_citation() -> None:
-    """Test streaming annotation added event with file_citation type extracts HostedFileContent."""
+    """Streaming `file_citation` should attach as a text annotation, matching non-streaming.
+
+    Previously the streaming path produced a standalone `HostedFileContent`, which then
+    serialized as `input_file` in assistant history and was rejected by the Responses API.
+    Annotations on text content roundtrip cleanly.
+    """
     client = OpenAIChatClient(model="test-model", api_key="test-key")
     chat_options = ChatOptions()
     function_call_ids: dict[int, tuple[str, str]] = {}
@@ -2727,15 +3014,19 @@ def test_streaming_annotation_added_with_file_citation() -> None:
 
     assert len(response.contents) == 1
     content = response.contents[0]
-    assert content.type == "hosted_file"
-    assert content.file_id == "file-xyz789"
-    assert content.additional_properties is not None
-    assert content.additional_properties.get("filename") == "sample.txt"
-    assert content.additional_properties.get("index") == 15
+    assert content.type == "text"
+    assert content.annotations is not None
+    assert len(content.annotations) == 1
+    annotation = content.annotations[0]
+    assert annotation["type"] == "citation"
+    assert annotation["file_id"] == "file-xyz789"
+    assert annotation["url"] == "sample.txt"
+    assert annotation["additional_properties"]["annotation_index"] == 1
+    assert annotation["additional_properties"]["index"] == 15
 
 
 def test_streaming_annotation_added_with_container_file_citation() -> None:
-    """Test streaming annotation added event with container_file_citation type."""
+    """Streaming `container_file_citation` should attach as a text annotation."""
     client = OpenAIChatClient(model="test-model", api_key="test-key")
     chat_options = ChatOptions()
     function_call_ids: dict[int, tuple[str, str]] = {}
@@ -2756,13 +3047,19 @@ def test_streaming_annotation_added_with_container_file_citation() -> None:
 
     assert len(response.contents) == 1
     content = response.contents[0]
-    assert content.type == "hosted_file"
-    assert content.file_id == "file-container123"
-    assert content.additional_properties is not None
-    assert content.additional_properties.get("container_id") == "container-456"
-    assert content.additional_properties.get("filename") == "data.csv"
-    assert content.additional_properties.get("start_index") == 10
-    assert content.additional_properties.get("end_index") == 50
+    assert content.type == "text"
+    assert content.annotations is not None
+    assert len(content.annotations) == 1
+    annotation = content.annotations[0]
+    assert annotation["type"] == "citation"
+    assert annotation["file_id"] == "file-container123"
+    assert annotation["url"] == "data.csv"
+    assert annotation["additional_properties"]["container_id"] == "container-456"
+    assert annotation["annotated_regions"] is not None
+    assert len(annotation["annotated_regions"]) == 1
+    region = annotation["annotated_regions"][0]
+    assert region["start_index"] == 10
+    assert region["end_index"] == 50
 
 
 def test_streaming_annotation_added_with_url_citation() -> None:
@@ -3962,6 +4259,12 @@ def test_with_callable_api_key() -> None:
             True,
             id="tool_choice_required",
         ),
+        param(
+            "tool_choice",
+            {"mode": "auto", "allowed_tools": ["get_weather"]},
+            True,
+            id="tool_choice_allowed_tools",
+        ),
         param("response_format", OutputStruct, True, id="response_format_pydantic"),
         param(
             "response_format",
@@ -4516,6 +4819,90 @@ async def test_prepare_options_excludes_continuation_token() -> None:
     assert run_options["background"] is True
 
 
+async def test_prepare_options_allowed_tools() -> None:
+    """Test that _prepare_options converts allowed_tools to OpenAI API format."""
+    client = OpenAIChatClient(model="test-model", api_key="test-key")
+
+    @tool
+    def get_weather(city: str) -> str:
+        """Get the weather for a city."""
+        return f"Sunny in {city}"
+
+    @tool
+    def search_docs(query: str) -> str:
+        """Search documentation."""
+        return f"Results for {query}"
+
+    messages = [Message(role="user", contents=[Content.from_text(text="Hello")])]
+    options: dict[str, Any] = {
+        "model": "test-model",
+        "tools": [get_weather, search_docs],
+        "tool_choice": {"mode": "auto", "allowed_tools": ["get_weather"]},
+    }
+
+    run_options = await client._prepare_options(messages, options)
+
+    assert run_options["tool_choice"] == {
+        "type": "allowed_tools",
+        "mode": "auto",
+        "tools": [{"type": "function", "name": "get_weather"}],
+    }
+
+
+async def test_prepare_options_allowed_tools_multiple() -> None:
+    """Test that _prepare_options converts multiple allowed_tools correctly."""
+    client = OpenAIChatClient(model="test-model", api_key="test-key")
+
+    @tool
+    def get_weather(city: str) -> str:
+        """Get the weather for a city."""
+        return f"Sunny in {city}"
+
+    @tool
+    def search_docs(query: str) -> str:
+        """Search documentation."""
+        return f"Results for {query}"
+
+    messages = [Message(role="user", contents=[Content.from_text(text="Hello")])]
+    options: dict[str, Any] = {
+        "model": "test-model",
+        "tools": [get_weather, search_docs],
+        "tool_choice": {"mode": "auto", "allowed_tools": ["get_weather", "search_docs"]},
+    }
+
+    run_options = await client._prepare_options(messages, options)
+
+    assert run_options["tool_choice"] == {
+        "type": "allowed_tools",
+        "mode": "auto",
+        "tools": [
+            {"type": "function", "name": "get_weather"},
+            {"type": "function", "name": "search_docs"},
+        ],
+    }
+
+
+async def test_prepare_options_auto_without_allowed_tools() -> None:
+    """Test that auto mode without allowed_tools still returns plain 'auto' string."""
+    client = OpenAIChatClient(model="test-model", api_key="test-key")
+
+    @tool
+    def get_weather(city: str) -> str:
+        """Get the weather for a city."""
+        return f"Sunny in {city}"
+
+    messages = [Message(role="user", contents=[Content.from_text(text="Hello")])]
+    options: dict[str, Any] = {
+        "model": "test-model",
+        "tools": [get_weather],
+        "tool_choice": {"mode": "auto"},
+    }
+
+    run_options = await client._prepare_options(messages, options)
+
+    assert run_options["tool_choice"] == "auto"
+
+
 # endregion
 
 

python/packages/openai/tests/openai/test_openai_chat_completion_client.py

@@ -1430,6 +1430,57 @@ def test_tool_choice_required_with_function_name(
     assert prepared_options["tool_choice"]["function"]["name"] == "get_weather"
 
 
+def test_tool_choice_allowed_tools_falls_back_to_mode(
+    openai_unit_test_env: dict[str, str],
+) -> None:
+    """Test that tool_choice with allowed_tools falls back to plain mode (Chat Completions API unsupported)."""
+    client = OpenAIChatCompletionClient()
+
+    messages = [Message(role="user", contents=["test"])]
+    options = {
+        "tools": [get_weather],
+        "tool_choice": {"mode": "auto", "allowed_tools": ["get_weather"]},
+    }
+
+    prepared_options = client._prepare_options(messages, options)
+
+    assert prepared_options["tool_choice"] == "auto"
+
+
+def test_tool_choice_allowed_tools_required_mode_falls_back(
+    openai_unit_test_env: dict[str, str],
+) -> None:
+    """Test that tool_choice with allowed_tools and required mode falls back to 'required'."""
+    client = OpenAIChatCompletionClient()
+
+    messages = [Message(role="user", contents=["test"])]
+    options = {
+        "tools": [get_weather],
+        "tool_choice": {"mode": "required", "allowed_tools": ["get_weather"]},
+    }
+
+    prepared_options = client._prepare_options(messages, options)
+
+    assert prepared_options["tool_choice"] == "required"
+
+
+def test_tool_choice_auto_dict_without_allowed_tools(
+    openai_unit_test_env: dict[str, str],
+) -> None:
+    """Test that tool_choice dict with mode auto and no allowed_tools falls through to plain 'auto'."""
+    client = OpenAIChatCompletionClient()
+
+    messages = [Message(role="user", contents=["test"])]
+    options = {
+        "tools": [get_weather],
+        "tool_choice": {"mode": "auto"},
+    }
+
+    prepared_options = client._prepare_options(messages, options)
+
+    assert prepared_options["tool_choice"] == "auto"
+
+
 def test_response_format_dict_passthrough(openai_unit_test_env: dict[str, str]) -> None:
     """Test that response_format as dict is passed through directly."""
     client = OpenAIChatCompletionClient()
@@ -1590,6 +1641,12 @@ class OutputStruct(BaseModel):
             False,
             id="tool_choice_required",
         ),
+        param(
+            "tool_choice",
+            {"mode": "auto", "allowed_tools": ["get_weather"]},
+            False,
+            id="tool_choice_allowed_tools",
+        ),
         param("response_format", OutputStruct, True, id="response_format_pydantic"),
         param(
             "response_format",

python/packages/orchestrations/pyproject.toml

@@ -4,7 +4,7 @@ description = "Orchestration patterns for Microsoft Agent Framework. Includes Se
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -23,7 +23,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
 ]
 
 [tool.uv]

python/packages/purview/pyproject.toml

@@ -4,7 +4,7 @@ description = "Microsoft Purview (Graph dataSecurityAndGovernance) integration f
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://github.com/microsoft/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -24,7 +24,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
     "azure-core>=1.30.0,<2",
     "httpx>=0.27.0,<0.29",
 ]

python/packages/redis/pyproject.toml

@@ -4,7 +4,7 @@ description = "Redis integration for Microsoft Agent Framework."
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.0.0b260428"
+version = "1.0.0b260429"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -23,7 +23,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core>=1.2.1,<2",
+    "agent-framework-core>=1.2.2,<2",
     "redis>=6.4.0,<7.2.1",
     "redisvl>=0.11.0,<0.16",
     "numpy>=2.2.6,<3"

python/pyproject.toml

@@ -4,7 +4,7 @@ description = "Microsoft Agent Framework for building AI Agents with Python. Thi
 authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
-version = "1.2.1"
+version = "1.2.2"
 license-files = ["LICENSE"]
 urls.homepage = "https://aka.ms/agent-framework"
 urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
@@ -23,7 +23,7 @@ classifiers = [
   "Typing :: Typed",
 ]
 dependencies = [
-    "agent-framework-core[all]==1.2.1",
+    "agent-framework-core[all]==1.2.2",
 ]
 
 [dependency-groups]
@@ -52,8 +52,9 @@ dev = [
 [tool.uv]
 package = false
 prerelease = "if-necessary-or-explicit"
-# Keep transitive litellm below the compromised 1.82.7/1.82.8 releases.
-constraint-dependencies = ["litellm<1.82.7"]
+# Security floors for transitive deps; overrides bypass litellm[proxy]'s strict pins.
+constraint-dependencies = ["litellm>=1.83.7", "fastapi-sso>=0.19.0"]
+override-dependencies = ["mcp[ws]>=1.27.0", "uvicorn[standard]>=0.34.0"]
 environments = [
     "sys_platform == 'darwin'",
     "sys_platform == 'linux'",
@@ -93,7 +94,6 @@ agent-framework-orchestrations = { workspace = true }
 agent-framework-purview = { workspace = true }
 agent-framework-redis = { workspace = true }
 agent-framework-azure-contentunderstanding = { workspace = true }
-litellm = { url = "https://files.pythonhosted.org/packages/57/77/0c6eca2cb049793ddf8ce9cdcd5123a35666c4962514788c4fc90edf1d3b/litellm-1.82.1-py3-none-any.whl" }
 
 [tool.ruff]
 line-length = 120

python/samples/03-workflows/declarative/invoke_http_request/main.py

@@ -0,0 +1,97 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Invoke HTTP Request sample - demonstrates the HttpRequestAction declarative action.
+
+This sample shows how to:
+  1. Configure a ``WorkflowFactory`` with a ``HttpRequestHandler`` so the YAML
+     ``HttpRequestAction`` can dispatch real HTTP calls.
+  2. Fetch JSON from a public REST endpoint (the GitHub repository API) and
+     bind the parsed response to a workflow variable.
+  3. Mirror the response body into the conversation via ``conversationId`` so
+     a downstream Foundry agent can answer questions about it using only that
+     conversation context.
+
+Security note:
+    ``DefaultHttpRequestHandler`` issues HTTP calls to whatever URL the
+    workflow author specifies and performs **no** allowlisting or SSRF
+    guards. For production use, replace it with a custom handler that
+    enforces an allowlist or DNS-rebinding-resistant policy and adds any
+    required authentication headers per call.
+
+Run with:
+    python -m samples.03-workflows.declarative.invoke_http_request.main
+"""
+
+import asyncio
+import os
+from pathlib import Path
+
+from agent_framework import Agent
+from agent_framework.declarative import (
+    DefaultHttpRequestHandler,
+    WorkflowFactory,
+)
+from agent_framework.foundry import FoundryChatClient
+from azure.identity import AzureCliCredential
+
+GITHUB_REPO_INFO_AGENT_INSTRUCTIONS = """\
+You answer the user's question about a GitHub repository using ONLY the JSON
+data already present in the conversation history. If the answer is not
+contained in the conversation, say so plainly rather than guessing. Be concise
+and helpful.
+"""
+
+
+async def main() -> None:
+    """Run the invoke HTTP request workflow."""
+    chat_client = FoundryChatClient(
+        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
+        model=os.environ["FOUNDRY_MODEL"],
+        credential=AzureCliCredential(),
+    )
+
+    # The agent has no tools — it answers the question about the GitHub
+    # repository using only the JSON data that ``HttpRequestAction`` adds to
+    # the conversation.
+    github_repo_info_agent = Agent(
+        client=chat_client,
+        name="GitHubRepoInfoAgent",
+        instructions=GITHUB_REPO_INFO_AGENT_INSTRUCTIONS,
+    )
+
+    agents = {"GitHubRepoInfoAgent": github_repo_info_agent}
+
+    # The default HttpRequestHandler is sufficient for this sample because
+    # the GitHub REST endpoint used here does not require authentication.
+    # For authenticated endpoints, supply a custom client_provider callback
+    # to DefaultHttpRequestHandler so each request can be routed through a
+    # pre-configured httpx.AsyncClient with the appropriate credentials.
+    async with DefaultHttpRequestHandler() as http_handler:
+        factory = WorkflowFactory(
+            agents=agents,
+            http_request_handler=http_handler,
+        )
+
+        workflow_path = Path(__file__).parent / "workflow.yaml"
+        workflow = factory.create_workflow_from_yaml_path(workflow_path)
+
+        print("=" * 60)
+        print("Invoke HTTP Request Workflow Demo")
+        print("=" * 60)
+        print()
+        print("Ask one question about the microsoft/agent-framework repo.")
+        print()
+
+        user_input = input("You: ").strip()  # noqa: ASYNC250
+        if not user_input:
+            user_input = "Please summarize the repository."
+
+        print("\nAgent: ", end="", flush=True)
+        async for event in workflow.run(user_input, stream=True):
+            if event.type == "output" and isinstance(event.data, str):
+                print(event.data, end="", flush=True)
+        print()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

python/samples/03-workflows/declarative/invoke_http_request/workflow.yaml

@@ -0,0 +1,57 @@
+#
+# This workflow demonstrates the HttpRequestAction declarative action.
+#
+# HttpRequestAction lets a workflow author issue an HTTP call directly from
+# YAML without writing any Python glue. It can:
+#
+#   - fetch data from external REST endpoints,
+#   - store the parsed response in a workflow variable, and
+#   - add the response body to the conversation so a downstream agent can
+#     answer questions based on it.
+#
+# This sample fetches public metadata for the microsoft/agent-framework
+# repository from the GitHub REST API (no authentication required) and uses
+# a Foundry agent to answer a single question about it.
+#
+# Example input:
+#   How many open issues does the repository have?
+#
+kind: Workflow
+trigger:
+
+  kind: OnConversationStart
+  id: workflow_invoke_http_request_demo
+  actions:
+
+    # Set the repository org/name used to form the request URL.
+    - kind: SetVariable
+      id: set_repo_name
+      variable: Local.RepoName
+      value: microsoft/agent-framework
+
+    # Invoke the GitHub repo API. The response body is parsed into
+    # Local.RepoInfo and also added to the conversation (via conversationId)
+    # so the agent below can answer questions based on it.
+    - kind: HttpRequestAction
+      id: fetch_repo_info
+      conversationId: =System.ConversationId
+      method: GET
+      url: =Concatenate("https://api.github.com/repos/", Local.RepoName)
+      headers:
+        Accept: application/vnd.github+json
+        User-Agent: agent-framework-sample
+      response: Local.RepoInfo
+
+    # Use the agent to answer the user's question using the conversation
+    # context (which now contains the GitHub JSON response). The user's
+    # original message is already in the conversation as System.LastMessage,
+    # and the executor's input fallback chain extracts its ``Text`` field
+    # automatically when ``input.messages`` is omitted.
+    - kind: InvokeAzureAgent
+      id: answer_question
+      conversationId: =System.ConversationId
+      agent:
+        name: GitHubRepoInfoAgent
+      output:
+        autoSend: true
+        messages: Local.AgentResponse

python/samples/05-end-to-end/chatkit-integration/frontend/package-lock.json

@@ -1199,6 +1199,19 @@
             "dev": true,
             "license": "ISC"
         },
+        "node_modules/picomatch": {
+            "version": "4.0.4",
+            "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.4.tgz",
+            "integrity": "sha512-QP88BAKvMam/3NxH6vj2o21R6MjxZUAd6nlwAS/pnGvN9IVLocLHxGYIzFhg6fUQ+5th6P4dv4eW9jX3DSIj7A==",
+            "dev": true,
+            "license": "MIT",
+            "engines": {
+                "node": ">=12"
+            },
+            "funding": {
+                "url": "https://github.com/sponsors/jonschlinkert"
+            }
+        },
         "node_modules/postcss": {
             "version": "8.5.10",
             "resolved": "https://registry.npmjs.org/postcss/-/postcss-8.5.10.tgz",
@@ -1345,19 +1358,6 @@
                 }
             }
         },
-        "node_modules/tinyglobby/node_modules/picomatch": {
-            "version": "4.0.3",
-            "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.3.tgz",
-            "integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==",
-            "dev": true,
-            "license": "MIT",
-            "engines": {
-                "node": ">=12"
-            },
-            "funding": {
-                "url": "https://github.com/sponsors/jonschlinkert"
-            }
-        },
         "node_modules/typescript": {
             "version": "5.9.3",
             "resolved": "https://registry.npmjs.org/typescript/-/typescript-5.9.3.tgz",
@@ -1464,19 +1464,6 @@
                     "optional": true
                 }
             }
-        },
-        "node_modules/vite/node_modules/picomatch": {
-            "version": "4.0.3",
-            "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.3.tgz",
-            "integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==",
-            "dev": true,
-            "license": "MIT",
-            "engines": {
-                "node": ">=12"
-            },
-            "funding": {
-                "url": "https://github.com/sponsors/jonschlinkert"
-            }
         }
     }
 }