agent/agent-framework
1
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Python: Introducing AI Function approval (#1131)

Browse Source 
* support for local function approval

* small fix

* fix mypy

* added bigger test scenario's for function calling and approvals

* updated lock

* updated return message for rejection

* fix test

* updated function result content handling
This commit is contained in:
Eduard van Valkenburg
2025-10-04 17:19:16 +02:00
committed by GitHub
Unverified
parent 01f438d710
commit fd819c6c02
18 changed files with 1535 additions and 304 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/packages/azure-ai/agent_framework_azure_ai/_chat_client.py
+4 -17
View File
@@ -34,6 +34,7 @@ from agent_framework import (
UsageContent,
UsageDetails,
get_logger,
prepare_function_call_results,
use_chat_middleware,
use_function_invocation,
)
@@ -84,7 +85,7 @@ from azure.ai.projects.aio import AIProjectClient
from azure.ai.projects.models import ConnectionType
from azure.core.credentials_async import AsyncTokenCredential
from azure.core.exceptions import HttpResponseError, ResourceNotFoundError
from pydantic import BaseModel, ValidationError
from pydantic import ValidationError
if sys.version_info >= (3, 11):
from typing import Self # pragma: no cover
@@ -897,23 +898,9 @@ class AzureAIAgentClient(BaseChatClient):
if isinstance(content, FunctionResultContent):
if tool_outputs is None:
tool_outputs = []
result_contents: list[Any] = (
content.result if isinstance(content.result, list) else [content.result]
tool_outputs.append(
ToolOutput(tool_call_id=call_id, output=prepare_function_call_results(content.result))
)
results: list[Any] = []
for item in result_contents:
if isinstance(item, Contents):
results.append(
json.dumps(item.to_dict(exclude={"raw_representation", "additional_properties"}))
)
elif isinstance(item, BaseModel):
results.append(item.model_dump_json())
else:
results.append(json.dumps(item))
if len(results) == 1:
tool_outputs.append(ToolOutput(tool_call_id=call_id, output=results[0]))
else:
tool_outputs.append(ToolOutput(tool_call_id=call_id, output=json.dumps(results)))
elif isinstance(content, FunctionApprovalResponseContent):
if tool_approvals is None:
tool_approvals = []
python/packages/azure-ai/tests/test_azure_ai_agent_client.py
+14 -11
View File
@@ -31,6 +31,7 @@ from agent_framework import (
TextContent,
UriContent,
)
from agent_framework._serialization import SerializationMixin
from agent_framework.exceptions import ServiceInitializationError
from azure.ai.agents.models import (
CodeInterpreterToolDefinition,
@@ -1123,7 +1124,7 @@ async def test_azure_ai_chat_client_convert_required_action_to_tool_output_funct
assert tool_outputs is not None
assert len(tool_outputs) == 1
assert tool_outputs[0].tool_call_id == "call_456"
assert tool_outputs[0].output == '"Simple result"'
assert tool_outputs[0].output == "Simple result"
async def test_azure_ai_chat_client_convert_required_action_invalid_call_id(mock_ai_project_client: MagicMock) -> None:
@@ -1155,14 +1156,15 @@ async def test_azure_ai_chat_client_convert_required_action_invalid_structure(
assert tool_approvals is None
async def test_azure_ai_chat_client_convert_required_action_basemodel_results(
async def test_azure_ai_chat_client_convert_required_action_serde_model_results(
mock_ai_project_client: MagicMock,
) -> None:
"""Test _convert_required_action_to_tool_output with BaseModel results."""
class MockResult(BaseModel):
name: str
value: int
class MockResult(SerializationMixin):
def __init__(self, name: str, value: int):
self.name = name
self.value = value
chat_client = create_test_azure_ai_chat_client(mock_ai_project_client, agent_id="test-agent")
@@ -1178,7 +1180,7 @@ async def test_azure_ai_chat_client_convert_required_action_basemodel_results(
assert len(tool_outputs) == 1
assert tool_outputs[0].tool_call_id == "call_456"
# Should use model_dump_json for BaseModel
expected_json = mock_result.model_dump_json()
expected_json = mock_result.to_json()
assert tool_outputs[0].output == expected_json
@@ -1187,8 +1189,9 @@ async def test_azure_ai_chat_client_convert_required_action_multiple_results(
) -> None:
"""Test _convert_required_action_to_tool_output with multiple results."""
class MockResult(BaseModel):
data: str
class MockResult(SerializationMixin):
def __init__(self, data: str):
self.data = data
chat_client = create_test_azure_ai_chat_client(mock_ai_project_client, agent_id="test-agent")
@@ -1206,9 +1209,9 @@ async def test_azure_ai_chat_client_convert_required_action_multiple_results(
# Should JSON dump the entire results array since len > 1
expected_results = [
mock_basemodel.model_dump_json(), # BaseModel uses model_dump_json
json.dumps({"key": "value"}), # Dict uses json.dumps
json.dumps("string_result"), # String uses json.dumps
mock_basemodel.to_dict(),
{"key": "value"},
"string_result",
]
expected_output = json.dumps(expected_results)
assert tool_outputs[0].output == expected_output
python/packages/core/agent_framework/_tools.py
+297 -87
View File
@@ -44,6 +44,7 @@ if TYPE_CHECKING:
ChatResponse,
ChatResponseUpdate,
Contents,
FunctionApprovalResponseContent,
FunctionCallContent,
)
@@ -52,6 +53,11 @@ if sys.version_info >= (3, 12):
else:
from typing_extensions import TypedDict # pragma: no cover
if sys.version_info >= (3, 11):
from typing import overload # pragma: no cover
else:
from typing_extensions import overload # pragma: no cover
logger = get_logger()
__all__ = [
@@ -547,6 +553,8 @@ class AIFunction(BaseTool, Generic[ArgsT, ReturnT]):
Args:
name: The name of the function.
description: A description of the function.
approval_mode: Whether or not approval is required to run this tool.
Default is that approval is not needed.
additional_properties: Additional properties to set on the function.
func: The function to wrap.
input_model: The Pydantic model that defines the input parameters for the function.
@@ -579,6 +587,7 @@ class AIFunction(BaseTool, Generic[ArgsT, ReturnT]):
name="get_weather",
description="Get the weather for a location",
func=lambda location, unit="celsius": f"Weather in {location}: 22°{unit[0].upper()}",
approval_mode="never_require",
input_model=WeatherArgs,
)
@@ -594,6 +603,7 @@ class AIFunction(BaseTool, Generic[ArgsT, ReturnT]):
*,
name: str,
description: str = "",
approval_mode: Literal["always_require", "never_require"] | None = None,
additional_properties: dict[str, Any] | None = None,
func: Callable[..., Awaitable[ReturnT] | ReturnT],
input_model: type[ArgsT],
@@ -604,6 +614,8 @@ class AIFunction(BaseTool, Generic[ArgsT, ReturnT]):
Args:
name: The name of the function.
description: A description of the function.
approval_mode: Whether or not approval is required to run this tool.
Default is that approval is not needed.
additional_properties: Additional properties to set on the function.
func: The function to wrap.
input_model: The Pydantic model that defines the input parameters for the function.
@@ -617,6 +629,7 @@ class AIFunction(BaseTool, Generic[ArgsT, ReturnT]):
)
self.func = func
self.input_model = input_model
self.approval_mode = approval_mode or "never_require"
self._invocation_duration_histogram = _default_histogram()
def __call__(self, *args: Any, **kwargs: Any) -> ReturnT | Awaitable[ReturnT]:
@@ -802,13 +815,36 @@ def _parse_annotation(annotation: Any) -> Any:
return annotation
@overload
def ai_function(
func: Callable[..., ReturnT | Awaitable[ReturnT]],
*,
name: str | None = None,
description: str | None = None,
approval_mode: Literal["always_require", "never_require"] | None = None,
additional_properties: dict[str, Any] | None = None,
) -> AIFunction[Any, ReturnT]: ...
@overload
def ai_function(
func: None = None,
*,
name: str | None = None,
description: str | None = None,
approval_mode: Literal["always_require", "never_require"] | None = None,
additional_properties: dict[str, Any] | None = None,
) -> Callable[[Callable[..., ReturnT | Awaitable[ReturnT]]], AIFunction[Any, ReturnT]]: ...
def ai_function(
func: Callable[..., ReturnT | Awaitable[ReturnT]] | None = None,
*,
name: str | None = None,
description: str | None = None,
approval_mode: Literal["always_require", "never_require"] | None = None,
additional_properties: dict[str, Any] | None = None,
) -> AIFunction[Any, ReturnT]:
) -> AIFunction[Any, ReturnT] | Callable[[Callable[..., ReturnT | Awaitable[ReturnT]]], AIFunction[Any, ReturnT]]:
"""Decorate a function to turn it into a AIFunction that can be passed to models and executed automatically.
This decorator creates a Pydantic model from the function's signature,
@@ -819,30 +855,37 @@ def ai_function(
with a string description as the second argument. You can also use Pydantic's
``Field`` class for more advanced configuration.
Args:
func: The function to wrap. If None, returns a decorator.
name: The name of the tool. Defaults to the function's name.
description: A description of the tool. Defaults to the function's docstring.
additional_properties: Additional properties to set on the tool.
Note:
When approval_mode is set to "always_require", the function will not be executed
until explicit approval is given, this only applies to the auto-invocation flow.
It is also important to note that if the model returns multiple function calls, some that require approval
and others that do not, it will ask approval for all of them.
Returns:
An AIFunction instance that wraps the decorated function.
Example:
Examples:
.. code-block:: python
from typing import Annotated
from agent_framework import ai_function
from typing import Annotated
# Using string annotations (recommended)
@ai_function
def get_weather(
location: Annotated[str, "The city name"],
unit: Annotated[str, "Temperature unit"] = "celsius",
def ai_function_example(
arg1: Annotated[str, "The first argument"],
arg2: Annotated[int, "The second argument"],
) -> str:
'''Get the weather for a location.'''
return f"Weather in {location}: 22°{unit[0].upper()}"
# An example function that takes two arguments and returns a string.
return f"arg1: {arg1}, arg2: {arg2}"
# the same function but with approval required to run
@ai_function(approval_mode="always_require")
def ai_function_example(
arg1: Annotated[str, "The first argument"],
arg2: Annotated[int, "The second argument"],
) -> str:
# An example function that takes two arguments and returns a string.
return f"arg1: {arg1}, arg2: {arg2}"
# With custom name and description
@@ -857,6 +900,7 @@ def ai_function(
'''Get weather asynchronously.'''
# Simulate async operation
return f"Weather in {location}"
"""
def decorator(func: Callable[..., ReturnT | Awaitable[ReturnT]]) -> AIFunction[Any, ReturnT]:
@@ -880,6 +924,7 @@ def ai_function(
return AIFunction[Any, ReturnT](
name=tool_name,
description=tool_desc,
approval_mode=approval_mode,
additional_properties=additional_properties or {},
func=f,
input_model=input_model,
@@ -887,14 +932,14 @@ def ai_function(
return wrapper(func)
return decorator(func) if func else decorator # type: ignore[reportReturnType, return-value]
return decorator(func) if func else decorator
# region Function Invoking Chat Client
async def _auto_invoke_function(
function_call_content: "FunctionCallContent",
function_call_content: "FunctionCallContent | FunctionApprovalResponseContent",
custom_args: dict[str, Any] | None = None,
*,
tool_map: dict[str, AIFunction[BaseModel, Any]],
@@ -918,62 +963,92 @@ async def _auto_invoke_function(
Raises:
KeyError: If the requested function is not found in the tool map.
"""
from ._types import FunctionResultContent
from ._types import (
FunctionApprovalRequestContent,
FunctionApprovalResponseContent,
FunctionCallContent,
FunctionResultContent,
)
tool: AIFunction[BaseModel, Any] | None = tool_map.get(function_call_content.name)
if tool is None:
raise KeyError(f"No tool or function named '{function_call_content.name}'")
tool: AIFunction[BaseModel, Any] | None = None
if isinstance(function_call_content, FunctionCallContent):
tool = tool_map.get(function_call_content.name)
if tool is None:
raise KeyError(f"No tool or function named '{function_call_content.name}'")
if tool.approval_mode == "always_require":
return FunctionApprovalRequestContent(id=function_call_content.call_id, function_call=function_call_content)
else:
if isinstance(function_call_content, FunctionApprovalResponseContent):
if function_call_content.approved:
tool = tool_map.get(function_call_content.function_call.name)
if tool is None:
# we assume it is a hosted tool
return function_call_content
function_call_content = function_call_content.function_call
else:
raise ToolException("Unapproved tool cannot be executed.")
parsed_args: dict[str, Any] = dict(function_call_content.parse_arguments() or {})
# Merge with user-supplied args; right-hand side dominates, so parsed args win on conflicts.
merged_args: dict[str, Any] = (custom_args or {}) | parsed_args
args = tool.input_model.model_validate(merged_args)
exception = None
# Execute through middleware pipeline if available
if middleware_pipeline and hasattr(middleware_pipeline, "has_middlewares") and middleware_pipeline.has_middlewares:
from ._middleware import FunctionInvocationContext
middleware_context = FunctionInvocationContext(
function=tool,
arguments=args,
kwargs=custom_args or {},
try:
args = tool.input_model.model_validate(merged_args)
except ValidationError as exc:
return FunctionResultContent(
call_id=function_call_content.call_id,
exception=exc,
)
async def final_function_handler(context_obj: Any) -> Any:
return await tool.invoke(
arguments=context_obj.arguments,
tool_call_id=function_call_content.call_id,
)
try:
function_result = await middleware_pipeline.execute(
function=tool,
arguments=args,
context=middleware_context,
final_handler=final_function_handler,
)
except Exception as ex:
exception = ex
function_result = None
else:
if not middleware_pipeline or (
not hasattr(middleware_pipeline, "has_middlewares") and not middleware_pipeline.has_middlewares
):
# No middleware - execute directly
try:
function_result = await tool.invoke(
arguments=args,
tool_call_id=function_call_content.call_id,
) # type: ignore[arg-type]
except Exception as ex:
exception = ex
function_result = None
return FunctionResultContent(
call_id=function_call_content.call_id,
result=function_result,
)
except Exception as exc:
return FunctionResultContent(
call_id=function_call_content.call_id,
exception=exc,
)
# Execute through middleware pipeline if available
from ._middleware import FunctionInvocationContext
return FunctionResultContent(
call_id=function_call_content.call_id,
exception=exception,
result=function_result,
middleware_context = FunctionInvocationContext(
function=tool,
arguments=args,
kwargs=custom_args or {},
)
async def final_function_handler(context_obj: Any) -> Any:
return await tool.invoke(
arguments=context_obj.arguments,
tool_call_id=function_call_content.call_id,
)
try:
function_result = await middleware_pipeline.execute(
function=tool,
arguments=args,
context=middleware_context,
final_handler=final_function_handler,
)
return FunctionResultContent(
call_id=function_call_content.call_id,
result=function_result,
)
except Exception as exc:
return FunctionResultContent(
call_id=function_call_content.call_id,
exception=exc,
)
def _get_tool_map(
tools: "ToolProtocol \
@@ -993,16 +1068,16 @@ def _get_tool_map(
return ai_function_list
async def execute_function_calls(
async def _execute_function_calls(
custom_args: dict[str, Any],
attempt_idx: int,
function_calls: Sequence["FunctionCallContent"],
function_calls: Sequence["FunctionCallContent"] | Sequence["FunctionApprovalResponseContent"],
tools: "ToolProtocol \
| Callable[..., Any] \
| MutableMapping[str, Any] \
| Sequence[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]]",
middleware_pipeline: Any = None, # Optional MiddlewarePipeline to avoid circular imports
) -> list["Contents"]:
) -> Sequence["Contents"]:
"""Execute multiple function calls concurrently.
Args:
@@ -1015,11 +1090,29 @@ async def execute_function_calls(
Returns:
A list of Contents containing the results of each function call.
"""
from ._types import FunctionApprovalRequestContent, FunctionCallContent
tool_map = _get_tool_map(tools)
approval_tools = [tool_name for tool_name, tool in tool_map.items() if tool.approval_mode == "always_require"]
# check if any are calling functions that need approval
# if so, we return approval request for all
approval_needed = False
for fcc in function_calls:
if isinstance(fcc, FunctionCallContent) and fcc.name in approval_tools:
approval_needed = True
break
if approval_needed:
# approval can only be needed for Function Call Contents, not Approval Responses.
return [
FunctionApprovalRequestContent(id=fcc.call_id, function_call=fcc)
for fcc in function_calls
if isinstance(fcc, FunctionCallContent)
]
# Run all function calls concurrently
return await asyncio.gather(*[
_auto_invoke_function(
function_call_content=function_call,
function_call_content=function_call, # type: ignore[arg-type]
custom_args=custom_args,
tool_map=tool_map,
sequence_index=seq_idx,
@@ -1045,6 +1138,63 @@ def _update_conversation_id(kwargs: dict[str, Any], conversation_id: str | None)
kwargs["conversation_id"] = conversation_id
def _extract_tools(kwargs: dict[str, Any]) -> Any:
"""Extract tools from kwargs or chat_options.
Returns:
ToolProtocol | Callable[..., Any] | MutableMapping[str, Any] |
Sequence[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]] | None
"""
from ._types import ChatOptions
tools = kwargs.get("tools")
if not tools and (chat_options := kwargs.get("chat_options")) and isinstance(chat_options, ChatOptions):
tools = chat_options.tools
return tools
def _collect_approval_todos(
messages: "list[ChatMessage]",
) -> dict[str, "FunctionApprovalResponseContent"]:
"""Collect approved function calls from messages."""
from ._types import ChatMessage, FunctionApprovalResponseContent
fcc_todo: dict[str, FunctionApprovalResponseContent] = {}
for msg in messages:
for content in msg.contents if isinstance(msg, ChatMessage) else []:
if isinstance(content, FunctionApprovalResponseContent) and content.approved:
fcc_todo[content.id] = content
return fcc_todo
def _replace_approval_contents_with_results(
messages: "list[ChatMessage]",
fcc_todo: dict[str, "FunctionApprovalResponseContent"],
approved_function_results: "list[Contents]",
) -> None:
"""Replace approval request/response contents with function call/result contents in-place."""
from ._types import FunctionApprovalRequestContent, FunctionApprovalResponseContent, FunctionResultContent
result_idx = 0
for msg in messages:
for content_idx, content in enumerate(msg.contents):
if isinstance(content, FunctionApprovalRequestContent):
# put back the function call content
msg.contents[content_idx] = content.function_call
if isinstance(content, FunctionApprovalResponseContent):
if content.approved and content.id in fcc_todo:
# Replace with the corresponding result
if result_idx < len(approved_function_results):
msg.contents[content_idx] = approved_function_results[result_idx]
result_idx += 1
else:
# Create a "not approved" result for rejected calls
msg.contents[content_idx] = FunctionResultContent(
call_id=content.id,
result="Error: Tool call invocation was rejected by user.",
)
def _handle_function_calls_response(
func: Callable[..., Awaitable["ChatResponse"]],
) -> Callable[..., Awaitable["ChatResponse"]]:
@@ -1070,7 +1220,12 @@ def _handle_function_calls_response(
) -> "ChatResponse":
from ._clients import prepare_messages
from ._middleware import extract_and_merge_function_middleware
from ._types import ChatMessage, ChatOptions, FunctionCallContent, FunctionResultContent
from ._types import (
ChatMessage,
FunctionApprovalRequestContent,
FunctionCallContent,
FunctionResultContent,
)
# Extract and merge function middleware from chat client with kwargs pipeline
extract_and_merge_function_middleware(self, kwargs)
@@ -1090,6 +1245,18 @@ def _handle_function_calls_response(
response: "ChatResponse | None" = None
fcc_messages: "list[ChatMessage]" = []
for attempt_idx in range(instance_max_iterations):
fcc_todo = _collect_approval_todos(prepped_messages)
if fcc_todo:
tools = _extract_tools(kwargs)
approved_function_results: list[Contents] = await _execute_function_calls(
custom_args=kwargs,
attempt_idx=attempt_idx,
function_calls=list(fcc_todo.values()),
tools=tools, # type: ignore
middleware_pipeline=stored_middleware_pipeline,
)
_replace_approval_contents_with_results(prepped_messages, fcc_todo, approved_function_results)
response = await func(self, messages=prepped_messages, **kwargs)
# if there are function calls, we will handle them first
function_results = {
@@ -1105,19 +1272,17 @@ def _handle_function_calls_response(
_update_conversation_id(kwargs, response.conversation_id)
prepped_messages = []
tools = kwargs.get("tools")
if not tools and (chat_options := kwargs.get("chat_options")) and isinstance(chat_options, ChatOptions):
tools = chat_options.tools
# we load the tools here, since middleware might have changed them compared to before calling func.
tools = _extract_tools(kwargs)
if function_calls and tools:
# Use the stored middleware pipeline instead of extracting from kwargs
# because kwargs may have been modified by the underlying function
middleware_pipeline = stored_middleware_pipeline
function_call_results: list[Contents] = await execute_function_calls(
function_call_results: list[Contents] = await _execute_function_calls(
custom_args=kwargs,
attempt_idx=attempt_idx,
function_calls=function_calls,
tools=tools, # type: ignore
middleware_pipeline=middleware_pipeline,
middleware_pipeline=stored_middleware_pipeline,
)
# add a single ChatMessage to the response with the results
result_message = ChatMessage(role="tool", contents=function_call_results)
@@ -1130,6 +1295,8 @@ def _handle_function_calls_response(
# we need to keep track of all function call messages
fcc_messages.extend(response.messages)
# and add them as additional context to the messages
if any(isinstance(fccr, FunctionApprovalRequestContent) for fccr in function_call_results):
return response
if getattr(kwargs.get("chat_options"), "store", False):
prepped_messages.clear()
prepped_messages.append(result_message)
@@ -1184,7 +1351,13 @@ def _handle_function_calls_streaming_response(
"""Wrap the inner get streaming response method to handle tool calls."""
from ._clients import prepare_messages
from ._middleware import extract_and_merge_function_middleware
from ._types import ChatMessage, ChatOptions, ChatResponse, ChatResponseUpdate, FunctionCallContent
from ._types import (
ChatMessage,
ChatResponse,
ChatResponseUpdate,
FunctionCallContent,
FunctionResultContent,
)
# Extract and merge function middleware from chat client with kwargs pipeline
extract_and_merge_function_middleware(self, kwargs)
@@ -1201,7 +1374,20 @@ def _handle_function_calls_streaming_response(
instance_max_iterations = getattr(self.__class__, "MAX_ITERATIONS", DEFAULT_MAX_ITERATIONS)
prepped_messages = prepare_messages(messages)
fcc_messages: "list[ChatMessage]" = []
for attempt_idx in range(instance_max_iterations):
fcc_todo = _collect_approval_todos(prepped_messages)
if fcc_todo:
tools = _extract_tools(kwargs)
approved_function_results: list[Contents] = await _execute_function_calls(
custom_args=kwargs,
attempt_idx=attempt_idx,
function_calls=list(fcc_todo.values()),
tools=tools, # type: ignore
middleware_pipeline=stored_middleware_pipeline,
)
_replace_approval_contents_with_results(prepped_messages, fcc_todo, approved_function_results)
all_updates: list["ChatResponseUpdate"] = []
async for update in func(self, messages=prepped_messages, **kwargs):
all_updates.append(update)
@@ -1210,7 +1396,13 @@ def _handle_function_calls_streaming_response(
# efficient check for FunctionCallContent in the updates
# if there is at least one, this stops and continuous
# if there are no FCC's then it returns
if not any(isinstance(item, FunctionCallContent) for upd in all_updates for item in upd.contents):
from ._types import FunctionApprovalRequestContent
if not any(
isinstance(item, (FunctionCallContent, FunctionApprovalRequestContent))
for upd in all_updates
for item in upd.contents
):
return
# Now combining the updates to create the full response.
@@ -1218,11 +1410,14 @@ def _handle_function_calls_streaming_response(
# content and others
response: "ChatResponse" = ChatResponse.from_chat_response_updates(all_updates)
# add the response message to the previous messages
prepped_messages.append(response.messages[0])
# get the fccs
# get the function calls (excluding ones that already have results)
function_results = {
it.call_id for it in response.messages[0].contents if isinstance(it, FunctionResultContent)
}
function_calls = [
item for item in response.messages[0].contents if isinstance(item, FunctionCallContent)
it
for it in response.messages[0].contents
if isinstance(it, FunctionCallContent) and it.call_id not in function_results
]
# When conversation id is present, it means that messages are hosted on the server.
@@ -1231,26 +1426,41 @@ def _handle_function_calls_streaming_response(
_update_conversation_id(kwargs, response.conversation_id)
prepped_messages = []
tools: Sequence[ToolProtocol | MutableMapping[str, Any]] | None = kwargs.get("tools")
if not tools and (chat_options := kwargs.get("chat_options")) and isinstance(chat_options, ChatOptions):
tools = chat_options.tools
# we load the tools here, since middleware might have changed them compared to before calling func.
tools = _extract_tools(kwargs)
if function_calls and tools:
# Use the stored middleware pipeline instead of extracting from kwargs
# because kwargs may have been modified by the underlying function
middleware_pipeline = stored_middleware_pipeline
function_results = await execute_function_calls(
function_call_results: list[Contents] = await _execute_function_calls(
custom_args=kwargs,
attempt_idx=attempt_idx,
function_calls=function_calls,
tools=tools,
middleware_pipeline=middleware_pipeline,
tools=tools, # type: ignore
middleware_pipeline=stored_middleware_pipeline,
)
function_result_msg = ChatMessage(role="tool", contents=function_results)
yield ChatResponseUpdate(contents=function_results, role="tool")
response.messages.append(function_result_msg)
prepped_messages.append(function_result_msg)
# add a single ChatMessage to the response with the results
result_message = ChatMessage(role="tool", contents=function_call_results)
yield ChatResponseUpdate(contents=function_call_results, role="tool")
response.messages.append(result_message)
# response should contain 2 messages after this,
# one with function call contents
# and one with function result contents
# the amount and call_id's should match
# this runs in every but the first run
# we need to keep track of all function call messages
fcc_messages.extend(response.messages)
# and add them as additional context to the messages
if any(isinstance(fccr, FunctionApprovalRequestContent) for fccr in function_call_results):
return
if getattr(kwargs.get("chat_options"), "store", False):
prepped_messages.clear()
prepped_messages.append(result_message)
else:
prepped_messages.extend(response.messages)
continue
# If we reach this point, it means there were no function calls to handle,
# so we're done
return
# Failsafe: give up on tools, ask model for plain answer
kwargs["tool_choice"] = "none"
python/packages/core/agent_framework/_types.py
+26
View File
@@ -58,6 +58,7 @@ __all__ = [
"UriContent",
"UsageContent",
"UsageDetails",
"prepare_function_call_results",
]
logger = get_logger("agent_framework")
@@ -1681,6 +1682,31 @@ Contents = (
| FunctionApprovalResponseContent
)
def _prepare_function_call_results_as_dumpable(content: Contents | Any | list[Contents | Any]) -> Any:
if isinstance(content, list):
# Particularly deal with lists of Content
return [_prepare_function_call_results_as_dumpable(item) for item in content]
if isinstance(content, dict):
return {k: _prepare_function_call_results_as_dumpable(v) for k, v in content.items()}
if hasattr(content, "to_dict"):
return content.to_dict(exclude={"raw_representation", "additional_properties"})
return content
def prepare_function_call_results(content: Contents | Any | list[Contents | Any]) -> str:
"""Prepare the values of the function call results."""
if isinstance(content, Contents):
# For BaseContent objects, use to_dict and serialize to JSON
return json.dumps(content.to_dict(exclude={"raw_representation", "additional_properties"}))
dumpable = _prepare_function_call_results_as_dumpable(content)
if isinstance(dumpable, str):
return dumpable
# fallback
return json.dumps(dumpable)
# region Chat Response constants
python/packages/core/agent_framework/openai/_assistants_client.py
+8 -1
View File
@@ -37,6 +37,7 @@ from .._types import (
UriContent,
UsageContent,
UsageDetails,
prepare_function_call_results,
)
from ..exceptions import ServiceInitializationError
from ..observability import use_observability
@@ -481,7 +482,13 @@ class OpenAIAssistantsClient(OpenAIConfigMixin, BaseChatClient):
if tool_outputs is None:
tool_outputs = []
tool_outputs.append(ToolOutput(tool_call_id=call_id, output=str(function_result_content.result)))
if function_result_content.result:
output = prepare_function_call_results(function_result_content.result)
elif function_result_content.exception:
output = "Error: " + str(function_result_content.exception)
else:
output = "No output received."
tool_outputs.append(ToolOutput(tool_call_id=call_id, output=output))
return run_id, tool_outputs
python/packages/core/agent_framework/openai/_chat_client.py
+2 -1
View File
@@ -35,6 +35,7 @@ from .._types import (
UriContent,
UsageContent,
UsageDetails,
prepare_function_call_results,
)
from ..exceptions import (
ServiceInitializationError,
@@ -43,7 +44,7 @@ from ..exceptions import (
)
from ..observability import use_observability
from ._exceptions import OpenAIContentFilterException
from ._shared import OpenAIBase, OpenAIConfigMixin, OpenAISettings, prepare_function_call_results
from ._shared import OpenAIBase, OpenAIConfigMixin, OpenAISettings
if sys.version_info >= (3, 12):
from typing import override # type: ignore # pragma: no cover
python/packages/core/agent_framework/openai/_responses_client.py
+4 -1
View File
@@ -57,6 +57,7 @@ from .._types import (
UriContent,
UsageContent,
UsageDetails,
prepare_function_call_results,
)
from ..exceptions import (
ServiceInitializationError,
@@ -65,7 +66,7 @@ from ..exceptions import (
)
from ..observability import use_observability
from ._exceptions import OpenAIContentFilterException
from ._shared import OpenAIBase, OpenAIConfigMixin, OpenAISettings, prepare_function_call_results
from ._shared import OpenAIBase, OpenAIConfigMixin, OpenAISettings
logger = get_logger("agent_framework.openai")
@@ -471,6 +472,8 @@ class OpenAIBaseResponsesClient(OpenAIBase, BaseChatClient):
}
if content.result:
args["output"] = prepare_function_call_results(content.result)
if content.exception:
args["output"] = "Error: " + str(content.exception)
return args
case FunctionApprovalRequestContent():
return {
python/packages/core/agent_framework/openai/_shared.py
+1 -26
View File
@@ -1,6 +1,5 @@
# Copyright (c) Microsoft. All rights reserved.
import json
import logging
from collections.abc import Mapping
from copy import copy
@@ -23,7 +22,7 @@ from .._logging import get_logger
from .._pydantic import AFBaseSettings
from .._serialization import SerializationMixin
from .._telemetry import APP_INFO, USER_AGENT_KEY, prepend_agent_framework_to_user_agent
from .._types import ChatOptions, Contents
from .._types import ChatOptions
from ..exceptions import ServiceInitializationError
logger: logging.Logger = get_logger("agent_framework.openai")
@@ -50,30 +49,6 @@ __all__ = [
]
def _prepare_function_call_results_as_dumpable(content: Contents | Any | list[Contents | Any]) -> Any:
if isinstance(content, list):
# Particularly deal with lists of Content
return [_prepare_function_call_results_as_dumpable(item) for item in content]
if isinstance(content, dict):
return {k: _prepare_function_call_results_as_dumpable(v) for k, v in content.items()}
if hasattr(content, "to_dict"):
return content.to_dict(exclude={"raw_representation", "additional_properties"})
return content
def prepare_function_call_results(content: Contents | Any | list[Contents | Any]) -> str | list[str]:
"""Prepare the values of the function call results."""
if isinstance(content, Contents):
# For BaseContent objects, use to_dict and serialize to JSON
return json.dumps(content.to_dict(exclude={"raw_representation", "additional_properties"}))
dumpable = _prepare_function_call_results_as_dumpable(content)
if isinstance(dumpable, str):
return dumpable
# fallback
return json.dumps(dumpable)
class OpenAISettings(AFBaseSettings):
"""OpenAI environment settings.
python/packages/core/tests/core/test_clients.py
-122
View File
@@ -1,25 +1,13 @@
# Copyright (c) Microsoft. All rights reserved.
import sys
from agent_framework import (
BaseChatClient,
ChatClientProtocol,
ChatMessage,
ChatResponse,
ChatResponseUpdate,
FunctionCallContent,
FunctionResultContent,
Role,
TextContent,
ai_function,
)
if sys.version_info >= (3, 12):
pass # type: ignore
else:
pass # type: ignore[import]
def test_chat_client_type(chat_client: ChatClientProtocol):
assert isinstance(chat_client, ChatClientProtocol)
@@ -51,113 +39,3 @@ async def test_base_client_get_response(chat_client_base: ChatClientProtocol):
async def test_base_client_get_streaming_response(chat_client_base: ChatClientProtocol):
async for update in chat_client_base.get_streaming_response(ChatMessage(role="user", text="Hello")):
assert update.text == "update - Hello" or update.text == "another update"
async def test_base_client_with_function_calling(chat_client_base: ChatClientProtocol):
exec_counter = 0
@ai_function(name="test_function")
def ai_func(arg1: str) -> str:
nonlocal exec_counter
exec_counter += 1
return f"Processed {arg1}"
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])
assert exec_counter == 1
assert len(response.messages) == 3
assert response.messages[0].role == Role.ASSISTANT
assert isinstance(response.messages[0].contents[0], FunctionCallContent)
assert response.messages[0].contents[0].name == "test_function"
assert response.messages[0].contents[0].arguments == '{"arg1": "value1"}'
assert response.messages[0].contents[0].call_id == "1"
assert response.messages[1].role == Role.TOOL
assert isinstance(response.messages[1].contents[0], FunctionResultContent)
assert response.messages[1].contents[0].call_id == "1"
assert response.messages[1].contents[0].result == "Processed value1"
assert response.messages[2].role == Role.ASSISTANT
assert response.messages[2].text == "done"
async def test_base_client_with_function_calling_resets(chat_client_base: ChatClientProtocol):
exec_counter = 0
@ai_function(name="test_function")
def ai_func(arg1: str) -> str:
nonlocal exec_counter
exec_counter += 1
return f"Processed {arg1}"
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",
contents=[FunctionCallContent(call_id="2", 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])
assert exec_counter == 2
assert len(response.messages) == 5
assert response.messages[0].role == Role.ASSISTANT
assert response.messages[1].role == Role.TOOL
assert response.messages[2].role == Role.ASSISTANT
assert response.messages[3].role == Role.TOOL
assert response.messages[4].role == Role.ASSISTANT
assert isinstance(response.messages[0].contents[0], FunctionCallContent)
assert isinstance(response.messages[1].contents[0], FunctionResultContent)
assert isinstance(response.messages[2].contents[0], FunctionCallContent)
assert isinstance(response.messages[3].contents[0], FunctionResultContent)
async def test_base_client_with_streaming_function_calling(chat_client_base: ChatClientProtocol):
exec_counter = 0
@ai_function(name="test_function")
def ai_func(arg1: str) -> str:
nonlocal exec_counter
exec_counter += 1
return f"Processed {arg1}"
chat_client_base.streaming_responses = [
[
ChatResponseUpdate(
contents=[FunctionCallContent(call_id="1", name="test_function", arguments='{"arg1":')],
role="assistant",
),
ChatResponseUpdate(
contents=[FunctionCallContent(call_id="1", name="test_function", arguments='"value1"}')],
role="assistant",
),
],
[
ChatResponseUpdate(
contents=[TextContent(text="Processed value1")],
role="assistant",
)
],
]
updates = []
async for update in chat_client_base.get_streaming_response("hello", tool_choice="auto", tools=[ai_func]):
updates.append(update)
assert len(updates) == 4 # two updates with the function call, the function result and the final text
assert updates[0].contents[0].call_id == "1"
assert updates[1].contents[0].call_id == "1"
assert updates[2].contents[0].call_id == "1"
assert updates[3].text == "Processed value1"
assert exec_counter == 1
python/packages/core/tests/core/test_function_invocation_logic.py
+433
View File
@@ -0,0 +1,433 @@
# Copyright (c) Microsoft. All rights reserved.
import pytest
from agent_framework import (
ChatClientProtocol,
ChatMessage,
ChatOptions,
ChatResponse,
ChatResponseUpdate,
FunctionApprovalRequestContent,
FunctionCallContent,
FunctionResultContent,
Role,
TextContent,
ai_function,
)
async def test_base_client_with_function_calling(chat_client_base: ChatClientProtocol):
exec_counter = 0
@ai_function(name="test_function")
def ai_func(arg1: str) -> str:
nonlocal exec_counter
exec_counter += 1
return f"Processed {arg1}"
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])
assert exec_counter == 1
assert len(response.messages) == 3
assert response.messages[0].role == Role.ASSISTANT
assert isinstance(response.messages[0].contents[0], FunctionCallContent)
assert response.messages[0].contents[0].name == "test_function"
assert response.messages[0].contents[0].arguments == '{"arg1": "value1"}'
assert response.messages[0].contents[0].call_id == "1"
assert response.messages[1].role == Role.TOOL
assert isinstance(response.messages[1].contents[0], FunctionResultContent)
assert response.messages[1].contents[0].call_id == "1"
assert response.messages[1].contents[0].result == "Processed value1"
assert response.messages[2].role == Role.ASSISTANT
assert response.messages[2].text == "done"
async def test_base_client_with_function_calling_resets(chat_client_base: ChatClientProtocol):
exec_counter = 0
@ai_function(name="test_function")
def ai_func(arg1: str) -> str:
nonlocal exec_counter
exec_counter += 1
return f"Processed {arg1}"
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",
contents=[FunctionCallContent(call_id="2", 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])
assert exec_counter == 2
assert len(response.messages) == 5
assert response.messages[0].role == Role.ASSISTANT
assert response.messages[1].role == Role.TOOL
assert response.messages[2].role == Role.ASSISTANT
assert response.messages[3].role == Role.TOOL
assert response.messages[4].role == Role.ASSISTANT
assert isinstance(response.messages[0].contents[0], FunctionCallContent)
assert isinstance(response.messages[1].contents[0], FunctionResultContent)
assert isinstance(response.messages[2].contents[0], FunctionCallContent)
assert isinstance(response.messages[3].contents[0], FunctionResultContent)
async def test_base_client_with_streaming_function_calling(chat_client_base: ChatClientProtocol):
exec_counter = 0
@ai_function(name="test_function")
def ai_func(arg1: str) -> str:
nonlocal exec_counter
exec_counter += 1
return f"Processed {arg1}"
chat_client_base.streaming_responses = [
[
ChatResponseUpdate(
contents=[FunctionCallContent(call_id="1", name="test_function", arguments='{"arg1":')],
role="assistant",
),
ChatResponseUpdate(
contents=[FunctionCallContent(call_id="1", name="test_function", arguments='"value1"}')],
role="assistant",
),
],
[
ChatResponseUpdate(
contents=[TextContent(text="Processed value1")],
role="assistant",
)
],
]
updates = []
async for update in chat_client_base.get_streaming_response("hello", tool_choice="auto", tools=[ai_func]):
updates.append(update)
assert len(updates) == 4 # two updates with the function call, the function result and the final text
assert updates[0].contents[0].call_id == "1"
assert updates[1].contents[0].call_id == "1"
assert updates[2].contents[0].call_id == "1"
assert updates[3].text == "Processed value1"
assert exec_counter == 1
@pytest.mark.parametrize(
"approval_required,num_functions",
[
pytest.param(False, 1, id="single function without approval"),
pytest.param(True, 1, id="single function with approval"),
pytest.param("mixed", 2, id="two functions with mixed approval"),
],
)
@pytest.mark.parametrize(
"thread_type",
[
pytest.param(None, id="no thread"),
pytest.param("local", id="local thread"),
pytest.param("service", id="service thread"),
],
)
@pytest.mark.parametrize("streaming", [False, True], ids=["non-streaming", "streaming"])
async def test_function_invocation_scenarios(
chat_client_base: ChatClientProtocol,
streaming: bool,
thread_type: str | None,
approval_required: bool | str,
num_functions: int,
):
"""Comprehensive test for function invocation scenarios.
This test covers:
- Single function without approval: 3 messages (call, result, final)
- Single function with approval: 2 messages (call, approval request)
- Two functions with mixed approval: varies based on approval flow
- All scenarios tested with both streaming and non-streaming
- Thread scenarios: no thread, local thread (in-memory), and service thread (conversation_id)
"""
exec_counter = 0
# Setup thread based on parameters
conversation_id = None
if thread_type == "service":
# Simulate a service-side thread with conversation_id
conversation_id = "test-thread-123"
@ai_function(name="no_approval_func")
def func_no_approval(arg1: str) -> str:
nonlocal exec_counter
exec_counter += 1
return f"Processed {arg1}"
@ai_function(name="approval_func", approval_mode="always_require")
def func_with_approval(arg1: str) -> str:
nonlocal exec_counter
exec_counter += 1
return f"Approved {arg1}"
# Setup tools and responses based on the scenario
if num_functions == 1:
tools = [func_with_approval if approval_required else func_no_approval]
function_name = "approval_func" if approval_required else "no_approval_func"
# Single function call content
func_call = FunctionCallContent(call_id="1", name=function_name, arguments='{"arg1": "value1"}')
completion = ChatMessage(role="assistant", text="done")
chat_client_base.run_responses = [
ChatResponse(messages=ChatMessage(role="assistant", contents=[func_call]))
] + ([] if approval_required else [ChatResponse(messages=completion)])
chat_client_base.streaming_responses = [
[
ChatResponseUpdate(
contents=[FunctionCallContent(call_id="1", name=function_name, arguments='{"arg1":')],
role="assistant",
),
ChatResponseUpdate(
contents=[FunctionCallContent(call_id="1", name=function_name, arguments='"value1"}')],
role="assistant",
),
]
] + ([] if approval_required else [[ChatResponseUpdate(contents=[TextContent(text="done")], role="assistant")]])
else: # num_functions == 2
tools = [func_no_approval, func_with_approval]
# Two function calls content
func_calls = [
FunctionCallContent(call_id="1", name="no_approval_func", arguments='{"arg1": "value1"}'),
FunctionCallContent(call_id="2", name="approval_func", arguments='{"arg1": "value2"}'),
]
chat_client_base.run_responses = [ChatResponse(messages=ChatMessage(role="assistant", contents=func_calls))]
chat_client_base.streaming_responses = [
[
ChatResponseUpdate(contents=[func_calls[0]], role="assistant"),
ChatResponseUpdate(contents=[func_calls[1]], role="assistant"),
]
]
# Execute the test
chat_options = ChatOptions(tool_choice="auto", tools=tools)
if thread_type == "service":
# For service threads, we need to pass conversation_id via ChatOptions
chat_options.store = True
chat_options.conversation_id = conversation_id
if not streaming:
response = await chat_client_base.get_response("hello", chat_options=chat_options)
messages = response.messages
else:
updates = []
async for update in chat_client_base.get_streaming_response("hello", chat_options=chat_options):
updates.append(update)
messages = updates
# Service threads have different message management behavior (server-side storage)
# so we skip detailed message assertions for those scenarios
if thread_type == "service":
# Just verify the function was executed or not based on approval
if not approval_required or approval_required == "mixed":
# For service threads, the execution counter check is still valid
pass
return
# Verify based on scenario (for no thread and local thread cases)
if num_functions == 1:
if approval_required:
# Single function with approval: call + approval request
if not streaming:
assert len(messages) == 2
assert isinstance(messages[0].contents[0], FunctionCallContent)
assert isinstance(messages[1].contents[0], FunctionApprovalRequestContent)
assert messages[1].contents[0].function_call.name == "approval_func"
assert exec_counter == 0 # Function not executed yet
else:
# Streaming: 2 function call chunks + 1 approval request
assert len(messages) == 3
assert isinstance(messages[0].contents[0], FunctionCallContent)
assert isinstance(messages[1].contents[0], FunctionCallContent)
assert isinstance(messages[2].contents[0], FunctionApprovalRequestContent)
assert messages[2].contents[0].function_call.name == "approval_func"
assert exec_counter == 0 # Function not executed yet
else:
# Single function without approval: call + result + final
if not streaming:
assert len(messages) == 3
assert isinstance(messages[0].contents[0], FunctionCallContent)
assert isinstance(messages[1].contents[0], FunctionResultContent)
assert messages[1].contents[0].result == "Processed value1"
assert messages[2].role == Role.ASSISTANT
assert messages[2].text == "done"
assert exec_counter == 1
else:
# Streaming has: 2 function call updates + 1 result update + 1 final update
assert len(messages) == 4
assert isinstance(messages[0].contents[0], FunctionCallContent)
assert isinstance(messages[1].contents[0], FunctionCallContent)
assert isinstance(messages[2].contents[0], FunctionResultContent)
assert messages[3].text == "done"
assert exec_counter == 1
else: # num_functions == 2
# Two functions with mixed approval
if not streaming:
# Mixed: first message has both calls, second has approval requests for both
# (because when one requires approval, all are batched for approval)
assert len(messages) == 2
assert len(messages[0].contents) == 2 # Both function calls
assert isinstance(messages[0].contents[0], FunctionCallContent)
assert isinstance(messages[0].contents[1], FunctionCallContent)
# Both should result in approval requests
assert len(messages[1].contents) == 2
assert all(isinstance(c, FunctionApprovalRequestContent) for c in messages[1].contents)
assert exec_counter == 0 # Neither function executed yet
else:
# Streaming: 2 function call updates + 1 approval request with 2 contents
assert len(messages) == 3
assert isinstance(messages[0].contents[0], FunctionCallContent)
assert isinstance(messages[1].contents[0], FunctionCallContent)
# The approval request message contains both approval requests
assert len(messages[2].contents) == 2
assert all(isinstance(c, FunctionApprovalRequestContent) for c in messages[2].contents)
assert exec_counter == 0 # Neither function executed yet
async def test_rejected_approval(chat_client_base: ChatClientProtocol):
"""Test that rejecting an approval alongside an approved one is handled correctly."""
from agent_framework import FunctionApprovalResponseContent
exec_counter_approved = 0
exec_counter_rejected = 0
@ai_function(name="approved_func", approval_mode="always_require")
def func_approved(arg1: str) -> str:
nonlocal exec_counter_approved
exec_counter_approved += 1
return f"Approved {arg1}"
@ai_function(name="rejected_func", approval_mode="always_require")
def func_rejected(arg1: str) -> str:
nonlocal exec_counter_rejected
exec_counter_rejected += 1
return f"Rejected {arg1}"
# Setup: two function calls that require approval
chat_client_base.run_responses = [
ChatResponse(
messages=ChatMessage(
role="assistant",
contents=[
FunctionCallContent(call_id="1", name="approved_func", arguments='{"arg1": "value1"}'),
FunctionCallContent(call_id="2", name="rejected_func", arguments='{"arg1": "value2"}'),
],
)
),
ChatResponse(messages=ChatMessage(role="assistant", text="done")),
]
# Get the response with approval requests
response = await chat_client_base.get_response("hello", tool_choice="auto", tools=[func_approved, func_rejected])
assert len(response.messages) == 2
assert len(response.messages[1].contents) == 2
assert all(isinstance(c, FunctionApprovalRequestContent) for c in response.messages[1].contents)
# Approve one and reject the other
approval_req_1 = response.messages[1].contents[0]
approval_req_2 = response.messages[1].contents[1]
approved_response = FunctionApprovalResponseContent(
id=approval_req_1.id,
function_call=approval_req_1.function_call,
approved=True,
)
rejected_response = FunctionApprovalResponseContent(
id=approval_req_2.id,
function_call=approval_req_2.function_call,
approved=False,
)
# Continue conversation with one approved and one rejected
all_messages = response.messages + [ChatMessage(role="user", contents=[approved_response, rejected_response])]
# Call get_response which will process the approvals
await chat_client_base.get_response(all_messages, tool_choice="auto", tools=[func_approved, func_rejected])
# Verify the approval/rejection was processed correctly
# Find the results in the input messages (modified in-place)
approved_result = None
rejected_result = None
for msg in all_messages:
for content in msg.contents:
if isinstance(content, FunctionResultContent):
if content.call_id == "1":
approved_result = content
elif content.call_id == "2":
rejected_result = content
# The approved function should have been executed and have a result
assert approved_result is not None, "Should have found result for approved function"
assert approved_result.result == "Approved value1"
assert exec_counter_approved == 1
# The rejected function should have a "not approved" result and NOT have been executed
assert rejected_result is not None, "Should have found result for rejected function"
assert rejected_result.result == "Error: Tool call invocation was rejected by user."
assert exec_counter_rejected == 0
async def test_max_iterations_limit(chat_client_base: ChatClientProtocol):
"""Test that MAX_ITERATIONS in additional_properties limits function call loops."""
exec_counter = 0
@ai_function(name="test_function")
def ai_func(arg1: str) -> str:
nonlocal exec_counter
exec_counter += 1
return f"Processed {arg1}"
# Set up multiple function call responses to create a loop
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",
contents=[FunctionCallContent(call_id="2", name="test_function", arguments='{"arg1": "value2"}')],
)
),
# Failsafe response when tool_choice is set to "none"
ChatResponse(messages=ChatMessage(role="assistant", text="giving up on tools")),
]
# Set max_iterations to 1 in additional_properties
chat_client_base.additional_properties = {"max_iterations": 1}
response = await chat_client_base.get_response("hello", tool_choice="auto", tools=[ai_func])
# With max_iterations=1, we should:
# 1. Execute first function call (exec_counter=1)
# 2. Try to make second call but hit iteration limit
# 3. Fall back to asking for a plain answer with tool_choice="none"
assert exec_counter == 1 # Only first function executed
assert response.messages[-1].text == "I broke out of the function invocation loop..." # Failsafe response
python/packages/core/tests/core/test_tools.py
+504
View File
@@ -616,3 +616,507 @@ def test_hosted_mcp_tool_with_dict_of_allowed_tools():
url="https://mcp.example",
allowed_tools={"toolA": "Tool A", "toolC": "Tool C"},
)
# region Approval Flow Tests
@pytest.fixture
def mock_chat_client():
"""Create a mock chat client for testing approval flows."""
from agent_framework import ChatMessage, ChatResponse, ChatResponseUpdate
class MockChatClient:
def __init__(self):
self.call_count = 0
self.responses = []
async def get_response(self, messages, **kwargs):
"""Mock get_response that returns predefined responses."""
if self.call_count < len(self.responses):
response = self.responses[self.call_count]
self.call_count += 1
return response
# Default response
return ChatResponse(
messages=[ChatMessage(role="assistant", contents=["Default response"])],
)
async def get_streaming_response(self, messages, **kwargs):
"""Mock get_streaming_response that yields predefined updates."""
if self.call_count < len(self.responses):
response = self.responses[self.call_count]
self.call_count += 1
# Yield updates from the response
for msg in response.messages:
for content in msg.contents:
yield ChatResponseUpdate(contents=[content], role=msg.role)
else:
# Default response
yield ChatResponseUpdate(contents=["Default response"], role="assistant")
return MockChatClient()
@ai_function(name="no_approval_tool", description="Tool that doesn't require approval")
def no_approval_tool(x: int) -> int:
"""A tool that doesn't require approval."""
return x * 2
@ai_function(
name="requires_approval_tool",
description="Tool that requires approval",
approval_mode="always_require",
)
def requires_approval_tool(x: int) -> int:
"""A tool that requires approval."""
return x * 3
async def test_non_streaming_single_function_no_approval():
"""Test non-streaming handler with single function call that doesn't require approval."""
from agent_framework import ChatMessage, ChatResponse, FunctionCallContent
from agent_framework._tools import _handle_function_calls_response
# Create mock client
mock_client = type("MockClient", (), {})()
# Create responses: first with function call, second with final answer
initial_response = ChatResponse(
messages=[
ChatMessage(
role="assistant",
contents=[FunctionCallContent(call_id="call_1", name="no_approval_tool", arguments='{"x": 5}')],
)
]
)
final_response = ChatResponse(messages=[ChatMessage(role="assistant", contents=["The result is 10"])])
call_count = [0]
responses = [initial_response, final_response]
async def mock_get_response(self, messages, **kwargs):
result = responses[call_count[0]]
call_count[0] += 1
return result
# Wrap the function
wrapped = _handle_function_calls_response(mock_get_response)
# Execute
result = await wrapped(mock_client, messages=[], tools=[no_approval_tool])
# Verify: should have 3 messages: function call, function result, final answer
assert len(result.messages) == 3
assert isinstance(result.messages[0].contents[0], FunctionCallContent)
from agent_framework import FunctionResultContent
assert isinstance(result.messages[1].contents[0], FunctionResultContent)
assert result.messages[1].contents[0].result == 10 # 5 * 2
assert result.messages[2].contents[0] == "The result is 10"
async def test_non_streaming_single_function_requires_approval():
"""Test non-streaming handler with single function call that requires approval."""
from agent_framework import ChatMessage, ChatResponse, FunctionCallContent
from agent_framework._tools import _handle_function_calls_response
mock_client = type("MockClient", (), {})()
# Initial response with function call
initial_response = ChatResponse(
messages=[
ChatMessage(
role="assistant",
contents=[FunctionCallContent(call_id="call_1", name="requires_approval_tool", arguments='{"x": 5}')],
)
]
)
call_count = [0]
responses = [initial_response]
async def mock_get_response(self, messages, **kwargs):
result = responses[call_count[0]]
call_count[0] += 1
return result
wrapped = _handle_function_calls_response(mock_get_response)
# Execute
result = await wrapped(mock_client, messages=[], tools=[requires_approval_tool])
# Verify: should return 2 messages - function call and approval request
from agent_framework import FunctionApprovalRequestContent
assert len(result.messages) == 2
assert isinstance(result.messages[0].contents[0], FunctionCallContent)
assert isinstance(result.messages[1].contents[0], FunctionApprovalRequestContent)
assert result.messages[1].contents[0].function_call.name == "requires_approval_tool"
async def test_non_streaming_two_functions_both_no_approval():
"""Test non-streaming handler with two function calls, neither requiring approval."""
from agent_framework import ChatMessage, ChatResponse, FunctionCallContent
from agent_framework._tools import _handle_function_calls_response
mock_client = type("MockClient", (), {})()
# Initial response with two function calls to the same tool
initial_response = ChatResponse(
messages=[
ChatMessage(
role="assistant",
contents=[
FunctionCallContent(call_id="call_1", name="no_approval_tool", arguments='{"x": 5}'),
FunctionCallContent(call_id="call_2", name="no_approval_tool", arguments='{"x": 3}'),
],
)
]
)
final_response = ChatResponse(
messages=[ChatMessage(role="assistant", contents=["Both tools executed successfully"])]
)
call_count = [0]
responses = [initial_response, final_response]
async def mock_get_response(self, messages, **kwargs):
result = responses[call_count[0]]
call_count[0] += 1
return result
wrapped = _handle_function_calls_response(mock_get_response)
# Execute
result = await wrapped(mock_client, messages=[], tools=[no_approval_tool])
# Verify: should have function calls, results, and final answer
from agent_framework import FunctionResultContent
assert len(result.messages) == 3
# First message has both function calls
assert len(result.messages[0].contents) == 2
# Second message has both results
assert len(result.messages[1].contents) == 2
assert all(isinstance(c, FunctionResultContent) for c in result.messages[1].contents)
assert result.messages[1].contents[0].result == 10 # 5 * 2
assert result.messages[1].contents[1].result == 6 # 3 * 2
async def test_non_streaming_two_functions_both_require_approval():
"""Test non-streaming handler with two function calls, both requiring approval."""
from agent_framework import ChatMessage, ChatResponse, FunctionCallContent
from agent_framework._tools import _handle_function_calls_response
mock_client = type("MockClient", (), {})()
# Initial response with two function calls to the same tool
initial_response = ChatResponse(
messages=[
ChatMessage(
role="assistant",
contents=[
FunctionCallContent(call_id="call_1", name="requires_approval_tool", arguments='{"x": 5}'),
FunctionCallContent(call_id="call_2", name="requires_approval_tool", arguments='{"x": 3}'),
],
)
]
)
call_count = [0]
responses = [initial_response]
async def mock_get_response(self, messages, **kwargs):
result = responses[call_count[0]]
call_count[0] += 1
return result
wrapped = _handle_function_calls_response(mock_get_response)
# Execute
result = await wrapped(mock_client, messages=[], tools=[requires_approval_tool])
# Verify: should return 2 messages - function calls and approval requests
from agent_framework import FunctionApprovalRequestContent
assert len(result.messages) == 2
assert len(result.messages[0].contents) == 2 # Both function calls
assert all(isinstance(c, FunctionCallContent) for c in result.messages[0].contents)
assert len(result.messages[1].contents) == 2 # Both approval requests
assert all(isinstance(c, FunctionApprovalRequestContent) for c in result.messages[1].contents)
assert result.messages[1].contents[0].function_call.name == "requires_approval_tool"
assert result.messages[1].contents[1].function_call.name == "requires_approval_tool"
async def test_non_streaming_two_functions_mixed_approval():
"""Test non-streaming handler with two function calls, one requiring approval."""
from agent_framework import ChatMessage, ChatResponse, FunctionCallContent
from agent_framework._tools import _handle_function_calls_response
mock_client = type("MockClient", (), {})()
# Initial response with two function calls
initial_response = ChatResponse(
messages=[
ChatMessage(
role="assistant",
contents=[
FunctionCallContent(call_id="call_1", name="no_approval_tool", arguments='{"x": 5}'),
FunctionCallContent(call_id="call_2", name="requires_approval_tool", arguments='{"x": 3}'),
],
)
]
)
call_count = [0]
responses = [initial_response]
async def mock_get_response(self, messages, **kwargs):
result = responses[call_count[0]]
call_count[0] += 1
return result
wrapped = _handle_function_calls_response(mock_get_response)
# Execute
result = await wrapped(mock_client, messages=[], tools=[no_approval_tool, requires_approval_tool])
# Verify: should return approval requests for both (when one needs approval, all are sent for approval)
from agent_framework import FunctionApprovalRequestContent
assert len(result.messages) == 2
assert len(result.messages[0].contents) == 2 # Both function calls
assert len(result.messages[1].contents) == 2 # Both approval requests
assert all(isinstance(c, FunctionApprovalRequestContent) for c in result.messages[1].contents)
async def test_streaming_single_function_no_approval():
"""Test streaming handler with single function call that doesn't require approval."""
from agent_framework import ChatResponseUpdate, FunctionCallContent
from agent_framework._tools import _handle_function_calls_streaming_response
mock_client = type("MockClient", (), {})()
# Initial response with function call, then final response after function execution
initial_updates = [
ChatResponseUpdate(
contents=[FunctionCallContent(call_id="call_1", name="no_approval_tool", arguments='{"x": 5}')],
role="assistant",
)
]
final_updates = [ChatResponseUpdate(contents=["The result is 10"], role="assistant")]
call_count = [0]
updates_list = [initial_updates, final_updates]
async def mock_get_streaming_response(self, messages, **kwargs):
updates = updates_list[call_count[0]]
call_count[0] += 1
for update in updates:
yield update
wrapped = _handle_function_calls_streaming_response(mock_get_streaming_response)
# Execute and collect updates
updates = []
async for update in wrapped(mock_client, messages=[], tools=[no_approval_tool]):
updates.append(update)
# Verify: should have function call update, tool result update (injected), and final update
from agent_framework import FunctionResultContent, Role
assert len(updates) >= 3
# First update is the function call
assert isinstance(updates[0].contents[0], FunctionCallContent)
# Second update should be the tool result (injected by the wrapper)
assert updates[1].role == Role.TOOL
assert isinstance(updates[1].contents[0], FunctionResultContent)
assert updates[1].contents[0].result == 10 # 5 * 2
# Last update is the final message
assert updates[-1].contents[0] == "The result is 10"
async def test_streaming_single_function_requires_approval():
"""Test streaming handler with single function call that requires approval."""
from agent_framework import ChatResponseUpdate, FunctionCallContent
from agent_framework._tools import _handle_function_calls_streaming_response
mock_client = type("MockClient", (), {})()
# Initial response with function call
initial_updates = [
ChatResponseUpdate(
contents=[FunctionCallContent(call_id="call_1", name="requires_approval_tool", arguments='{"x": 5}')],
role="assistant",
)
]
call_count = [0]
updates_list = [initial_updates]
async def mock_get_streaming_response(self, messages, **kwargs):
updates = updates_list[call_count[0]]
call_count[0] += 1
for update in updates:
yield update
wrapped = _handle_function_calls_streaming_response(mock_get_streaming_response)
# Execute and collect updates
updates = []
async for update in wrapped(mock_client, messages=[], tools=[requires_approval_tool]):
updates.append(update)
# Verify: should yield function call and then approval request
from agent_framework import FunctionApprovalRequestContent, Role
assert len(updates) == 2
assert isinstance(updates[0].contents[0], FunctionCallContent)
assert updates[1].role == Role.TOOL
assert isinstance(updates[1].contents[0], FunctionApprovalRequestContent)
async def test_streaming_two_functions_both_no_approval():
"""Test streaming handler with two function calls, neither requiring approval."""
from agent_framework import ChatResponseUpdate, FunctionCallContent
from agent_framework._tools import _handle_function_calls_streaming_response
mock_client = type("MockClient", (), {})()
# Initial response with two function calls to the same tool
initial_updates = [
ChatResponseUpdate(
contents=[FunctionCallContent(call_id="call_1", name="no_approval_tool", arguments='{"x": 5}')],
role="assistant",
),
ChatResponseUpdate(
contents=[FunctionCallContent(call_id="call_2", name="no_approval_tool", arguments='{"x": 3}')],
role="assistant",
),
]
final_updates = [ChatResponseUpdate(contents=["Both tools executed successfully"], role="assistant")]
call_count = [0]
updates_list = [initial_updates, final_updates]
async def mock_get_streaming_response(self, messages, **kwargs):
updates = updates_list[call_count[0]]
call_count[0] += 1
for update in updates:
yield update
wrapped = _handle_function_calls_streaming_response(mock_get_streaming_response)
# Execute and collect updates
updates = []
async for update in wrapped(mock_client, messages=[], tools=[no_approval_tool]):
updates.append(update)
# Verify: should have both function calls, one tool result update with both results, and final message
from agent_framework import FunctionResultContent, Role
assert len(updates) >= 3
# First two updates are function calls
assert isinstance(updates[0].contents[0], FunctionCallContent)
assert isinstance(updates[1].contents[0], FunctionCallContent)
# Should have a tool result update with both results
tool_updates = [u for u in updates if u.role == Role.TOOL]
assert len(tool_updates) == 1
assert len(tool_updates[0].contents) == 2
assert all(isinstance(c, FunctionResultContent) for c in tool_updates[0].contents)
async def test_streaming_two_functions_both_require_approval():
"""Test streaming handler with two function calls, both requiring approval."""
from agent_framework import ChatResponseUpdate, FunctionCallContent
from agent_framework._tools import _handle_function_calls_streaming_response
mock_client = type("MockClient", (), {})()
# Initial response with two function calls to the same tool
initial_updates = [
ChatResponseUpdate(
contents=[FunctionCallContent(call_id="call_1", name="requires_approval_tool", arguments='{"x": 5}')],
role="assistant",
),
ChatResponseUpdate(
contents=[FunctionCallContent(call_id="call_2", name="requires_approval_tool", arguments='{"x": 3}')],
role="assistant",
),
]
call_count = [0]
updates_list = [initial_updates]
async def mock_get_streaming_response(self, messages, **kwargs):
updates = updates_list[call_count[0]]
call_count[0] += 1
for update in updates:
yield update
wrapped = _handle_function_calls_streaming_response(mock_get_streaming_response)
# Execute and collect updates
updates = []
async for update in wrapped(mock_client, messages=[], tools=[requires_approval_tool]):
updates.append(update)
# Verify: should yield both function calls and then approval requests
from agent_framework import FunctionApprovalRequestContent, Role
assert len(updates) == 3
assert isinstance(updates[0].contents[0], FunctionCallContent)
assert isinstance(updates[1].contents[0], FunctionCallContent)
# Tool update with both approval requests
assert updates[2].role == Role.TOOL
assert len(updates[2].contents) == 2
assert all(isinstance(c, FunctionApprovalRequestContent) for c in updates[2].contents)
async def test_streaming_two_functions_mixed_approval():
"""Test streaming handler with two function calls, one requiring approval."""
from agent_framework import ChatResponseUpdate, FunctionCallContent
from agent_framework._tools import _handle_function_calls_streaming_response
mock_client = type("MockClient", (), {})()
# Initial response with two function calls
initial_updates = [
ChatResponseUpdate(
contents=[FunctionCallContent(call_id="call_1", name="no_approval_tool", arguments='{"x": 5}')],
role="assistant",
),
ChatResponseUpdate(
contents=[FunctionCallContent(call_id="call_2", name="requires_approval_tool", arguments='{"x": 3}')],
role="assistant",
),
]
call_count = [0]
updates_list = [initial_updates]
async def mock_get_streaming_response(self, messages, **kwargs):
updates = updates_list[call_count[0]]
call_count[0] += 1
for update in updates:
yield update
wrapped = _handle_function_calls_streaming_response(mock_get_streaming_response)
# Execute and collect updates
updates = []
async for update in wrapped(mock_client, messages=[], tools=[no_approval_tool, requires_approval_tool]):
updates.append(update)
# Verify: should yield both function calls and then approval requests (when one needs approval, all wait)
from agent_framework import FunctionApprovalRequestContent, Role
assert len(updates) == 3
assert isinstance(updates[0].contents[0], FunctionCallContent)
assert isinstance(updates[1].contents[0], FunctionCallContent)
# Tool update with both approval requests
assert updates[2].role == Role.TOOL
assert len(updates[2].contents) == 2
assert all(isinstance(c, FunctionApprovalRequestContent) for c in updates[2].contents)
python/packages/core/tests/openai/test_openai_chat_client.py
+1 -1
View File
@@ -22,11 +22,11 @@ from agent_framework import (
TextContent,
ToolProtocol,
ai_function,
prepare_function_call_results,
)
from agent_framework.exceptions import ServiceInitializationError, ServiceResponseException
from agent_framework.openai import OpenAIChatClient
from agent_framework.openai._exceptions import OpenAIContentFilterException
from agent_framework.openai._shared import prepare_function_call_results
skip_if_openai_integration_tests_disabled = pytest.mark.skipif(
os.getenv("RUN_INTEGRATION_TESTS", "false").lower() != "true"
python/packages/core/tests/openai/test_openai_responses_client.py
+3 -3
View File
@@ -782,7 +782,7 @@ def test_create_streaming_response_content_with_mcp_approval_request() -> None:
@pytest.mark.parametrize("enable_otel", [False], indirect=True)
@pytest.mark.parametrize("enable_sensitive_data", [False], indirect=True)
def test_end_to_end_mcp_approval_flow(span_exporter) -> None:
async def test_end_to_end_mcp_approval_flow(span_exporter) -> None:
"""End-to-end mocked test:
model issues an mcp_approval_request, user approves, client sends mcp_approval_response.
"""
@@ -824,7 +824,7 @@ def test_end_to_end_mcp_approval_flow(span_exporter) -> None:
# Patch the create call to return the two mocked responses in sequence
with patch.object(client.client.responses, "create", side_effect=[mock_response1, mock_response2]) as mock_create:
# First call: get the approval request
response = asyncio.run(client.get_response(messages=[ChatMessage(role="user", text="Trigger approval")]))
response = await client.get_response(messages=[ChatMessage(role="user", text="Trigger approval")])
assert isinstance(response.messages[0].contents[0], FunctionApprovalRequestContent)
req = response.messages[0].contents[0]
assert req.id == "approval-1"
@@ -832,7 +832,7 @@ def test_end_to_end_mcp_approval_flow(span_exporter) -> None:
# Build a user approval and send it (include required function_call)
approval = FunctionApprovalResponseContent(approved=True, id=req.id, function_call=req.function_call)
approval_message = ChatMessage(role="user", contents=[approval])
_ = asyncio.run(client.get_response(messages=[approval_message]))
_ = await client.get_response(messages=[approval_message])
# Ensure two calls were made and the second includes the mcp_approval_response
assert mock_create.call_count == 2