Python: Rebase durable task feature branch with main (#2806)

This commit is contained in:
Laveesh Rohra
2025-12-17 14:02:36 -08:00
committed by GitHub
Unverified
parent a48a8dd524
commit 87a38bc7da
227 changed files with 11969 additions and 2638 deletions
+28 -10
View File
@@ -10,7 +10,7 @@ from pytest import fixture
@fixture
def enable_otel(request: Any) -> bool:
def enable_instrumentation(request: Any) -> bool:
"""Fixture that returns a boolean indicating if Otel is enabled."""
return request.param if hasattr(request, "param") else True
@@ -22,20 +22,31 @@ def enable_sensitive_data(request: Any) -> bool:
@fixture
def span_exporter(monkeypatch, enable_otel: bool, enable_sensitive_data: bool) -> Generator[SpanExporter]:
def span_exporter(monkeypatch, enable_instrumentation: bool, enable_sensitive_data: bool) -> Generator[SpanExporter]:
"""Fixture to remove environment variables for ObservabilitySettings."""
env_vars = [
"ENABLE_OTEL",
"ENABLE_INSTRUMENTATION",
"ENABLE_SENSITIVE_DATA",
"OTLP_ENDPOINT",
"APPLICATIONINSIGHTS_CONNECTION_STRING",
"ENABLE_CONSOLE_EXPORTERS",
"OTEL_EXPORTER_OTLP_ENDPOINT",
"OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",
"OTEL_EXPORTER_OTLP_METRICS_ENDPOINT",
"OTEL_EXPORTER_OTLP_LOGS_ENDPOINT",
"OTEL_EXPORTER_OTLP_PROTOCOL",
"OTEL_EXPORTER_OTLP_HEADERS",
"OTEL_EXPORTER_OTLP_TRACES_HEADERS",
"OTEL_EXPORTER_OTLP_METRICS_HEADERS",
"OTEL_EXPORTER_OTLP_LOGS_HEADERS",
"OTEL_SERVICE_NAME",
"OTEL_SERVICE_VERSION",
"OTEL_RESOURCE_ATTRIBUTES",
]
for key in env_vars:
monkeypatch.delenv(key, raising=False) # type: ignore
monkeypatch.setenv("ENABLE_OTEL", str(enable_otel)) # type: ignore
if not enable_otel:
monkeypatch.setenv("ENABLE_INSTRUMENTATION", str(enable_instrumentation)) # type: ignore
if not enable_instrumentation:
# we overwrite sensitive data for tests
enable_sensitive_data = False
monkeypatch.setenv("ENABLE_SENSITIVE_DATA", str(enable_sensitive_data)) # type: ignore
@@ -51,15 +62,22 @@ def span_exporter(monkeypatch, enable_otel: bool, enable_sensitive_data: bool) -
# recreate observability settings with values from above and no file.
observability_settings = observability.ObservabilitySettings(env_file_path="test.env")
observability_settings._configure() # pyright: ignore[reportPrivateUsage]
# Configure providers manually without calling _configure() to avoid OTLP imports
if enable_instrumentation or enable_sensitive_data:
from opentelemetry.sdk.trace import TracerProvider
tracer_provider = TracerProvider(resource=observability_settings._resource)
trace.set_tracer_provider(tracer_provider)
monkeypatch.setattr(observability, "OBSERVABILITY_SETTINGS", observability_settings, raising=False) # type: ignore
with (
patch("agent_framework.observability.OBSERVABILITY_SETTINGS", observability_settings),
patch("agent_framework.observability.setup_observability"),
patch("agent_framework.observability.configure_otel_providers"),
):
exporter = InMemorySpanExporter()
if enable_otel or enable_sensitive_data:
if enable_instrumentation or enable_sensitive_data:
tracer_provider = trace.get_tracer_provider()
if not hasattr(tracer_provider, "add_span_processor"):
raise RuntimeError("Tracer provider does not support adding span processors.")
@@ -21,9 +21,11 @@ from agent_framework import (
ChatResponse,
Context,
ContextProvider,
FunctionCallContent,
HostedCodeInterpreterTool,
Role,
TextContent,
ai_function,
)
from agent_framework._mcp import MCPTool
from agent_framework.exceptions import AgentExecutionException
@@ -595,3 +597,38 @@ async def test_chat_agent_with_local_mcp_tools(chat_client: ChatClientProtocol)
# Test async context manager with MCP tools
async with agent:
pass
async def test_agent_tool_receives_thread_in_kwargs(chat_client_base: Any) -> None:
"""Verify tool execution receives 'thread' inside **kwargs when function is called by client."""
captured: dict[str, Any] = {}
@ai_function(name="echo_thread_info")
def echo_thread_info(text: str, **kwargs: Any) -> str: # type: ignore[reportUnknownParameterType]
thread = kwargs.get("thread")
captured["has_thread"] = thread is not None
captured["has_message_store"] = thread.message_store is not None if isinstance(thread, AgentThread) else False
return f"echo: {text}"
# Make the base client emit a function call for our tool
chat_client_base.run_responses = [
ChatResponse(
messages=ChatMessage(
role="assistant",
contents=[FunctionCallContent(call_id="1", name="echo_thread_info", arguments='{"text": "hello"}')],
)
),
ChatResponse(messages=ChatMessage(role="assistant", text="done")),
]
agent = ChatAgent(
chat_client=chat_client_base, tools=[echo_thread_info], chat_message_store_factory=ChatMessageStore
)
thread = agent.get_new_thread()
result = await agent.run("hello", thread=thread)
assert result.text == "done"
assert captured.get("has_thread") is True
assert captured.get("has_message_store") is True
@@ -1,5 +1,7 @@
# Copyright (c) Microsoft. All rights reserved.
from collections.abc import Awaitable, Callable
import pytest
from agent_framework import (
@@ -16,6 +18,7 @@ from agent_framework import (
TextContent,
ai_function,
)
from agent_framework._middleware import FunctionInvocationContext, FunctionMiddleware
async def test_base_client_with_function_calling(chat_client_base: ChatClientProtocol):
@@ -2206,3 +2209,175 @@ async def test_streaming_error_recovery_resets_counter(chat_client_base: ChatCli
assert len(error_results) >= 1
assert len(success_results) >= 1
assert call_count == 2 # Both calls executed
class TerminateLoopMiddleware(FunctionMiddleware):
"""Middleware that sets terminate=True to exit the function calling loop."""
async def process(
self, context: FunctionInvocationContext, next_handler: Callable[[FunctionInvocationContext], Awaitable[None]]
) -> None:
# Set result to a simple value - the framework will wrap it in FunctionResultContent
context.result = "terminated by middleware"
context.terminate = True
async def test_terminate_loop_single_function_call(chat_client_base: ChatClientProtocol):
"""Test that terminate_loop=True exits the function calling loop after single function call."""
exec_counter = 0
@ai_function(name="test_function")
def ai_func(arg1: str) -> str:
nonlocal exec_counter
exec_counter += 1
return f"Processed {arg1}"
# Queue up two responses: function call, then final text
# If terminate_loop works, only the first response should be consumed
chat_client_base.run_responses = [
ChatResponse(
messages=ChatMessage(
role="assistant",
contents=[FunctionCallContent(call_id="1", name="test_function", arguments='{"arg1": "value1"}')],
)
),
ChatResponse(messages=ChatMessage(role="assistant", text="done")),
]
response = await chat_client_base.get_response(
"hello",
tool_choice="auto",
tools=[ai_func],
middleware=[TerminateLoopMiddleware()],
)
# Function should NOT have been executed - middleware intercepted it
assert exec_counter == 0
# There should be 2 messages: assistant with function call, tool result from middleware
# The loop should NOT have continued to call the LLM again
assert len(response.messages) == 2
assert response.messages[0].role == Role.ASSISTANT
assert isinstance(response.messages[0].contents[0], FunctionCallContent)
assert response.messages[1].role == Role.TOOL
assert isinstance(response.messages[1].contents[0], FunctionResultContent)
assert response.messages[1].contents[0].result == "terminated by middleware"
# Verify the second response is still in the queue (wasn't consumed)
assert len(chat_client_base.run_responses) == 1
class SelectiveTerminateMiddleware(FunctionMiddleware):
"""Only terminates for terminating_function."""
async def process(
self, context: FunctionInvocationContext, next_handler: Callable[[FunctionInvocationContext], Awaitable[None]]
) -> None:
if context.function.name == "terminating_function":
# Set result to a simple value - the framework will wrap it in FunctionResultContent
context.result = "terminated by middleware"
context.terminate = True
else:
await next_handler(context)
async def test_terminate_loop_multiple_function_calls_one_terminates(chat_client_base: ChatClientProtocol):
"""Test that any(terminate_loop=True) exits loop even with multiple function calls."""
normal_call_count = 0
terminating_call_count = 0
@ai_function(name="normal_function")
def normal_func(arg1: str) -> str:
nonlocal normal_call_count
normal_call_count += 1
return f"Normal {arg1}"
@ai_function(name="terminating_function")
def terminating_func(arg1: str) -> str:
nonlocal terminating_call_count
terminating_call_count += 1
return f"Terminating {arg1}"
# Queue up two responses: parallel function calls, then final text
chat_client_base.run_responses = [
ChatResponse(
messages=ChatMessage(
role="assistant",
contents=[
FunctionCallContent(call_id="1", name="normal_function", arguments='{"arg1": "value1"}'),
FunctionCallContent(call_id="2", name="terminating_function", arguments='{"arg1": "value2"}'),
],
)
),
ChatResponse(messages=ChatMessage(role="assistant", text="done")),
]
response = await chat_client_base.get_response(
"hello",
tool_choice="auto",
tools=[normal_func, terminating_func],
middleware=[SelectiveTerminateMiddleware()],
)
# normal_function should have executed (middleware calls next_handler)
# terminating_function should NOT have executed (middleware intercepts it)
assert normal_call_count == 1
assert terminating_call_count == 0
# There should be 2 messages: assistant with function calls, tool results
# The loop should NOT have continued to call the LLM again
assert len(response.messages) == 2
assert response.messages[0].role == Role.ASSISTANT
assert len(response.messages[0].contents) == 2
assert response.messages[1].role == Role.TOOL
# Both function results should be present
assert len(response.messages[1].contents) == 2
# Verify the second response is still in the queue (wasn't consumed)
assert len(chat_client_base.run_responses) == 1
async def test_terminate_loop_streaming_single_function_call(chat_client_base: ChatClientProtocol):
"""Test that terminate_loop=True exits the streaming function calling loop."""
exec_counter = 0
@ai_function(name="test_function")
def ai_func(arg1: str) -> str:
nonlocal exec_counter
exec_counter += 1
return f"Processed {arg1}"
# Queue up two streaming responses
chat_client_base.streaming_responses = [
[
ChatResponseUpdate(
contents=[FunctionCallContent(call_id="1", name="test_function", arguments='{"arg1": "value1"}')],
role="assistant",
),
],
[
ChatResponseUpdate(
contents=[TextContent(text="done")],
role="assistant",
)
],
]
updates = []
async for update in chat_client_base.get_streaming_response(
"hello",
tool_choice="auto",
tools=[ai_func],
middleware=[TerminateLoopMiddleware()],
):
updates.append(update)
# Function should NOT have been executed - middleware intercepted it
assert exec_counter == 0
# Should have function call update and function result update
# The loop should NOT have continued to call the LLM again
assert len(updates) == 2
# Verify the second streaming response is still in the queue (wasn't consumed)
assert len(chat_client_base.streaming_responses) == 1
+355 -112
View File
@@ -9,7 +9,7 @@ import pytest
from mcp import types
from mcp.client.session import ClientSession
from mcp.shared.exceptions import McpError
from pydantic import AnyUrl, ValidationError
from pydantic import AnyUrl, BaseModel, ValidationError
from agent_framework import (
ChatMessage,
@@ -75,17 +75,21 @@ def test_mcp_call_tool_result_to_ai_contents():
mcp_result = types.CallToolResult(
content=[
types.TextContent(type="text", text="Result text"),
types.ImageContent(type="image", data="data:image/png;base64,xyz", mimeType="image/png"),
types.ImageContent(type="image", data="xyz", mimeType="image/png"),
types.ImageContent(type="image", data=b"abc", mimeType="image/webp"),
]
)
ai_contents = _mcp_call_tool_result_to_ai_contents(mcp_result)
assert len(ai_contents) == 2
assert len(ai_contents) == 3
assert isinstance(ai_contents[0], TextContent)
assert ai_contents[0].text == "Result text"
assert isinstance(ai_contents[1], DataContent)
assert ai_contents[1].uri == "data:image/png;base64,xyz"
assert ai_contents[1].media_type == "image/png"
assert isinstance(ai_contents[2], DataContent)
assert ai_contents[2].uri == "data:image/webp;base64,abc"
assert ai_contents[2].media_type == "image/webp"
def test_mcp_call_tool_result_with_meta_error():
@@ -183,7 +187,7 @@ def test_mcp_call_tool_result_regression_successful_workflow():
mcp_result = types.CallToolResult(
content=[
types.TextContent(type="text", text="Success message"),
types.ImageContent(type="image", data="data:image/jpeg;base64,abc123", mimeType="image/jpeg"),
types.ImageContent(type="image", data="abc123", mimeType="image/jpeg"),
]
)
@@ -218,7 +222,8 @@ def test_mcp_content_types_to_ai_content_text():
def test_mcp_content_types_to_ai_content_image():
"""Test conversion of MCP image content to AI content."""
mcp_content = types.ImageContent(type="image", data="data:image/jpeg;base64,abc", mimeType="image/jpeg")
mcp_content = types.ImageContent(type="image", data="abc", mimeType="image/jpeg")
mcp_content = types.ImageContent(type="image", data=b"abc", mimeType="image/jpeg")
ai_content = _mcp_type_to_ai_content(mcp_content)[0]
assert isinstance(ai_content, DataContent)
@@ -229,7 +234,7 @@ def test_mcp_content_types_to_ai_content_image():
def test_mcp_content_types_to_ai_content_audio():
"""Test conversion of MCP audio content to AI content."""
mcp_content = types.AudioContent(type="audio", data="data:audio/wav;base64,def", mimeType="audio/wav")
mcp_content = types.AudioContent(type="audio", data="def", mimeType="audio/wav")
ai_content = _mcp_type_to_ai_content(mcp_content)[0]
assert isinstance(ai_content, DataContent)
@@ -357,122 +362,360 @@ def test_chat_message_to_mcp_types():
assert isinstance(mcp_contents[1], types.ImageContent)
def test_get_input_model_from_mcp_tool():
"""Test creation of input model from MCP tool."""
tool = types.Tool(
name="test_tool",
description="A test tool",
inputSchema={
"type": "object",
"properties": {"param1": {"type": "string"}, "param2": {"type": "number"}},
"required": ["param1"],
},
)
model = _get_input_model_from_mcp_tool(tool)
# Create an instance to verify the model works
instance = model(param1="test", param2=42)
assert instance.param1 == "test"
assert instance.param2 == 42
# Test validation
with pytest.raises(ValidationError): # Missing required param1
model(param2=42)
def test_get_input_model_from_mcp_tool_with_nested_object():
"""Test creation of input model from MCP tool with nested object property."""
tool = types.Tool(
name="get_customer_detail",
description="Get customer details",
inputSchema={
"type": "object",
"properties": {
"params": {
"type": "object",
"properties": {"customer_id": {"type": "integer"}},
"required": ["customer_id"],
@pytest.mark.parametrize(
"test_id,input_schema,valid_data,expected_values,invalid_data,validation_check",
[
# Basic types with required/optional fields
(
"basic_types",
{
"type": "object",
"properties": {"param1": {"type": "string"}, "param2": {"type": "number"}},
"required": ["param1"],
},
{"param1": "test", "param2": 42},
{"param1": "test", "param2": 42},
{"param2": 42}, # Missing required param1
None,
),
# Nested object
(
"nested_object",
{
"type": "object",
"properties": {
"params": {
"type": "object",
"properties": {"customer_id": {"type": "integer"}},
"required": ["customer_id"],
}
},
"required": ["params"],
},
{"params": {"customer_id": 251}},
{"params.customer_id": 251},
{"params": {}}, # Missing required customer_id
lambda instance: isinstance(instance.params, BaseModel),
),
# $ref resolution
(
"ref_schema",
{
"type": "object",
"properties": {"params": {"$ref": "#/$defs/CustomerIdParam"}},
"required": ["params"],
"$defs": {
"CustomerIdParam": {
"type": "object",
"properties": {"customer_id": {"type": "integer"}},
"required": ["customer_id"],
}
},
},
{"params": {"customer_id": 251}},
{"params.customer_id": 251},
{"params": {}}, # Missing required customer_id
lambda instance: isinstance(instance.params, BaseModel),
),
# Array of strings (typed)
(
"array_of_strings",
{
"type": "object",
"properties": {
"tags": {
"type": "array",
"description": "List of tags",
"items": {"type": "string"},
}
},
"required": ["tags"],
},
{"tags": ["tag1", "tag2", "tag3"]},
{"tags": ["tag1", "tag2", "tag3"]},
None, # No validation error test for this case
None,
),
# Array of integers (typed)
(
"array_of_integers",
{
"type": "object",
"properties": {
"numbers": {
"type": "array",
"description": "List of integers",
"items": {"type": "integer"},
}
},
"required": ["numbers"],
},
{"numbers": [1, 2, 3]},
{"numbers": [1, 2, 3]},
None,
None,
),
# Array of objects (complex nested)
(
"array_of_objects",
{
"type": "object",
"properties": {
"users": {
"type": "array",
"description": "List of users",
"items": {
"type": "object",
"properties": {
"id": {"type": "integer", "description": "User ID"},
"name": {"type": "string", "description": "User name"},
},
"required": ["id", "name"],
},
}
},
"required": ["users"],
},
{"users": [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]},
{"users[0].id": 1, "users[0].name": "Alice", "users[1].id": 2, "users[1].name": "Bob"},
{"users": [{"id": 1}]}, # Missing required 'name'
lambda instance: all(isinstance(user, BaseModel) for user in instance.users),
),
# Deeply nested objects (3+ levels)
(
"deeply_nested",
{
"type": "object",
"properties": {
"query": {
"type": "object",
"properties": {
"filters": {
"type": "object",
"properties": {
"date_range": {
"type": "object",
"properties": {
"start": {"type": "string"},
"end": {"type": "string"},
},
"required": ["start", "end"],
},
"categories": {"type": "array", "items": {"type": "string"}},
},
"required": ["date_range"],
}
},
"required": ["filters"],
}
},
"required": ["query"],
},
{
"query": {
"filters": {
"date_range": {"start": "2024-01-01", "end": "2024-12-31"},
"categories": ["tech", "science"],
}
}
},
"required": ["params"],
},
)
model = _get_input_model_from_mcp_tool(tool)
{
"query.filters.date_range.start": "2024-01-01",
"query.filters.date_range.end": "2024-12-31",
"query.filters.categories": ["tech", "science"],
},
{"query": {"filters": {"date_range": {}}}}, # Missing required start and end
None,
),
# Complex $ref with nested structure
(
"ref_nested_structure",
{
"type": "object",
"properties": {"order": {"$ref": "#/$defs/OrderParams"}},
"required": ["order"],
"$defs": {
"OrderParams": {
"type": "object",
"properties": {
"customer": {"$ref": "#/$defs/Customer"},
"items": {"type": "array", "items": {"$ref": "#/$defs/OrderItem"}},
},
"required": ["customer", "items"],
},
"Customer": {
"type": "object",
"properties": {"id": {"type": "integer"}, "email": {"type": "string"}},
"required": ["id", "email"],
},
"OrderItem": {
"type": "object",
"properties": {"product_id": {"type": "string"}, "quantity": {"type": "integer"}},
"required": ["product_id", "quantity"],
},
},
},
{
"order": {
"customer": {"id": 123, "email": "test@example.com"},
"items": [{"product_id": "prod1", "quantity": 2}],
}
},
{
"order.customer.id": 123,
"order.customer.email": "test@example.com",
"order.items[0].product_id": "prod1",
"order.items[0].quantity": 2,
},
{"order": {"customer": {"id": 123}, "items": []}}, # Missing email
lambda instance: isinstance(instance.order.customer, BaseModel),
),
# Mixed types (primitives, arrays, nested objects)
(
"mixed_types",
{
"type": "object",
"properties": {
"simple_string": {"type": "string"},
"simple_number": {"type": "integer"},
"string_array": {"type": "array", "items": {"type": "string"}},
"nested_config": {
"type": "object",
"properties": {
"enabled": {"type": "boolean"},
"options": {"type": "array", "items": {"type": "string"}},
},
"required": ["enabled"],
},
},
"required": ["simple_string", "nested_config"],
},
{
"simple_string": "test",
"simple_number": 42,
"string_array": ["a", "b"],
"nested_config": {"enabled": True, "options": ["opt1", "opt2"]},
},
{
"simple_string": "test",
"simple_number": 42,
"string_array": ["a", "b"],
"nested_config.enabled": True,
"nested_config.options": ["opt1", "opt2"],
},
None,
None,
),
# Empty schema (no properties)
(
"empty_schema",
{"type": "object", "properties": {}},
{},
{},
None,
None,
),
# All primitive types
(
"all_primitives",
{
"type": "object",
"properties": {
"string_field": {"type": "string"},
"integer_field": {"type": "integer"},
"number_field": {"type": "number"},
"boolean_field": {"type": "boolean"},
},
},
{"string_field": "test", "integer_field": 42, "number_field": 3.14, "boolean_field": True},
{"string_field": "test", "integer_field": 42, "number_field": 3.14, "boolean_field": True},
None,
None,
),
# Edge case: unresolvable $ref (fallback to dict)
(
"unresolvable_ref",
{
"type": "object",
"properties": {"data": {"$ref": "#/$defs/NonExistent"}},
"$defs": {},
},
{"data": {"key": "value"}},
{"data": {"key": "value"}},
None,
None,
),
# Edge case: array without items schema (fallback to bare list)
(
"array_no_items",
{
"type": "object",
"properties": {"items": {"type": "array"}},
},
{"items": [1, "two", 3.0]},
{"items": [1, "two", 3.0]},
None,
None,
),
# Edge case: object without properties (fallback to dict)
(
"object_no_properties",
{
"type": "object",
"properties": {"config": {"type": "object"}},
},
{"config": {"arbitrary": "data", "nested": {"key": "value"}}},
{"config": {"arbitrary": "data", "nested": {"key": "value"}}},
None,
None,
),
],
)
def test_get_input_model_from_mcp_tool_parametrized(
test_id, input_schema, valid_data, expected_values, invalid_data, validation_check
):
"""Parametrized test for JSON schema to Pydantic model conversion.
# Create an instance to verify the model works with nested objects
instance = model(params={"customer_id": 251})
assert instance.params == {"customer_id": 251}
assert isinstance(instance.params, dict)
This test covers various edge cases including:
- Basic types with required/optional fields
- Nested objects
- $ref resolution
- Typed arrays (strings, integers, objects)
- Deeply nested structures
- Complex $ref with nested structures
- Mixed types
# Verify model_dump produces the correct nested structure
dumped = instance.model_dump()
assert dumped == {"params": {"customer_id": 251}}
def test_get_input_model_from_mcp_tool_with_ref_schema():
"""Test creation of input model from MCP tool with $ref schema.
This simulates a FastMCP tool that uses Pydantic models with $ref in the schema.
The schema should be resolved and nested objects should be preserved.
To add a new test case, add a tuple to the parametrize decorator with:
- test_id: A descriptive name for the test case
- input_schema: The JSON schema (inputSchema dict)
- valid_data: Valid data to instantiate the model
- expected_values: Dict of expected values (supports dot notation for nested access)
- invalid_data: Invalid data to test validation errors (None to skip)
- validation_check: Optional callable to perform additional validation checks
"""
# This is similar to what FastMCP generates when you have:
# async def get_customer_detail(params: CustomerIdParam) -> CustomerDetail
tool = types.Tool(
name="get_customer_detail",
description="Get customer details",
inputSchema={
"type": "object",
"properties": {"params": {"$ref": "#/$defs/CustomerIdParam"}},
"required": ["params"],
"$defs": {
"CustomerIdParam": {
"type": "object",
"properties": {"customer_id": {"type": "integer"}},
"required": ["customer_id"],
}
},
},
)
tool = types.Tool(name="test_tool", description="A test tool", inputSchema=input_schema)
model = _get_input_model_from_mcp_tool(tool)
# Create an instance to verify the model works with $ref schemas
instance = model(params={"customer_id": 251})
assert instance.params == {"customer_id": 251}
assert isinstance(instance.params, dict)
# Test valid data
instance = model(**valid_data)
# Verify model_dump produces the correct nested structure
dumped = instance.model_dump()
assert dumped == {"params": {"customer_id": 251}}
# Check expected values
for field_path, expected_value in expected_values.items():
# Support dot notation and array indexing for nested access
current = instance
parts = field_path.replace("]", "").replace("[", ".").split(".")
for part in parts:
current = current[int(part)] if part.isdigit() else getattr(current, part)
assert current == expected_value, f"Field {field_path} = {current}, expected {expected_value}"
# Run additional validation checks if provided
if validation_check:
assert validation_check(instance), f"Validation check failed for {test_id}"
def test_get_input_model_from_mcp_tool_with_simple_array():
"""Test array with simple items schema (items schema should be preserved in json_schema_extra)."""
tool = types.Tool(
name="simple_array_tool",
description="Tool with simple array",
inputSchema={
"type": "object",
"properties": {
"tags": {
"type": "array",
"description": "List of tags",
"items": {"type": "string"}, # Simple string array
}
},
"required": ["tags"],
},
)
model = _get_input_model_from_mcp_tool(tool)
# Create an instance
instance = model(tags=["tag1", "tag2", "tag3"])
assert instance.tags == ["tag1", "tag2", "tag3"]
# Verify JSON schema still preserves items for simple types
json_schema = model.model_json_schema()
tags_property = json_schema["properties"]["tags"]
assert "items" in tags_property
assert tags_property["items"]["type"] == "string"
# Test invalid data if provided
if invalid_data is not None:
with pytest.raises(ValidationError):
model(**invalid_data)
def test_get_input_model_from_mcp_prompt():
@@ -193,7 +193,8 @@ class TestChatAgentFunctionBasedMiddleware:
# Create a message to start the conversation
messages = [ChatMessage(role=Role.USER, text="test message")]
# Set up chat client to return a function call
# Set up chat client to return a function call, then a final response
# If terminate works correctly, only the first response should be consumed
chat_client.responses = [
ChatResponse(
messages=[
@@ -204,7 +205,8 @@ class TestChatAgentFunctionBasedMiddleware:
],
)
]
)
),
ChatResponse(messages=[ChatMessage(role=Role.ASSISTANT, text="this should not be consumed")]),
]
# Create the test function with the expected signature
@@ -222,7 +224,11 @@ class TestChatAgentFunctionBasedMiddleware:
# Verify that function was not called and only middleware executed
assert execution_order == ["middleware_before", "middleware_after"]
assert "function_called" not in execution_order
assert execution_order == ["middleware_before", "middleware_after"]
# Verify the chat client was only called once (no extra LLM call after termination)
assert chat_client.call_count == 1
# Verify the second response is still in the queue (wasn't consumed)
assert len(chat_client.responses) == 1
async def test_function_middleware_with_post_termination(self, chat_client: "MockChatClient") -> None:
"""Test that function middleware can terminate execution after calling next()."""
@@ -242,7 +248,8 @@ class TestChatAgentFunctionBasedMiddleware:
# Create a message to start the conversation
messages = [ChatMessage(role=Role.USER, text="test message")]
# Set up chat client to return a function call
# Set up chat client to return a function call, then a final response
# If terminate works correctly, only the first response should be consumed
chat_client.responses = [
ChatResponse(
messages=[
@@ -253,7 +260,8 @@ class TestChatAgentFunctionBasedMiddleware:
],
)
]
)
),
ChatResponse(messages=[ChatMessage(role=Role.ASSISTANT, text="this should not be consumed")]),
]
# Create the test function with the expected signature
@@ -273,6 +281,11 @@ class TestChatAgentFunctionBasedMiddleware:
assert "function_called" in execution_order
assert execution_order == ["middleware_before", "function_called", "middleware_after"]
# Verify the chat client was only called once (no extra LLM call after termination)
assert chat_client.call_count == 1
# Verify the second response is still in the queue (wasn't consumed)
assert len(chat_client.responses) == 1
async def test_function_based_agent_middleware_with_chat_agent(self, chat_client: "MockChatClient") -> None:
"""Test function-based agent middleware with ChatAgent."""
execution_order: list[str] = []
@@ -33,8 +33,8 @@ from agent_framework.observability import (
ChatMessageListTimestampFilter,
OtelAttr,
get_function_span,
use_agent_observability,
use_observability,
use_agent_instrumentation,
use_instrumentation,
)
# region Test constants
@@ -157,7 +157,7 @@ def test_start_span_with_tool_call_id(span_exporter: InMemorySpanExporter):
assert span.attributes[OtelAttr.TOOL_TYPE] == "function"
# region Test use_observability decorator
# region Test use_instrumentation decorator
def test_decorator_with_valid_class():
@@ -175,7 +175,7 @@ def test_decorator_with_valid_class():
return gen()
# Apply the decorator
decorated_class = use_observability(MockChatClient)
decorated_class = use_instrumentation(MockChatClient)
assert hasattr(decorated_class, OPEN_TELEMETRY_CHAT_CLIENT_MARKER)
@@ -187,7 +187,7 @@ def test_decorator_with_missing_methods():
# Apply the decorator - should not raise an error
with pytest.raises(ChatClientInitializationError):
use_observability(MockChatClient)
use_instrumentation(MockChatClient)
def test_decorator_with_partial_methods():
@@ -200,7 +200,7 @@ def test_decorator_with_partial_methods():
return Mock()
with pytest.raises(ChatClientInitializationError):
use_observability(MockChatClient)
use_instrumentation(MockChatClient)
# region Test telemetry decorator with mock client
@@ -235,7 +235,7 @@ def mock_chat_client():
@pytest.mark.parametrize("enable_sensitive_data", [True, False], indirect=True)
async def test_chat_client_observability(mock_chat_client, span_exporter: InMemorySpanExporter, enable_sensitive_data):
"""Test that when diagnostics are enabled, telemetry is applied."""
client = use_observability(mock_chat_client)()
client = use_instrumentation(mock_chat_client)()
messages = [ChatMessage(role=Role.USER, text="Test message")]
span_exporter.clear()
@@ -258,8 +258,8 @@ async def test_chat_client_observability(mock_chat_client, span_exporter: InMemo
async def test_chat_client_streaming_observability(
mock_chat_client, span_exporter: InMemorySpanExporter, enable_sensitive_data
):
"""Test streaming telemetry through the use_observability decorator."""
client = use_observability(mock_chat_client)()
"""Test streaming telemetry through the use_instrumentation decorator."""
client = use_instrumentation(mock_chat_client)()
messages = [ChatMessage(role=Role.USER, text="Test")]
span_exporter.clear()
# Collect all yielded updates
@@ -282,7 +282,7 @@ async def test_chat_client_streaming_observability(
async def test_chat_client_without_model_id_observability(mock_chat_client, span_exporter: InMemorySpanExporter):
"""Test telemetry shouldn't fail when the model_id is not provided for unknown reason."""
client = use_observability(mock_chat_client)()
client = use_instrumentation(mock_chat_client)()
messages = [ChatMessage(role=Role.USER, text="Test")]
span_exporter.clear()
response = await client.get_response(messages=messages)
@@ -301,7 +301,7 @@ async def test_chat_client_streaming_without_model_id_observability(
mock_chat_client, span_exporter: InMemorySpanExporter
):
"""Test streaming telemetry shouldn't fail when the model_id is not provided for unknown reason."""
client = use_observability(mock_chat_client)()
client = use_instrumentation(mock_chat_client)()
messages = [ChatMessage(role=Role.USER, text="Test")]
span_exporter.clear()
# Collect all yielded updates
@@ -329,7 +329,7 @@ def test_prepend_user_agent_with_none_value():
assert AGENT_FRAMEWORK_USER_AGENT in str(result["User-Agent"])
# region Test use_agent_observability decorator
# region Test use_agent_instrumentation decorator
def test_agent_decorator_with_valid_class():
@@ -337,7 +337,7 @@ def test_agent_decorator_with_valid_class():
# Create a mock class with the required methods
class MockChatClientAgent:
AGENT_SYSTEM_NAME = "test_agent_system"
AGENT_PROVIDER_NAME = "test_agent_system"
def __init__(self):
self.id = "test_agent_id"
@@ -358,7 +358,7 @@ def test_agent_decorator_with_valid_class():
return AgentThread()
# Apply the decorator
decorated_class = use_agent_observability(MockChatClientAgent)
decorated_class = use_agent_instrumentation(MockChatClientAgent)
assert hasattr(decorated_class, OPEN_TELEMETRY_AGENT_MARKER)
@@ -367,19 +367,19 @@ def test_agent_decorator_with_missing_methods():
"""Test that agent decorator handles classes missing required methods gracefully."""
class MockAgent:
AGENT_SYSTEM_NAME = "test_agent_system"
AGENT_PROVIDER_NAME = "test_agent_system"
# Apply the decorator - should not raise an error
with pytest.raises(AgentInitializationError):
use_agent_observability(MockAgent)
use_agent_instrumentation(MockAgent)
def test_agent_decorator_with_partial_methods():
"""Test agent decorator when only one method is present."""
from agent_framework.observability import use_agent_observability
from agent_framework.observability import use_agent_instrumentation
class MockAgent:
AGENT_SYSTEM_NAME = "test_agent_system"
AGENT_PROVIDER_NAME = "test_agent_system"
def __init__(self):
self.id = "test_agent_id"
@@ -390,7 +390,7 @@ def test_agent_decorator_with_partial_methods():
return Mock()
with pytest.raises(AgentInitializationError):
use_agent_observability(MockAgent)
use_agent_instrumentation(MockAgent)
# region Test agent telemetry decorator with mock agent
@@ -401,7 +401,7 @@ def mock_chat_agent():
"""Create a mock chat client agent for testing."""
class MockChatClientAgent:
AGENT_SYSTEM_NAME = "test_agent_system"
AGENT_PROVIDER_NAME = "test_agent_system"
def __init__(self):
self.id = "test_agent_id"
@@ -433,7 +433,7 @@ async def test_agent_instrumentation_enabled(
):
"""Test that when agent diagnostics are enabled, telemetry is applied."""
agent = use_agent_observability(mock_chat_agent)()
agent = use_agent_instrumentation(mock_chat_agent)()
span_exporter.clear()
response = await agent.run("Test message")
@@ -457,8 +457,8 @@ async def test_agent_instrumentation_enabled(
async def test_agent_streaming_response_with_diagnostics_enabled_via_decorator(
mock_chat_agent: AgentProtocol, span_exporter: InMemorySpanExporter, enable_sensitive_data
):
"""Test agent streaming telemetry through the use_agent_observability decorator."""
agent = use_agent_observability(mock_chat_agent)()
"""Test agent streaming telemetry through the use_agent_instrumentation decorator."""
agent = use_agent_instrumentation(mock_chat_agent)()
span_exporter.clear()
updates = []
async for update in agent.run_stream("Test message"):
@@ -522,3 +522,393 @@ async def test_function_call_with_error_handling(span_exporter: InMemorySpanExpo
exception_message = exception_event.attributes["exception.message"]
assert isinstance(exception_message, str)
assert "Function execution failed" in exception_message
# region Test OTEL environment variable parsing
@pytest.mark.skipif(
True,
reason="Skipping OTLP exporter tests - optional dependency not installed by default",
)
def test_get_exporters_from_env_with_grpc_endpoint(monkeypatch):
"""Test _get_exporters_from_env with OTEL_EXPORTER_OTLP_ENDPOINT (gRPC)."""
from agent_framework.observability import _get_exporters_from_env
monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317")
monkeypatch.setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc")
exporters = _get_exporters_from_env()
# Should return 3 exporters (trace, metrics, logs)
assert len(exporters) == 3
@pytest.mark.skipif(
True,
reason="Skipping OTLP exporter tests - optional dependency not installed by default",
)
def test_get_exporters_from_env_with_http_endpoint(monkeypatch):
"""Test _get_exporters_from_env with OTEL_EXPORTER_OTLP_ENDPOINT (HTTP)."""
from agent_framework.observability import _get_exporters_from_env
monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4318")
monkeypatch.setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "http")
exporters = _get_exporters_from_env()
# Should return 3 exporters (trace, metrics, logs)
assert len(exporters) == 3
@pytest.mark.skipif(
True,
reason="Skipping OTLP exporter tests - optional dependency not installed by default",
)
def test_get_exporters_from_env_with_individual_endpoints(monkeypatch):
"""Test _get_exporters_from_env with individual signal endpoints."""
from agent_framework.observability import _get_exporters_from_env
monkeypatch.setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "http://localhost:4317")
monkeypatch.setenv("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", "http://localhost:4318")
monkeypatch.setenv("OTEL_EXPORTER_OTLP_LOGS_ENDPOINT", "http://localhost:4319")
monkeypatch.setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc")
exporters = _get_exporters_from_env()
# Should return 3 exporters (trace, metrics, logs)
assert len(exporters) == 3
@pytest.mark.skipif(
True,
reason="Skipping OTLP exporter tests - optional dependency not installed by default",
)
def test_get_exporters_from_env_with_headers(monkeypatch):
"""Test _get_exporters_from_env with OTEL_EXPORTER_OTLP_HEADERS."""
from agent_framework.observability import _get_exporters_from_env
monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317")
monkeypatch.setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc")
monkeypatch.setenv("OTEL_EXPORTER_OTLP_HEADERS", "key1=value1,key2=value2")
exporters = _get_exporters_from_env()
# Should return 3 exporters with headers
assert len(exporters) == 3
@pytest.mark.skipif(
True,
reason="Skipping OTLP exporter tests - optional dependency not installed by default",
)
def test_get_exporters_from_env_with_signal_specific_headers(monkeypatch):
"""Test _get_exporters_from_env with signal-specific headers."""
from agent_framework.observability import _get_exporters_from_env
monkeypatch.setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "http://localhost:4317")
monkeypatch.setenv("OTEL_EXPORTER_OTLP_TRACES_HEADERS", "trace-key=trace-value")
monkeypatch.setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc")
exporters = _get_exporters_from_env()
# Should have at least the traces exporter
assert len(exporters) >= 1
@pytest.mark.skipif(
True,
reason="Skipping OTLP exporter tests - optional dependency not installed by default",
)
def test_get_exporters_from_env_without_env_vars(monkeypatch):
"""Test _get_exporters_from_env returns empty list when no env vars set."""
from agent_framework.observability import _get_exporters_from_env
# Clear all OTEL env vars
for key in [
"OTEL_EXPORTER_OTLP_ENDPOINT",
"OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",
"OTEL_EXPORTER_OTLP_METRICS_ENDPOINT",
"OTEL_EXPORTER_OTLP_LOGS_ENDPOINT",
]:
monkeypatch.delenv(key, raising=False)
exporters = _get_exporters_from_env()
# Should return empty list
assert len(exporters) == 0
@pytest.mark.skipif(
True,
reason="Skipping OTLP exporter tests - optional dependency not installed by default",
)
def test_get_exporters_from_env_missing_grpc_dependency(monkeypatch):
"""Test _get_exporters_from_env raises ImportError when gRPC exporters not installed."""
from agent_framework.observability import _get_exporters_from_env
monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317")
monkeypatch.setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc")
# Mock the import to raise ImportError
original_import = __builtins__.__import__
def mock_import(name, *args, **kwargs):
if "opentelemetry.exporter.otlp.proto.grpc" in name:
raise ImportError("No module named 'opentelemetry.exporter.otlp.proto.grpc'")
return original_import(name, *args, **kwargs)
monkeypatch.setattr(__builtins__, "__import__", mock_import)
with pytest.raises(ImportError, match="opentelemetry-exporter-otlp-proto-grpc"):
_get_exporters_from_env()
# region Test create_resource
def test_create_resource_from_env(monkeypatch):
"""Test create_resource reads OTEL environment variables."""
from agent_framework.observability import create_resource
monkeypatch.setenv("OTEL_SERVICE_NAME", "test-service")
monkeypatch.setenv("OTEL_SERVICE_VERSION", "1.0.0")
monkeypatch.setenv("OTEL_RESOURCE_ATTRIBUTES", "deployment.environment=production,host.name=server1")
resource = create_resource()
assert resource.attributes["service.name"] == "test-service"
assert resource.attributes["service.version"] == "1.0.0"
assert resource.attributes["deployment.environment"] == "production"
assert resource.attributes["host.name"] == "server1"
def test_create_resource_with_parameters_override_env(monkeypatch):
"""Test create_resource parameters override environment variables."""
from agent_framework.observability import create_resource
monkeypatch.setenv("OTEL_SERVICE_NAME", "env-service")
monkeypatch.setenv("OTEL_SERVICE_VERSION", "0.1.0")
resource = create_resource(service_name="param-service", service_version="2.0.0")
# Parameters should override env vars
assert resource.attributes["service.name"] == "param-service"
assert resource.attributes["service.version"] == "2.0.0"
def test_create_resource_with_custom_attributes(monkeypatch):
"""Test create_resource accepts custom attributes."""
from agent_framework.observability import create_resource
resource = create_resource(custom_attr="custom_value", another_attr=123)
assert resource.attributes["custom_attr"] == "custom_value"
assert resource.attributes["another_attr"] == 123
# region Test _create_otlp_exporters
@pytest.mark.skipif(
True,
reason="Skipping OTLP exporter tests - optional dependency not installed by default",
)
def test_create_otlp_exporters_grpc_with_single_endpoint():
"""Test _create_otlp_exporters creates gRPC exporters with single endpoint."""
from agent_framework.observability import _create_otlp_exporters
exporters = _create_otlp_exporters(endpoint="http://localhost:4317", protocol="grpc")
# Should return 3 exporters (trace, metrics, logs)
assert len(exporters) == 3
@pytest.mark.skipif(
True,
reason="Skipping OTLP exporter tests - optional dependency not installed by default",
)
def test_create_otlp_exporters_http_with_single_endpoint():
"""Test _create_otlp_exporters creates HTTP exporters with single endpoint."""
from agent_framework.observability import _create_otlp_exporters
exporters = _create_otlp_exporters(endpoint="http://localhost:4318", protocol="http")
# Should return 3 exporters (trace, metrics, logs)
assert len(exporters) == 3
@pytest.mark.skipif(
True,
reason="Skipping OTLP exporter tests - optional dependency not installed by default",
)
def test_create_otlp_exporters_with_individual_endpoints():
"""Test _create_otlp_exporters with individual signal endpoints."""
from agent_framework.observability import _create_otlp_exporters
exporters = _create_otlp_exporters(
protocol="grpc",
traces_endpoint="http://localhost:4317",
metrics_endpoint="http://localhost:4318",
logs_endpoint="http://localhost:4319",
)
# Should return 3 exporters
assert len(exporters) == 3
@pytest.mark.skipif(
True,
reason="Skipping OTLP exporter tests - optional dependency not installed by default",
)
def test_create_otlp_exporters_with_headers():
"""Test _create_otlp_exporters with headers."""
from agent_framework.observability import _create_otlp_exporters
exporters = _create_otlp_exporters(
endpoint="http://localhost:4317", protocol="grpc", headers={"Authorization": "Bearer token"}
)
# Should return 3 exporters with headers
assert len(exporters) == 3
@pytest.mark.skipif(
True,
reason="Skipping OTLP exporter tests - optional dependency not installed by default",
)
def test_create_otlp_exporters_grpc_missing_dependency():
"""Test _create_otlp_exporters raises ImportError when gRPC exporters not installed."""
import sys
from unittest.mock import patch
from agent_framework.observability import _create_otlp_exporters
# Mock the import to raise ImportError
with (
patch.dict(sys.modules, {"opentelemetry.exporter.otlp.proto.grpc.trace_exporter": None}),
pytest.raises(ImportError, match="opentelemetry-exporter-otlp-proto-grpc"),
):
_create_otlp_exporters(endpoint="http://localhost:4317", protocol="grpc")
# region Test configure_otel_providers with views
@pytest.mark.skipif(
True,
reason="Skipping OTLP exporter tests - optional dependency not installed by default",
)
def test_configure_otel_providers_with_views(monkeypatch):
"""Test configure_otel_providers accepts views parameter."""
from opentelemetry.sdk.metrics import View
from opentelemetry.sdk.metrics.view import DropAggregation
from agent_framework.observability import configure_otel_providers
# Clear all OTEL env vars
for key in [
"OTEL_EXPORTER_OTLP_ENDPOINT",
"OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",
"OTEL_EXPORTER_OTLP_METRICS_ENDPOINT",
"OTEL_EXPORTER_OTLP_LOGS_ENDPOINT",
]:
monkeypatch.delenv(key, raising=False)
# Create a view that drops all metrics
views = [View(instrument_name="*", aggregation=DropAggregation())]
# Should not raise an error
configure_otel_providers(views=views)
@pytest.mark.skipif(
True,
reason="Skipping OTLP exporter tests - optional dependency not installed by default",
)
def test_configure_otel_providers_without_views(monkeypatch):
"""Test configure_otel_providers works without views parameter."""
from agent_framework.observability import configure_otel_providers
# Clear all OTEL env vars
for key in [
"OTEL_EXPORTER_OTLP_ENDPOINT",
"OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",
"OTEL_EXPORTER_OTLP_METRICS_ENDPOINT",
"OTEL_EXPORTER_OTLP_LOGS_ENDPOINT",
]:
monkeypatch.delenv(key, raising=False)
# Should not raise an error with default empty views
configure_otel_providers()
# region Test console exporters opt-in
def test_console_exporters_opt_in_false(monkeypatch):
"""Test console exporters are not added when ENABLE_CONSOLE_EXPORTERS is false."""
from agent_framework.observability import ObservabilitySettings
monkeypatch.setenv("ENABLE_CONSOLE_EXPORTERS", "false")
monkeypatch.delenv("OTEL_EXPORTER_OTLP_ENDPOINT", raising=False)
settings = ObservabilitySettings(env_file_path="test.env")
assert settings.enable_console_exporters is False
def test_console_exporters_opt_in_true(monkeypatch):
"""Test console exporters are added when ENABLE_CONSOLE_EXPORTERS is true."""
from agent_framework.observability import ObservabilitySettings
monkeypatch.setenv("ENABLE_CONSOLE_EXPORTERS", "true")
settings = ObservabilitySettings(env_file_path="test.env")
assert settings.enable_console_exporters is True
def test_console_exporters_default_false(monkeypatch):
"""Test console exporters default to False when not set."""
from agent_framework.observability import ObservabilitySettings
monkeypatch.delenv("ENABLE_CONSOLE_EXPORTERS", raising=False)
settings = ObservabilitySettings(env_file_path="test.env")
assert settings.enable_console_exporters is False
# region Test _parse_headers helper
def test_parse_headers_valid():
"""Test _parse_headers with valid header string."""
from agent_framework.observability import _parse_headers
headers = _parse_headers("key1=value1,key2=value2")
assert headers == {"key1": "value1", "key2": "value2"}
def test_parse_headers_with_spaces():
"""Test _parse_headers handles spaces around keys and values."""
from agent_framework.observability import _parse_headers
headers = _parse_headers("key1 = value1 , key2 = value2 ")
assert headers == {"key1": "value1", "key2": "value2"}
def test_parse_headers_empty_string():
"""Test _parse_headers with empty string."""
from agent_framework.observability import _parse_headers
headers = _parse_headers("")
assert headers == {}
def test_parse_headers_invalid_format():
"""Test _parse_headers ignores invalid pairs."""
from agent_framework.observability import _parse_headers
headers = _parse_headers("key1=value1,invalid,key2=value2")
# Should only include valid pairs
assert headers == {"key1": "value1", "key2": "value2"}
+192 -2
View File
@@ -1,5 +1,5 @@
# Copyright (c) Microsoft. All rights reserved.
from typing import Any
from typing import Annotated, Any, Literal
from unittest.mock import Mock
import pytest
@@ -14,7 +14,7 @@ from agent_framework import (
ToolProtocol,
ai_function,
)
from agent_framework._tools import _parse_inputs
from agent_framework._tools import _parse_annotation, _parse_inputs
from agent_framework.exceptions import ToolException
from agent_framework.observability import OtelAttr
@@ -128,6 +128,95 @@ def test_ai_function_decorator_in_class():
assert test_tool(1, 2) == 3
def test_ai_function_with_literal_type_parameter():
"""Test ai_function decorator with Literal type parameter (issue #2891)."""
@ai_function
def search_flows(category: Literal["Data", "Security", "Network"], issue: str) -> str:
"""Search flows by category."""
return f"{category}: {issue}"
assert isinstance(search_flows, AIFunction)
schema = search_flows.parameters()
assert schema == {
"properties": {
"category": {"enum": ["Data", "Security", "Network"], "title": "Category", "type": "string"},
"issue": {"title": "Issue", "type": "string"},
},
"required": ["category", "issue"],
"title": "search_flows_input",
"type": "object",
}
# Verify invocation works
assert search_flows("Data", "test issue") == "Data: test issue"
def test_ai_function_with_literal_type_in_class_method():
"""Test ai_function decorator with Literal type parameter in a class method (issue #2891)."""
class MyTools:
@ai_function
def search_flows(self, category: Literal["Data", "Security", "Network"], issue: str) -> str:
"""Search flows by category."""
return f"{category}: {issue}"
tools = MyTools()
search_tool = tools.search_flows
assert isinstance(search_tool, AIFunction)
schema = search_tool.parameters()
assert schema == {
"properties": {
"category": {"enum": ["Data", "Security", "Network"], "title": "Category", "type": "string"},
"issue": {"title": "Issue", "type": "string"},
},
"required": ["category", "issue"],
"title": "search_flows_input",
"type": "object",
}
# Verify invocation works
assert search_tool("Security", "test issue") == "Security: test issue"
def test_ai_function_with_literal_int_type():
"""Test ai_function decorator with Literal int type parameter."""
@ai_function
def set_priority(priority: Literal[1, 2, 3], task: str) -> str:
"""Set priority for a task."""
return f"Priority {priority}: {task}"
assert isinstance(set_priority, AIFunction)
schema = set_priority.parameters()
assert schema == {
"properties": {
"priority": {"enum": [1, 2, 3], "title": "Priority", "type": "integer"},
"task": {"title": "Task", "type": "string"},
},
"required": ["priority", "task"],
"title": "set_priority_input",
"type": "object",
}
assert set_priority(1, "important task") == "Priority 1: important task"
def test_ai_function_with_literal_and_annotated():
"""Test ai_function decorator with Literal type combined with Annotated for description."""
@ai_function
def categorize(
category: Annotated[Literal["A", "B", "C"], "The category to assign"],
name: str,
) -> str:
"""Categorize an item."""
return f"{category}: {name}"
assert isinstance(categorize, AIFunction)
schema = categorize.parameters()
# Literal type inside Annotated should preserve enum values
assert schema["properties"]["category"]["enum"] == ["A", "B", "C"]
assert categorize("A", "test") == "A: test"
async def test_ai_function_decorator_shared_state():
"""Test that decorated methods maintain shared state across multiple calls and tool usage."""
@@ -1334,3 +1423,104 @@ async def test_streaming_two_functions_mixed_approval():
assert updates[2].role == Role.ASSISTANT
assert len(updates[2].contents) == 2
assert all(isinstance(c, FunctionApprovalRequestContent) for c in updates[2].contents)
async def test_ai_function_with_kwargs_injection():
"""Test that ai_function correctly handles kwargs injection and hides them from schema."""
@ai_function
def tool_with_kwargs(x: int, **kwargs: Any) -> str:
"""A tool that accepts kwargs."""
user_id = kwargs.get("user_id", "unknown")
return f"x={x}, user={user_id}"
# Verify schema does not include kwargs
assert tool_with_kwargs.parameters() == {
"properties": {"x": {"title": "X", "type": "integer"}},
"required": ["x"],
"title": "tool_with_kwargs_input",
"type": "object",
}
# Verify direct invocation works
assert tool_with_kwargs(1, user_id="user1") == "x=1, user=user1"
# Verify invoke works with injected args
result = await tool_with_kwargs.invoke(
arguments=tool_with_kwargs.input_model(x=5),
user_id="user2",
)
assert result == "x=5, user=user2"
# Verify invoke works without injected args (uses default)
result_default = await tool_with_kwargs.invoke(
arguments=tool_with_kwargs.input_model(x=10),
)
assert result_default == "x=10, user=unknown"
# region _parse_annotation tests
def test_parse_annotation_with_literal_type():
"""Test that _parse_annotation returns Literal types unchanged (issue #2891)."""
from typing import get_args, get_origin
# Literal with string values
literal_annotation = Literal["Data", "Security", "Network"]
result = _parse_annotation(literal_annotation)
assert result is literal_annotation
assert get_origin(result) is Literal
assert get_args(result) == ("Data", "Security", "Network")
def test_parse_annotation_with_literal_int_type():
"""Test that _parse_annotation returns Literal int types unchanged."""
from typing import get_args, get_origin
literal_annotation = Literal[1, 2, 3]
result = _parse_annotation(literal_annotation)
assert result is literal_annotation
assert get_origin(result) is Literal
assert get_args(result) == (1, 2, 3)
def test_parse_annotation_with_literal_bool_type():
"""Test that _parse_annotation returns Literal bool types unchanged."""
from typing import get_args, get_origin
literal_annotation = Literal[True, False]
result = _parse_annotation(literal_annotation)
assert result is literal_annotation
assert get_origin(result) is Literal
assert get_args(result) == (True, False)
def test_parse_annotation_with_simple_types():
"""Test that _parse_annotation returns simple types unchanged."""
assert _parse_annotation(str) is str
assert _parse_annotation(int) is int
assert _parse_annotation(float) is float
assert _parse_annotation(bool) is bool
def test_parse_annotation_with_annotated_and_literal():
"""Test that Annotated[Literal[...], description] works correctly."""
from typing import get_args, get_origin
# When Literal is inside Annotated, it should still be preserved
annotated_literal = Annotated[Literal["A", "B", "C"], "The category"]
result = _parse_annotation(annotated_literal)
# The Annotated type should be preserved
origin = get_origin(result)
assert origin is Annotated
args = get_args(result)
# First arg is the Literal type
literal_type = args[0]
assert get_origin(literal_type) is Literal
assert get_args(literal_type) == ("A", "B", "C")
# endregion
@@ -832,7 +832,7 @@ def test_create_streaming_response_content_with_mcp_approval_request() -> None:
assert fa.function_call.name == "do_stream_action"
@pytest.mark.parametrize("enable_otel", [False], indirect=True)
@pytest.mark.parametrize("enable_instrumentation", [False], indirect=True)
@pytest.mark.parametrize("enable_sensitive_data", [False], indirect=True)
async def test_end_to_end_mcp_approval_flow(span_exporter) -> None:
"""End-to-end mocked test:
@@ -993,6 +993,110 @@ def test_streaming_response_basic_structure() -> None:
assert response.raw_representation is mock_event
def test_streaming_annotation_added_with_file_path() -> None:
"""Test streaming annotation added event with file_path type extracts HostedFileContent."""
client = OpenAIResponsesClient(model_id="test-model", api_key="test-key")
chat_options = ChatOptions()
function_call_ids: dict[int, tuple[str, str]] = {}
mock_event = MagicMock()
mock_event.type = "response.output_text.annotation.added"
mock_event.annotation_index = 0
mock_event.annotation = {
"type": "file_path",
"file_id": "file-abc123",
"index": 42,
}
response = client._create_streaming_response_content(mock_event, chat_options, function_call_ids)
assert len(response.contents) == 1
content = response.contents[0]
assert isinstance(content, HostedFileContent)
assert content.file_id == "file-abc123"
assert content.additional_properties is not None
assert content.additional_properties.get("annotation_index") == 0
assert content.additional_properties.get("index") == 42
def test_streaming_annotation_added_with_file_citation() -> None:
"""Test streaming annotation added event with file_citation type extracts HostedFileContent."""
client = OpenAIResponsesClient(model_id="test-model", api_key="test-key")
chat_options = ChatOptions()
function_call_ids: dict[int, tuple[str, str]] = {}
mock_event = MagicMock()
mock_event.type = "response.output_text.annotation.added"
mock_event.annotation_index = 1
mock_event.annotation = {
"type": "file_citation",
"file_id": "file-xyz789",
"filename": "sample.txt",
"index": 15,
}
response = client._create_streaming_response_content(mock_event, chat_options, function_call_ids)
assert len(response.contents) == 1
content = response.contents[0]
assert isinstance(content, HostedFileContent)
assert content.file_id == "file-xyz789"
assert content.additional_properties is not None
assert content.additional_properties.get("filename") == "sample.txt"
assert content.additional_properties.get("index") == 15
def test_streaming_annotation_added_with_container_file_citation() -> None:
"""Test streaming annotation added event with container_file_citation type."""
client = OpenAIResponsesClient(model_id="test-model", api_key="test-key")
chat_options = ChatOptions()
function_call_ids: dict[int, tuple[str, str]] = {}
mock_event = MagicMock()
mock_event.type = "response.output_text.annotation.added"
mock_event.annotation_index = 2
mock_event.annotation = {
"type": "container_file_citation",
"file_id": "file-container123",
"container_id": "container-456",
"filename": "data.csv",
"start_index": 10,
"end_index": 50,
}
response = client._create_streaming_response_content(mock_event, chat_options, function_call_ids)
assert len(response.contents) == 1
content = response.contents[0]
assert isinstance(content, HostedFileContent)
assert content.file_id == "file-container123"
assert content.additional_properties is not None
assert content.additional_properties.get("container_id") == "container-456"
assert content.additional_properties.get("filename") == "data.csv"
assert content.additional_properties.get("start_index") == 10
assert content.additional_properties.get("end_index") == 50
def test_streaming_annotation_added_with_unknown_type() -> None:
"""Test streaming annotation added event with unknown type is ignored."""
client = OpenAIResponsesClient(model_id="test-model", api_key="test-key")
chat_options = ChatOptions()
function_call_ids: dict[int, tuple[str, str]] = {}
mock_event = MagicMock()
mock_event.type = "response.output_text.annotation.added"
mock_event.annotation_index = 0
mock_event.annotation = {
"type": "url_citation",
"url": "https://example.com",
}
response = client._create_streaming_response_content(mock_event, chat_options, function_call_ids)
# url_citation should not produce HostedFileContent
assert len(response.contents) == 0
def test_service_response_exception_includes_original_error_details() -> None:
"""Test that ServiceResponseException messages include original error details in the new format."""
client = OpenAIResponsesClient(model_id="test-model", api_key="test-key")
@@ -22,5 +22,5 @@ def test_datetime_in_tool_results() -> None:
result = _to_otel_part(content)
parsed = json.loads(result["response"])
# Datetime should be converted to string
assert isinstance(parsed["timestamp"], str)
# Datetime should be converted to string in the result field
assert isinstance(parsed["result"]["timestamp"], str)
@@ -3,6 +3,7 @@
from typing import Any, cast
import pytest
from typing_extensions import Never
from agent_framework import (
AgentExecutorRequest,
@@ -52,6 +53,55 @@ def test_concurrent_builder_rejects_duplicate_executors() -> None:
ConcurrentBuilder().participants([a, b])
def test_concurrent_builder_rejects_duplicate_executors_from_factories() -> None:
"""Test that duplicate executor IDs from factories are detected at build time."""
def create_dup1() -> Executor:
return _FakeAgentExec("dup", "A")
def create_dup2() -> Executor:
return _FakeAgentExec("dup", "B") # same executor id
builder = ConcurrentBuilder().register_participants([create_dup1, create_dup2])
with pytest.raises(ValueError, match="Duplicate executor ID 'dup' detected in workflow."):
builder.build()
def test_concurrent_builder_rejects_mixed_participants_and_factories() -> None:
"""Test that mixing .participants() and .register_participants() raises an error."""
# Case 1: participants first, then register_participants
with pytest.raises(ValueError, match="Cannot mix .participants"):
(
ConcurrentBuilder()
.participants([_FakeAgentExec("a", "A")])
.register_participants([lambda: _FakeAgentExec("b", "B")])
)
# Case 2: register_participants first, then participants
with pytest.raises(ValueError, match="Cannot mix .participants"):
(
ConcurrentBuilder()
.register_participants([lambda: _FakeAgentExec("a", "A")])
.participants([_FakeAgentExec("b", "B")])
)
def test_concurrent_builder_rejects_multiple_calls_to_participants() -> None:
"""Test that multiple calls to .participants() raises an error."""
with pytest.raises(ValueError, match=r"participants\(\) has already been called"):
(ConcurrentBuilder().participants([_FakeAgentExec("a", "A")]).participants([_FakeAgentExec("b", "B")]))
def test_concurrent_builder_rejects_multiple_calls_to_register_participants() -> None:
"""Test that multiple calls to .register_participants() raises an error."""
with pytest.raises(ValueError, match=r"register_participants\(\) has already been called"):
(
ConcurrentBuilder()
.register_participants([lambda: _FakeAgentExec("a", "A")])
.register_participants([lambda: _FakeAgentExec("b", "B")])
)
async def test_concurrent_default_aggregator_emits_single_user_and_assistants() -> None:
# Three synthetic agent executors
e1 = _FakeAgentExec("agentA", "Alpha")
@@ -159,6 +209,138 @@ def test_concurrent_custom_aggregator_uses_callback_name_for_id() -> None:
assert aggregator.id == "summarize"
async def test_concurrent_with_aggregator_executor_instance() -> None:
"""Test with_aggregator using an Executor instance (not factory)."""
class CustomAggregator(Executor):
@handler
async def aggregate(self, results: list[AgentExecutorResponse], ctx: WorkflowContext[Never, str]) -> None:
texts: list[str] = []
for r in results:
msgs: list[ChatMessage] = r.agent_run_response.messages
texts.append(msgs[-1].text if msgs else "")
await ctx.yield_output(" & ".join(sorted(texts)))
e1 = _FakeAgentExec("agentA", "One")
e2 = _FakeAgentExec("agentB", "Two")
aggregator_instance = CustomAggregator(id="instance_aggregator")
wf = ConcurrentBuilder().participants([e1, e2]).with_aggregator(aggregator_instance).build()
completed = False
output: str | None = None
async for ev in wf.run_stream("prompt: instance test"):
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
completed = True
elif isinstance(ev, WorkflowOutputEvent):
output = cast(str, ev.data)
if completed and output is not None:
break
assert completed
assert output is not None
assert isinstance(output, str)
assert output == "One & Two"
async def test_concurrent_with_aggregator_executor_factory() -> None:
"""Test with_aggregator using an Executor factory."""
class CustomAggregator(Executor):
@handler
async def aggregate(self, results: list[AgentExecutorResponse], ctx: WorkflowContext[Never, str]) -> None:
texts: list[str] = []
for r in results:
msgs: list[ChatMessage] = r.agent_run_response.messages
texts.append(msgs[-1].text if msgs else "")
await ctx.yield_output(" | ".join(sorted(texts)))
e1 = _FakeAgentExec("agentA", "One")
e2 = _FakeAgentExec("agentB", "Two")
wf = (
ConcurrentBuilder()
.participants([e1, e2])
.register_aggregator(lambda: CustomAggregator(id="custom_aggregator"))
.build()
)
completed = False
output: str | None = None
async for ev in wf.run_stream("prompt: factory test"):
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
completed = True
elif isinstance(ev, WorkflowOutputEvent):
output = cast(str, ev.data)
if completed and output is not None:
break
assert completed
assert output is not None
assert isinstance(output, str)
assert output == "One | Two"
async def test_concurrent_with_aggregator_executor_factory_with_default_id() -> None:
"""Test with_aggregator using an Executor class directly as factory (with default __init__ parameters)."""
class CustomAggregator(Executor):
def __init__(self, id: str = "default_aggregator") -> None:
super().__init__(id)
@handler
async def aggregate(self, results: list[AgentExecutorResponse], ctx: WorkflowContext[Never, str]) -> None:
texts: list[str] = []
for r in results:
msgs: list[ChatMessage] = r.agent_run_response.messages
texts.append(msgs[-1].text if msgs else "")
await ctx.yield_output(" | ".join(sorted(texts)))
e1 = _FakeAgentExec("agentA", "One")
e2 = _FakeAgentExec("agentB", "Two")
wf = ConcurrentBuilder().participants([e1, e2]).register_aggregator(CustomAggregator).build()
completed = False
output: str | None = None
async for ev in wf.run_stream("prompt: factory test"):
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
completed = True
elif isinstance(ev, WorkflowOutputEvent):
output = cast(str, ev.data)
if completed and output is not None:
break
assert completed
assert output is not None
assert isinstance(output, str)
assert output == "One | Two"
def test_concurrent_builder_rejects_multiple_calls_to_with_aggregator() -> None:
"""Test that multiple calls to .with_aggregator() raises an error."""
def summarize(results: list[AgentExecutorResponse]) -> str: # type: ignore[override]
return str(len(results))
with pytest.raises(ValueError, match=r"with_aggregator\(\) has already been called"):
(ConcurrentBuilder().with_aggregator(summarize).with_aggregator(summarize))
def test_concurrent_builder_rejects_multiple_calls_to_register_aggregator() -> None:
"""Test that multiple calls to .register_aggregator() raises an error."""
class CustomAggregator(Executor):
pass
with pytest.raises(ValueError, match=r"register_aggregator\(\) has already been called"):
(
ConcurrentBuilder()
.register_aggregator(lambda: CustomAggregator(id="agg1"))
.register_aggregator(lambda: CustomAggregator(id="agg2"))
)
async def test_concurrent_checkpoint_resume_round_trip() -> None:
storage = InMemoryCheckpointStorage()
@@ -278,3 +460,92 @@ async def test_concurrent_checkpoint_runtime_overrides_buildtime() -> None:
assert len(runtime_checkpoints) > 0, "Runtime storage should have checkpoints"
assert len(buildtime_checkpoints) == 0, "Build-time storage should have no checkpoints when overridden"
def test_concurrent_builder_rejects_empty_participant_factories() -> None:
with pytest.raises(ValueError):
ConcurrentBuilder().register_participants([])
async def test_concurrent_builder_reusable_after_build_with_participants() -> None:
"""Test that the builder can be reused to build multiple identical workflows with participants()."""
e1 = _FakeAgentExec("agentA", "One")
e2 = _FakeAgentExec("agentB", "Two")
builder = ConcurrentBuilder().participants([e1, e2])
builder.build()
assert builder._participants[0] is e1 # type: ignore
assert builder._participants[1] is e2 # type: ignore
assert builder._participant_factories == [] # type: ignore
async def test_concurrent_builder_reusable_after_build_with_factories() -> None:
"""Test that the builder can be reused to build multiple workflows with register_participants()."""
call_count = 0
def create_agent_executor_a() -> Executor:
nonlocal call_count
call_count += 1
return _FakeAgentExec("agentA", "One")
def create_agent_executor_b() -> Executor:
nonlocal call_count
call_count += 1
return _FakeAgentExec("agentB", "Two")
builder = ConcurrentBuilder().register_participants([create_agent_executor_a, create_agent_executor_b])
# Build the first workflow
wf1 = builder.build()
assert builder._participants == [] # type: ignore
assert len(builder._participant_factories) == 2 # type: ignore
assert call_count == 2
# Build the second workflow
wf2 = builder.build()
assert call_count == 4
# Verify that the two workflows have different executor instances
assert wf1.executors["agentA"] is not wf2.executors["agentA"]
assert wf1.executors["agentB"] is not wf2.executors["agentB"]
async def test_concurrent_with_register_participants() -> None:
"""Test workflow creation using register_participants with factories."""
def create_agent1() -> Executor:
return _FakeAgentExec("agentA", "Alpha")
def create_agent2() -> Executor:
return _FakeAgentExec("agentB", "Beta")
def create_agent3() -> Executor:
return _FakeAgentExec("agentC", "Gamma")
wf = ConcurrentBuilder().register_participants([create_agent1, create_agent2, create_agent3]).build()
completed = False
output: list[ChatMessage] | None = None
async for ev in wf.run_stream("test prompt"):
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
completed = True
elif isinstance(ev, WorkflowOutputEvent):
output = cast(list[ChatMessage], ev.data)
if completed and output is not None:
break
assert completed
assert output is not None
messages: list[ChatMessage] = output
# Expect one user message + one assistant message per participant
assert len(messages) == 1 + 3
assert messages[0].role == Role.USER
assert "test prompt" in messages[0].text
assistant_texts = {m.text for m in messages[1:]}
assert assistant_texts == {"Alpha", "Beta", "Gamma"}
assert all(m.role == Role.ASSISTANT for m in messages[1:])
@@ -25,7 +25,12 @@ from agent_framework import (
from agent_framework._mcp import MCPTool
from agent_framework._workflows import AgentRunEvent
from agent_framework._workflows import _handoff as handoff_module # type: ignore
from agent_framework._workflows._handoff import _clone_chat_agent # type: ignore[reportPrivateUsage]
from agent_framework._workflows._checkpoint_encoding import decode_checkpoint_value, encode_checkpoint_value
from agent_framework._workflows._handoff import (
_clone_chat_agent, # type: ignore[reportPrivateUsage]
_ConversationWithUserInput,
_UserInputGateway,
)
from agent_framework._workflows._workflow_builder import WorkflowBuilder
@@ -218,7 +223,7 @@ async def test_handoff_preserves_complex_additional_properties(complex_metadata:
workflow = (
HandoffBuilder(participants=[triage, specialist])
.set_coordinator("triage")
.set_coordinator(triage)
.with_termination_condition(lambda conv: sum(1 for msg in conv if msg.role == Role.USER) >= 2)
.build()
)
@@ -281,7 +286,7 @@ async def test_tool_call_handoff_detection_with_text_hint():
triage = _RecordingAgent(name="triage", handoff_to="specialist", text_handoff=True)
specialist = _RecordingAgent(name="specialist")
workflow = HandoffBuilder(participants=[triage, specialist]).set_coordinator("triage").build()
workflow = HandoffBuilder(participants=[triage, specialist]).set_coordinator(triage).build()
await _drain(workflow.run_stream("Package arrived broken"))
@@ -296,7 +301,7 @@ async def test_autonomous_interaction_mode_yields_output_without_user_request():
workflow = (
HandoffBuilder(participants=[triage, specialist])
.set_coordinator("triage")
.set_coordinator(triage)
.with_interaction_mode("autonomous", autonomous_turn_limit=1)
.build()
)
@@ -428,13 +433,13 @@ def test_build_fails_without_coordinator():
triage = _RecordingAgent(name="triage")
specialist = _RecordingAgent(name="specialist")
with pytest.raises(ValueError, match="coordinator must be defined before build"):
with pytest.raises(ValueError, match=r"Must call set_coordinator\(...\) before building the workflow."):
HandoffBuilder(participants=[triage, specialist]).build()
def test_build_fails_without_participants():
"""Verify that build() raises ValueError when no participants are provided."""
with pytest.raises(ValueError, match="No participants provided"):
with pytest.raises(ValueError, match="No participants or participant_factories have been configured."):
HandoffBuilder().build()
@@ -605,7 +610,7 @@ async def test_return_to_previous_enabled():
workflow = (
HandoffBuilder(participants=[triage, specialist_a, specialist_b])
.set_coordinator("triage")
.set_coordinator(triage)
.enable_return_to_previous(True)
.with_termination_condition(lambda conv: sum(1 for m in conv if m.role == Role.USER) >= 3)
.build()
@@ -638,7 +643,7 @@ def test_handoff_builder_sets_start_executor_once(monkeypatch: pytest.MonkeyPatc
workflow = (
HandoffBuilder(participants=[coordinator, specialist])
.set_coordinator("coordinator")
.set_coordinator(coordinator)
.with_termination_condition(lambda conv: len(conv) > 0)
.build()
)
@@ -698,7 +703,7 @@ async def test_handoff_builder_with_request_info():
# Build workflow with request info enabled
workflow = (
HandoffBuilder(participants=[coordinator, specialist])
.set_coordinator("coordinator")
.set_coordinator(coordinator)
.with_termination_condition(lambda conv: len([m for m in conv if m.role == Role.USER]) >= 1)
.with_request_info()
.build()
@@ -775,3 +780,893 @@ async def test_return_to_previous_state_serialization():
# Verify current_agent_id was restored
assert coordinator2._current_agent_id == "specialist_a", "Current agent should be restored from checkpoint" # type: ignore[reportPrivateUsage]
# region Participant Factory Tests
def test_handoff_builder_rejects_empty_participant_factories():
"""Test that HandoffBuilder rejects empty participant_factories dictionary."""
# Empty factories are rejected immediately when calling participant_factories()
with pytest.raises(ValueError, match=r"participant_factories cannot be empty"):
HandoffBuilder().participant_factories({})
with pytest.raises(ValueError, match=r"No participants or participant_factories have been configured"):
HandoffBuilder(participant_factories={}).build()
def test_handoff_builder_rejects_mixing_participants_and_factories():
"""Test that mixing participants and participant_factories in __init__ raises an error."""
triage = _RecordingAgent(name="triage")
with pytest.raises(ValueError, match="Cannot mix .participants"):
HandoffBuilder(participants=[triage], participant_factories={"triage": lambda: triage})
def test_handoff_builder_rejects_mixing_participants_and_participant_factories_methods():
"""Test that mixing .participants() and .participant_factories() raises an error."""
triage = _RecordingAgent(name="triage")
# Case 1: participants first, then participant_factories
with pytest.raises(ValueError, match="Cannot mix .participants"):
HandoffBuilder(participants=[triage]).participant_factories({
"specialist": lambda: _RecordingAgent(name="specialist")
})
# Case 2: participant_factories first, then participants
with pytest.raises(ValueError, match="Cannot mix .participants"):
HandoffBuilder(participant_factories={"triage": lambda: triage}).participants([
_RecordingAgent(name="specialist")
])
# Case 3: participants(), then participant_factories()
with pytest.raises(ValueError, match="Cannot mix .participants"):
HandoffBuilder().participants([triage]).participant_factories({
"specialist": lambda: _RecordingAgent(name="specialist")
})
# Case 4: participant_factories(), then participants()
with pytest.raises(ValueError, match="Cannot mix .participants"):
HandoffBuilder().participant_factories({"triage": lambda: triage}).participants([
_RecordingAgent(name="specialist")
])
# Case 5: mix during initialization
with pytest.raises(ValueError, match="Cannot mix .participants"):
HandoffBuilder(
participants=[triage], participant_factories={"specialist": lambda: _RecordingAgent(name="specialist")}
)
def test_handoff_builder_rejects_multiple_calls_to_participant_factories():
"""Test that multiple calls to .participant_factories() raises an error."""
with pytest.raises(ValueError, match=r"participant_factories\(\) has already been called"):
(
HandoffBuilder()
.participant_factories({"agent1": lambda: _RecordingAgent(name="agent1")})
.participant_factories({"agent2": lambda: _RecordingAgent(name="agent2")})
)
def test_handoff_builder_rejects_multiple_calls_to_participants():
"""Test that multiple calls to .participants() raises an error."""
with pytest.raises(ValueError, match="participants have already been assigned"):
(HandoffBuilder().participants([_RecordingAgent(name="agent1")]).participants([_RecordingAgent(name="agent2")]))
def test_handoff_builder_rejects_duplicate_factories():
"""Test that multiple calls to participant_factories are rejected."""
factories = {
"triage": lambda: _RecordingAgent(name="triage"),
"specialist": lambda: _RecordingAgent(name="specialist"),
}
# Multiple calls to participant_factories should fail
builder = HandoffBuilder(participant_factories=factories)
with pytest.raises(ValueError, match=r"participant_factories\(\) has already been called"):
builder.participant_factories({"triage": lambda: _RecordingAgent(name="triage2")})
def test_handoff_builder_rejects_instance_coordinator_with_factories():
"""Test that using an agent instance for set_coordinator when using factories raises an error."""
def create_triage() -> _RecordingAgent:
return _RecordingAgent(name="triage")
def create_specialist() -> _RecordingAgent:
return _RecordingAgent(name="specialist")
# Create an agent instance
coordinator_instance = _RecordingAgent(name="coordinator")
with pytest.raises(ValueError, match=r"Call participants\(\.\.\.\) before coordinator\(\.\.\.\)"):
(
HandoffBuilder(
participant_factories={"triage": create_triage, "specialist": create_specialist}
).set_coordinator(coordinator_instance) # Instance, not factory name
)
def test_handoff_builder_rejects_factory_name_coordinator_with_instances():
"""Test that using a factory name for set_coordinator when using instances raises an error."""
triage = _RecordingAgent(name="triage")
specialist = _RecordingAgent(name="specialist")
with pytest.raises(
ValueError, match="coordinator factory name 'triage' is not part of the participant_factories list"
):
(
HandoffBuilder(participants=[triage, specialist]).set_coordinator(
"triage"
) # String factory name, not instance
)
def test_handoff_builder_rejects_mixed_types_in_add_handoff_source():
"""Test that add_handoff rejects factory name source with instance-based participants."""
triage = _RecordingAgent(name="triage")
specialist = _RecordingAgent(name="specialist")
with pytest.raises(TypeError, match="Cannot mix factory names \\(str\\) and AgentProtocol/Executor instances"):
(
HandoffBuilder(participants=[triage, specialist])
.set_coordinator(triage)
.add_handoff("triage", specialist) # String source with instance participants
)
def test_handoff_builder_accepts_all_factory_names_in_add_handoff():
"""Test that add_handoff accepts all factory names when using participant_factories."""
def create_triage() -> _RecordingAgent:
return _RecordingAgent(name="triage")
def create_specialist_a() -> _RecordingAgent:
return _RecordingAgent(name="specialist_a")
def create_specialist_b() -> _RecordingAgent:
return _RecordingAgent(name="specialist_b")
# This should work - all strings with participant_factories
builder = (
HandoffBuilder(
participant_factories={
"triage": create_triage,
"specialist_a": create_specialist_a,
"specialist_b": create_specialist_b,
}
)
.set_coordinator("triage")
.add_handoff("triage", ["specialist_a", "specialist_b"])
)
workflow = builder.build()
assert "triage" in workflow.executors
assert "specialist_a" in workflow.executors
assert "specialist_b" in workflow.executors
def test_handoff_builder_accepts_all_instances_in_add_handoff():
"""Test that add_handoff accepts all instances when using participants."""
triage = _RecordingAgent(name="triage", handoff_to="specialist_a")
specialist_a = _RecordingAgent(name="specialist_a")
specialist_b = _RecordingAgent(name="specialist_b")
# This should work - all instances with participants
builder = (
HandoffBuilder(participants=[triage, specialist_a, specialist_b])
.set_coordinator(triage)
.add_handoff(triage, [specialist_a, specialist_b])
)
workflow = builder.build()
assert "triage" in workflow.executors
assert "specialist_a" in workflow.executors
assert "specialist_b" in workflow.executors
async def test_handoff_with_participant_factories():
"""Test workflow creation using participant_factories."""
call_count = 0
def create_triage() -> _RecordingAgent:
nonlocal call_count
call_count += 1
return _RecordingAgent(name="triage", handoff_to="specialist")
def create_specialist() -> _RecordingAgent:
nonlocal call_count
call_count += 1
return _RecordingAgent(name="specialist")
workflow = (
HandoffBuilder(participant_factories={"triage": create_triage, "specialist": create_specialist})
.set_coordinator("triage")
.with_termination_condition(lambda conv: sum(1 for m in conv if m.role == Role.USER) >= 2)
.build()
)
# Factories should be called during build
assert call_count == 2
events = await _drain(workflow.run_stream("Need help"))
requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
assert requests
# Follow-up message
events = await _drain(workflow.send_responses_streaming({requests[-1].request_id: "More details"}))
outputs = [ev for ev in events if isinstance(ev, WorkflowOutputEvent)]
assert outputs
async def test_handoff_participant_factories_reusable_builder():
"""Test that the builder can be reused to build multiple workflows with factories."""
call_count = 0
def create_triage() -> _RecordingAgent:
nonlocal call_count
call_count += 1
return _RecordingAgent(name="triage", handoff_to="specialist")
def create_specialist() -> _RecordingAgent:
nonlocal call_count
call_count += 1
return _RecordingAgent(name="specialist")
builder = HandoffBuilder(
participant_factories={"triage": create_triage, "specialist": create_specialist}
).set_coordinator("triage")
# Build first workflow
wf1 = builder.build()
assert call_count == 2
# Build second workflow
wf2 = builder.build()
assert call_count == 4
# Verify that the two workflows have different agent instances
assert wf1.executors["triage"] is not wf2.executors["triage"]
assert wf1.executors["specialist"] is not wf2.executors["specialist"]
async def test_handoff_with_participant_factories_and_add_handoff():
"""Test that .add_handoff() works correctly with participant_factories."""
def create_triage() -> _RecordingAgent:
return _RecordingAgent(name="triage", handoff_to="specialist_a")
def create_specialist_a() -> _RecordingAgent:
return _RecordingAgent(name="specialist_a", handoff_to="specialist_b")
def create_specialist_b() -> _RecordingAgent:
return _RecordingAgent(name="specialist_b")
workflow = (
HandoffBuilder(
participant_factories={
"triage": create_triage,
"specialist_a": create_specialist_a,
"specialist_b": create_specialist_b,
}
)
.set_coordinator("triage")
.add_handoff("triage", ["specialist_a", "specialist_b"])
.add_handoff("specialist_a", "specialist_b")
.with_termination_condition(lambda conv: sum(1 for m in conv if m.role == Role.USER) >= 3)
.build()
)
# Start conversation - triage hands off to specialist_a
events = await _drain(workflow.run_stream("Initial request"))
requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
assert requests
# Verify specialist_a executor exists and was called
assert "specialist_a" in workflow.executors
# Second user message - specialist_a hands off to specialist_b
events = await _drain(workflow.send_responses_streaming({requests[-1].request_id: "Need escalation"}))
requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
assert requests
# Verify specialist_b executor exists
assert "specialist_b" in workflow.executors
async def test_handoff_participant_factories_with_checkpointing():
"""Test checkpointing with participant_factories."""
from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage
storage = InMemoryCheckpointStorage()
def create_triage() -> _RecordingAgent:
return _RecordingAgent(name="triage", handoff_to="specialist")
def create_specialist() -> _RecordingAgent:
return _RecordingAgent(name="specialist")
workflow = (
HandoffBuilder(participant_factories={"triage": create_triage, "specialist": create_specialist})
.set_coordinator("triage")
.with_checkpointing(storage)
.with_termination_condition(lambda conv: sum(1 for m in conv if m.role == Role.USER) >= 2)
.build()
)
# Run workflow and capture output
events = await _drain(workflow.run_stream("checkpoint test"))
requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
assert requests
events = await _drain(workflow.send_responses_streaming({requests[-1].request_id: "follow up"}))
outputs = [ev for ev in events if isinstance(ev, WorkflowOutputEvent)]
assert outputs, "Should have workflow output after termination condition is met"
# List checkpoints - just verify they were created
checkpoints = await storage.list_checkpoints()
assert checkpoints, "Checkpoints should be created during workflow execution"
def test_handoff_set_coordinator_with_factory_name():
"""Test that set_coordinator accepts factory name as string."""
def create_triage() -> _RecordingAgent:
return _RecordingAgent(name="triage")
def create_specialist() -> _RecordingAgent:
return _RecordingAgent(name="specialist")
builder = HandoffBuilder(
participant_factories={"triage": create_triage, "specialist": create_specialist}
).set_coordinator("triage")
workflow = builder.build()
assert "triage" in workflow.executors
def test_handoff_add_handoff_with_factory_names():
"""Test that add_handoff accepts factory names as strings."""
def create_triage() -> _RecordingAgent:
return _RecordingAgent(name="triage", handoff_to="specialist_a")
def create_specialist_a() -> _RecordingAgent:
return _RecordingAgent(name="specialist_a")
def create_specialist_b() -> _RecordingAgent:
return _RecordingAgent(name="specialist_b")
builder = (
HandoffBuilder(
participant_factories={
"triage": create_triage,
"specialist_a": create_specialist_a,
"specialist_b": create_specialist_b,
}
)
.set_coordinator("triage")
.add_handoff("triage", ["specialist_a", "specialist_b"])
)
workflow = builder.build()
assert "triage" in workflow.executors
assert "specialist_a" in workflow.executors
assert "specialist_b" in workflow.executors
async def test_handoff_participant_factories_autonomous_mode():
"""Test autonomous mode with participant_factories."""
def create_triage() -> _RecordingAgent:
return _RecordingAgent(name="triage", handoff_to="specialist")
def create_specialist() -> _RecordingAgent:
return _RecordingAgent(name="specialist")
workflow = (
HandoffBuilder(participant_factories={"triage": create_triage, "specialist": create_specialist})
.set_coordinator("triage")
.with_interaction_mode("autonomous", autonomous_turn_limit=2)
.build()
)
events = await _drain(workflow.run_stream("Issue"))
outputs = [ev for ev in events if isinstance(ev, WorkflowOutputEvent)]
assert outputs, "Autonomous mode should yield output"
requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
assert not requests, "Autonomous mode should not request user input"
async def test_handoff_participant_factories_with_request_info():
"""Test that .with_request_info() works with participant_factories."""
def create_triage() -> _RecordingAgent:
return _RecordingAgent(name="triage")
def create_specialist() -> _RecordingAgent:
return _RecordingAgent(name="specialist")
builder = (
HandoffBuilder(participant_factories={"triage": create_triage, "specialist": create_specialist})
.set_coordinator("triage")
.with_request_info(agents=["triage"])
)
workflow = builder.build()
assert "triage" in workflow.executors
def test_handoff_participant_factories_invalid_coordinator_name():
"""Test that set_coordinator raises error for non-existent factory name."""
def create_triage() -> _RecordingAgent:
return _RecordingAgent(name="triage")
with pytest.raises(
ValueError, match="coordinator factory name 'nonexistent' is not part of the participant_factories list"
):
(HandoffBuilder(participant_factories={"triage": create_triage}).set_coordinator("nonexistent").build())
def test_handoff_participant_factories_invalid_handoff_target():
"""Test that add_handoff raises error for non-existent target factory name."""
def create_triage() -> _RecordingAgent:
return _RecordingAgent(name="triage")
def create_specialist() -> _RecordingAgent:
return _RecordingAgent(name="specialist")
with pytest.raises(ValueError, match="Target factory name 'nonexistent' is not in the participant_factories list"):
(
HandoffBuilder(participant_factories={"triage": create_triage, "specialist": create_specialist})
.set_coordinator("triage")
.add_handoff("triage", "nonexistent")
.build()
)
async def test_handoff_participant_factories_enable_return_to_previous():
"""Test return_to_previous works with participant_factories."""
def create_triage() -> _RecordingAgent:
return _RecordingAgent(name="triage", handoff_to="specialist_a")
def create_specialist_a() -> _RecordingAgent:
return _RecordingAgent(name="specialist_a", handoff_to="specialist_b")
def create_specialist_b() -> _RecordingAgent:
return _RecordingAgent(name="specialist_b")
workflow = (
HandoffBuilder(
participant_factories={
"triage": create_triage,
"specialist_a": create_specialist_a,
"specialist_b": create_specialist_b,
}
)
.set_coordinator("triage")
.add_handoff("triage", ["specialist_a", "specialist_b"])
.add_handoff("specialist_a", "specialist_b")
.enable_return_to_previous(True)
.with_termination_condition(lambda conv: sum(1 for m in conv if m.role == Role.USER) >= 3)
.build()
)
# Start conversation - triage hands off to specialist_a
events = await _drain(workflow.run_stream("Initial request"))
requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
assert requests
# Second user message - specialist_a hands off to specialist_b
events = await _drain(workflow.send_responses_streaming({requests[-1].request_id: "Need escalation"}))
requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
assert requests
# Third user message - should route back to specialist_b (return to previous)
events = await _drain(workflow.send_responses_streaming({requests[-1].request_id: "Follow up"}))
outputs = [ev for ev in events if isinstance(ev, WorkflowOutputEvent)]
assert outputs or [ev for ev in events if isinstance(ev, RequestInfoEvent)]
# endregion Participant Factory Tests
async def test_handoff_user_input_request_checkpoint_excludes_conversation():
"""Test that HandoffUserInputRequest serialization excludes conversation to prevent duplication.
Issue #2667: When checkpointing a workflow with a pending HandoffUserInputRequest,
the conversation field gets serialized twice: once in the RequestInfoEvent's data
and once in the coordinator's conversation state. On restore, this causes duplicate
messages.
The fix is to exclude the conversation field during checkpoint serialization since
the conversation is already preserved in the coordinator's state.
"""
# Create a conversation history
conversation = [
ChatMessage(role=Role.USER, text="Hello"),
ChatMessage(role=Role.ASSISTANT, text="Hi there!"),
ChatMessage(role=Role.USER, text="Help me"),
]
# Create a HandoffUserInputRequest with the conversation
request = HandoffUserInputRequest(
conversation=conversation,
awaiting_agent_id="specialist_agent",
prompt="Please provide your input",
source_executor_id="gateway",
)
# Encode the request (simulating checkpoint save)
encoded = encode_checkpoint_value(request)
# Verify conversation is NOT in the encoded output
# The fix should exclude conversation from serialization
assert isinstance(encoded, dict)
# If using MODEL_MARKER strategy (to_dict/from_dict)
if "__af_model__" in encoded or "__af_dataclass__" in encoded:
value = encoded.get("value", {})
assert "conversation" not in value, "conversation should be excluded from checkpoint serialization"
# Decode the request (simulating checkpoint restore)
decoded = decode_checkpoint_value(encoded)
# Verify the decoded request is a HandoffUserInputRequest
assert isinstance(decoded, HandoffUserInputRequest)
# Verify other fields are preserved
assert decoded.awaiting_agent_id == "specialist_agent"
assert decoded.prompt == "Please provide your input"
assert decoded.source_executor_id == "gateway"
# Conversation should be an empty list after deserialization
# (will be reconstructed from coordinator state on restore)
assert decoded.conversation == []
async def test_handoff_user_input_request_roundtrip_preserves_metadata():
"""Test that non-conversation fields survive checkpoint roundtrip."""
request = HandoffUserInputRequest(
conversation=[ChatMessage(role=Role.USER, text="test")],
awaiting_agent_id="test_agent",
prompt="Enter your response",
source_executor_id="test_gateway",
)
# Roundtrip through checkpoint encoding
encoded = encode_checkpoint_value(request)
decoded = decode_checkpoint_value(encoded)
assert isinstance(decoded, HandoffUserInputRequest)
assert decoded.awaiting_agent_id == request.awaiting_agent_id
assert decoded.prompt == request.prompt
assert decoded.source_executor_id == request.source_executor_id
async def test_request_info_event_with_handoff_user_input_request():
"""Test RequestInfoEvent serialization with HandoffUserInputRequest data."""
conversation = [
ChatMessage(role=Role.USER, text="Hello"),
ChatMessage(role=Role.ASSISTANT, text="How can I help?"),
]
request = HandoffUserInputRequest(
conversation=conversation,
awaiting_agent_id="specialist",
prompt="Provide input",
source_executor_id="gateway",
)
# Create a RequestInfoEvent wrapping the request
event = RequestInfoEvent(
request_id="test-request-123",
source_executor_id="gateway",
request_data=request,
response_type=object,
)
# Serialize the event
event_dict = event.to_dict()
# Verify the data field doesn't contain conversation
data_encoded = event_dict["data"]
if isinstance(data_encoded, dict) and ("__af_model__" in data_encoded or "__af_dataclass__" in data_encoded):
value = data_encoded.get("value", {})
assert "conversation" not in value
# Deserialize and verify
restored_event = RequestInfoEvent.from_dict(event_dict)
assert isinstance(restored_event.data, HandoffUserInputRequest)
assert restored_event.data.awaiting_agent_id == "specialist"
assert restored_event.data.conversation == []
async def test_handoff_user_input_request_to_dict_excludes_conversation():
"""Test that to_dict() method excludes conversation field."""
conversation = [
ChatMessage(role=Role.USER, text="Hello"),
ChatMessage(role=Role.ASSISTANT, text="Hi!"),
]
request = HandoffUserInputRequest(
conversation=conversation,
awaiting_agent_id="agent1",
prompt="Enter input",
source_executor_id="gateway",
)
# Call to_dict directly
data = request.to_dict()
# Verify conversation is excluded
assert "conversation" not in data
assert data["awaiting_agent_id"] == "agent1"
assert data["prompt"] == "Enter input"
assert data["source_executor_id"] == "gateway"
async def test_handoff_user_input_request_from_dict_creates_empty_conversation():
"""Test that from_dict() creates an instance with empty conversation."""
data = {
"awaiting_agent_id": "agent1",
"prompt": "Enter input",
"source_executor_id": "gateway",
}
request = HandoffUserInputRequest.from_dict(data)
assert request.conversation == []
assert request.awaiting_agent_id == "agent1"
assert request.prompt == "Enter input"
assert request.source_executor_id == "gateway"
async def test_user_input_gateway_resume_handles_empty_conversation():
"""Test that _UserInputGateway.resume_from_user handles post-restore scenario.
After checkpoint restore, the HandoffUserInputRequest will have an empty
conversation. The gateway should handle this by sending only the new user
messages to the coordinator.
"""
from unittest.mock import AsyncMock
# Create a gateway
gateway = _UserInputGateway(
starting_agent_id="coordinator",
prompt="Enter input",
id="test-gateway",
)
# Simulate post-restore: request with empty conversation
restored_request = HandoffUserInputRequest(
conversation=[], # Empty after restore
awaiting_agent_id="specialist",
prompt="Enter input",
source_executor_id="test-gateway",
)
# Create mock context
mock_ctx = MagicMock()
mock_ctx.send_message = AsyncMock()
# Call resume_from_user with a user response
await gateway.resume_from_user(restored_request, "New user message", mock_ctx)
# Verify send_message was called
mock_ctx.send_message.assert_called_once()
# Get the message that was sent
call_args = mock_ctx.send_message.call_args
sent_message = call_args[0][0]
# Verify it's a _ConversationWithUserInput
assert isinstance(sent_message, _ConversationWithUserInput)
# Verify it contains only the new user message (not any history)
assert len(sent_message.full_conversation) == 1
assert sent_message.full_conversation[0].role == Role.USER
assert sent_message.full_conversation[0].text == "New user message"
async def test_user_input_gateway_resume_with_full_conversation():
"""Test that _UserInputGateway.resume_from_user handles normal flow correctly.
In normal flow (no checkpoint restore), the HandoffUserInputRequest has
the full conversation. The gateway should send the full conversation
plus the new user messages.
"""
from unittest.mock import AsyncMock
# Create a gateway
gateway = _UserInputGateway(
starting_agent_id="coordinator",
prompt="Enter input",
id="test-gateway",
)
# Normal flow: request with full conversation
normal_request = HandoffUserInputRequest(
conversation=[
ChatMessage(role=Role.USER, text="Hello"),
ChatMessage(role=Role.ASSISTANT, text="Hi!"),
],
awaiting_agent_id="specialist",
prompt="Enter input",
source_executor_id="test-gateway",
)
# Create mock context
mock_ctx = MagicMock()
mock_ctx.send_message = AsyncMock()
# Call resume_from_user with a user response
await gateway.resume_from_user(normal_request, "Follow up message", mock_ctx)
# Verify send_message was called
mock_ctx.send_message.assert_called_once()
# Get the message that was sent
call_args = mock_ctx.send_message.call_args
sent_message = call_args[0][0]
# Verify it's a _ConversationWithUserInput
assert isinstance(sent_message, _ConversationWithUserInput)
# Verify it contains the full conversation plus new user message
assert len(sent_message.full_conversation) == 3
assert sent_message.full_conversation[0].text == "Hello"
assert sent_message.full_conversation[1].text == "Hi!"
assert sent_message.full_conversation[2].text == "Follow up message"
async def test_coordinator_handle_user_input_post_restore():
"""Test that _HandoffCoordinator.handle_user_input handles post-restore correctly.
After checkpoint restore, the coordinator has its conversation restored,
and the gateway sends only the new user messages. The coordinator should
append these to its existing conversation rather than replacing.
"""
from unittest.mock import AsyncMock
from agent_framework._workflows._handoff import _HandoffCoordinator
# Create a coordinator with pre-existing conversation (simulating restored state)
coordinator = _HandoffCoordinator(
starting_agent_id="triage",
specialist_ids={"specialist_a": "specialist_a"},
input_gateway_id="gateway",
termination_condition=lambda conv: False,
id="test-coordinator",
)
# Simulate restored conversation
coordinator._conversation = [
ChatMessage(role=Role.USER, text="Hello"),
ChatMessage(role=Role.ASSISTANT, text="Hi there!"),
ChatMessage(role=Role.USER, text="Help me"),
ChatMessage(role=Role.ASSISTANT, text="Sure, what do you need?"),
]
# Create mock context
mock_ctx = MagicMock()
mock_ctx.send_message = AsyncMock()
# Simulate post-restore: only new user message with explicit flag
incoming = _ConversationWithUserInput(
full_conversation=[ChatMessage(role=Role.USER, text="I need shipping help")],
is_post_restore=True,
)
# Handle the user input
await coordinator.handle_user_input(incoming, mock_ctx)
# Verify conversation was appended, not replaced
assert len(coordinator._conversation) == 5
assert coordinator._conversation[0].text == "Hello"
assert coordinator._conversation[1].text == "Hi there!"
assert coordinator._conversation[2].text == "Help me"
assert coordinator._conversation[3].text == "Sure, what do you need?"
assert coordinator._conversation[4].text == "I need shipping help"
async def test_coordinator_handle_user_input_normal_flow():
"""Test that _HandoffCoordinator.handle_user_input handles normal flow correctly.
In normal flow (no restore), the gateway sends the full conversation.
The coordinator should replace its conversation with the incoming one.
"""
from unittest.mock import AsyncMock
from agent_framework._workflows._handoff import _HandoffCoordinator
# Create a coordinator
coordinator = _HandoffCoordinator(
starting_agent_id="triage",
specialist_ids={"specialist_a": "specialist_a"},
input_gateway_id="gateway",
termination_condition=lambda conv: False,
id="test-coordinator",
)
# Set some initial conversation
coordinator._conversation = [
ChatMessage(role=Role.USER, text="Old message"),
]
# Create mock context
mock_ctx = MagicMock()
mock_ctx.send_message = AsyncMock()
# Normal flow: full conversation including new user message (is_post_restore=False by default)
incoming = _ConversationWithUserInput(
full_conversation=[
ChatMessage(role=Role.USER, text="Hello"),
ChatMessage(role=Role.ASSISTANT, text="Hi!"),
ChatMessage(role=Role.USER, text="New message"),
],
is_post_restore=False,
)
# Handle the user input
await coordinator.handle_user_input(incoming, mock_ctx)
# Verify conversation was replaced (normal flow with full history)
assert len(coordinator._conversation) == 3
assert coordinator._conversation[0].text == "Hello"
assert coordinator._conversation[1].text == "Hi!"
assert coordinator._conversation[2].text == "New message"
async def test_coordinator_handle_user_input_multiple_consecutive_user_messages():
"""Test that multiple consecutive USER messages in normal flow are handled correctly.
This is a regression test for the edge case where a user submits multiple consecutive
USER messages. The explicit is_post_restore flag ensures this doesn't get incorrectly
detected as a post-restore scenario.
"""
from unittest.mock import AsyncMock
from agent_framework._workflows._handoff import _HandoffCoordinator
# Create a coordinator with existing conversation
coordinator = _HandoffCoordinator(
starting_agent_id="triage",
specialist_ids={"specialist_a": "specialist_a"},
input_gateway_id="gateway",
termination_condition=lambda conv: False,
id="test-coordinator",
)
# Set existing conversation with 4 messages
coordinator._conversation = [
ChatMessage(role=Role.USER, text="Original message 1"),
ChatMessage(role=Role.ASSISTANT, text="Response 1"),
ChatMessage(role=Role.USER, text="Original message 2"),
ChatMessage(role=Role.ASSISTANT, text="Response 2"),
]
# Create mock context
mock_ctx = MagicMock()
mock_ctx.send_message = AsyncMock()
# Normal flow: User sends multiple consecutive USER messages
# This should REPLACE the conversation, not append to it
incoming = _ConversationWithUserInput(
full_conversation=[
ChatMessage(role=Role.USER, text="New user message 1"),
ChatMessage(role=Role.USER, text="New user message 2"),
],
is_post_restore=False, # Explicit flag - this is normal flow
)
# Handle the user input
await coordinator.handle_user_input(incoming, mock_ctx)
# Verify conversation was REPLACED (not appended)
# Without the explicit flag, the old heuristic might incorrectly append
assert len(coordinator._conversation) == 2
assert coordinator._conversation[0].text == "New user message 1"
assert coordinator._conversation[1].text == "New user message 2"
@@ -876,3 +876,204 @@ def test_magentic_builder_does_not_have_human_input_hook():
"MagenticBuilder should not have with_human_input_hook - "
"use with_plan_review() or with_human_input_on_stall() instead"
)
# region Message Deduplication Tests
async def test_magentic_no_duplicate_messages_with_conversation_history():
"""Test that passing list[ChatMessage] does not create duplicate messages in chat_history.
When a frontend passes conversation history as list[ChatMessage], the last message
(task) should not be duplicated in the orchestrator's chat_history.
"""
manager = FakeManager(max_round_count=10)
manager.satisfied_after_signoff = True # Complete immediately after first agent response
wf = MagenticBuilder().participants(agentA=_DummyExec("agentA")).with_standard_manager(manager).build()
# Simulate frontend passing conversation history
conversation: list[ChatMessage] = [
ChatMessage(role=Role.USER, text="previous question"),
ChatMessage(role=Role.ASSISTANT, text="previous answer"),
ChatMessage(role=Role.USER, text="current task"),
]
# Get orchestrator to inspect chat_history after run
orchestrator = None
for executor in wf.executors.values():
if isinstance(executor, MagenticOrchestratorExecutor):
orchestrator = executor
break
events: list[WorkflowEvent] = []
async for event in wf.run_stream(conversation):
events.append(event)
if isinstance(event, WorkflowStatusEvent) and event.state in (
WorkflowRunState.IDLE,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
):
break
assert orchestrator is not None
assert orchestrator._context is not None # type: ignore[reportPrivateUsage]
# Count occurrences of each message text in chat_history
history = orchestrator._context.chat_history # type: ignore[reportPrivateUsage]
user_task_count = sum(1 for msg in history if msg.text == "current task")
prev_question_count = sum(1 for msg in history if msg.text == "previous question")
prev_answer_count = sum(1 for msg in history if msg.text == "previous answer")
# Each input message should appear exactly once (no duplicates)
assert prev_question_count == 1, f"Expected 1 'previous question', got {prev_question_count}"
assert prev_answer_count == 1, f"Expected 1 'previous answer', got {prev_answer_count}"
assert user_task_count == 1, f"Expected 1 'current task', got {user_task_count}"
async def test_magentic_agent_executor_no_duplicate_messages_on_broadcast():
"""Test that MagenticAgentExecutor does not duplicate messages from broadcasts.
When the orchestrator broadcasts the task ledger to all agents, each agent
should receive it exactly once, not multiple times.
"""
backing_executor = _DummyExec("backing")
agent_exec = MagenticAgentExecutor(backing_executor, "agentA")
# Simulate orchestrator sending a broadcast message
broadcast_msg = ChatMessage(
role=Role.ASSISTANT,
text="Task ledger content",
author_name="magentic_manager",
)
# Simulate the same message being received multiple times (e.g., from checkpoint restore + live)
from agent_framework._workflows._magentic import _MagenticResponseMessage
response1 = _MagenticResponseMessage(body=broadcast_msg, broadcast=True)
response2 = _MagenticResponseMessage(body=broadcast_msg, broadcast=True)
# Create a mock context
from unittest.mock import AsyncMock, MagicMock
mock_context = MagicMock()
mock_context.send_message = AsyncMock()
# Call the handler twice with the same message
await agent_exec.handle_response_message(response1, mock_context) # type: ignore[arg-type]
await agent_exec.handle_response_message(response2, mock_context) # type: ignore[arg-type]
# Count how many times the broadcast message appears
history = agent_exec._chat_history # type: ignore[reportPrivateUsage]
broadcast_count = sum(1 for msg in history if msg.text == "Task ledger content")
# Each broadcast should be recorded (this is expected behavior - broadcasts are additive)
# The test documents current behavior. If dedup is needed, this assertion would change.
assert broadcast_count == 2, (
f"Expected 2 broadcasts (current behavior is additive), got {broadcast_count}. "
"If deduplication is required, update the handler logic."
)
async def test_magentic_context_no_duplicate_on_reset():
"""Test that MagenticContext.reset() clears chat_history without leaving duplicates."""
ctx = MagenticContext(
task=ChatMessage(role=Role.USER, text="task"),
participant_descriptions={"Alice": "Researcher"},
)
# Add some history
ctx.chat_history.append(ChatMessage(role=Role.ASSISTANT, text="response1"))
ctx.chat_history.append(ChatMessage(role=Role.ASSISTANT, text="response2"))
assert len(ctx.chat_history) == 2
# Reset
ctx.reset()
# Verify clean slate
assert len(ctx.chat_history) == 0, "chat_history should be empty after reset"
# Add new history
ctx.chat_history.append(ChatMessage(role=Role.ASSISTANT, text="new_response"))
assert len(ctx.chat_history) == 1, "Should have exactly 1 message after adding to reset context"
async def test_magentic_start_message_messages_list_integrity():
"""Test that _MagenticStartMessage preserves message list without internal duplication."""
conversation: list[ChatMessage] = [
ChatMessage(role=Role.USER, text="msg1"),
ChatMessage(role=Role.ASSISTANT, text="msg2"),
ChatMessage(role=Role.USER, text="msg3"),
]
start_msg = _MagenticStartMessage(conversation)
# Verify messages list is preserved
assert len(start_msg.messages) == 3, f"Expected 3 messages, got {len(start_msg.messages)}"
# Verify task is the last message (not a copy)
assert start_msg.task is start_msg.messages[-1], "task should be the same object as messages[-1]"
assert start_msg.task.text == "msg3"
async def test_magentic_checkpoint_restore_no_duplicate_history():
"""Test that checkpoint restore does not create duplicate messages in chat_history."""
manager = FakeManager(max_round_count=10)
storage = InMemoryCheckpointStorage()
wf = (
MagenticBuilder()
.participants(agentA=_DummyExec("agentA"))
.with_standard_manager(manager)
.with_checkpointing(storage)
.build()
)
# Run with conversation history to create initial checkpoint
conversation: list[ChatMessage] = [
ChatMessage(role=Role.USER, text="history_msg"),
ChatMessage(role=Role.USER, text="task_msg"),
]
async for event in wf.run_stream(conversation):
if isinstance(event, WorkflowStatusEvent) and event.state in (
WorkflowRunState.IDLE,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
):
break
# Get checkpoint
checkpoints = await storage.list_checkpoints()
assert len(checkpoints) > 0, "Should have created checkpoints"
latest_checkpoint = checkpoints[-1]
# Load checkpoint and verify no duplicates in shared state
checkpoint_data = await storage.load_checkpoint(latest_checkpoint.checkpoint_id)
assert checkpoint_data is not None
# Check the magentic_context in the checkpoint
for _, executor_state in checkpoint_data.metadata.items():
if isinstance(executor_state, dict) and "magentic_context" in executor_state:
ctx_data = executor_state["magentic_context"]
chat_history = ctx_data.get("chat_history", [])
# Count unique messages by text
texts = [
msg.get("text") or (msg.get("contents", [{}])[0].get("text") if msg.get("contents") else None)
for msg in chat_history
]
text_counts: dict[str, int] = {}
for text in texts:
if text:
text_counts[text] = text_counts.get(text, 0) + 1
# Input messages should not be duplicated
assert text_counts.get("history_msg", 0) <= 1, (
f"'history_msg' appears {text_counts.get('history_msg', 0)} times in checkpoint - expected <= 1"
)
assert text_counts.get("task_msg", 0) <= 1, (
f"'task_msg' appears {text_counts.get('task_msg', 0)} times in checkpoint - expected <= 1"
)
# endregion
@@ -9,18 +9,23 @@ from agent_framework import (
AgentRunResponse,
AgentRunResponseUpdate,
AgentRunUpdateEvent,
AgentThread,
ChatMessage,
ChatMessageStore,
DataContent,
Executor,
FunctionApprovalRequestContent,
FunctionApprovalResponseContent,
FunctionCallContent,
Role,
TextContent,
UriContent,
UsageContent,
UsageDetails,
WorkflowAgent,
WorkflowBuilder,
WorkflowContext,
executor,
handler,
response_handler,
)
@@ -75,6 +80,31 @@ class RequestingExecutor(Executor):
await ctx.add_event(AgentRunUpdateEvent(executor_id=self.id, data=update))
class ConversationHistoryCapturingExecutor(Executor):
"""Executor that captures the received conversation history for verification."""
def __init__(self, id: str):
super().__init__(id=id)
self.received_messages: list[ChatMessage] = []
@handler
async def handle_message(self, messages: list[ChatMessage], ctx: WorkflowContext[list[ChatMessage]]) -> None:
# Capture all received messages
self.received_messages = list(messages)
# Count messages by role for the response
message_count = len(messages)
response_text = f"Received {message_count} messages"
response_message = ChatMessage(role=Role.ASSISTANT, contents=[TextContent(text=response_text)])
streaming_update = AgentRunResponseUpdate(
contents=[TextContent(text=response_text)], role=Role.ASSISTANT, message_id=str(uuid.uuid4())
)
await ctx.add_event(AgentRunUpdateEvent(executor_id=self.id, data=streaming_update))
await ctx.send_message([response_message])
class TestWorkflowAgent:
"""Test cases for WorkflowAgent end-to-end functionality."""
@@ -257,6 +287,240 @@ class TestWorkflowAgent:
with pytest.raises(ValueError, match="Workflow's start executor cannot handle list\\[ChatMessage\\]"):
workflow.as_agent()
async def test_workflow_as_agent_yield_output_surfaces_as_agent_response(self) -> None:
"""Test that ctx.yield_output() in a workflow executor surfaces as agent output when using .as_agent().
This validates the fix for issue #2813: WorkflowOutputEvent should be converted to
AgentRunResponseUpdate when the workflow is wrapped via .as_agent().
"""
@executor
async def yielding_executor(messages: list[ChatMessage], ctx: WorkflowContext) -> None:
# Extract text from input for demonstration
input_text = messages[0].text if messages else "no input"
await ctx.yield_output(f"processed: {input_text}")
workflow = WorkflowBuilder().set_start_executor(yielding_executor).build()
# Run directly - should return WorkflowOutputEvent in result
direct_result = await workflow.run([ChatMessage(role=Role.USER, contents=[TextContent(text="hello")])])
direct_outputs = direct_result.get_outputs()
assert len(direct_outputs) == 1
assert direct_outputs[0] == "processed: hello"
# Run as agent - yield_output should surface as agent response message
agent = workflow.as_agent("test-agent")
agent_result = await agent.run("hello")
assert isinstance(agent_result, AgentRunResponse)
assert len(agent_result.messages) == 1
assert agent_result.messages[0].text == "processed: hello"
async def test_workflow_as_agent_yield_output_surfaces_in_run_stream(self) -> None:
"""Test that ctx.yield_output() surfaces as AgentRunResponseUpdate when streaming."""
@executor
async def yielding_executor(messages: list[ChatMessage], ctx: WorkflowContext) -> None:
await ctx.yield_output("first output")
await ctx.yield_output("second output")
workflow = WorkflowBuilder().set_start_executor(yielding_executor).build()
agent = workflow.as_agent("test-agent")
updates: list[AgentRunResponseUpdate] = []
async for update in agent.run_stream("hello"):
updates.append(update)
# Should have received updates for both yield_output calls
texts = [u.text for u in updates if u.text]
assert "first output" in texts
assert "second output" in texts
async def test_workflow_as_agent_yield_output_with_content_types(self) -> None:
"""Test that yield_output preserves different content types (TextContent, DataContent, etc.)."""
@executor
async def content_yielding_executor(messages: list[ChatMessage], ctx: WorkflowContext) -> None:
# Yield different content types
await ctx.yield_output(TextContent(text="text content"))
await ctx.yield_output(DataContent(data=b"binary data", media_type="application/octet-stream"))
await ctx.yield_output(UriContent(uri="https://example.com/image.png", media_type="image/png"))
workflow = WorkflowBuilder().set_start_executor(content_yielding_executor).build()
agent = workflow.as_agent("content-test-agent")
result = await agent.run("test")
assert isinstance(result, AgentRunResponse)
assert len(result.messages) == 3
# Verify each content type is preserved
assert isinstance(result.messages[0].contents[0], TextContent)
assert result.messages[0].contents[0].text == "text content"
assert isinstance(result.messages[1].contents[0], DataContent)
assert result.messages[1].contents[0].media_type == "application/octet-stream"
assert isinstance(result.messages[2].contents[0], UriContent)
assert result.messages[2].contents[0].uri == "https://example.com/image.png"
async def test_workflow_as_agent_yield_output_with_chat_message(self) -> None:
"""Test that yield_output with ChatMessage preserves the message structure."""
@executor
async def chat_message_executor(messages: list[ChatMessage], ctx: WorkflowContext) -> None:
msg = ChatMessage(
role=Role.ASSISTANT,
contents=[TextContent(text="response text")],
author_name="custom-author",
)
await ctx.yield_output(msg)
workflow = WorkflowBuilder().set_start_executor(chat_message_executor).build()
agent = workflow.as_agent("chat-msg-agent")
result = await agent.run("test")
assert len(result.messages) == 1
assert result.messages[0].role == Role.ASSISTANT
assert result.messages[0].text == "response text"
assert result.messages[0].author_name == "custom-author"
async def test_workflow_as_agent_yield_output_sets_raw_representation(self) -> None:
"""Test that yield_output sets raw_representation with the original data."""
# A custom object to verify raw_representation preserves the original data
class CustomData:
def __init__(self, value: int):
self.value = value
def __str__(self) -> str:
return f"CustomData({self.value})"
@executor
async def raw_yielding_executor(messages: list[ChatMessage], ctx: WorkflowContext) -> None:
# Yield different types of data
await ctx.yield_output("simple string")
await ctx.yield_output(TextContent(text="text content"))
custom = CustomData(42)
await ctx.yield_output(custom)
workflow = WorkflowBuilder().set_start_executor(raw_yielding_executor).build()
agent = workflow.as_agent("raw-test-agent")
updates: list[AgentRunResponseUpdate] = []
async for update in agent.run_stream("test"):
updates.append(update)
# Should have 3 updates
assert len(updates) == 3
# Verify raw_representation is set for each update
assert updates[0].raw_representation == "simple string"
assert isinstance(updates[1].raw_representation, TextContent)
assert updates[1].raw_representation.text == "text content"
assert isinstance(updates[2].raw_representation, CustomData)
assert updates[2].raw_representation.value == 42
async def test_thread_conversation_history_included_in_workflow_run(self) -> None:
"""Test that conversation history from thread is included when running WorkflowAgent.
This verifies that when a thread with existing messages is provided to agent.run(),
the workflow receives the complete conversation history (thread history + new messages).
"""
# Create an executor that captures all received messages
capturing_executor = ConversationHistoryCapturingExecutor(id="capturing")
workflow = WorkflowBuilder().set_start_executor(capturing_executor).build()
agent = WorkflowAgent(workflow=workflow, name="Thread History Test Agent")
# Create a thread with existing conversation history
history_messages = [
ChatMessage(role=Role.USER, text="Previous user message"),
ChatMessage(role=Role.ASSISTANT, text="Previous assistant response"),
]
message_store = ChatMessageStore(messages=history_messages)
thread = AgentThread(message_store=message_store)
# Run the agent with the thread and a new message
new_message = "New user question"
await agent.run(new_message, thread=thread)
# Verify the executor received both history AND new message
assert len(capturing_executor.received_messages) == 3
# Verify the order: history first, then new message
assert capturing_executor.received_messages[0].text == "Previous user message"
assert capturing_executor.received_messages[1].text == "Previous assistant response"
assert capturing_executor.received_messages[2].text == "New user question"
async def test_thread_conversation_history_included_in_workflow_stream(self) -> None:
"""Test that conversation history from thread is included when streaming WorkflowAgent.
This verifies that run_stream also includes thread history.
"""
# Create an executor that captures all received messages
capturing_executor = ConversationHistoryCapturingExecutor(id="capturing_stream")
workflow = WorkflowBuilder().set_start_executor(capturing_executor).build()
agent = WorkflowAgent(workflow=workflow, name="Thread Stream Test Agent")
# Create a thread with existing conversation history
history_messages = [
ChatMessage(role=Role.SYSTEM, text="You are a helpful assistant"),
ChatMessage(role=Role.USER, text="Hello"),
ChatMessage(role=Role.ASSISTANT, text="Hi there!"),
]
message_store = ChatMessageStore(messages=history_messages)
thread = AgentThread(message_store=message_store)
# Stream from the agent with the thread and a new message
async for _ in agent.run_stream("How are you?", thread=thread):
pass
# Verify the executor received all messages (3 from history + 1 new)
assert len(capturing_executor.received_messages) == 4
# Verify the order
assert capturing_executor.received_messages[0].text == "You are a helpful assistant"
assert capturing_executor.received_messages[1].text == "Hello"
assert capturing_executor.received_messages[2].text == "Hi there!"
assert capturing_executor.received_messages[3].text == "How are you?"
async def test_empty_thread_works_correctly(self) -> None:
"""Test that an empty thread (no message store) works correctly."""
capturing_executor = ConversationHistoryCapturingExecutor(id="empty_thread_test")
workflow = WorkflowBuilder().set_start_executor(capturing_executor).build()
agent = WorkflowAgent(workflow=workflow, name="Empty Thread Test Agent")
# Create an empty thread
thread = AgentThread()
# Run with the empty thread
await agent.run("Just a new message", thread=thread)
# Should only receive the new message
assert len(capturing_executor.received_messages) == 1
assert capturing_executor.received_messages[0].text == "Just a new message"
async def test_checkpoint_storage_passed_to_workflow(self) -> None:
"""Test that checkpoint_storage parameter is passed through to the workflow."""
from agent_framework import InMemoryCheckpointStorage
capturing_executor = ConversationHistoryCapturingExecutor(id="checkpoint_test")
workflow = WorkflowBuilder().set_start_executor(capturing_executor).build()
agent = WorkflowAgent(workflow=workflow, name="Checkpoint Test Agent")
# Create checkpoint storage
checkpoint_storage = InMemoryCheckpointStorage()
# Run with checkpoint storage enabled
async for _ in agent.run_stream("Test message", checkpoint_storage=checkpoint_storage):
pass
# Drain workflow events to get checkpoint
# The workflow should have created checkpoints
checkpoints = await checkpoint_storage.list_checkpoints(workflow.id)
assert len(checkpoints) > 0, "Checkpoints should have been created when checkpoint_storage is provided"
class TestWorkflowAgentMergeUpdates:
"""Test cases specifically for the WorkflowAgent.merge_updates static method."""
@@ -293,6 +293,20 @@ def test_register_duplicate_name_raises_error():
builder.register_executor(lambda: MockExecutor(id="executor_2"), name="MyExecutor")
def test_register_duplicate_id_raises_error():
"""Test that registering duplicate id raises an error."""
builder = WorkflowBuilder()
# Register first executor
builder.register_executor(lambda: MockExecutor(id="executor"), name="MyExecutor1")
builder.register_executor(lambda: MockExecutor(id="executor"), name="MyExecutor2")
builder.set_start_executor("MyExecutor1")
# Registering second executor with same ID should raise ValueError
with pytest.raises(ValueError, match="Executor with ID 'executor' has already been created."):
builder.build()
def test_register_agent_basic():
"""Test basic agent registration with lazy initialization."""
builder = WorkflowBuilder()
@@ -483,7 +497,13 @@ def test_mixing_eager_and_lazy_initialization_error():
builder.register_executor(lambda: MockExecutor(id="Lazy"), name="Lazy")
# Mixing eager and lazy should raise an error during add_edge
with pytest.raises(ValueError, match="Both source and target must be either names"):
with pytest.raises(
ValueError,
match=(
r"Both source and target must be either registered factory names \(str\) "
r"or Executor/AgentProtocol instances\."
),
):
builder.add_edge(eager_executor, "Lazy")
@@ -0,0 +1,492 @@
# Copyright (c) Microsoft. All rights reserved.
from collections.abc import AsyncIterable
from typing import Annotated, Any
from agent_framework import (
AgentRunResponse,
AgentRunResponseUpdate,
AgentThread,
BaseAgent,
ChatMessage,
ConcurrentBuilder,
GroupChatBuilder,
GroupChatStateSnapshot,
HandoffBuilder,
Role,
SequentialBuilder,
TextContent,
WorkflowRunState,
WorkflowStatusEvent,
ai_function,
)
from agent_framework._workflows._const import WORKFLOW_RUN_KWARGS_KEY
# Track kwargs received by tools during test execution
_received_kwargs: list[dict[str, Any]] = []
def _reset_received_kwargs() -> None:
"""Reset the kwargs tracker before each test."""
_received_kwargs.clear()
@ai_function
def tool_with_kwargs(
action: Annotated[str, "The action to perform"],
**kwargs: Any,
) -> str:
"""A test tool that captures kwargs for verification."""
_received_kwargs.append(dict(kwargs))
custom_data = kwargs.get("custom_data", {})
user_token = kwargs.get("user_token", {})
return f"Executed {action} with custom_data={custom_data}, user={user_token.get('user_name', 'unknown')}"
class _KwargsCapturingAgent(BaseAgent):
"""Test agent that captures kwargs passed to run/run_stream."""
captured_kwargs: list[dict[str, Any]]
def __init__(self, name: str = "test_agent") -> None:
super().__init__(name=name, description="Test agent for kwargs capture")
self.captured_kwargs = []
async def run(
self,
messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
*,
thread: AgentThread | None = None,
**kwargs: Any,
) -> AgentRunResponse:
self.captured_kwargs.append(dict(kwargs))
return AgentRunResponse(messages=[ChatMessage(role=Role.ASSISTANT, text=f"{self.display_name} response")])
async def run_stream(
self,
messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
*,
thread: AgentThread | None = None,
**kwargs: Any,
) -> AsyncIterable[AgentRunResponseUpdate]:
self.captured_kwargs.append(dict(kwargs))
yield AgentRunResponseUpdate(contents=[TextContent(text=f"{self.display_name} response")])
class _EchoAgent(BaseAgent):
"""Simple agent that echoes back for workflow completion."""
async def run(
self,
messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
*,
thread: AgentThread | None = None,
**kwargs: Any,
) -> AgentRunResponse:
return AgentRunResponse(messages=[ChatMessage(role=Role.ASSISTANT, text=f"{self.display_name} reply")])
async def run_stream(
self,
messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
*,
thread: AgentThread | None = None,
**kwargs: Any,
) -> AsyncIterable[AgentRunResponseUpdate]:
yield AgentRunResponseUpdate(contents=[TextContent(text=f"{self.display_name} reply")])
# region Sequential Builder Tests
async def test_sequential_kwargs_flow_to_agent() -> None:
"""Test that kwargs passed to SequentialBuilder workflow flow through to agent."""
agent = _KwargsCapturingAgent(name="seq_agent")
workflow = SequentialBuilder().participants([agent]).build()
custom_data = {"endpoint": "https://api.example.com", "version": "v1"}
user_token = {"user_name": "alice", "access_level": "admin"}
async for event in workflow.run_stream(
"test message",
custom_data=custom_data,
user_token=user_token,
):
if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
break
# Verify agent received kwargs
assert len(agent.captured_kwargs) >= 1, "Agent should have been invoked at least once"
received = agent.captured_kwargs[0]
assert "custom_data" in received, "Agent should receive custom_data kwarg"
assert "user_token" in received, "Agent should receive user_token kwarg"
assert received["custom_data"] == custom_data
assert received["user_token"] == user_token
async def test_sequential_kwargs_flow_to_multiple_agents() -> None:
"""Test that kwargs flow to all agents in a sequential workflow."""
agent1 = _KwargsCapturingAgent(name="agent1")
agent2 = _KwargsCapturingAgent(name="agent2")
workflow = SequentialBuilder().participants([agent1, agent2]).build()
custom_data = {"key": "value"}
async for event in workflow.run_stream("test", custom_data=custom_data):
if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
break
# Both agents should have received kwargs
assert len(agent1.captured_kwargs) >= 1, "First agent should be invoked"
assert len(agent2.captured_kwargs) >= 1, "Second agent should be invoked"
assert agent1.captured_kwargs[0].get("custom_data") == custom_data
assert agent2.captured_kwargs[0].get("custom_data") == custom_data
async def test_sequential_run_kwargs_flow() -> None:
"""Test that kwargs flow through workflow.run() (non-streaming)."""
agent = _KwargsCapturingAgent(name="run_agent")
workflow = SequentialBuilder().participants([agent]).build()
_ = await workflow.run("test message", custom_data={"test": True})
assert len(agent.captured_kwargs) >= 1
assert agent.captured_kwargs[0].get("custom_data") == {"test": True}
# endregion
# region Concurrent Builder Tests
async def test_concurrent_kwargs_flow_to_agents() -> None:
"""Test that kwargs flow to all agents in a concurrent workflow."""
agent1 = _KwargsCapturingAgent(name="concurrent1")
agent2 = _KwargsCapturingAgent(name="concurrent2")
workflow = ConcurrentBuilder().participants([agent1, agent2]).build()
custom_data = {"batch_id": "123"}
user_token = {"user_name": "bob"}
async for event in workflow.run_stream(
"concurrent test",
custom_data=custom_data,
user_token=user_token,
):
if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
break
# Both agents should have received kwargs
assert len(agent1.captured_kwargs) >= 1, "First concurrent agent should be invoked"
assert len(agent2.captured_kwargs) >= 1, "Second concurrent agent should be invoked"
for agent in [agent1, agent2]:
received = agent.captured_kwargs[0]
assert received.get("custom_data") == custom_data
assert received.get("user_token") == user_token
# endregion
# region GroupChat Builder Tests
async def test_groupchat_kwargs_flow_to_agents() -> None:
"""Test that kwargs flow to agents in a group chat workflow."""
agent1 = _KwargsCapturingAgent(name="chat1")
agent2 = _KwargsCapturingAgent(name="chat2")
# Simple selector that takes GroupChatStateSnapshot
turn_count = 0
def simple_selector(state: GroupChatStateSnapshot) -> str | None:
nonlocal turn_count
turn_count += 1
if turn_count > 2: # Stop after 2 turns
return None
# state is a Mapping - access via dict syntax
names = list(state["participants"].keys())
return names[(turn_count - 1) % len(names)]
workflow = (
GroupChatBuilder().participants(chat1=agent1, chat2=agent2).set_select_speakers_func(simple_selector).build()
)
custom_data = {"session_id": "group123"}
async for event in workflow.run_stream("group chat test", custom_data=custom_data):
if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
break
# At least one agent should have received kwargs
all_kwargs = agent1.captured_kwargs + agent2.captured_kwargs
assert len(all_kwargs) >= 1, "At least one agent should be invoked in group chat"
for received in all_kwargs:
assert received.get("custom_data") == custom_data
# endregion
# region SharedState Verification Tests
async def test_kwargs_stored_in_shared_state() -> None:
"""Test that kwargs are stored in SharedState with the correct key."""
from agent_framework import Executor, WorkflowContext, handler
stored_kwargs: dict[str, Any] | None = None
class _SharedStateInspector(Executor):
@handler
async def inspect(self, msgs: list[ChatMessage], ctx: WorkflowContext[list[ChatMessage]]) -> None:
nonlocal stored_kwargs
stored_kwargs = await ctx.get_shared_state(WORKFLOW_RUN_KWARGS_KEY)
await ctx.send_message(msgs)
inspector = _SharedStateInspector(id="inspector")
workflow = SequentialBuilder().participants([inspector]).build()
async for event in workflow.run_stream("test", my_kwarg="my_value", another=123):
if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
break
assert stored_kwargs is not None, "kwargs should be stored in SharedState"
assert stored_kwargs.get("my_kwarg") == "my_value"
assert stored_kwargs.get("another") == 123
async def test_empty_kwargs_stored_as_empty_dict() -> None:
"""Test that empty kwargs are stored as empty dict in SharedState."""
from agent_framework import Executor, WorkflowContext, handler
stored_kwargs: Any = "NOT_CHECKED"
class _SharedStateChecker(Executor):
@handler
async def check(self, msgs: list[ChatMessage], ctx: WorkflowContext[list[ChatMessage]]) -> None:
nonlocal stored_kwargs
stored_kwargs = await ctx.get_shared_state(WORKFLOW_RUN_KWARGS_KEY)
await ctx.send_message(msgs)
checker = _SharedStateChecker(id="checker")
workflow = SequentialBuilder().participants([checker]).build()
# Run without any kwargs
async for event in workflow.run_stream("test"):
if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
break
# SharedState should have empty dict when no kwargs provided
assert stored_kwargs == {}, f"Expected empty dict, got: {stored_kwargs}"
# endregion
# region Edge Cases
async def test_kwargs_with_none_values() -> None:
"""Test that kwargs with None values are passed through correctly."""
agent = _KwargsCapturingAgent(name="none_test")
workflow = SequentialBuilder().participants([agent]).build()
async for event in workflow.run_stream("test", optional_param=None, other_param="value"):
if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
break
assert len(agent.captured_kwargs) >= 1
received = agent.captured_kwargs[0]
assert "optional_param" in received
assert received["optional_param"] is None
assert received["other_param"] == "value"
async def test_kwargs_with_complex_nested_data() -> None:
"""Test that complex nested data structures flow through correctly."""
agent = _KwargsCapturingAgent(name="nested_test")
workflow = SequentialBuilder().participants([agent]).build()
complex_data = {
"level1": {
"level2": {
"level3": ["a", "b", "c"],
"number": 42,
},
"list": [1, 2, {"nested": True}],
},
"tuple_like": [1, 2, 3],
}
async for event in workflow.run_stream("test", complex_data=complex_data):
if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
break
assert len(agent.captured_kwargs) >= 1
received = agent.captured_kwargs[0]
assert received.get("complex_data") == complex_data
async def test_kwargs_preserved_across_workflow_reruns() -> None:
"""Test that kwargs are correctly isolated between workflow runs."""
agent = _KwargsCapturingAgent(name="rerun_test")
# Build separate workflows for each run to avoid "already running" error
workflow1 = SequentialBuilder().participants([agent]).build()
workflow2 = SequentialBuilder().participants([agent]).build()
# First run
async for event in workflow1.run_stream("run1", run_id="first"):
if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
break
# Second run with different kwargs (using fresh workflow)
async for event in workflow2.run_stream("run2", run_id="second"):
if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
break
assert len(agent.captured_kwargs) >= 2
assert agent.captured_kwargs[0].get("run_id") == "first"
assert agent.captured_kwargs[1].get("run_id") == "second"
# endregion
# region Handoff Builder Tests
async def test_handoff_kwargs_flow_to_agents() -> None:
"""Test that kwargs flow to agents in a handoff workflow."""
agent1 = _KwargsCapturingAgent(name="coordinator")
agent2 = _KwargsCapturingAgent(name="specialist")
workflow = (
HandoffBuilder()
.participants([agent1, agent2])
.set_coordinator(agent1)
.with_interaction_mode("autonomous")
.build()
)
custom_data = {"session_id": "handoff123"}
async for event in workflow.run_stream("handoff test", custom_data=custom_data):
if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
break
# Coordinator agent should have received kwargs
assert len(agent1.captured_kwargs) >= 1, "Coordinator should be invoked in handoff"
assert agent1.captured_kwargs[0].get("custom_data") == custom_data
# endregion
# region Magentic Builder Tests
async def test_magentic_kwargs_flow_to_agents() -> None:
"""Test that kwargs flow to agents in a magentic workflow via MagenticAgentExecutor."""
from agent_framework import MagenticBuilder
from agent_framework._workflows._magentic import (
MagenticContext,
MagenticManagerBase,
_MagenticProgressLedger,
_MagenticProgressLedgerItem,
)
# Create a mock manager that completes after one round
class _MockManager(MagenticManagerBase):
def __init__(self) -> None:
super().__init__(max_stall_count=3, max_reset_count=None, max_round_count=2)
self.task_ledger = None
async def plan(self, context: MagenticContext) -> ChatMessage:
return ChatMessage(role=Role.ASSISTANT, text="Plan: Test task", author_name="manager")
async def replan(self, context: MagenticContext) -> ChatMessage:
return ChatMessage(role=Role.ASSISTANT, text="Replan: Test task", author_name="manager")
async def create_progress_ledger(self, context: MagenticContext) -> _MagenticProgressLedger:
# Return completed on first call
return _MagenticProgressLedger(
is_request_satisfied=_MagenticProgressLedgerItem(answer=True, reason="Done"),
is_progress_being_made=_MagenticProgressLedgerItem(answer=True, reason="Progress"),
is_in_loop=_MagenticProgressLedgerItem(answer=False, reason="Not looping"),
instruction_or_question=_MagenticProgressLedgerItem(answer="Complete", reason="Done"),
next_speaker=_MagenticProgressLedgerItem(answer="agent1", reason="First"),
)
async def prepare_final_answer(self, context: MagenticContext) -> ChatMessage:
return ChatMessage(role=Role.ASSISTANT, text="Final answer", author_name="manager")
agent = _KwargsCapturingAgent(name="agent1")
manager = _MockManager()
workflow = MagenticBuilder().participants(agent1=agent).with_standard_manager(manager=manager).build()
custom_data = {"session_id": "magentic123"}
async for event in workflow.run_stream("magentic test", custom_data=custom_data):
if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
break
# The workflow completes immediately via prepare_final_answer without invoking agents
# because is_request_satisfied=True. This test verifies the kwargs storage path works.
# A more comprehensive integration test would require the manager to select an agent.
async def test_magentic_kwargs_stored_in_shared_state() -> None:
"""Test that kwargs are stored in SharedState when using MagenticWorkflow.run_stream()."""
from agent_framework import MagenticBuilder
from agent_framework._workflows._magentic import (
MagenticContext,
MagenticManagerBase,
_MagenticProgressLedger,
_MagenticProgressLedgerItem,
)
class _MockManager(MagenticManagerBase):
def __init__(self) -> None:
super().__init__(max_stall_count=3, max_reset_count=None, max_round_count=1)
self.task_ledger = None
async def plan(self, context: MagenticContext) -> ChatMessage:
return ChatMessage(role=Role.ASSISTANT, text="Plan", author_name="manager")
async def replan(self, context: MagenticContext) -> ChatMessage:
return ChatMessage(role=Role.ASSISTANT, text="Replan", author_name="manager")
async def create_progress_ledger(self, context: MagenticContext) -> _MagenticProgressLedger:
return _MagenticProgressLedger(
is_request_satisfied=_MagenticProgressLedgerItem(answer=True, reason="Done"),
is_progress_being_made=_MagenticProgressLedgerItem(answer=True, reason="Progress"),
is_in_loop=_MagenticProgressLedgerItem(answer=False, reason="Not looping"),
instruction_or_question=_MagenticProgressLedgerItem(answer="Done", reason="Done"),
next_speaker=_MagenticProgressLedgerItem(answer="agent1", reason="First"),
)
async def prepare_final_answer(self, context: MagenticContext) -> ChatMessage:
return ChatMessage(role=Role.ASSISTANT, text="Final", author_name="manager")
agent = _KwargsCapturingAgent(name="agent1")
manager = _MockManager()
magentic_workflow = MagenticBuilder().participants(agent1=agent).with_standard_manager(manager=manager).build()
# Use MagenticWorkflow.run_stream() which goes through the kwargs attachment path
custom_data = {"magentic_key": "magentic_value"}
async for event in magentic_workflow.run_stream("test task", custom_data=custom_data):
if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
break
# Verify the workflow completed (kwargs were stored, even if agent wasn't invoked)
# The test validates the code path through MagenticWorkflow.run_stream -> _MagenticStartMessage
# endregion
@@ -229,8 +229,10 @@ async def test_trace_context_handling(span_exporter: InMemorySpanExporter) -> No
assert processing_span.attributes.get("message.payload_type") == "str"
@pytest.mark.parametrize("enable_otel", [False], indirect=True)
async def test_trace_context_disabled_when_tracing_disabled(enable_otel, span_exporter: InMemorySpanExporter) -> None:
@pytest.mark.parametrize("enable_instrumentation", [False], indirect=True)
async def test_trace_context_disabled_when_tracing_disabled(
enable_instrumentation, span_exporter: InMemorySpanExporter
) -> None:
"""Test that no trace context is added when tracing is disabled."""
# Tracing should be disabled by default
executor = MockExecutor("test-executor")
@@ -433,7 +435,7 @@ async def test_workflow_error_handling_in_tracing(span_exporter: InMemorySpanExp
assert workflow_span.status.status_code.name == "ERROR"
@pytest.mark.parametrize("enable_otel", [False], indirect=True)
@pytest.mark.parametrize("enable_instrumentation", [False], indirect=True)
async def test_message_trace_context_serialization(span_exporter: InMemorySpanExporter) -> None:
"""Test that message trace context is properly serialized/deserialized."""
ctx = InProcRunnerContext(InMemoryCheckpointStorage())