Merge branch 'main' into copilot/fix-azure-functions-worker-crashes

This commit is contained in:
Laveesh Rohra
2026-03-02 13:08:05 -08:00
committed by GitHub
Unverified
121 changed files with 5159 additions and 320 deletions
@@ -612,11 +612,11 @@ class AgentFunctionApp(DFAppBase):
context: Durable Functions orchestration context invoking the agent.
agent_name: Name of the agent registered on this app.
Raises:
ValueError: If the requested agent has not been registered.
Returns:
DurableAIAgent[AgentTask] wrapper bound to the orchestration context.
Raises:
ValueError: If the requested agent has not been registered.
"""
normalized_name = str(agent_name)
@@ -81,6 +81,16 @@ OptionsCoT = TypeVar(
)
def _get_tool_name(tool: Any) -> str | None:
"""Extract a tool's name from either an object with a .name attribute or a dict tool definition."""
if isinstance(tool, dict):
func = tool.get("function")
if isinstance(func, dict):
return func.get("name")
return None
return getattr(tool, "name", None)
def _merge_options(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
"""Merge two options dicts, with override values taking precedence.
@@ -97,8 +107,8 @@ def _merge_options(base: dict[str, Any], override: dict[str, Any]) -> dict[str,
continue
if key == "tools" and result.get("tools"):
# Combine tool lists, avoiding duplicates by name
existing_names = {getattr(t, "name", None) for t in result["tools"]}
unique_new = [t for t in value if getattr(t, "name", None) not in existing_names]
existing_names = {_get_tool_name(t) for t in result["tools"]} - {None}
unique_new = [t for t in value if _get_tool_name(t) not in existing_names]
result["tools"] = list(result["tools"]) + unique_new
elif key == "logit_bias" and result.get("logit_bias"):
# Merge logit_bias dicts
@@ -935,6 +945,11 @@ class RawAgent(BaseAgent, Generic[OptionsCoT]): # type: ignore[misc]
session.service_session_id = conv_id
return update
def _finalizer(updates: Sequence[AgentResponseUpdate]) -> AgentResponse[Any]:
ctx = ctx_holder["ctx"]
rf = ctx.get("chat_options", {}).get("response_format") if ctx else (options.get("response_format") if options else None)
return self._finalize_response_updates(updates, response_format=rf)
return (
ResponseStream
.from_awaitable(_get_stream())
@@ -943,9 +958,7 @@ class RawAgent(BaseAgent, Generic[OptionsCoT]): # type: ignore[misc]
map_chat_to_agent_update,
agent_name=self.name,
),
finalizer=partial(
self._finalize_response_updates, response_format=options.get("response_format") if options else None
),
finalizer=_finalizer,
)
.with_transform_hook(_propagate_conversation_id)
.with_result_hook(_post_hook)
@@ -93,13 +93,13 @@ def detect_media_type_from_base64(
This will look at the actual data to determine the media_type and not at the URI prefix.
Will also not compare those two values.
Raises:
ValueError: If not exactly 1 of data_bytes, data_str, or data_uri is provided, or if base64 decoding fails.
Returns:
The detected media type (e.g., 'image/png', 'audio/wav', 'application/pdf')
or None if the format is not recognized.
Raises:
ValueError: If not exactly 1 of data_bytes, data_str, or data_uri is provided, or if base64 decoding fails.
Examples:
.. code-block:: python
@@ -670,6 +670,9 @@ class Content:
additional_properties: Optional additional properties.
raw_representation: Optional raw representation from an underlying implementation.
Returns:
A Content instance with type="data" for data URIs or type="uri" for external URIs.
Raises:
ContentError: If the URI is not valid.
@@ -693,9 +696,6 @@ class Content:
raw_base64_string
}"
)
Returns:
A Content instance with type="data" for data URIs or type="uri" for external URIs.
"""
return cls(
**_validate_uri(uri, media_type),
@@ -270,6 +270,11 @@ class WorkflowAgent(BaseAgent):
output_events.append(event)
result = self._convert_workflow_events_to_agent_response(response_id, output_events)
# Set the response on the context so after_run providers (e.g. InMemoryHistoryProvider)
# can persist the response messages alongside input messages.
session_context._response = result # type: ignore[assignment]
await self._run_after_providers(session=provider_session, context=session_context)
return result
@@ -322,12 +327,20 @@ class WorkflowAgent(BaseAgent):
# combine the messages
session_messages: list[Message] = session_context.get_messages(include_input=True)
all_updates: list[AgentResponseUpdate] = []
async for event in self._run_core(
session_messages, checkpoint_id, checkpoint_storage, streaming=True, **kwargs
):
updates = self._convert_workflow_event_to_agent_response_updates(response_id, event)
for update in updates:
all_updates.append(update)
yield update
# Build the final response from collected updates so after_run providers
# (e.g. InMemoryHistoryProvider) can persist the response messages.
if all_updates:
session_context._response = AgentResponse.from_updates(all_updates) # type: ignore[assignment]
await self._run_after_providers(session=provider_session, context=session_context)
async def _run_core(
@@ -360,7 +360,7 @@ class AgentExecutor(Executor):
Returns:
The complete AgentResponse, or None if waiting for user input.
"""
run_kwargs, options = self._prepare_agent_run_args(ctx.get_state(WORKFLOW_RUN_KWARGS_KEY) or {})
run_kwargs, options = self._prepare_agent_run_args(ctx.get_state(WORKFLOW_RUN_KWARGS_KEY, {}))
updates: list[AgentResponseUpdate] = []
streamed_user_input_requests: list[Content] = []
@@ -415,6 +415,10 @@ class AgentExecutor(Executor):
return response
# Parameters that are explicitly passed to agent.run() by AgentExecutor
# and must not appear in **run_kwargs to avoid TypeError from duplicate values.
_RESERVED_RUN_PARAMS: frozenset[str] = frozenset({"session", "stream", "messages"})
@staticmethod
def _prepare_agent_run_args(raw_run_kwargs: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any] | None]:
"""Prepare kwargs and options for agent.run(), avoiding duplicate option passing.
@@ -423,8 +427,23 @@ class AgentExecutor(Executor):
`options.additional_function_arguments`. If workflow kwargs include an
`options` key, merge it into the final options object and remove it from
kwargs before spreading `**run_kwargs`.
Reserved parameters (session, stream, messages) that are explicitly
managed by AgentExecutor are stripped from run_kwargs to prevent
``TypeError: got multiple values for keyword argument`` collisions.
"""
run_kwargs = dict(raw_run_kwargs)
# Strip reserved params that AgentExecutor passes explicitly to agent.run().
for key in AgentExecutor._RESERVED_RUN_PARAMS:
if key in run_kwargs:
logger.warning(
"Workflow kwarg '%s' is reserved by AgentExecutor and will be ignored. "
"Remove it from workflow.run() kwargs to silence this warning.",
key,
)
run_kwargs.pop(key)
options_from_workflow = run_kwargs.pop("options", None)
workflow_additional_args = run_kwargs.pop("additional_function_arguments", None)
@@ -6,6 +6,7 @@ import functools
import inspect
import logging
import types
import typing
from collections.abc import Awaitable, Callable
from typing import Any, TypeVar, overload
@@ -722,20 +723,30 @@ def _validate_handler_signature(
if not skip_message_annotation and message_param.annotation == inspect.Parameter.empty:
raise ValueError(f"Handler {func.__name__} must have a type annotation for the message parameter")
# Resolve string annotations from `from __future__ import annotations`.
# Fall back to raw annotations if resolution fails (e.g. unresolvable forward refs,
# AttributeError, or RecursionError), so registration failures are easier to diagnose.
try:
type_hints = typing.get_type_hints(func)
except Exception:
type_hints = {p.name: p.annotation for p in params}
# Validate ctx parameter is WorkflowContext and extract type args
ctx_param = params[2]
if skip_message_annotation and ctx_param.annotation == inspect.Parameter.empty:
ctx_annotation = type_hints.get(ctx_param.name, ctx_param.annotation)
if skip_message_annotation and ctx_annotation == inspect.Parameter.empty:
# When explicit types are provided via @handler(input=..., output=...),
# the ctx parameter doesn't need a type annotation - types come from the decorator.
output_types: list[type[Any] | types.UnionType] = []
workflow_output_types: list[type[Any] | types.UnionType] = []
else:
output_types, workflow_output_types = validate_workflow_context_annotation(
ctx_param.annotation, f"parameter '{ctx_param.name}'", "Handler"
ctx_annotation, f"parameter '{ctx_param.name}'", "Handler"
)
message_type = message_param.annotation if message_param.annotation != inspect.Parameter.empty else None
ctx_annotation = ctx_param.annotation
message_type = type_hints.get(message_param.name, message_param.annotation)
if message_type == inspect.Parameter.empty:
message_type = None
return message_type, ctx_annotation, output_types, workflow_output_types
@@ -345,9 +345,14 @@ class Workflow(DictConvertible):
self._runner.context.reset_for_new_run()
self._state.clear()
# Store run kwargs in State so executors can access them
# Always store (even empty dict) so retrieval is deterministic
self._state.set(WORKFLOW_RUN_KWARGS_KEY, run_kwargs or {})
# 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.
if run_kwargs is not None:
self._state.set(WORKFLOW_RUN_KWARGS_KEY, run_kwargs)
elif reset_context:
self._state.set(WORKFLOW_RUN_KWARGS_KEY, {})
self._state.commit() # Commit immediately so kwargs are available
# Set streaming mode after reset
@@ -369,7 +374,6 @@ class Workflow(DictConvertible):
with _framework_event_origin():
pending_status = WorkflowEvent.status(WorkflowRunState.IN_PROGRESS_PENDING_REQUESTS)
yield pending_status
# Workflow runs until idle - emit final status based on whether requests are pending
if saw_request:
with _framework_event_origin():
@@ -564,6 +568,10 @@ class Workflow(DictConvertible):
initial_executor_fn=initial_executor_fn,
reset_context=reset_context,
streaming=streaming,
# Empty **kwargs (no caller-provided kwargs) is collapsed to None so that
# continuation calls without explicit kwargs preserve the original run's kwargs.
# A non-empty kwargs dict (even one with empty values like {"key": {}})
# is passed through and will overwrite stored kwargs.
run_kwargs=kwargs if kwargs else None,
):
if event.type == "output" and not self._should_yield_output_event(event):
@@ -385,7 +385,7 @@ class WorkflowExecutor(Executor):
try:
# Get kwargs from parent workflow's State to propagate to subworkflow
parent_kwargs: dict[str, Any] = ctx.get_state(WORKFLOW_RUN_KWARGS_KEY) or {}
parent_kwargs: dict[str, Any] = ctx.get_state(WORKFLOW_RUN_KWARGS_KEY, {})
# Run the sub-workflow and collect all events, passing parent kwargs
result = await self.workflow.run(input_data, **parent_kwargs)
@@ -180,7 +180,7 @@ class AzureOpenAIResponsesClient( # type: ignore[misc]
client: AzureOpenAIResponsesClient[MyOptions] = AzureOpenAIResponsesClient()
response = await client.get_response("Hello", options={"my_custom_option": "value"})
"""
if model_id := kwargs.pop("model_id", None) and not deployment_name:
if (model_id := kwargs.pop("model_id", None)) and not deployment_name:
deployment_name = str(model_id)
# Project client path: create OpenAI client from an Azure AI Foundry project
@@ -16,6 +16,8 @@ from typing import TYPE_CHECKING, Any, Generic, Literal, TypedDict, cast
from openai import AsyncOpenAI
from openai.types.beta.threads import (
FileCitationDeltaAnnotation,
FilePathDeltaAnnotation,
ImageURLContentBlockParam,
ImageURLParam,
MessageContentPartParam,
@@ -39,12 +41,14 @@ from .._tools import (
normalize_tools,
)
from .._types import (
Annotation,
ChatOptions,
ChatResponse,
ChatResponseUpdate,
Content,
Message,
ResponseStream,
TextSpanRegion,
UsageDetails,
)
from ..observability import ChatTelemetryLayer
@@ -554,9 +558,53 @@ class OpenAIAssistantsClient( # type: ignore[misc]
for delta_block in delta.content or []:
if isinstance(delta_block, TextDeltaBlock) and delta_block.text and delta_block.text.value:
text_content = Content.from_text(delta_block.text.value)
if delta_block.text.annotations:
text_content.annotations = []
for annotation in delta_block.text.annotations:
if isinstance(annotation, FileCitationDeltaAnnotation):
ann: Annotation = Annotation(
type="citation",
additional_properties={
"text": annotation.text,
"index": annotation.index,
},
raw_representation=annotation,
)
if annotation.file_citation and annotation.file_citation.file_id:
ann["file_id"] = annotation.file_citation.file_id
if annotation.start_index is not None and annotation.end_index is not None:
ann["annotated_regions"] = [
TextSpanRegion(
type="text_span",
start_index=annotation.start_index,
end_index=annotation.end_index,
)
]
text_content.annotations.append(ann)
elif isinstance(annotation, FilePathDeltaAnnotation):
ann = Annotation(
type="citation",
additional_properties={
"text": annotation.text,
"index": annotation.index,
},
raw_representation=annotation,
)
if annotation.file_path and annotation.file_path.file_id:
ann["file_id"] = annotation.file_path.file_id
if annotation.start_index is not None and annotation.end_index is not None:
ann["annotated_regions"] = [
TextSpanRegion(
type="text_span",
start_index=annotation.start_index,
end_index=annotation.end_index,
)
]
text_content.annotations.append(ann)
yield ChatResponseUpdate(
role=role, # type: ignore[arg-type]
contents=[Content.from_text(delta_block.text.value)],
contents=[text_content],
conversation_id=thread_id,
message_id=response_id,
raw_representation=response.data,
@@ -43,6 +43,8 @@ from .._tools import (
FunctionInvocationConfiguration,
FunctionInvocationLayer,
FunctionTool,
ToolTypes,
normalize_tools,
)
from .._types import (
Annotation,
@@ -425,21 +427,24 @@ class RawOpenAIResponsesClient( # type: ignore[misc]
# region Prep methods
def _prepare_tools_for_openai(self, tools: Sequence[Any] | None) -> list[Any]:
def _prepare_tools_for_openai(
self, tools: ToolTypes | Callable[..., Any] | Sequence[ToolTypes | Callable[..., Any]] | None
) -> list[Any]:
"""Prepare tools for the OpenAI Responses API.
Converts FunctionTool to Responses API format. All other tools pass through unchanged.
Args:
tools: Sequence of tools to prepare.
tools: A single tool or sequence of tools to prepare.
Returns:
List of tool parameters ready for the OpenAI API.
"""
if not tools:
tools_list = normalize_tools(tools)
if not tools_list:
return []
response_tools: list[Any] = []
for tool in tools:
for tool in tools_list:
if isinstance(tool, FunctionTool):
params = tool.parameters()
params["additionalProperties"] = False
@@ -90,6 +90,29 @@ def test_init_model_id_constructor(azure_openai_unit_test_env: dict[str, str]) -
assert isinstance(azure_responses_client, SupportsChatGetResponse)
def test_init_model_id_kwarg(azure_openai_unit_test_env: dict[str, str]) -> None:
"""Test that model_id kwarg correctly sets the deployment name (issue #4299)."""
azure_responses_client = AzureOpenAIResponsesClient(model_id="gpt-4o")
assert azure_responses_client.model_id == "gpt-4o"
assert isinstance(azure_responses_client, SupportsChatGetResponse)
def test_init_model_id_kwarg_does_not_override_deployment_name(azure_openai_unit_test_env: dict[str, str]) -> None:
"""Test that deployment_name takes precedence over model_id kwarg (issue #4299)."""
azure_responses_client = AzureOpenAIResponsesClient(deployment_name="my-deployment", model_id="gpt-4o")
assert azure_responses_client.model_id == "my-deployment"
assert isinstance(azure_responses_client, SupportsChatGetResponse)
def test_init_model_id_kwarg_none(azure_openai_unit_test_env: dict[str, str]) -> None:
"""Test that model_id=None does not override the env-var deployment name."""
azure_responses_client = AzureOpenAIResponsesClient(model_id=None)
assert azure_responses_client.model_id == azure_openai_unit_test_env["AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME"]
def test_init_with_default_header(azure_openai_unit_test_env: dict[str, str]) -> None:
default_headers = {"X-Unit-Test": "test-guid"}
+199 -1
View File
@@ -25,7 +25,7 @@ from agent_framework import (
SupportsChatGetResponse,
tool,
)
from agent_framework._agents import _merge_options, _sanitize_agent_name
from agent_framework._agents import _get_tool_name, _merge_options, _sanitize_agent_name
from agent_framework._mcp import MCPTool
@@ -97,6 +97,58 @@ async def test_chat_client_agent_run_streaming(client: SupportsChatGetResponse)
assert result.text == "test streaming response another update"
async def test_chat_client_agent_streaming_response_format_from_default_options(
client: SupportsChatGetResponse,
) -> None:
"""AgentResponse.value must be parsed when response_format is set in default_options and streaming."""
from pydantic import BaseModel
class Greeting(BaseModel):
greeting: str
json_text = '{"greeting": "Hello"}'
client.streaming_responses.append( # type: ignore[attr-defined]
[ChatResponseUpdate(contents=[Content.from_text(json_text)], role="assistant", finish_reason="stop")]
)
agent = Agent(client=client, default_options={"response_format": Greeting})
stream = agent.run("Hello", stream=True)
async for _ in stream:
pass
result = await stream.get_final_response()
assert result.text == json_text
assert result.value is not None
assert isinstance(result.value, Greeting)
assert result.value.greeting == "Hello"
async def test_chat_client_agent_streaming_response_format_from_run_options(
client: SupportsChatGetResponse,
) -> None:
"""AgentResponse.value must be parsed when response_format is passed via run() options kwarg."""
from pydantic import BaseModel
class Greeting(BaseModel):
greeting: str
json_text = '{"greeting": "Hi"}'
client.streaming_responses.append( # type: ignore[attr-defined]
[ChatResponseUpdate(contents=[Content.from_text(json_text)], role="assistant", finish_reason="stop")]
)
agent = Agent(client=client)
stream = agent.run("Hello", stream=True, options={"response_format": Greeting})
async for _ in stream:
pass
result = await stream.get_final_response()
assert result.text == json_text
assert result.value is not None
assert isinstance(result.value, Greeting)
assert result.value.greeting == "Hi"
async def test_chat_client_agent_create_session(client: SupportsChatGetResponse) -> None:
agent = Agent(client=client)
session = agent.create_session()
@@ -880,6 +932,152 @@ def test_merge_options_tools_combined():
assert "tool2" in tool_names
def test_merge_options_dict_tools_combined():
"""Test _merge_options combines dict-defined tool lists without duplicates."""
base = {
"tools": [
{"type": "function", "function": {"name": "tool_a"}},
]
}
override = {
"tools": [
{"type": "function", "function": {"name": "tool_b"}},
]
}
result = _merge_options(base, override)
assert len(result["tools"]) == 2
names = [_get_tool_name(t) for t in result["tools"]]
assert "tool_a" in names
assert "tool_b" in names
def test_merge_options_dict_tools_deduplicates():
"""Test _merge_options deduplicates dict-defined tools by function name."""
base = {
"tools": [
{"type": "function", "function": {"name": "tool_a"}},
]
}
override = {
"tools": [
{"type": "function", "function": {"name": "tool_a"}},
{"type": "function", "function": {"name": "tool_b"}},
]
}
result = _merge_options(base, override)
assert len(result["tools"]) == 2
names = [_get_tool_name(t) for t in result["tools"]]
assert names.count("tool_a") == 1
assert "tool_b" in names
def test_merge_options_mixed_tools_combined():
"""Test _merge_options combines object and dict-defined tools."""
class MockTool:
def __init__(self, name):
self.name = name
base = {"tools": [MockTool("tool_a")]}
override = {
"tools": [
{"type": "function", "function": {"name": "tool_b"}},
]
}
result = _merge_options(base, override)
assert len(result["tools"]) == 2
names = [_get_tool_name(t) for t in result["tools"]]
assert "tool_a" in names
assert "tool_b" in names
def test_merge_options_mixed_tools_deduplicates():
"""Test _merge_options deduplicates when a dict tool and object tool share the same name."""
class MockTool:
def __init__(self, name):
self.name = name
base = {"tools": [MockTool("tool_a")]}
override = {
"tools": [
{"type": "function", "function": {"name": "tool_a"}},
]
}
result = _merge_options(base, override)
assert len(result["tools"]) == 1
assert _get_tool_name(result["tools"][0]) == "tool_a"
def test_merge_options_nameless_tools_not_deduplicated():
"""Test that tools with no extractable name (None) are not falsely deduplicated."""
base = {
"tools": [
{"type": "function"}, # no 'function.name' -> _get_tool_name returns None
]
}
override = {
"tools": [
{"type": "function"}, # also returns None
]
}
result = _merge_options(base, override)
# Both nameless tools should be kept (None is excluded from dedup set)
assert len(result["tools"]) == 2
def test_get_tool_name_dict_no_function_key():
"""_get_tool_name returns None for a dict without a 'function' key."""
assert _get_tool_name({"type": "function"}) is None
def test_get_tool_name_dict_function_not_dict():
"""_get_tool_name returns None when 'function' value is not a dict."""
assert _get_tool_name({"function": "not_a_dict"}) is None
def test_get_tool_name_dict_function_no_name():
"""_get_tool_name returns None when 'function' dict has no 'name' key."""
assert _get_tool_name({"function": {"description": "does stuff"}}) is None
def test_get_tool_name_object_no_name_attr():
"""_get_tool_name returns None for an object without a 'name' attribute."""
assert _get_tool_name(object()) is None
def test_get_tool_name_non_dict_non_object():
"""_get_tool_name returns None for non-dict inputs like int or string."""
assert _get_tool_name(42) is None
assert _get_tool_name("tool_name") is None
def test_get_tool_name_valid_dict():
"""_get_tool_name extracts name from a well-formed dict tool."""
tool_dict = {"type": "function", "function": {"name": "my_tool"}}
assert _get_tool_name(tool_dict) == "my_tool"
def test_get_tool_name_valid_object():
"""_get_tool_name extracts name from an object with a name attribute."""
class MockTool:
def __init__(self, name):
self.name = name
assert _get_tool_name(MockTool("my_tool")) == "my_tool"
def test_merge_options_logit_bias_merged():
"""Test _merge_options merges logit_bias dicts."""
base = {"logit_bias": {"token1": 1.0}}
@@ -7,6 +7,8 @@ from unittest.mock import AsyncMock, MagicMock
import pytest
from openai.types.beta.threads import MessageDeltaEvent, Run, TextDeltaBlock
from openai.types.beta.threads.file_citation_delta_annotation import FileCitationDeltaAnnotation
from openai.types.beta.threads.file_path_delta_annotation import FilePathDeltaAnnotation
from openai.types.beta.threads.runs import RunStep
from pydantic import Field
@@ -443,6 +445,120 @@ async def test_process_stream_events_message_delta_text(mock_async_openai: Magic
assert update.raw_representation == mock_message_delta
async def test_process_stream_events_message_delta_text_with_file_citation_annotations(
mock_async_openai: MagicMock,
) -> None:
"""Test _process_stream_events maps file citation annotations from TextDeltaBlock."""
client = create_test_openai_assistants_client(mock_async_openai)
mock_annotation = FileCitationDeltaAnnotation(
index=0,
type="file_citation",
file_citation={"file_id": "file-abc123"},
start_index=10,
end_index=24,
text="【4:0†source】",
)
mock_delta_block = MagicMock(spec=TextDeltaBlock)
mock_delta_block.text = MagicMock()
mock_delta_block.text.value = "Some text 【4:0†source】 more text"
mock_delta_block.text.annotations = [mock_annotation]
mock_delta = MagicMock()
mock_delta.role = "assistant"
mock_delta.content = [mock_delta_block]
mock_message_delta = MagicMock(spec=MessageDeltaEvent)
mock_message_delta.delta = mock_delta
mock_response = MagicMock()
mock_response.event = "thread.message.delta"
mock_response.data = mock_message_delta
async def async_iterator() -> Any:
yield mock_response
mock_stream = MagicMock()
mock_stream.__aenter__ = AsyncMock(return_value=async_iterator())
mock_stream.__aexit__ = AsyncMock(return_value=None)
thread_id = "thread-789"
updates: list[ChatResponseUpdate] = []
async for update in client._process_stream_events(mock_stream, thread_id): # type: ignore
updates.append(update)
assert len(updates) == 1
update = updates[0]
assert update.text == "Some text 【4:0†source】 more text"
assert update.contents is not None
content = update.contents[0]
assert content.annotations is not None
assert len(content.annotations) == 1
ann = content.annotations[0]
assert ann["type"] == "citation"
assert ann["file_id"] == "file-abc123"
assert ann["annotated_regions"] is not None
assert ann["annotated_regions"][0]["start_index"] == 10
assert ann["annotated_regions"][0]["end_index"] == 24
assert ann["additional_properties"]["text"] == "【4:0†source】"
async def test_process_stream_events_message_delta_text_with_file_path_annotations(
mock_async_openai: MagicMock,
) -> None:
"""Test _process_stream_events maps file path annotations from TextDeltaBlock."""
client = create_test_openai_assistants_client(mock_async_openai)
mock_annotation = FilePathDeltaAnnotation(
index=0,
type="file_path",
file_path={"file_id": "file-xyz789"},
start_index=5,
end_index=20,
text="sandbox:/path/to/file",
)
mock_delta_block = MagicMock(spec=TextDeltaBlock)
mock_delta_block.text = MagicMock()
mock_delta_block.text.value = "Here sandbox:/path/to/file is the file"
mock_delta_block.text.annotations = [mock_annotation]
mock_delta = MagicMock()
mock_delta.role = "assistant"
mock_delta.content = [mock_delta_block]
mock_message_delta = MagicMock(spec=MessageDeltaEvent)
mock_message_delta.delta = mock_delta
mock_response = MagicMock()
mock_response.event = "thread.message.delta"
mock_response.data = mock_message_delta
async def async_iterator() -> Any:
yield mock_response
mock_stream = MagicMock()
mock_stream.__aenter__ = AsyncMock(return_value=async_iterator())
mock_stream.__aexit__ = AsyncMock(return_value=None)
thread_id = "thread-annotation"
updates: list[ChatResponseUpdate] = []
async for update in client._process_stream_events(mock_stream, thread_id): # type: ignore
updates.append(update)
assert len(updates) == 1
content = updates[0].contents[0]
assert content.annotations is not None
assert len(content.annotations) == 1
ann = content.annotations[0]
assert ann["type"] == "citation"
assert ann["file_id"] == "file-xyz789"
assert ann["annotated_regions"] is not None
assert ann["annotated_regions"][0]["start_index"] == 5
assert ann["annotated_regions"][0]["end_index"] == 20
async def test_process_stream_events_requires_action(mock_async_openai: MagicMock) -> None:
"""Test _process_stream_events with thread.run.requires_action event."""
client = create_test_openai_assistants_client(mock_async_openai)
@@ -1193,6 +1193,52 @@ def test_prepare_tools_for_openai_with_mcp() -> None:
assert "require_approval" in mcp
def test_prepare_tools_for_openai_single_function_tool() -> None:
"""Test that a single FunctionTool (not wrapped in a list) is handled correctly."""
client = OpenAIResponsesClient(model_id="test-model", api_key="test-key")
@tool
def hello(name: str) -> str:
"""Say hello."""
return name
resp_tools = client._prepare_tools_for_openai(hello)
assert isinstance(resp_tools, list)
assert len(resp_tools) == 1
tool_def = resp_tools[0]
assert tool_def["type"] == "function"
assert tool_def["name"] == "hello"
assert tool_def["strict"] is False
assert "parameters" in tool_def
params = tool_def["parameters"]
assert isinstance(params, dict)
assert params.get("type") == "object"
assert "properties" in params
assert "name" in params["properties"]
assert params["properties"]["name"]["type"] == "string"
def test_prepare_tools_for_openai_single_dict_tool() -> None:
"""Test that a single dict tool (not wrapped in a list) is handled correctly."""
client = OpenAIResponsesClient(model_id="test-model", api_key="test-key")
web_tool = OpenAIResponsesClient.get_web_search_tool(search_context_size="low")
resp_tools = client._prepare_tools_for_openai(web_tool)
assert isinstance(resp_tools, list)
assert len(resp_tools) == 1
assert "type" in resp_tools[0]
assert resp_tools[0]["search_context_size"] == "low"
def test_prepare_tools_for_openai_none() -> None:
"""Test that passing None returns an empty list."""
client = OpenAIResponsesClient(model_id="test-model", api_key="test-key")
resp_tools = client._prepare_tools_for_openai(None)
assert isinstance(resp_tools, list)
assert len(resp_tools) == 0
def test_parse_response_from_openai_with_mcp_approval_request() -> None:
"""Test that a non-streaming mcp_approval_request is parsed into FunctionApprovalRequestContent."""
client = OpenAIResponsesClient(model_id="test-model", api_key="test-key")
@@ -1,7 +1,10 @@
# Copyright (c) Microsoft. All rights reserved.
import logging
from collections.abc import AsyncIterable, Awaitable
from typing import Any
from typing import TYPE_CHECKING, Any
import pytest
from agent_framework import (
AgentExecutor,
@@ -18,6 +21,9 @@ from agent_framework._workflows._agent_executor import AgentExecutorResponse
from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage
from agent_framework.orchestrations import SequentialBuilder
if TYPE_CHECKING:
from _pytest.logging import LogCaptureFixture
class _CountingAgent(BaseAgent):
"""Agent that echoes messages with a counter to verify session state persistence."""
@@ -251,3 +257,85 @@ async def test_agent_executor_save_and_restore_state_directly() -> None:
# Verify session was restored with correct session_id
restored_session = new_executor._session # type: ignore[reportPrivateUsage]
assert restored_session.session_id == session.session_id
async def test_agent_executor_run_with_session_kwarg_does_not_raise() -> None:
"""Passing session= via workflow.run() should not cause a duplicate-keyword TypeError (#4295)."""
agent = _CountingAgent(id="session_kwarg_agent", name="SessionKwargAgent")
executor = AgentExecutor(agent, id="session_kwarg_exec")
workflow = SequentialBuilder(participants=[executor]).build()
# This previously raised: TypeError: run() got multiple values for keyword argument 'session'
result = await workflow.run("hello", session="user-supplied-value")
assert result is not None
assert agent.call_count == 1
async def test_agent_executor_run_streaming_with_stream_kwarg_does_not_raise() -> None:
"""Passing stream= via workflow.run() kwargs should not cause a duplicate-keyword TypeError."""
agent = _CountingAgent(id="stream_kwarg_agent", name="StreamKwargAgent")
executor = AgentExecutor(agent, id="stream_kwarg_exec")
workflow = SequentialBuilder(participants=[executor]).build()
# stream=True at workflow level triggers streaming mode (returns async iterable)
events = []
async for event in workflow.run("hello", stream=True):
events.append(event)
assert len(events) > 0
assert agent.call_count == 1
@pytest.mark.parametrize("reserved_kwarg", ["session", "stream", "messages"])
async def test_prepare_agent_run_args_strips_reserved_kwargs(
reserved_kwarg: str, caplog: "LogCaptureFixture"
) -> None:
"""_prepare_agent_run_args must remove reserved kwargs and log a warning."""
raw = {reserved_kwarg: "should-be-stripped", "custom_key": "keep-me"}
with caplog.at_level(logging.WARNING):
run_kwargs, options = AgentExecutor._prepare_agent_run_args(raw)
assert reserved_kwarg not in run_kwargs
assert "custom_key" in run_kwargs
assert options is not None
assert options["additional_function_arguments"]["custom_key"] == "keep-me"
assert any(reserved_kwarg in record.message for record in caplog.records)
async def test_prepare_agent_run_args_preserves_non_reserved_kwargs() -> None:
"""Non-reserved workflow kwargs should pass through unchanged."""
raw = {"custom_param": "value", "another": 42}
run_kwargs, options = AgentExecutor._prepare_agent_run_args(raw)
assert run_kwargs["custom_param"] == "value"
assert run_kwargs["another"] == 42
async def test_prepare_agent_run_args_strips_all_reserved_kwargs_at_once(
caplog: "LogCaptureFixture",
) -> None:
"""All reserved kwargs should be stripped when supplied together, each emitting a warning."""
raw = {"session": "x", "stream": True, "messages": [], "custom": 1}
with caplog.at_level(logging.WARNING):
run_kwargs, options = AgentExecutor._prepare_agent_run_args(raw)
assert "session" not in run_kwargs
assert "stream" not in run_kwargs
assert "messages" not in run_kwargs
assert run_kwargs["custom"] == 1
assert options is not None
assert options["additional_function_arguments"]["custom"] == 1
warned_keys = {r.message.split("'")[1] for r in caplog.records if "reserved" in r.message.lower()}
assert warned_keys == {"session", "stream", "messages"}
async def test_agent_executor_run_with_messages_kwarg_does_not_raise() -> None:
"""Passing messages= via workflow.run() kwargs should not cause a duplicate-keyword TypeError."""
agent = _CountingAgent(id="messages_kwarg_agent", name="MessagesKwargAgent")
executor = AgentExecutor(agent, id="messages_kwarg_exec")
workflow = SequentialBuilder(participants=[executor]).build()
result = await workflow.run("hello", messages=["stale"])
assert result is not None
assert agent.call_count == 1
@@ -0,0 +1,124 @@
# Copyright (c) Microsoft. All rights reserved.
from __future__ import annotations
from typing import Any
import pytest
from pydantic import BaseModel
from agent_framework import Executor, WorkflowContext, handler
class MyTypeA(BaseModel):
pass
class MyTypeB(BaseModel):
pass
class MyTypeC(BaseModel):
pass
class TestExecutorFutureAnnotations:
"""Test suite for Executor with from __future__ import annotations."""
def test_handler_decorator_future_annotations(self):
"""Test @handler decorator works with stringified annotations (issue #3898)."""
class MyExecutor(Executor):
@handler
async def example(self, input: str, ctx: WorkflowContext[MyTypeA, MyTypeB]) -> None:
pass
exec_instance = MyExecutor(id="test")
assert str in exec_instance._handlers
spec = exec_instance._handler_specs[0]
assert spec["message_type"] is str
assert spec["output_types"] == [MyTypeA]
assert spec["workflow_output_types"] == [MyTypeB]
def test_handler_decorator_future_annotations_single_type_arg(self):
"""Test @handler with single type argument and future annotations."""
class MyExecutor(Executor):
@handler
async def example(self, input: int, ctx: WorkflowContext[MyTypeA]) -> None:
pass
exec_instance = MyExecutor(id="test")
assert int in exec_instance._handlers
spec = exec_instance._handler_specs[0]
assert spec["message_type"] is int
assert spec["output_types"] == [MyTypeA]
def test_handler_decorator_future_annotations_complex(self):
"""Test @handler with complex type annotations and future annotations."""
class MyExecutor(Executor):
@handler
async def example(self, data: dict[str, Any], ctx: WorkflowContext[list[str]]) -> None:
pass
exec_instance = MyExecutor(id="test")
spec = exec_instance._handler_specs[0]
assert spec["message_type"] == dict[str, Any]
assert spec["output_types"] == [list[str]]
def test_handler_decorator_future_annotations_bare_context(self):
"""Test @handler with bare WorkflowContext and future annotations."""
class MyExecutor(Executor):
@handler
async def example(self, input: str, ctx: WorkflowContext) -> None:
pass
exec_instance = MyExecutor(id="test")
assert str in exec_instance._handlers
spec = exec_instance._handler_specs[0]
assert spec["output_types"] == []
assert spec["workflow_output_types"] == []
def test_handler_decorator_future_annotations_explicit_types(self):
"""Test @handler with explicit type parameters under future annotations."""
class MyExecutor(Executor):
@handler(input=str, output=MyTypeA)
async def example(self, input, ctx) -> None:
pass
exec_instance = MyExecutor(id="test")
assert str in exec_instance._handlers
spec = exec_instance._handler_specs[0]
assert spec["message_type"] is str
assert spec["output_types"] == [MyTypeA]
def test_handler_decorator_future_annotations_union_context(self):
"""Test @handler with union type context annotations and future annotations."""
class MyExecutor(Executor):
@handler
async def example(self, input: str, ctx: WorkflowContext[MyTypeA | MyTypeB, MyTypeC]) -> None:
pass
exec_instance = MyExecutor(id="test")
assert str in exec_instance._handlers
spec = exec_instance._handler_specs[0]
assert spec["output_types"] == [MyTypeA, MyTypeB]
assert spec["workflow_output_types"] == [MyTypeC]
def test_handler_unresolvable_annotation_raises(self):
"""Test that an unresolvable forward-reference annotation raises ValueError.
When get_type_hints fails (e.g. NameError for NonExistentType), the code falls back
to raw string annotations. The ctx parameter's raw string annotation is then not
recognised as a valid WorkflowContext type, so a ValueError is still raised.
"""
with pytest.raises(ValueError):
class Bad(Executor):
@handler
async def example(self, input: NonExistentType, ctx: WorkflowContext[MyTypeA, MyTypeB]) -> None: # noqa: F821
pass
@@ -578,6 +578,92 @@ class TestWorkflowAgent:
assert "first message" in texts
assert "second message" in texts
async def test_multi_turn_session_stores_responses(self) -> None:
"""Test that WorkflowAgent stores response messages in session history (issue #1694).
Previously, session_context._response was not set before running after_run
providers, so InMemoryHistoryProvider never persisted response messages.
On subsequent runs the workflow only received prior user inputs, not prior
assistant responses, breaking multi-turn conversations.
"""
capturing_executor = ConversationHistoryCapturingExecutor(id="multi_turn_test", streaming=False)
workflow = WorkflowBuilder(start_executor=capturing_executor).build()
agent = workflow.as_agent(name="Multi Turn Agent")
session = AgentSession()
# First turn
await agent.run("My name is Bob", session=session)
# Second turn — the executor should see prior user+assistant messages plus new input
await agent.run("What is my name?", session=session)
received = capturing_executor.received_messages
roles = [m.role for m in received]
texts = [m.text for m in received]
# History should include: user("My name is Bob"), assistant(response), user("What is my name?")
assert len(received) == 3, f"Expected 3 messages (user, assistant, user), got {len(received)}: {roles}"
assert roles[0] == "user"
assert "My name is Bob" in (texts[0] or "")
assert roles[1] == "assistant"
assert roles[2] == "user"
assert "What is my name?" in (texts[2] or "")
async def test_multi_turn_session_stores_responses_streaming(self) -> None:
"""Streaming variant: WorkflowAgent stores response messages in session history."""
capturing_executor = ConversationHistoryCapturingExecutor(id="multi_turn_stream_test", streaming=True)
workflow = WorkflowBuilder(start_executor=capturing_executor).build()
agent = workflow.as_agent(name="Multi Turn Stream Agent")
session = AgentSession()
# First turn (streaming)
stream = agent.run("Hello", stream=True, session=session)
async for _ in stream:
pass
await stream.get_final_response()
# Second turn — should include prior history
stream2 = agent.run("Follow up", stream=True, session=session)
async for _ in stream2:
pass
await stream2.get_final_response()
received = capturing_executor.received_messages
roles = [m.role for m in received]
assert len(received) == 3, f"Expected 3 messages, got {len(received)}: {roles}"
assert roles[0] == "user"
assert roles[1] == "assistant"
assert roles[2] == "user"
async def test_multi_turn_session_roundtrip_serialization(self) -> None:
"""Test that session can be serialized/deserialized and multi-turn still works."""
capturing_executor = ConversationHistoryCapturingExecutor(id="roundtrip_test", streaming=False)
workflow = WorkflowBuilder(start_executor=capturing_executor).build()
agent = workflow.as_agent(name="Roundtrip Agent")
session = AgentSession()
# First turn
await agent.run("My name is Bob", session=session)
# Serialize and deserialize the session
serialized = session.to_dict()
restored_session = AgentSession.from_dict(serialized)
# Second turn with restored session
await agent.run("What is my name?", session=restored_session)
received = capturing_executor.received_messages
roles = [m.role for m in received]
texts = [m.text for m in received]
assert len(received) == 3, f"Expected 3 messages, got {len(received)}: {roles}"
assert roles[0] == "user"
assert "My name is Bob" in (texts[0] or "")
assert roles[1] == "assistant"
assert roles[2] == "user"
assert "What is my name?" in (texts[2] or "")
async def test_workflow_agent_keeps_explicit_context_providers(self) -> None:
"""Test that WorkflowAgent does not append defaults when context providers are explicitly provided."""
workflow = WorkflowBuilder(
@@ -446,6 +446,215 @@ async def test_kwargs_with_complex_nested_data() -> None:
assert received.get("complex_data") == complex_data
async def test_kwargs_preserved_on_response_continuation() -> None:
"""Test that run kwargs are preserved when continuing a paused workflow with run(responses=...).
Regression test for #4293: kwargs were overwritten to {} on continuation calls.
"""
class _ApprovalCapturingAgent(BaseAgent):
"""Agent that pauses for approval on first call and captures kwargs on every call."""
captured_kwargs: list[dict[str, Any]]
_asked: bool
def __init__(self) -> None:
super().__init__(name="approval_agent", description="Test agent")
self.captured_kwargs = []
self._asked = False
def run(
self,
messages: str | Content | Message | Sequence[str | Content | Message] | None = None,
*,
stream: bool = False,
session: AgentSession | None = None,
**kwargs: Any,
) -> Awaitable[AgentResponse] | ResponseStream[AgentResponseUpdate, AgentResponse]:
self.captured_kwargs.append(dict(kwargs))
if not self._asked:
self._asked = True
async def _pause() -> AgentResponse:
call = Content.from_function_call(call_id="c1", name="do_thing", arguments="{}")
req = Content.from_function_approval_request(id="r1", function_call=call)
return AgentResponse(messages=[Message("assistant", [req])])
return _pause()
async def _done() -> AgentResponse:
return AgentResponse(messages=[Message("assistant", ["done"])])
return _done()
from agent_framework import WorkflowBuilder
agent = _ApprovalCapturingAgent()
workflow = WorkflowBuilder(start_executor=agent, output_executors=[agent]).build()
# Initial run with kwargs — workflow should pause for approval
result = await workflow.run("go", custom_data={"token": "abc"})
request_events = result.get_request_info_events()
assert len(request_events) == 1
# Continue with responses only — no new kwargs
approval = request_events[0]
await workflow.run(
responses={approval.request_id: approval.data.to_function_approval_response(True)}
)
# Both calls should have received the original kwargs
assert len(agent.captured_kwargs) == 2
assert agent.captured_kwargs[0].get("custom_data") == {"token": "abc"}
assert agent.captured_kwargs[1].get("custom_data") == {"token": "abc"}, (
f"kwargs should be preserved on continuation, got: {agent.captured_kwargs[1]}"
)
async def test_kwargs_overridden_on_response_continuation() -> None:
"""Test that explicitly provided kwargs override prior kwargs on continuation."""
class _ApprovalCapturingAgent(BaseAgent):
captured_kwargs: list[dict[str, Any]]
_asked: bool
def __init__(self) -> None:
super().__init__(name="approval_agent", description="Test agent")
self.captured_kwargs = []
self._asked = False
def run(
self,
messages: str | Content | Message | Sequence[str | Content | Message] | None = None,
*,
stream: bool = False,
session: AgentSession | None = None,
**kwargs: Any,
) -> Awaitable[AgentResponse] | ResponseStream[AgentResponseUpdate, AgentResponse]:
self.captured_kwargs.append(dict(kwargs))
if not self._asked:
self._asked = True
async def _pause() -> AgentResponse:
call = Content.from_function_call(call_id="c1", name="do_thing", arguments="{}")
req = Content.from_function_approval_request(id="r1", function_call=call)
return AgentResponse(messages=[Message("assistant", [req])])
return _pause()
async def _done() -> AgentResponse:
return AgentResponse(messages=[Message("assistant", ["done"])])
return _done()
from agent_framework import WorkflowBuilder
agent = _ApprovalCapturingAgent()
workflow = WorkflowBuilder(start_executor=agent, output_executors=[agent]).build()
result = await workflow.run("go", custom_data={"token": "abc"})
request_events = result.get_request_info_events()
approval = request_events[0]
# Continue with responses AND new kwargs — should override
await workflow.run(
responses={approval.request_id: approval.data.to_function_approval_response(True)},
custom_data={"token": "xyz"},
)
assert len(agent.captured_kwargs) == 2
assert agent.captured_kwargs[0].get("custom_data") == {"token": "abc"}
assert agent.captured_kwargs[1].get("custom_data") == {"token": "xyz"}
async def test_kwargs_empty_value_passed_on_continuation() -> None:
"""Test that explicitly passing a kwarg with an empty value on continuation overrides prior kwargs.
This exercises the boundary where the caller provides kwargs (e.g., custom_data={})
that differ from the original run. Because the kwargs dict is non-empty (it has a key),
it passes the `kwargs if kwargs else None` gate and the `is not None` check, so it
overwrites the previously stored kwargs.
"""
class _ApprovalCapturingAgent(BaseAgent):
captured_kwargs: list[dict[str, Any]]
_asked: bool
def __init__(self) -> None:
super().__init__(name="approval_agent", description="Test agent")
self.captured_kwargs = []
self._asked = False
def run(
self,
messages: str | Content | Message | Sequence[str | Content | Message] | None = None,
*,
stream: bool = False,
session: AgentSession | None = None,
**kwargs: Any,
) -> Awaitable[AgentResponse] | ResponseStream[AgentResponseUpdate, AgentResponse]:
self.captured_kwargs.append(dict(kwargs))
if not self._asked:
self._asked = True
async def _pause() -> AgentResponse:
call = Content.from_function_call(call_id="c1", name="do_thing", arguments="{}")
req = Content.from_function_approval_request(id="r1", function_call=call)
return AgentResponse(messages=[Message("assistant", [req])])
return _pause()
async def _done() -> AgentResponse:
return AgentResponse(messages=[Message("assistant", ["done"])])
return _done()
from agent_framework import WorkflowBuilder
agent = _ApprovalCapturingAgent()
workflow = WorkflowBuilder(start_executor=agent, output_executors=[agent]).build()
# Initial run with non-empty kwargs
result = await workflow.run("go", custom_data={"token": "abc"})
request_events = result.get_request_info_events()
assert len(request_events) == 1
# Continue with custom_data={} — explicitly clearing the value.
# kwargs={"custom_data": {}} is truthy (has a key), so run_kwargs is set.
approval = request_events[0]
await workflow.run(
responses={approval.request_id: approval.data.to_function_approval_response(True)},
custom_data={},
)
assert len(agent.captured_kwargs) == 2
assert agent.captured_kwargs[0].get("custom_data") == {"token": "abc"}
# The continuation explicitly set custom_data={}, overriding the original
assert agent.captured_kwargs[1].get("custom_data") == {}
async def test_kwargs_reset_context_stores_empty_dict() -> None:
"""Test that reset_context=True with no kwargs stores an empty dict.
This exercises the `elif reset_context` branch that ensures WORKFLOW_RUN_KWARGS_KEY
is always populated after a fresh run, even when no kwargs are provided.
"""
agent = _KwargsCapturingAgent(name="reset_ctx_test")
workflow = SequentialBuilder(participants=[agent]).build()
# Run with no kwargs and reset_context=True (the default for a fresh run)
async for event in workflow.run("test", stream=True):
if event.type == "status" and event.state == WorkflowRunState.IDLE:
break
assert len(agent.captured_kwargs) >= 1
# The only kwarg should be the framework-injected 'options' (no user-provided kwargs)
received = agent.captured_kwargs[0]
assert "custom_data" not in received
assert received.get("options") is None
async def test_kwargs_preserved_across_workflow_reruns() -> None:
"""Test that kwargs are correctly isolated between workflow runs."""
agent = _KwargsCapturingAgent(name="rerun_test")
@@ -92,8 +92,7 @@ class DeploymentManager:
break
# Get event from queue with short timeout
event = await asyncio.wait_for(event_queue.get(), timeout=0.1)
yield event
yield await asyncio.wait_for(event_queue.get(), timeout=0.1)
except asyncio.TimeoutError:
# No event in queue, continue waiting
continue
@@ -21,6 +21,13 @@ from .models._discovery_models import EntityInfo
logger = logging.getLogger(__name__)
def _get_event_type(event: Any) -> str | None:
"""Safely get the type of an event, handling both objects and dicts."""
if isinstance(event, dict):
return event.get("type")
return getattr(event, "type", None)
class EntityNotFoundError(Exception):
"""Raised when an entity is not found."""
@@ -264,7 +271,7 @@ class AgentFrameworkExecutor:
elif entity_info.type == "workflow":
async for event in self._execute_workflow(entity_obj, request, trace_collector):
# Log request_info event (type='request_info') for debugging HIL flow
if event.type == "request_info":
if _get_event_type(event) == "request_info":
logger.info(
"🔔 [EXECUTOR] request_info event (type='request_info') detected from workflow!"
)
@@ -330,19 +337,22 @@ class AgentFrameworkExecutor:
# Agent must have run() method - use stream=True for streaming
if hasattr(agent, "run") and callable(agent.run):
# Use Agent Framework's run() with stream=True for streaming
# Capture the stream reference so we can call get_final_response()
# after iteration. This triggers result hooks (after_run providers
# like InMemoryHistoryProvider) that persist conversation history.
run_kwargs: dict[str, Any] = {"stream": True}
if session:
async for update in agent.run(user_message, stream=True, session=session):
for trace_event in trace_collector.get_pending_events():
yield trace_event
run_kwargs["session"] = session
yield update
else:
async for update in agent.run(user_message, stream=True):
for trace_event in trace_collector.get_pending_events():
yield trace_event
stream = agent.run(user_message, **run_kwargs)
async for update in stream:
for trace_event in trace_collector.get_pending_events():
yield trace_event
yield update
yield update
# Finalize stream to trigger result hooks (saves conversation history)
await stream.get_final_response()
else:
raise ValueError("Agent must implement run() method")
@@ -471,7 +481,7 @@ class AgentFrameworkExecutor:
checkpoint_storage=checkpoint_storage,
):
# Enrich new request_info events that may come from subsequent HIL requests
if event.type == "request_info":
if _get_event_type(event) == "request_info":
self._enrich_request_info_event_with_response_schema(event, workflow)
for trace_event in trace_collector.get_pending_events():
@@ -493,7 +503,7 @@ class AgentFrameworkExecutor:
checkpoint_id=checkpoint_id,
checkpoint_storage=checkpoint_storage,
):
if event.type == "request_info":
if _get_event_type(event) == "request_info":
self._enrich_request_info_event_with_response_schema(event, workflow)
for trace_event in trace_collector.get_pending_events():
@@ -517,7 +527,7 @@ class AgentFrameworkExecutor:
parsed_input = await self._parse_workflow_input(workflow, request.input)
async for event in workflow.run(parsed_input, stream=True, checkpoint_storage=checkpoint_storage):
if event.type == "request_info":
if _get_event_type(event) == "request_info":
self._enrich_request_info_event_with_response_schema(event, workflow)
for trace_event in trace_collector.get_pending_events():
@@ -741,6 +741,51 @@ async def test_full_pipeline_workflow_output_event_serialization():
assert len(output_events) >= 3, f"Expected 3+ output events for yield_output calls, got {len(output_events)}"
async def test_workflow_error_yields_dict_event_without_crash():
"""Test that workflow errors don't crash execute_entity (#3983).
When a workflow raises an exception, _execute_workflow yields a raw dict
{"type": "error", ...}. The execute_entity caller must handle both dict
events and object events without crashing on attribute access.
"""
from unittest.mock import AsyncMock, MagicMock
from agent_framework_devui.models._discovery_models import EntityInfo
discovery = MagicMock(spec=EntityDiscovery)
mapper = MessageMapper()
executor = AgentFrameworkExecutor(discovery, mapper)
entity_info = EntityInfo(id="bad_wf", name="bad_wf", type="workflow", framework="agent_framework")
discovery.get_entity_info.return_value = entity_info
# Mock workflow whose run() raises
mock_workflow = MagicMock()
mock_workflow.name = "bad_wf"
def failing_run(*args, **kwargs):
raise RuntimeError("Sorry, something went wrong.")
mock_workflow.run = failing_run
discovery.load_entity = AsyncMock(return_value=mock_workflow)
request = AgentFrameworkRequest(
model="test",
input="hello",
metadata={"entity_id": "bad_wf"},
)
events = []
# This should NOT raise AttributeError: 'dict' object has no attribute 'type'
async for event in executor.execute_entity("bad_wf", request):
events.append(event)
# Should get at least one error event
assert len(events) > 0
error_events = [e for e in events if isinstance(e, dict) and e.get("type") == "error"]
assert len(error_events) > 0, f"Expected error dict events, got: {events}"
if __name__ == "__main__":
# Simple test runner
async def run_tests():
@@ -197,7 +197,7 @@ class HandoffAgentExecutor(AgentExecutor):
def __init__(
self,
agent: SupportsAgentRun,
agent: Agent,
handoffs: Sequence[HandoffConfiguration],
*,
agent_session: AgentSession | None = None,
@@ -210,7 +210,7 @@ class HandoffAgentExecutor(AgentExecutor):
"""Initialize the HandoffAgentExecutor.
Args:
agent: The agent to execute
agent: The ``Agent`` instance to execute
handoffs: Sequence of handoff configurations defining target agents
agent_session: Optional AgentSession that manages the agent's execution context
is_start_agent: Whether this agent is the starting agent in the handoff workflow.
@@ -240,20 +240,18 @@ class HandoffAgentExecutor(AgentExecutor):
def _prepare_agent_with_handoffs(
self,
agent: SupportsAgentRun,
agent: Agent,
handoffs: Sequence[HandoffConfiguration],
) -> SupportsAgentRun:
) -> Agent:
"""Prepare an agent by adding handoff tools for the specified target agents.
Args:
agent: The agent to prepare
agent: The ``Agent`` instance to prepare
handoffs: Sequence of handoff configurations defining target agents
Returns:
A new AgentExecutor instance with handoff tools added
A cloned ``Agent`` instance with handoff tools added
"""
if not isinstance(agent, Agent):
raise TypeError("Handoff can only be applied to Agent. Please ensure the agent is a Agent instance.")
# Clone the agent to avoid mutating the original
cloned_agent = self._clone_chat_agent(agent) # type: ignore
@@ -701,13 +699,15 @@ class HandoffBuilder:
approach to multi-agent collaboration. Handoffs can be configured using `.add_handoff`. If
none are specified, all agents can hand off to all others by default (making a mesh topology).
Participants must be agents. Support for custom executors is not available in handoff workflows.
Participants must be ``Agent`` instances. ``SupportsAgentRun`` protocol implementors that
are not ``Agent`` subclasses are not supported because handoff workflows require cloning,
tool injection, and middleware capabilities only available on ``Agent``.
Outputs:
The final conversation history as a list of Message once the group chat completes.
Note:
1. Agents in handoff workflows must be Agent instances and support local tool calls.
1. Agents in handoff workflows must be ``Agent`` instances and support local tool calls.
2. Handoff doesn't support intermediate outputs from agents. All outputs are returned as
they become available. This is because agents in handoff workflows are not considered
sub-agents of a central orchestrator, thus all outputs are directly emitted.
@@ -717,7 +717,7 @@ class HandoffBuilder:
self,
*,
name: str | None = None,
participants: Sequence[SupportsAgentRun] | None = None,
participants: Sequence[Agent] | None = None,
description: str | None = None,
checkpoint_storage: CheckpointStorage | None = None,
termination_condition: TerminationCondition | None = None,
@@ -734,7 +734,7 @@ class HandoffBuilder:
Args:
name: Optional workflow identifier used in logging and debugging.
If not provided, a default name will be generated.
participants: Optional list of agents that will participate in the handoff workflow.
participants: Optional list of ``Agent`` instances that will participate in the handoff workflow.
You can also call `.participants([...])` later. Each participant must have a
unique identifier (`.name` is preferred if set, otherwise `.id` is used).
description: Optional human-readable description explaining the workflow's
@@ -747,7 +747,7 @@ class HandoffBuilder:
self._description = description
# Participant related members
self._participants: dict[str, SupportsAgentRun] = {}
self._participants: dict[str, Agent] = {}
self._start_id: str | None = None
if participants:
@@ -768,11 +768,11 @@ class HandoffBuilder:
# Termination related members
self._termination_condition: Callable[[list[Message]], bool | Awaitable[bool]] | None = termination_condition
def participants(self, participants: Sequence[SupportsAgentRun]) -> "HandoffBuilder":
def participants(self, participants: Sequence[Agent]) -> "HandoffBuilder":
"""Register the agents that will participate in the handoff workflow.
Args:
participants: Sequence of SupportsAgentRun instances. Each must have a unique identifier.
participants: Sequence of ``Agent`` instances. Each must have a unique identifier.
(`.name` is preferred if set, otherwise `.id` is used).
Returns:
@@ -781,7 +781,7 @@ class HandoffBuilder:
Raises:
ValueError: If participants is empty, contains duplicates, or `.participants()`
has already been called.
TypeError: If participants are not SupportsAgentRun instances.
TypeError: If participants are not ``Agent`` instances.
Example:
@@ -804,14 +804,15 @@ class HandoffBuilder:
if not participants:
raise ValueError("participants cannot be empty")
named: dict[str, SupportsAgentRun] = {}
named: dict[str, Agent] = {}
for participant in participants:
if isinstance(participant, SupportsAgentRun):
resolved_id = self._resolve_to_id(participant)
else:
if not isinstance(participant, Agent):
raise TypeError(
f"Participants must be SupportsAgentRun or Executor instances. Got {type(participant).__name__}."
f"Participants must be Agent instances. Got {type(participant).__name__}. "
"Handoff workflows require Agent because they rely on cloning, tool injection, "
"and middleware capabilities."
)
resolved_id = self._resolve_to_id(participant)
if resolved_id in named:
raise ValueError(f"Duplicate participant name '{resolved_id}' detected")
@@ -823,8 +824,8 @@ class HandoffBuilder:
def add_handoff(
self,
source: SupportsAgentRun,
targets: Sequence[SupportsAgentRun],
source: Agent,
targets: Sequence[Agent],
*,
description: str | None = None,
) -> "HandoffBuilder":
@@ -905,7 +906,7 @@ class HandoffBuilder:
return self
def with_start_agent(self, agent: SupportsAgentRun) -> "HandoffBuilder":
def with_start_agent(self, agent: Agent) -> "HandoffBuilder":
"""Set the agent that will initiate the handoff workflow.
If not specified, the first registered participant will be used as the starting agent.
@@ -929,7 +930,7 @@ class HandoffBuilder:
def with_autonomous_mode(
self,
*,
agents: Sequence[SupportsAgentRun] | Sequence[str] | None = None,
agents: Sequence[Agent] | Sequence[str] | None = None,
prompts: dict[str, str] | None = None,
turn_limits: dict[str, int] | None = None,
) -> "HandoffBuilder":
@@ -943,7 +944,7 @@ class HandoffBuilder:
Args:
agents: Optional list of agents to enable autonomous mode for. Can be:
- Factory names (str): If using participant factories
- SupportsAgentRun instances: The actual agent objects
- SupportsAgentRun / Agent instances: The actual agent objects
- If not provided, all agents will operate in autonomous mode.
prompts: Optional mapping of agent identifiers/factory names to custom prompts to use when continuing
in autonomous mode. If not provided, a default prompt will be used.
@@ -1092,22 +1093,22 @@ class HandoffBuilder:
# region Internal Helper Methods
def _resolve_agents(self) -> dict[str, SupportsAgentRun]:
def _resolve_agents(self) -> dict[str, Agent]:
"""Resolve participant instances into agent instances.
Returns:
Map of executor IDs to `SupportsAgentRun` instances
Map of executor IDs to ``Agent`` instances
"""
if not self._participants:
raise ValueError("No participants provided. Call .participants() first.")
return self._participants
def _resolve_handoffs(self, agents: dict[str, SupportsAgentRun]) -> dict[str, list[HandoffConfiguration]]:
def _resolve_handoffs(self, agents: dict[str, Agent]) -> dict[str, list[HandoffConfiguration]]:
"""Resolve handoff configurations to executor IDs.
Args:
agents: Map of agent IDs to `SupportsAgentRun` instances
agents: Map of agent IDs to ``Agent`` instances
Returns:
Map of executor IDs to list of HandoffConfiguration instances
@@ -1154,13 +1155,13 @@ class HandoffBuilder:
def _resolve_executors(
self,
agents: dict[str, SupportsAgentRun],
agents: dict[str, Agent],
handoffs: dict[str, list[HandoffConfiguration]],
) -> dict[str, HandoffAgentExecutor]:
"""Resolve agents into HandoffAgentExecutors.
Args:
agents: Map of agent IDs to `SupportsAgentRun` instances
agents: Map of agent IDs to ``Agent`` instances
handoffs: Map of executor IDs to list of HandoffConfiguration instances
Returns:
@@ -1091,3 +1091,29 @@ async def test_auto_handoff_middleware_calls_next_for_non_handoff_tool() -> None
call_next.assert_awaited_once()
assert context.result is None
def test_handoff_builder_rejects_non_agent_supports_agent_run():
"""Verify that participants() rejects SupportsAgentRun implementations that are not Agent instances."""
from agent_framework import AgentResponse, AgentSession, SupportsAgentRun
class FakeAgentRun:
def __init__(self, id, name):
self.id = id
self.name = name
self.description = "d"
async def run(self, messages=None, *, stream=False, session=None, **kwargs):
return AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("ok")])])
def create_session(self, **kwargs):
return AgentSession()
def get_session(self, *, service_session_id, **kwargs):
return AgentSession(service_session_id=service_session_id)
fake = FakeAgentRun("a", "A")
assert isinstance(fake, SupportsAgentRun)
with pytest.raises(TypeError, match="Participants must be Agent instances"):
HandoffBuilder().participants([fake])
+2
View File
@@ -148,6 +148,8 @@ ignore = [
"**/tests/**" = ["D", "INP", "TD", "ERA001", "RUF", "S"]
"samples/**" = ["D", "INP", "ERA001", "RUF", "S", "T201", "CPY"]
"*.ipynb" = ["CPY", "E501"]
# RUF070: Assignment before yield is intentional - context manager must exit before yielding
"**/agent_framework/_workflows/_workflow.py" = ["RUF070"]
[tool.ruff.format]
docstring-code-format = true
@@ -94,7 +94,9 @@ class EchoingChatClient(BaseChatClient[OptionsT]):
response_text = f"{response_text} {suffix}"
stream_delay_seconds = float(options.get("stream_delay_seconds", 0.05))
response_message = Message(role="assistant", contents=[Content.from_text(response_text)])
response_message = Message(
role="assistant", contents=[Content.from_text(response_text)]
)
response = ChatResponse(
messages=[response_message],
@@ -146,7 +148,7 @@ async def main() -> None:
# Use the chat client directly
print("Using chat client directly:")
direct_response = await echo_client.get_response(
"Hello, custom chat client!",
[Message(role="user", text="Hello, custom chat client!")],
options={
"uppercase": True,
"suffix": "(CUSTOM OPTIONS)",
@@ -10,7 +10,14 @@ import logging
import os
from typing import Annotated
from agent_framework import Agent, Executor, WorkflowBuilder, WorkflowContext, handler, tool
from agent_framework import (
Agent,
Executor,
WorkflowBuilder,
WorkflowContext,
handler,
tool,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.devui import serve
from dotenv import load_dotenv
@@ -30,7 +37,9 @@ def get_weather(
"""Get the weather for a given location."""
conditions = ["sunny", "cloudy", "rainy", "stormy"]
temperature = 53
return f"The weather in {location} is {conditions[0]} with a high of {temperature}°C."
return (
f"The weather in {location} is {conditions[0]} with a high of {temperature}°C."
)
@tool(approval_mode="never_require")
@@ -59,7 +68,9 @@ class AddExclamation(Executor):
"""Add exclamation mark to text."""
@handler
async def add_exclamation(self, text: str, ctx: WorkflowContext[Never, str]) -> None:
async def add_exclamation(
self, text: str, ctx: WorkflowContext[Never, str]
) -> None:
"""Add exclamation and yield as workflow output."""
result = f"{text}!"
await ctx.yield_output(result)
@@ -74,9 +85,9 @@ def main():
# Create Azure OpenAI chat client
client = AzureOpenAIChatClient(
api_key=os.environ.get("AZURE_OPENAI_API_KEY"),
azure_endpoint=os.environ.get("AZURE_OPENAI_ENDPOINT"),
deployment_name=os.environ["AZURE_OPENAI_CHAT_DEPLOYMENT_NAME"],
endpoint=os.environ.get("AZURE_OPENAI_ENDPOINT"),
api_version=os.environ.get("AZURE_OPENAI_API_VERSION", "2024-10-21"),
model_id=os.environ.get("AZURE_OPENAI_CHAT_DEPLOYMENT_NAME", "gpt-4o"),
)
# Create agents
+8
View File
@@ -160,6 +160,14 @@ Sequential orchestration uses a few small adapter nodes for plumbing:
These may appear in event streams (executor_invoked/executor_completed). They're analogous to
concurrents dispatcher and aggregator and can be ignored if you only care about agent activity.
### AzureOpenAIResponsesClient vs AzureAIAgent
Workflow and orchestration samples use `AzureOpenAIResponsesClient` rather than the CRUD-style `AzureAIAgent` client. The key difference:
- **`AzureOpenAIResponsesClient`** — A lightweight client that uses the underlying Agent Service V2 (Responses API) for non-CRUD-style agents. Orchestrations use this client because agents are created locally and do not require server-side lifecycle management (create/update/delete). This is the recommended client for orchestration patterns (Sequential, Concurrent, Handoff, GroupChat, Magentic).
- **`AzureAIAgent`** — A CRUD-style client for server-managed agents. Use this when you need persistent, server-side agent definitions with features like file search, code interpreter sessions, or thread management provided by the Azure AI Agent Service.
### Environment Variables
Workflow samples that use `AzureOpenAIResponsesClient` expect:
@@ -18,10 +18,11 @@ Run with:
"""
import asyncio
import os
from pathlib import Path
from typing import Any
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.declarative import WorkflowFactory
from azure.identity import AzureCliCredential
from pydantic import BaseModel, Field
@@ -196,8 +197,12 @@ def format_order_confirmation(order_data: dict[str, Any], order_calculation: dic
async def main():
"""Run the agent to function tool workflow."""
# Create Azure OpenAI client
chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())
# Create Azure OpenAI Responses client
chat_client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
# Create the order analysis agent with structured output
order_analysis_agent = chat_client.as_agent(
@@ -6,7 +6,7 @@ This sample demonstrates an agent with function tools responding to user queries
The workflow showcases:
- **Function Tools**: Agent equipped with tools to query menu data
- **Real Azure OpenAI Agent**: Uses `AzureOpenAIChatClient` to create an agent with tools
- **Real Azure OpenAI Agent**: Uses `AzureOpenAIResponsesClient` to create an agent with tools
- **Agent Registration**: Shows how to register agents with the `WorkflowFactory`
## Tools
@@ -72,7 +72,11 @@ Session Complete
```python
# Create the agent with tools
client = AzureOpenAIChatClient(credential=AzureCliCredential())
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
menu_agent = client.as_agent(
name="MenuAgent",
instructions="You are a helpful restaurant menu assistant...",
@@ -92,6 +92,10 @@ from agent_framework.orchestrations import (
These may appear in event streams (executor_invoked/executor_completed). They're analogous to concurrent's dispatcher and aggregator and can be ignored if you only care about agent activity.
## Why AzureOpenAIResponsesClient?
Orchestration samples use `AzureOpenAIResponsesClient` rather than the CRUD-style `AzureAIAgent` client. Orchestrations create agents locally and do not require server-side lifecycle management (create/update/delete). `AzureOpenAIResponsesClient` is a lightweight client that uses the underlying Agent Service V2 (Responses API) for non-CRUD-style agents, which is ideal for orchestration patterns like Sequential, Concurrent, Handoff, GroupChat, and Magentic.
## Environment Variables
Orchestration samples that use `AzureOpenAIResponsesClient` expect:
@@ -1,10 +1,11 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from typing import Any
from agent_framework import Message
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import ConcurrentBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -26,14 +27,20 @@ Demonstrates:
- Workflow completion when idle with no pending work
Prerequisites:
- Azure OpenAI access configured for AzureOpenAIChatClient (use az login + env vars)
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
- Familiarity with Workflow events (WorkflowEvent)
"""
async def main() -> None:
# 1) Create three domain agents using AzureOpenAIChatClient
client = AzureOpenAIChatClient(credential=AzureCliCredential())
# 1) Create three domain agents using AzureOpenAIResponsesClient
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
researcher = client.as_agent(
instructions=(
@@ -1,6 +1,7 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from typing import Any
from agent_framework import (
@@ -12,7 +13,7 @@ from agent_framework import (
WorkflowContext,
handler,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import ConcurrentBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -29,21 +30,23 @@ and emit AgentExecutorResponse outputs, which allows reuse of the high-level
ConcurrentBuilder API and the default aggregator.
Demonstrates:
- Executors that create their Agent in __init__ (via AzureOpenAIChatClient)
- Executors that create their Agent in __init__ (via AzureOpenAIResponsesClient)
- A @handler that converts AgentExecutorRequest -> AgentExecutorResponse
- ConcurrentBuilder(participants=[...]) to build fan-out/fan-in
- Default aggregator returning list[Message] (one user + one assistant per agent)
- Workflow completion when all participants become idle
Prerequisites:
- Azure OpenAI configured for AzureOpenAIChatClient (az login + required env vars)
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
class ResearcherExec(Executor):
agent: Agent
def __init__(self, client: AzureOpenAIChatClient, id: str = "researcher"):
def __init__(self, client: AzureOpenAIResponsesClient, id: str = "researcher"):
self.agent = client.as_agent(
instructions=(
"You're an expert market and product researcher. Given a prompt, provide concise, factual insights,"
@@ -63,7 +66,7 @@ class ResearcherExec(Executor):
class MarketerExec(Executor):
agent: Agent
def __init__(self, client: AzureOpenAIChatClient, id: str = "marketer"):
def __init__(self, client: AzureOpenAIResponsesClient, id: str = "marketer"):
self.agent = client.as_agent(
instructions=(
"You're a creative marketing strategist. Craft compelling value propositions and target messaging"
@@ -83,7 +86,7 @@ class MarketerExec(Executor):
class LegalExec(Executor):
agent: Agent
def __init__(self, client: AzureOpenAIChatClient, id: str = "legal"):
def __init__(self, client: AzureOpenAIResponsesClient, id: str = "legal"):
self.agent = client.as_agent(
instructions=(
"You're a cautious legal/compliance reviewer. Highlight constraints, disclaimers, and policy concerns"
@@ -101,7 +104,11 @@ class LegalExec(Executor):
async def main() -> None:
client = AzureOpenAIChatClient(credential=AzureCliCredential())
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
researcher = ResearcherExec(client)
marketer = MarketerExec(client)
@@ -1,10 +1,11 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from typing import Any
from agent_framework import Message
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import ConcurrentBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -17,7 +18,7 @@ Sample: Concurrent Orchestration with Custom Aggregator
Build a concurrent workflow with ConcurrentBuilder that fans out one prompt to
multiple domain agents and fans in their responses. Override the default
aggregator with a custom async callback that uses AzureOpenAIChatClient.get_response()
aggregator with a custom async callback that uses AzureOpenAIResponsesClient.get_response()
to synthesize a concise, consolidated summary from the experts' outputs.
The workflow completes when all participants become idle.
@@ -28,12 +29,18 @@ Demonstrates:
- Workflow output yielded with the synthesized summary string
Prerequisites:
- Azure OpenAI configured for AzureOpenAIChatClient (az login + required env vars)
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
async def main() -> None:
client = AzureOpenAIChatClient(credential=AzureCliCredential())
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
researcher = client.as_agent(
instructions=(
@@ -1,6 +1,7 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from typing import cast
from agent_framework import (
@@ -8,7 +9,7 @@ from agent_framework import (
AgentResponseUpdate,
Message,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import GroupChatBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -25,7 +26,9 @@ What it does:
- Coordinates a researcher and writer agent to solve tasks collaboratively
Prerequisites:
- OpenAI environment variables configured for OpenAIChatClient
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
ORCHESTRATOR_AGENT_INSTRUCTIONS = """
@@ -39,8 +42,12 @@ Guidelines:
async def main() -> None:
# Create a chat client using Azure OpenAI and Azure CLI credentials for all agents
client = AzureOpenAIChatClient(credential=AzureCliCredential())
# Create a Responses client using Azure OpenAI and Azure CLI credentials for all agents
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
# Orchestrator agent that manages the conversation
# Note: This agent (and the underlying chat client) must support structured outputs.
@@ -2,6 +2,7 @@
import asyncio
import logging
import os
from typing import cast
from agent_framework import (
@@ -9,7 +10,7 @@ from agent_framework import (
AgentResponseUpdate,
Message,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import GroupChatBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -38,15 +39,21 @@ Participants represent:
- Doctor from Scandinavia (public health, equity, societal support)
Prerequisites:
- OpenAI environment variables configured for OpenAIChatClient
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
# Load environment variables from .env file
load_dotenv()
def _get_chat_client() -> AzureOpenAIChatClient:
return AzureOpenAIChatClient(credential=AzureCliCredential())
def _get_chat_client() -> AzureOpenAIResponsesClient:
return AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
async def main() -> None:
@@ -1,6 +1,7 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from typing import cast
from agent_framework import (
@@ -8,7 +9,7 @@ from agent_framework import (
AgentResponseUpdate,
Message,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import GroupChatBuilder, GroupChatState
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -24,7 +25,9 @@ What it does:
- Uses a pure Python function to control speaker selection based on conversation state
Prerequisites:
- OpenAI environment variables configured for OpenAIChatClient
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
@@ -36,8 +39,12 @@ def round_robin_selector(state: GroupChatState) -> str:
async def main() -> None:
# Create a chat client using Azure OpenAI and Azure CLI credentials for all agents
client = AzureOpenAIChatClient(credential=AzureCliCredential())
# Create a Responses client using Azure OpenAI and Azure CLI credentials for all agents
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
# Participant agents
expert = Agent(
@@ -2,6 +2,7 @@
import asyncio
import logging
import os
from typing import cast
from agent_framework import (
@@ -10,7 +11,7 @@ from agent_framework import (
Message,
resolve_agent_id,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import HandoffBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -28,8 +29,9 @@ Routing Pattern:
User -> Coordinator -> Specialist (iterates N times) -> Handoff -> Final Output
Prerequisites:
- `az login` (Azure CLI authentication)
- Environment variables for AzureOpenAIChatClient (AZURE_OPENAI_ENDPOINT, etc.)
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run `az login` before executing the sample.
Key Concepts:
- Autonomous interaction mode: agents iterate until they handoff
@@ -41,7 +43,7 @@ load_dotenv()
def create_agents(
client: AzureOpenAIChatClient,
client: AzureOpenAIResponsesClient,
) -> tuple[Agent, Agent, Agent]:
"""Create coordinator and specialists for autonomous iteration."""
coordinator = client.as_agent(
@@ -77,7 +79,11 @@ def create_agents(
async def main() -> None:
"""Run an autonomous handoff workflow with specialist iteration enabled."""
client = AzureOpenAIChatClient(credential=AzureCliCredential())
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
coordinator, research_agent, summary_agent = create_agents(client)
# Build the workflow with autonomous mode
@@ -1,6 +1,7 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from typing import Annotated, cast
from agent_framework import (
@@ -11,7 +12,7 @@ from agent_framework import (
WorkflowRunState,
tool,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import HandoffAgentUserRequest, HandoffBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -25,8 +26,9 @@ A handoff workflow defines a pattern that assembles agents in a mesh topology, a
them to transfer control to each other based on the conversation context.
Prerequisites:
- `az login` (Azure CLI authentication)
- Environment variables configured for AzureOpenAIChatClient (AZURE_OPENAI_ENDPOINT, etc.)
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run `az login` before executing the sample.
Key Concepts:
- Auto-registered handoff tools: HandoffBuilder automatically creates handoff tools
@@ -58,11 +60,11 @@ def process_return(order_number: Annotated[str, "Order number to process return
return f"Return initiated successfully for order {order_number}. You will receive return instructions via email."
def create_agents(client: AzureOpenAIChatClient) -> tuple[Agent, Agent, Agent, Agent]:
def create_agents(client: AzureOpenAIResponsesClient) -> tuple[Agent, Agent, Agent, Agent]:
"""Create and configure the triage and specialist agents.
Args:
client: The AzureOpenAIChatClient to use for creating agents.
client: The AzureOpenAIResponsesClient to use for creating agents.
Returns:
Tuple of (triage_agent, refund_agent, order_agent, return_agent)
@@ -192,8 +194,12 @@ async def main() -> None:
the demo reproducible and testable. In a production application, you would
replace the scripted_responses with actual user input collection.
"""
# Initialize the Azure OpenAI chat client
client = AzureOpenAIChatClient(credential=AzureCliCredential())
# Initialize the Azure OpenAI Responses client
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
# Create all agents: triage + specialists
triage, refund, order, support = create_agents(client)
@@ -3,6 +3,7 @@
import asyncio
import json
import logging
import os
from typing import cast
from agent_framework import (
@@ -11,8 +12,9 @@ from agent_framework import (
Message,
WorkflowEvent,
)
from agent_framework.openai import OpenAIChatClient, OpenAIResponsesClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import GroupChatRequestSentEvent, MagenticBuilder, MagenticProgressLedger
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
logging.basicConfig(level=logging.WARNING)
@@ -40,7 +42,9 @@ energy efficiency and CO2 emissions of several ML models, streams intermediate
events, and prints the final answer. The workflow completes when idle.
Prerequisites:
- OpenAI credentials configured for `OpenAIChatClient` and `OpenAIResponsesClient`.
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
# Load environment variables from .env file
@@ -48,25 +52,29 @@ load_dotenv()
async def main() -> None:
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
researcher_agent = Agent(
name="ResearcherAgent",
description="Specialist in research and information gathering",
instructions=(
"You are a Researcher. You find information without additional computation or quantitative analysis."
),
# This agent requires the gpt-4o-search-preview model to perform web searches.
client=OpenAIChatClient(model_id="gpt-4o-search-preview"),
client=client,
)
# Create code interpreter tool using instance method
coder_client = OpenAIResponsesClient()
code_interpreter_tool = coder_client.get_code_interpreter_tool()
code_interpreter_tool = client.get_code_interpreter_tool()
coder_agent = Agent(
name="CoderAgent",
description="A helpful assistant that writes and executes code to process and analyze data.",
instructions="You solve questions using code. Please provide detailed analysis and computation process.",
client=coder_client,
client=client,
tools=code_interpreter_tool,
)
@@ -75,7 +83,7 @@ async def main() -> None:
name="MagenticManager",
description="Orchestrator that coordinates the research and coding workflow",
instructions="You coordinate a team to complete complex tasks efficiently.",
client=OpenAIChatClient(),
client=client,
)
print("\nBuilding Magentic Workflow...")
@@ -2,6 +2,7 @@
import asyncio
import json
import os
from datetime import datetime
from pathlib import Path
from typing import cast
@@ -14,9 +15,9 @@ from agent_framework import (
WorkflowEvent,
WorkflowRunState,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import MagenticBuilder, MagenticPlanReviewRequest
from azure.identity._credentials import AzureCliCredential
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
# Load environment variables from .env file
@@ -38,7 +39,9 @@ Concepts highlighted here:
`responses` mapping so we can inject the stored human reply during restoration.
Prerequisites:
- OpenAI environment variables configured for `OpenAIChatClient`.
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
TASK = (
@@ -61,14 +64,22 @@ def build_workflow(checkpoint_storage: FileCheckpointStorage):
name="ResearcherAgent",
description="Collects background facts and references for the project.",
instructions=("You are the research lead. Gather crisp bullet points the team should know."),
client=AzureOpenAIChatClient(credential=AzureCliCredential()),
client=AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
),
)
writer = Agent(
name="WriterAgent",
description="Synthesizes the final brief for stakeholders.",
instructions=("You convert the research notes into a structured brief with milestones and risks."),
client=AzureOpenAIChatClient(credential=AzureCliCredential()),
client=AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
),
)
# Create a manager agent for orchestration
@@ -76,7 +87,11 @@ def build_workflow(checkpoint_storage: FileCheckpointStorage):
name="MagenticManager",
description="Orchestrator that coordinates the research and writing workflow",
instructions="You coordinate a team to complete complex tasks efficiently.",
client=AzureOpenAIChatClient(credential=AzureCliCredential()),
client=AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
),
)
# The builder wires in the Magentic orchestrator, sets the plan review path, and
@@ -2,6 +2,7 @@
import asyncio
import json
import os
from collections.abc import AsyncIterable
from typing import cast
@@ -11,8 +12,9 @@ from agent_framework import (
Message,
WorkflowEvent,
)
from agent_framework.openai import OpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import MagenticBuilder, MagenticPlanReviewRequest, MagenticPlanReviewResponse
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
# Load environment variables from .env file
@@ -35,7 +37,9 @@ Plan review options:
- revise(feedback): Provide textual feedback to modify the plan
Prerequisites:
- OpenAI credentials configured for `OpenAIChatClient`.
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
# Keep track of the last response to format output nicely in streaming mode
@@ -96,25 +100,31 @@ async def process_event_stream(stream: AsyncIterable[WorkflowEvent]) -> dict[str
async def main() -> None:
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
researcher_agent = Agent(
name="ResearcherAgent",
description="Specialist in research and information gathering",
instructions="You are a Researcher. You find information and gather facts.",
client=OpenAIChatClient(model_id="gpt-4o"),
client=client,
)
analyst_agent = Agent(
name="AnalystAgent",
description="Data analyst who processes and summarizes research findings",
instructions="You are an Analyst. You analyze findings and create summaries.",
client=OpenAIChatClient(model_id="gpt-4o"),
client=client,
)
manager_agent = Agent(
name="MagenticManager",
description="Orchestrator that coordinates the workflow",
instructions="You coordinate a team to complete tasks efficiently.",
client=OpenAIChatClient(model_id="gpt-4o"),
client=client,
)
print("\nBuilding Magentic Workflow with Human Plan Review...")
@@ -1,10 +1,11 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from typing import cast
from agent_framework import Message
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import SequentialBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -28,13 +29,19 @@ Note on internal adapters:
You can safely ignore them when focusing on agent progress.
Prerequisites:
- Azure OpenAI access configured for AzureOpenAIChatClient (use az login + env vars)
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
async def main() -> None:
# 1) Create agents
client = AzureOpenAIChatClient(credential=AzureCliCredential())
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
writer = client.as_agent(
instructions=("You are a concise copywriter. Provide a single, punchy marketing sentence based on the prompt."),
@@ -1,6 +1,7 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from typing import Any
from agent_framework import (
@@ -10,7 +11,7 @@ from agent_framework import (
WorkflowContext,
handler,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import SequentialBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -32,7 +33,9 @@ Custom executor contract:
- Emit the updated conversation via ctx.send_message([...])
Prerequisites:
- Azure OpenAI access configured for AzureOpenAIChatClient (use az login + env vars)
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
@@ -62,7 +65,11 @@ class Summarizer(Executor):
async def main() -> None:
# 1) Create a content agent
client = AzureOpenAIChatClient(credential=AzureCliCredential())
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
content = client.as_agent(
instructions="Produce a concise paragraph answering the user's request.",
name="content",
@@ -53,9 +53,10 @@ class BatchCompletion:
AgentInstruction = (
"You are validating exactly one Python sample.\n"
"Analyze the sample code and execute it. Determine if it runs successfully, fails, or times out.\n"
"Analyze the sample code and execute it. Based on the execution result, determine if it "
"runs successfully, fails, or times out. Feel free to install any required dependencies.\n"
"The sample can be interactive. If it is interactive, respond to the sample when prompted "
"based on your analysis of the code. You do not need to consult human on what to respond\n"
"based on your analysis of the code. You do not need to consult human on what to respond.\n"
"Return ONLY valid JSON with this schema:\n"
"{\n"
' "status": "success|failure|timeout|error",\n'
+14 -2
View File
@@ -21,6 +21,14 @@ def generate_report(results: list[RunResult]) -> Report:
Returns:
Report object with aggregated statistics
"""
# Sort results: failures, timeouts, errors first, then successes
status_priority = {
RunStatus.FAILURE: 0,
RunStatus.TIMEOUT: 1,
RunStatus.ERROR: 2,
RunStatus.SUCCESS: 3,
}
sorted_results = sorted(results, key=lambda r: status_priority[r.status])
return Report(
timestamp=datetime.now(),
@@ -29,7 +37,7 @@ def generate_report(results: list[RunResult]) -> Report:
failure_count=sum(1 for r in results if r.status == RunStatus.FAILURE),
timeout_count=sum(1 for r in results if r.status == RunStatus.TIMEOUT),
error_count=sum(1 for r in results if r.status == RunStatus.ERROR),
results=results,
results=sorted_results,
)
@@ -84,9 +92,13 @@ def print_summary(report: Report) -> None:
print(f" [PASS] Success: {report.success_count}")
print(f" [FAIL] Failure: {report.failure_count}")
print(f" [TIMEOUT] Timeout: {report.timeout_count}")
print(f" [ERROR] Error: {report.error_count}")
print(f" [ERR] Errors: {report.error_count}")
print("=" * 80)
# Print JSON output for GitHub Actions visibility
print("\nJSON Report:")
print(json.dumps(report.to_dict(), indent=2))
class GenerateReportExecutor(Executor):
"""Executor that generates the final validation report."""