diff --git a/python/semantic_kernel/agents/azure_ai/agent_thread_actions.py b/python/semantic_kernel/agents/azure_ai/agent_thread_actions.py index 62fb798bb1..7015c1ef4e 100644 --- a/python/semantic_kernel/agents/azure_ai/agent_thread_actions.py +++ b/python/semantic_kernel/agents/azure_ai/agent_thread_actions.py @@ -68,6 +68,8 @@ from semantic_kernel.agents.open_ai.assistant_content_generation import merge_st from semantic_kernel.agents.open_ai.function_action_result import FunctionActionResult from semantic_kernel.agents.open_ai.run_polling_options import RunPollingOptions from semantic_kernel.connectors.ai.function_calling_utils import kernel_function_metadata_to_function_call_format +from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior +from semantic_kernel.connectors.ai.function_choice_type import FunctionChoiceType from semantic_kernel.contents.chat_message_content import ChatMessageContent from semantic_kernel.contents.function_call_content import FunctionCallContent from semantic_kernel.contents.utils.author_role import AuthorRole @@ -124,6 +126,7 @@ class AgentThreadActions: parallel_tool_calls: bool | None = None, metadata: dict[str, str] | None = None, polling_options: RunPollingOptions | None = None, + function_choice_behavior: FunctionChoiceBehavior | None = None, **kwargs: Any, ) -> AsyncIterable[tuple[bool, "ChatMessageContent"]]: """Invoke the message in the thread. @@ -139,7 +142,9 @@ class AgentThreadActions: additional_messages: The additional messages to add to the thread. Only supports messages with role = User or Assistant. https://platform.openai.com/docs/api-reference/runs/createRun#runs-createrun-additional_messages - tools: The tools. + tools: The SDK-level tools (e.g. CodeInterpreter, FileSearch, AzureAISearch). When provided, + overrides the tools from the agent definition. Does not affect kernel function availability; + use function_choice_behavior for that. temperature: The temperature. top_p: The top p. max_prompt_tokens: The max prompt tokens. @@ -150,6 +155,9 @@ class AgentThreadActions: metadata: The metadata. polling_options: The polling options defined at the run-level. These will override the agent-level polling options. + function_choice_behavior: Controls which kernel functions are allowed to execute during this run. + Use FunctionChoiceBehavior.Auto(filters={"included_functions": [...]}) to restrict to specific + functions. Only Auto is supported; other types will raise an error. kwargs: Additional keyword arguments. Returns: @@ -158,7 +166,11 @@ class AgentThreadActions: arguments = KernelArguments() if arguments is None else KernelArguments(**arguments, **kwargs) kernel = kernel or agent.kernel - tools = cls._get_tools(agent=agent, kernel=kernel) # type: ignore + cls._validate_function_choice_behavior(function_choice_behavior) + + tools = cls._get_tools( + agent=agent, kernel=kernel, tools_override=tools, function_choice_behavior=function_choice_behavior + ) # type: ignore base_instructions = await agent.format_instructions(kernel=kernel, arguments=arguments) @@ -232,7 +244,11 @@ class AgentThreadActions: chat_history = ChatHistory() if kwargs.get("chat_history") is None else kwargs["chat_history"] _ = await cls._invoke_function_calls( - kernel=kernel, fccs=fccs, chat_history=chat_history, arguments=arguments + kernel=kernel, + fccs=fccs, + chat_history=chat_history, + arguments=arguments, + function_choice_behavior=function_choice_behavior, ) tool_outputs = cls._format_tool_outputs(fccs, chat_history) @@ -467,6 +483,7 @@ class AgentThreadActions: temperature: float | None = None, top_p: float | None = None, truncation_strategy: TruncationObject | None = None, + function_choice_behavior: FunctionChoiceBehavior | None = None, **kwargs: Any, ) -> AsyncIterable["StreamingChatMessageContent"]: """Invoke the agent stream and yield ChatMessageContent continuously. @@ -489,10 +506,15 @@ class AgentThreadActions: formed from the streamed chunks. parallel_tool_calls: Whether to configure parallel tool calls. response_format: The response format. - tools: The tools. + tools: The SDK-level tools (e.g. CodeInterpreter, FileSearch, AzureAISearch). When provided, + overrides the tools from the agent definition. Does not affect kernel function availability; + use function_choice_behavior for that. temperature: The temperature. top_p: The top p. truncation_strategy: The truncation strategy. + function_choice_behavior: Controls which kernel functions are allowed to execute during this run. + Use FunctionChoiceBehavior.Auto(filters={"included_functions": [...]}) to restrict to specific + functions. Only Auto is supported; other types will raise an error. kwargs: Additional keyword arguments. Returns: @@ -502,7 +524,11 @@ class AgentThreadActions: kernel = kernel or agent.kernel arguments = agent._merge_arguments(arguments) - tools = cls._get_tools(agent=agent, kernel=kernel) # type: ignore + cls._validate_function_choice_behavior(function_choice_behavior) + + tools = cls._get_tools( + agent=agent, kernel=kernel, tools_override=tools, function_choice_behavior=function_choice_behavior + ) # type: ignore base_instructions = await agent.format_instructions(kernel=kernel, arguments=arguments) @@ -549,6 +575,7 @@ class AgentThreadActions: arguments=arguments, function_steps=function_steps, active_messages=active_messages, + function_choice_behavior=function_choice_behavior, ): if content: yield content @@ -564,6 +591,7 @@ class AgentThreadActions: function_steps: dict[str, FunctionCallContent], active_messages: dict[str, RunStep], output_messages: "list[ChatMessageContent] | None" = None, + function_choice_behavior: FunctionChoiceBehavior | None = None, ) -> AsyncIterable["StreamingChatMessageContent"]: """Process events from the main stream and delegate tool output handling as needed.""" thread_msg_id = None @@ -671,6 +699,7 @@ class AgentThreadActions: run=run, function_steps=function_steps, arguments=arguments, + function_choice_behavior=function_choice_behavior, ) if action_result is None: raise RuntimeError( @@ -959,12 +988,74 @@ class AgentThreadActions: } return [tool for tool in new_tools if tool.get("function", {}).get("name") not in existing_names] + @staticmethod + def _validate_function_choice_behavior( + function_choice_behavior: FunctionChoiceBehavior | None, + ) -> None: + """Validate the function choice behavior is compatible with agent invocations.""" + if function_choice_behavior is None: + return + if function_choice_behavior.type_ != FunctionChoiceType.AUTO: + raise AgentInvokeException( + f"FunctionChoiceBehavior with type '{function_choice_behavior.type_}' is not supported for agent " + "invocations. Use FunctionChoiceBehavior.Auto(filters=...) to control which kernel functions " + "are available." + ) + if not function_choice_behavior.auto_invoke_kernel_functions: + raise AgentInvokeException( + "FunctionChoiceBehavior.Auto(auto_invoke=False) is not supported for agent invocations. " + "The agent run loop manages tool invocation; disabling auto_invoke is not compatible." + ) + valid_filter_keys: set[str] = { + "excluded_plugins", + "included_plugins", + "excluded_functions", + "included_functions", + } + if function_choice_behavior.filters is not None: + if not function_choice_behavior.filters: + raise AgentInvokeException( + "FunctionChoiceBehavior filters must not be empty. Provide at least one filter key " + f"from {sorted(valid_filter_keys)}, or omit filters entirely to include all " + "kernel functions." + ) + unknown_keys = {str(k) for k in function_choice_behavior.filters} - valid_filter_keys + if unknown_keys: + raise AgentInvokeException( + f"Unknown filter key(s): {sorted(unknown_keys)}. " + f"Valid filter keys are: {sorted(valid_filter_keys)}." + ) + @classmethod - def _get_tools(cls: type[_T], agent: "AzureAIAgent", kernel: "Kernel") -> list[dict[str, Any] | ToolDefinition]: - """Get the tools for the agent.""" - tools: list[Any] = list(agent.definition.tools) - funcs = kernel.get_full_list_of_function_metadata() - cls._validate_function_tools_registered(tools, funcs) + def _get_tools( + cls: type[_T], + agent: "AzureAIAgent", + kernel: "Kernel", + tools_override: list[ToolDefinition] | None = None, + function_choice_behavior: FunctionChoiceBehavior | None = None, + ) -> list[dict[str, Any] | ToolDefinition]: + """Get the tools for the agent. + + Args: + agent: The agent instance. + kernel: The kernel to use for function metadata. + tools_override: When provided, overrides agent.definition.tools (SDK-level tools only). + function_choice_behavior: When provided, filters which kernel functions are included. + """ + tools: list[Any] = list(tools_override) if tools_override is not None else list(agent.definition.tools) + + # Always validate against the full kernel function list to catch truly + # unregistered functions, regardless of FCB filtering. + all_funcs = kernel.get_full_list_of_function_metadata() + cls._validate_function_tools_registered(tools, all_funcs) + + # Determine which kernel functions to advertise based on function_choice_behavior + if function_choice_behavior is not None and not function_choice_behavior.enable_kernel_functions: + funcs: list[KernelFunctionMetadata] = [] + elif function_choice_behavior is not None and function_choice_behavior.filters: + funcs = kernel.get_list_of_function_metadata(function_choice_behavior.filters) + else: + funcs = all_funcs dict_defs = [kernel_function_metadata_to_function_call_format(f) for f in funcs] deduped_defs = cls._deduplicate_tools(tools, dict_defs) tools.extend(deduped_defs) @@ -1071,6 +1162,7 @@ class AgentThreadActions: fccs: list["FunctionCallContent"], chat_history: "ChatHistory", arguments: KernelArguments, + function_choice_behavior: FunctionChoiceBehavior | None = None, ) -> list["AutoFunctionInvocationContext | None"]: """Invoke the function calls.""" return await asyncio.gather( @@ -1079,6 +1171,7 @@ class AgentThreadActions: function_call=function_call, chat_history=chat_history, arguments=arguments, + function_behavior=function_choice_behavior, ) for function_call in fccs ], @@ -1111,6 +1204,7 @@ class AgentThreadActions: run: ThreadRun, function_steps: dict[str, "FunctionCallContent"], arguments: KernelArguments, + function_choice_behavior: FunctionChoiceBehavior | None = None, **kwargs: Any, ) -> FunctionActionResult | None: """Handle the requires action event for a streaming run.""" @@ -1121,7 +1215,11 @@ class AgentThreadActions: chat_history = ChatHistory() if kwargs.get("chat_history") is None else kwargs["chat_history"] results = await cls._invoke_function_calls( - kernel=kernel, fccs=fccs, chat_history=chat_history, arguments=arguments + kernel=kernel, + fccs=fccs, + chat_history=chat_history, + arguments=arguments, + function_choice_behavior=function_choice_behavior, ) function_result_streaming_content = merge_streaming_function_results( diff --git a/python/semantic_kernel/agents/azure_ai/azure_ai_agent.py b/python/semantic_kernel/agents/azure_ai/azure_ai_agent.py index 44af0f22fe..d181ae5738 100644 --- a/python/semantic_kernel/agents/azure_ai/azure_ai_agent.py +++ b/python/semantic_kernel/agents/azure_ai/azure_ai_agent.py @@ -41,6 +41,7 @@ from semantic_kernel.agents.azure_ai.azure_ai_channel import AzureAIChannel from semantic_kernel.agents.channels.agent_channel import AgentChannel from semantic_kernel.agents.open_ai.run_polling_options import RunPollingOptions from semantic_kernel.connectors.ai.function_calling_utils import kernel_function_metadata_to_function_call_format +from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior from semantic_kernel.contents.chat_message_content import ChatMessageContent from semantic_kernel.contents.utils.author_role import AuthorRole from semantic_kernel.exceptions.agent_exceptions import ( @@ -647,6 +648,7 @@ class AzureAIAgent(DeclarativeSpecMixin, Agent): parallel_tool_calls: bool | None = None, metadata: dict[str, str] | None = None, polling_options: RunPollingOptions | None = None, + function_choice_behavior: FunctionChoiceBehavior | None = None, **kwargs: Any, ) -> AgentResponseItem[ChatMessageContent]: """Get a response from the agent on a thread. @@ -671,6 +673,8 @@ class AzureAIAgent(DeclarativeSpecMixin, Agent): parallel_tool_calls: Whether to allow parallel tool calls. metadata: Metadata for the agent. polling_options: The polling options for the agent. + function_choice_behavior: The function choice behavior to control which kernel + functions are available. Only Auto is supported; other types will raise an error. **kwargs: Additional keyword arguments. Returns: @@ -716,6 +720,7 @@ class AzureAIAgent(DeclarativeSpecMixin, Agent): thread_id=thread.id, kernel=kernel, arguments=arguments, + function_choice_behavior=function_choice_behavior, **run_level_params, # type: ignore ): if is_visible and response.metadata.get("code") is not True: @@ -752,6 +757,7 @@ class AzureAIAgent(DeclarativeSpecMixin, Agent): parallel_tool_calls: bool | None = None, metadata: dict[str, str] | None = None, polling_options: RunPollingOptions | None = None, + function_choice_behavior: FunctionChoiceBehavior | None = None, **kwargs: Any, ) -> AsyncIterable[AgentResponseItem[ChatMessageContent]]: """Invoke the agent on the specified thread. @@ -777,6 +783,8 @@ class AzureAIAgent(DeclarativeSpecMixin, Agent): parallel_tool_calls: Whether to allow parallel tool calls. polling_options: The polling options for the agent. metadata: Metadata for the agent. + function_choice_behavior: The function choice behavior to control which kernel + functions are available. Only Auto is supported; other types will raise an error. **kwargs: Additional keyword arguments. Yields: @@ -821,6 +829,7 @@ class AzureAIAgent(DeclarativeSpecMixin, Agent): thread_id=thread.id, kernel=kernel, arguments=arguments, + function_choice_behavior=function_choice_behavior, **run_level_params, # type: ignore ): message.metadata["thread_id"] = thread.id @@ -856,6 +865,7 @@ class AzureAIAgent(DeclarativeSpecMixin, Agent): response_format: AgentsApiResponseFormatOption | None = None, parallel_tool_calls: bool | None = None, metadata: dict[str, str] | None = None, + function_choice_behavior: FunctionChoiceBehavior | None = None, **kwargs: Any, ) -> AsyncIterable[AgentResponseItem["StreamingChatMessageContent"]]: """Invoke the agent on the specified thread with a stream of messages. @@ -881,6 +891,8 @@ class AzureAIAgent(DeclarativeSpecMixin, Agent): response_format: Response format for the agent. parallel_tool_calls: Whether to allow parallel tool calls. metadata: Metadata for the agent. + function_choice_behavior: The function choice behavior to control which kernel + functions are available. Only Auto is supported; other types will raise an error. **kwargs: Additional keyword arguments. Yields: @@ -928,6 +940,7 @@ class AzureAIAgent(DeclarativeSpecMixin, Agent): output_messages=collected_messages, kernel=kernel, arguments=arguments, + function_choice_behavior=function_choice_behavior, **run_level_params, # type: ignore ): # Before yielding the current streamed message, emit any new full messages first diff --git a/python/semantic_kernel/agents/open_ai/assistant_thread_actions.py b/python/semantic_kernel/agents/open_ai/assistant_thread_actions.py index 3a6679df64..6941d8f45c 100644 --- a/python/semantic_kernel/agents/open_ai/assistant_thread_actions.py +++ b/python/semantic_kernel/agents/open_ai/assistant_thread_actions.py @@ -34,6 +34,8 @@ from semantic_kernel.agents.open_ai.assistant_content_generation import ( from semantic_kernel.agents.open_ai.function_action_result import FunctionActionResult from semantic_kernel.agents.open_ai.run_polling_options import RunPollingOptions from semantic_kernel.connectors.ai.function_calling_utils import kernel_function_metadata_to_function_call_format +from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior +from semantic_kernel.connectors.ai.function_choice_type import FunctionChoiceType from semantic_kernel.contents.file_reference_content import FileReferenceContent from semantic_kernel.contents.function_call_content import FunctionCallContent from semantic_kernel.contents.streaming_file_reference_content import StreamingFileReferenceContent @@ -154,6 +156,7 @@ class AssistantThreadActions: top_p: float | None = None, truncation_strategy: "TruncationStrategy | None" = None, polling_options: RunPollingOptions | None = None, + function_choice_behavior: FunctionChoiceBehavior | None = None, **kwargs: Any, ) -> AsyncIterable[tuple[bool, "ChatMessageContent"]]: """Invoke the assistant. @@ -173,12 +176,17 @@ class AssistantThreadActions: parallel_tool_calls: The parallel tool calls. reasoning_effort: The reasoning effort. response_format: The response format. - tools: The tools. + tools: The SDK-level tools (e.g. CodeInterpreter, FileSearch). When provided, + overrides the tools from the agent definition. Does not affect kernel function availability; + use function_choice_behavior for that. temperature: The temperature. top_p: The top p. truncation_strategy: The truncation strategy. polling_options: The polling options defined at the run-level. These will override the agent-level polling options. + function_choice_behavior: Controls which kernel functions are allowed to execute during this run. + Use FunctionChoiceBehavior.Auto(filters={"included_functions": [...]}) to restrict to specific + functions. Only Auto is supported; other types will raise an error. kwargs: Additional keyword arguments. Returns: @@ -187,7 +195,11 @@ class AssistantThreadActions: arguments = KernelArguments() if arguments is None else KernelArguments(**arguments, **kwargs) kernel = kernel or agent.kernel - tools = cls._get_tools(agent=agent, kernel=kernel) # type: ignore + cls._validate_function_choice_behavior(function_choice_behavior) + + tools = cls._get_tools( + agent=agent, kernel=kernel, tools_override=tools, function_choice_behavior=function_choice_behavior + ) # type: ignore base_instructions = await agent.format_instructions(kernel=kernel, arguments=arguments) @@ -260,7 +272,11 @@ class AssistantThreadActions: chat_history = ChatHistory() _ = await cls._invoke_function_calls( - kernel=kernel, fccs=fccs, chat_history=chat_history, arguments=arguments + kernel=kernel, + fccs=fccs, + chat_history=chat_history, + arguments=arguments, + function_choice_behavior=function_choice_behavior, ) tool_outputs = cls._format_tool_outputs(fccs, chat_history) @@ -374,6 +390,7 @@ class AssistantThreadActions: temperature: float | None = None, top_p: float | None = None, truncation_strategy: "TruncationStrategy | None" = None, + function_choice_behavior: FunctionChoiceBehavior | None = None, **kwargs: Any, ) -> AsyncIterable["StreamingChatMessageContent"]: """Invoke the assistant. @@ -396,10 +413,15 @@ class AssistantThreadActions: parallel_tool_calls: The parallel tool calls. reasoning_effort: The reasoning effort. response_format: The response format. - tools: The tools. + tools: The SDK-level tools (e.g. CodeInterpreter, FileSearch). When provided, + overrides the tools from the agent definition. Does not affect kernel function availability; + use function_choice_behavior for that. temperature: The temperature. top_p: The top p. truncation_strategy: The truncation strategy. + function_choice_behavior: Controls which kernel functions are allowed to execute during this run. + Use FunctionChoiceBehavior.Auto(filters={"included_functions": [...]}) to restrict to specific + functions. Only Auto is supported; other types will raise an error. kwargs: Additional keyword arguments. Returns: @@ -408,7 +430,11 @@ class AssistantThreadActions: arguments = KernelArguments() if arguments is None else KernelArguments(**arguments, **kwargs) kernel = kernel or agent.kernel - tools = cls._get_tools(agent=agent, kernel=kernel) # type: ignore + cls._validate_function_choice_behavior(function_choice_behavior) + + tools = cls._get_tools( + agent=agent, kernel=kernel, tools_override=tools, function_choice_behavior=function_choice_behavior + ) # type: ignore base_instructions = await agent.format_instructions(kernel=kernel, arguments=arguments) @@ -496,6 +522,7 @@ class AssistantThreadActions: run, function_steps, arguments, + function_choice_behavior=function_choice_behavior, ) if action_result is None: raise AgentInvokeException( @@ -553,6 +580,7 @@ class AssistantThreadActions: run: "Run", function_steps: dict[str, "FunctionCallContent"], arguments: KernelArguments, + function_choice_behavior: FunctionChoiceBehavior | None = None, **kwargs: Any, ) -> FunctionActionResult | None: """Handle the requires action event for a streaming run.""" @@ -563,7 +591,11 @@ class AssistantThreadActions: chat_history = ChatHistory() if kwargs.get("chat_history") is None else kwargs["chat_history"] results = await cls._invoke_function_calls( - kernel=kernel, fccs=fccs, chat_history=chat_history, arguments=arguments + kernel=kernel, + fccs=fccs, + chat_history=chat_history, + arguments=arguments, + function_choice_behavior=function_choice_behavior, ) function_result_streaming_content = merge_streaming_function_results( @@ -658,6 +690,7 @@ class AssistantThreadActions: fccs: list["FunctionCallContent"], chat_history: "ChatHistory", arguments: KernelArguments, + function_choice_behavior: FunctionChoiceBehavior | None = None, ) -> list["AutoFunctionInvocationContext | None"]: """Invoke the function calls.""" return await asyncio.gather( @@ -666,6 +699,7 @@ class AssistantThreadActions: function_call=function_call, chat_history=chat_history, arguments=arguments, + function_behavior=function_choice_behavior, ) for function_call in fccs ], @@ -837,22 +871,80 @@ class AssistantThreadActions: if tool_definition := cls.tool_metadata.get(tool): yield from tool_definition + @staticmethod + def _validate_function_choice_behavior( + function_choice_behavior: FunctionChoiceBehavior | None, + ) -> None: + """Validate the function choice behavior is compatible with agent invocations.""" + if function_choice_behavior is None: + return + if function_choice_behavior.type_ != FunctionChoiceType.AUTO: + raise AgentInvokeException( + f"FunctionChoiceBehavior with type '{function_choice_behavior.type_}' is not supported for agent " + "invocations. Use FunctionChoiceBehavior.Auto(filters=...) to control which kernel functions " + "are available." + ) + if not function_choice_behavior.auto_invoke_kernel_functions: + raise AgentInvokeException( + "FunctionChoiceBehavior.Auto(auto_invoke=False) is not supported for agent invocations. " + "The agent run loop manages tool invocation; disabling auto_invoke is not compatible." + ) + valid_filter_keys: set[str] = { + "excluded_plugins", + "included_plugins", + "excluded_functions", + "included_functions", + } + if function_choice_behavior.filters is not None: + if not function_choice_behavior.filters: + raise AgentInvokeException( + "FunctionChoiceBehavior filters must not be empty. Provide at least one filter key " + f"from {sorted(valid_filter_keys)}, or omit filters entirely to include all " + "kernel functions." + ) + unknown_keys = {str(k) for k in function_choice_behavior.filters} - valid_filter_keys + if unknown_keys: + raise AgentInvokeException( + f"Unknown filter key(s): {sorted(unknown_keys)}. " + f"Valid filter keys are: {sorted(valid_filter_keys)}." + ) + @classmethod - def _get_tools(cls: type[_T], agent: "OpenAIAssistantAgent", kernel: "Kernel") -> list[dict[str, str]]: + def _get_tools( + cls: type[_T], + agent: "OpenAIAssistantAgent", + kernel: "Kernel", + tools_override: "list[AssistantToolParam] | None" = None, + function_choice_behavior: FunctionChoiceBehavior | None = None, + ) -> list[dict[str, str]]: """Get the list of tools for the assistant. + Args: + agent: The assistant agent. + kernel: The kernel to use for function metadata. + tools_override: When provided, overrides agent.definition.tools (SDK-level tools only). + function_choice_behavior: When provided, filters which kernel functions are included. + Returns: The list of tools. """ tools: list[Any] = [] - for tool in agent.definition.tools: + source_tools = tools_override if tools_override is not None else agent.definition.tools + for tool in source_tools: if isinstance(tool, CodeInterpreterTool): tools.append({"type": "code_interpreter"}) elif isinstance(tool, FileSearchTool): tools.append({"type": "file_search"}) - funcs = agent.kernel.get_full_list_of_function_metadata() + # Determine kernel function metadata based on function_choice_behavior + if function_choice_behavior is not None and not function_choice_behavior.enable_kernel_functions: + funcs = [] + elif function_choice_behavior is not None and function_choice_behavior.filters: + funcs = kernel.get_list_of_function_metadata(function_choice_behavior.filters) + else: + funcs = kernel.get_full_list_of_function_metadata() + tools.extend([kernel_function_metadata_to_function_call_format(f) for f in funcs]) return tools diff --git a/python/semantic_kernel/agents/open_ai/openai_assistant_agent.py b/python/semantic_kernel/agents/open_ai/openai_assistant_agent.py index a1daaa75f5..123f19aaf4 100644 --- a/python/semantic_kernel/agents/open_ai/openai_assistant_agent.py +++ b/python/semantic_kernel/agents/open_ai/openai_assistant_agent.py @@ -35,6 +35,7 @@ from semantic_kernel.agents.channels.agent_channel import AgentChannel from semantic_kernel.agents.channels.open_ai_assistant_channel import OpenAIAssistantChannel from semantic_kernel.agents.open_ai.assistant_thread_actions import AssistantThreadActions from semantic_kernel.agents.open_ai.run_polling_options import RunPollingOptions +from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior from semantic_kernel.connectors.ai.open_ai.settings.open_ai_settings import OpenAISettings from semantic_kernel.connectors.utils.structured_output_schema import generate_structured_output_response_format_schema from semantic_kernel.contents.chat_message_content import ChatMessageContent @@ -758,6 +759,7 @@ class OpenAIAssistantAgent(DeclarativeSpecMixin, Agent): top_p: float | None = None, truncation_strategy: "TruncationStrategy | None" = None, polling_options: RunPollingOptions | None = None, + function_choice_behavior: "FunctionChoiceBehavior | None" = None, **kwargs: Any, ) -> AgentResponseItem[ChatMessageContent]: """Get a response from the agent on a thread. @@ -783,6 +785,8 @@ class OpenAIAssistantAgent(DeclarativeSpecMixin, Agent): top_p: The top p. truncation_strategy: The truncation strategy. polling_options: The polling options at the run-level. + function_choice_behavior: The function choice behavior to control which kernel + functions are available. Only Auto is supported; other types will raise an error. kwargs: Additional keyword arguments. Returns: @@ -829,6 +833,7 @@ class OpenAIAssistantAgent(DeclarativeSpecMixin, Agent): thread_id=thread.id, kernel=kernel, arguments=arguments, + function_choice_behavior=function_choice_behavior, **run_level_params, # type: ignore ): if is_visible and response.metadata.get("code") is not True: @@ -866,6 +871,7 @@ class OpenAIAssistantAgent(DeclarativeSpecMixin, Agent): top_p: float | None = None, truncation_strategy: "TruncationStrategy | None" = None, polling_options: RunPollingOptions | None = None, + function_choice_behavior: "FunctionChoiceBehavior | None" = None, **kwargs: Any, ) -> AsyncIterable[AgentResponseItem[ChatMessageContent]]: """Invoke the agent. @@ -892,6 +898,8 @@ class OpenAIAssistantAgent(DeclarativeSpecMixin, Agent): top_p: The top p. truncation_strategy: The truncation strategy. polling_options: The polling options at the run-level. + function_choice_behavior: The function choice behavior to control which kernel + functions are available. Only Auto is supported; other types will raise an error. kwargs: Additional keyword arguments. Yields: @@ -937,6 +945,7 @@ class OpenAIAssistantAgent(DeclarativeSpecMixin, Agent): thread_id=thread.id, kernel=kernel, arguments=arguments, + function_choice_behavior=function_choice_behavior, **run_level_params, # type: ignore ): message.metadata["thread_id"] = thread.id @@ -973,6 +982,7 @@ class OpenAIAssistantAgent(DeclarativeSpecMixin, Agent): temperature: float | None = None, top_p: float | None = None, truncation_strategy: "TruncationStrategy | None" = None, + function_choice_behavior: "FunctionChoiceBehavior | None" = None, **kwargs: Any, ) -> AsyncIterable[AgentResponseItem[StreamingChatMessageContent]]: """Invoke the agent. @@ -999,6 +1009,8 @@ class OpenAIAssistantAgent(DeclarativeSpecMixin, Agent): temperature: The temperature. top_p: The top p. truncation_strategy: The truncation strategy. + function_choice_behavior: The function choice behavior to control which kernel + functions are available. Only Auto is supported; other types will raise an error. kwargs: Additional keyword arguments. Yields: @@ -1047,6 +1059,7 @@ class OpenAIAssistantAgent(DeclarativeSpecMixin, Agent): output_messages=collected_messages, kernel=kernel, arguments=arguments, + function_choice_behavior=function_choice_behavior, **run_level_params, # type: ignore ): # Before yielding the current streamed message, emit any new full messages first diff --git a/python/tests/unit/agents/azure_ai_agent/test_agent_thread_actions.py b/python/tests/unit/agents/azure_ai_agent/test_agent_thread_actions.py index 000491d090..0916183e14 100644 --- a/python/tests/unit/agents/azure_ai_agent/test_agent_thread_actions.py +++ b/python/tests/unit/agents/azure_ai_agent/test_agent_thread_actions.py @@ -3,6 +3,7 @@ from datetime import datetime, timezone from unittest.mock import AsyncMock, MagicMock, patch +import pytest from azure.ai.agents.models import ( MessageTextContent, MessageTextDetails, @@ -26,9 +27,14 @@ from pytest import fixture from semantic_kernel.agents.azure_ai.agent_thread_actions import AgentThreadActions from semantic_kernel.agents.azure_ai.azure_ai_agent import AzureAIAgent +from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior from semantic_kernel.contents import FunctionCallContent, FunctionResultContent, TextContent from semantic_kernel.contents.chat_message_content import ChatMessageContent from semantic_kernel.contents.utils.author_role import AuthorRole +from semantic_kernel.exceptions.agent_exceptions import AgentInvokeException +from semantic_kernel.functions.kernel_arguments import KernelArguments +from semantic_kernel.functions.kernel_function_decorator import kernel_function +from semantic_kernel.functions.kernel_plugin import KernelPlugin from semantic_kernel.kernel import Kernel @@ -354,3 +360,303 @@ async def test_agent_thread_actions_invoke_stream(ai_project_client, ai_agent_de collected_messages.append(content) assert isinstance(content, ChatMessageContent) assert content.metadata.get("message_id") == "msg_1" + + +# region Security tests for tools override and function_choice_behavior + + +async def test_validate_function_choice_behavior_rejects_required(): + """Required FCB is not supported for agent invocations.""" + with pytest.raises(AgentInvokeException, match="not supported"): + AgentThreadActions._validate_function_choice_behavior(FunctionChoiceBehavior.Required()) + + +async def test_validate_function_choice_behavior_accepts_auto(): + """Auto FCB should be accepted without error.""" + AgentThreadActions._validate_function_choice_behavior(FunctionChoiceBehavior.Auto()) + + +async def test_validate_function_choice_behavior_rejects_none_invoke(): + """NoneInvoke FCB is not supported for agent invocations.""" + with pytest.raises(AgentInvokeException, match="not supported"): + AgentThreadActions._validate_function_choice_behavior(FunctionChoiceBehavior.NoneInvoke()) + + +async def test_validate_function_choice_behavior_accepts_none(): + """None (no FCB) should be accepted.""" + AgentThreadActions._validate_function_choice_behavior(None) + + +async def test_validate_function_choice_behavior_rejects_auto_invoke_false(): + """Auto with auto_invoke=False is not supported for agent invocations.""" + with pytest.raises(AgentInvokeException, match="auto_invoke"): + AgentThreadActions._validate_function_choice_behavior(FunctionChoiceBehavior.Auto(auto_invoke=False)) + + +async def test_validate_function_choice_behavior_rejects_empty_filters(): + """Empty filters dict should be rejected.""" + fcb = FunctionChoiceBehavior.Auto() + fcb.filters = {} + with pytest.raises(AgentInvokeException, match="must not be empty"): + AgentThreadActions._validate_function_choice_behavior(fcb) + + +async def test_validate_function_choice_behavior_rejects_unknown_filter_keys(): + """Unknown filter keys should be rejected.""" + fcb = FunctionChoiceBehavior.Auto() + # Bypass Pydantic validation to simulate a mistyped key reaching the validator + object.__setattr__(fcb, "filters", {"include_functions": ["foo"]}) + with pytest.raises(AgentInvokeException, match="Unknown filter key"): + AgentThreadActions._validate_function_choice_behavior(fcb) + + +async def test_validate_function_choice_behavior_accepts_valid_filters(): + """Valid filter keys should be accepted.""" + AgentThreadActions._validate_function_choice_behavior( + FunctionChoiceBehavior.Auto(filters={"included_functions": ["plugin-func"]}) + ) + + +async def test_get_tools_with_tools_override(ai_project_client, ai_agent_definition): + """When tools_override is provided, it should replace agent.definition.tools.""" + from azure.ai.agents.models import CodeInterpreterToolDefinition + + agent = AzureAIAgent(client=ai_project_client, definition=ai_agent_definition) + kernel = MagicMock(spec=Kernel) + kernel.get_full_list_of_function_metadata.return_value = [] + + override_tool = CodeInterpreterToolDefinition() + tools = AgentThreadActions._get_tools(agent=agent, kernel=kernel, tools_override=[override_tool]) + # Should contain the override tool, not agent.definition.tools + assert any( + (isinstance(t, CodeInterpreterToolDefinition) or (isinstance(t, dict) and t.get("type") == "code_interpreter")) + for t in tools + ) + + +async def test_get_tools_with_fcb_filters(ai_project_client, ai_agent_definition): + """When function_choice_behavior has filters, only matching functions should be included.""" + agent = AzureAIAgent(client=ai_project_client, definition=ai_agent_definition) + kernel = MagicMock(spec=Kernel) + + # Simulate filtered metadata + mock_metadata = MagicMock() + mock_metadata.fully_qualified_name = "Plugin-AllowedFunc" + mock_metadata.name = "AllowedFunc" + mock_metadata.plugin_name = "Plugin" + mock_metadata.description = "An allowed function" + mock_metadata.parameters = [] + mock_metadata.is_prompt = False + mock_metadata.return_parameter = MagicMock() + mock_metadata.return_parameter.description = "" + mock_metadata.return_parameter.type_ = "str" + mock_metadata.additional_properties = {} + + kernel.get_list_of_function_metadata.return_value = [mock_metadata] + kernel.get_full_list_of_function_metadata.return_value = [] + + fcb = FunctionChoiceBehavior.Auto(filters={"included_functions": ["Plugin-AllowedFunc"]}) + AgentThreadActions._get_tools(agent=agent, kernel=kernel, function_choice_behavior=fcb) + # Should have called get_list_of_function_metadata with the filters + kernel.get_list_of_function_metadata.assert_called_once_with(fcb.filters) + + +async def test_get_tools_with_fcb_disable_kernel_functions(ai_project_client, ai_agent_definition): + """When enable_kernel_functions=False, no kernel functions should be included.""" + agent = AzureAIAgent(client=ai_project_client, definition=ai_agent_definition) + kernel = MagicMock(spec=Kernel) + + fcb = FunctionChoiceBehavior.Auto(enable_kernel_functions=False) + AgentThreadActions._get_tools(agent=agent, kernel=kernel, function_choice_behavior=fcb) + # Full list is called for validation, but filtered list should not be called + kernel.get_full_list_of_function_metadata.assert_called_once() + kernel.get_list_of_function_metadata.assert_not_called() + + +async def test_invoke_function_calls_passes_function_behavior(): + """_invoke_function_calls should pass function_behavior to kernel.invoke_function_call.""" + mock_kernel = AsyncMock(spec=Kernel) + mock_kernel.invoke_function_call.return_value = None + + fcc = FunctionCallContent(name="Plugin-Func", arguments={}, id="call1") + from semantic_kernel.contents.chat_history import ChatHistory + + chat_history = ChatHistory() + fcb = FunctionChoiceBehavior.Auto(filters={"included_functions": ["Plugin-Func"]}) + + await AgentThreadActions._invoke_function_calls( + kernel=mock_kernel, + fccs=[fcc], + chat_history=chat_history, + arguments=KernelArguments(), + function_choice_behavior=fcb, + ) + + mock_kernel.invoke_function_call.assert_awaited_once() + call_kwargs = mock_kernel.invoke_function_call.call_args + assert call_kwargs.kwargs.get("function_behavior") is fcb + + +async def test_invoke_function_calls_passes_disabled_kernel_functions(): + """_invoke_function_calls should pass enable_kernel_functions=False FCB to kernel.""" + mock_kernel = AsyncMock(spec=Kernel) + mock_kernel.invoke_function_call.return_value = None + + fcc = FunctionCallContent(name="Plugin-Func", arguments={}, id="call1") + from semantic_kernel.contents.chat_history import ChatHistory + + chat_history = ChatHistory() + fcb = FunctionChoiceBehavior.Auto(enable_kernel_functions=False) + + await AgentThreadActions._invoke_function_calls( + kernel=mock_kernel, + fccs=[fcc], + chat_history=chat_history, + arguments=KernelArguments(), + function_choice_behavior=fcb, + ) + + mock_kernel.invoke_function_call.assert_awaited_once() + call_kwargs = mock_kernel.invoke_function_call.call_args + passed_behavior = call_kwargs.kwargs.get("function_behavior") + assert passed_behavior is fcb + assert not passed_behavior.enable_kernel_functions + + +async def test_invoke_function_calls_blocks_disallowed_function(): + """A real Kernel should block a function call not in the FCB allowlist. + + This verifies that the enforcement in kernel.invoke_function_call actually + rejects a disallowed function name when filters are provided, rather than + only asserting that the kwarg is forwarded. + """ + from semantic_kernel.contents.chat_history import ChatHistory + from semantic_kernel.functions.kernel_function_from_method import KernelFunctionFromMethod + + @kernel_function + def allowed_func() -> str: + return "allowed" + + @kernel_function + def disallowed_func() -> str: + return "disallowed" + + kernel = Kernel() + kernel.add_plugin( + KernelPlugin( + name="Plugin", + functions=[ + KernelFunctionFromMethod(method=allowed_func, plugin_name="Plugin"), + KernelFunctionFromMethod(method=disallowed_func, plugin_name="Plugin"), + ], + ) + ) + + fcb = FunctionChoiceBehavior.Auto(filters={"included_functions": ["Plugin-allowed_func"]}) + + # Call a function NOT in the allowlist + fcc = FunctionCallContent( + name="Plugin-disallowed_func", + plugin_name="Plugin", + function_name="disallowed_func", + arguments={}, + id="call1", + ) + chat_history = ChatHistory() + + result = await kernel.invoke_function_call( + function_call=fcc, + chat_history=chat_history, + function_behavior=fcb, + ) + # invoke_function_call catches the FunctionExecutionException and returns None, + # adding an error message to chat_history instead of raising. + assert result is None + assert len(chat_history.messages) == 1 + result_item = chat_history.messages[0].items[0] + assert "not part of the provided tools" in str(result_item.result) + + +async def test_invoke_function_calls_allows_permitted_function(): + """A real Kernel should allow a function call that IS in the FCB allowlist.""" + from semantic_kernel.contents.chat_history import ChatHistory + from semantic_kernel.functions.kernel_function_from_method import KernelFunctionFromMethod + + @kernel_function + def allowed_func() -> str: + return "ok" + + @kernel_function + def other_func() -> str: + return "other" + + kernel = Kernel() + kernel.add_plugin( + KernelPlugin( + name="Plugin", + functions=[ + KernelFunctionFromMethod(method=allowed_func, plugin_name="Plugin"), + KernelFunctionFromMethod(method=other_func, plugin_name="Plugin"), + ], + ) + ) + + fcb = FunctionChoiceBehavior.Auto(filters={"included_functions": ["Plugin-allowed_func"]}) + + fcc = FunctionCallContent( + name="Plugin-allowed_func", + plugin_name="Plugin", + function_name="allowed_func", + arguments={}, + id="call1", + ) + chat_history = ChatHistory() + + await kernel.invoke_function_call( + function_call=fcc, + chat_history=chat_history, + function_behavior=fcb, + ) + # Should succeed — the function result should be in chat_history + assert len(chat_history.messages) == 1 + result_item = chat_history.messages[0].items[0] + assert "ok" in str(result_item.result) + + +async def test_invoke_raises_for_non_auto_fcb(ai_project_client, ai_agent_definition): + """Calling AgentThreadActions.invoke() with a non-Auto FCB should raise before any API call.""" + agent = AzureAIAgent(client=ai_project_client, definition=ai_agent_definition) + agent.client.agents = AsyncMock() + + with pytest.raises(AgentInvokeException, match="not supported"): + async for _ in AgentThreadActions.invoke( + agent=agent, + thread_id="thread123", + kernel=Kernel(), + function_choice_behavior=FunctionChoiceBehavior.Required(), + ): + pass + + # No API calls should have been made + agent.client.agents.runs.create.assert_not_awaited() + + +async def test_invoke_stream_raises_for_non_auto_fcb(ai_project_client, ai_agent_definition): + """Calling AgentThreadActions.invoke_stream() with a non-Auto FCB should raise before any API call.""" + agent = AzureAIAgent(client=ai_project_client, definition=ai_agent_definition) + agent.client.agents = AsyncMock() + + with pytest.raises(AgentInvokeException, match="not supported"): + async for _ in AgentThreadActions.invoke_stream( + agent=agent, + thread_id="thread123", + kernel=Kernel(), + function_choice_behavior=FunctionChoiceBehavior.NoneInvoke(), + ): + pass + + # No API calls should have been made + agent.client.agents.create_stream.assert_not_called() + + +# endregion diff --git a/python/tests/unit/agents/azure_ai_agent/test_azure_ai_agent.py b/python/tests/unit/agents/azure_ai_agent/test_azure_ai_agent.py index b5dc1178b6..dff8f210f0 100644 --- a/python/tests/unit/agents/azure_ai_agent/test_azure_ai_agent.py +++ b/python/tests/unit/agents/azure_ai_agent/test_azure_ai_agent.py @@ -9,6 +9,7 @@ from azure.core.credentials_async import AsyncTokenCredential from semantic_kernel.agents.agent import AgentResponseItem from semantic_kernel.agents.azure_ai.azure_ai_agent import AzureAIAgent, AzureAIAgentThread from semantic_kernel.agents.channels.agent_channel import AgentChannel +from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior from semantic_kernel.contents.chat_history import ChatHistory from semantic_kernel.contents.chat_message_content import ChatMessageContent from semantic_kernel.contents.function_call_content import FunctionCallContent @@ -398,3 +399,80 @@ def test_create_client_raises_if_no_endpoint(): assert "Azure AI endpoint" in str(e) else: assert False, "Expected AgentInitializationException to be raised" + + +async def test_azure_ai_agent_get_response_passes_function_choice_behavior(ai_project_client, ai_agent_definition): + agent = AzureAIAgent(client=ai_project_client, definition=ai_agent_definition) + thread = AsyncMock(spec=AzureAIAgentThread) + fcb = FunctionChoiceBehavior.Auto() + captured_kwargs = {} + + async def fake_invoke(*args, **kwargs): + captured_kwargs.update(kwargs) + yield True, ChatMessageContent(role=AuthorRole.ASSISTANT, content="content") + + with patch( + "semantic_kernel.agents.azure_ai.agent_thread_actions.AgentThreadActions.invoke", + side_effect=fake_invoke, + ): + await agent.get_response(messages="message", thread=thread, function_choice_behavior=fcb) + + assert captured_kwargs.get("function_choice_behavior") is fcb + + +async def test_azure_ai_agent_invoke_passes_function_choice_behavior(ai_project_client, ai_agent_definition): + agent = AzureAIAgent(client=ai_project_client, definition=ai_agent_definition) + thread = AsyncMock(spec=AzureAIAgentThread) + fcb = FunctionChoiceBehavior.Auto() + captured_kwargs = {} + + async def fake_invoke(*args, **kwargs): + captured_kwargs.update(kwargs) + yield True, ChatMessageContent(role=AuthorRole.ASSISTANT, content="content") + + with patch( + "semantic_kernel.agents.azure_ai.agent_thread_actions.AgentThreadActions.invoke", + side_effect=fake_invoke, + ): + async for _ in agent.invoke(messages="message", thread=thread, function_choice_behavior=fcb): + pass + + assert captured_kwargs.get("function_choice_behavior") is fcb + + +async def test_azure_ai_agent_invoke_stream_passes_function_choice_behavior(ai_project_client, ai_agent_definition): + agent = AzureAIAgent(client=ai_project_client, definition=ai_agent_definition) + thread = AsyncMock(spec=AzureAIAgentThread) + fcb = FunctionChoiceBehavior.Auto() + captured_kwargs = {} + + async def fake_invoke(*args, **kwargs): + captured_kwargs.update(kwargs) + yield ChatMessageContent(role=AuthorRole.ASSISTANT, content="content") + + with patch( + "semantic_kernel.agents.azure_ai.agent_thread_actions.AgentThreadActions.invoke_stream", + side_effect=fake_invoke, + ): + async for _ in agent.invoke_stream(messages="message", thread=thread, function_choice_behavior=fcb): + pass + + assert captured_kwargs.get("function_choice_behavior") is fcb + + +async def test_azure_ai_agent_get_response_no_fcb_passes_none(ai_project_client, ai_agent_definition): + agent = AzureAIAgent(client=ai_project_client, definition=ai_agent_definition) + thread = AsyncMock(spec=AzureAIAgentThread) + captured_kwargs = {} + + async def fake_invoke(*args, **kwargs): + captured_kwargs.update(kwargs) + yield True, ChatMessageContent(role=AuthorRole.ASSISTANT, content="content") + + with patch( + "semantic_kernel.agents.azure_ai.agent_thread_actions.AgentThreadActions.invoke", + side_effect=fake_invoke, + ): + await agent.get_response(messages="message", thread=thread) + + assert captured_kwargs.get("function_choice_behavior") is None diff --git a/python/tests/unit/agents/openai_assistant/test_assistant_thread_actions.py b/python/tests/unit/agents/openai_assistant/test_assistant_thread_actions.py index 1bb688bb42..f4758dd13a 100644 --- a/python/tests/unit/agents/openai_assistant/test_assistant_thread_actions.py +++ b/python/tests/unit/agents/openai_assistant/test_assistant_thread_actions.py @@ -55,6 +55,7 @@ from semantic_kernel.agents.open_ai.assistant_thread_actions import AssistantThr from semantic_kernel.agents.open_ai.function_action_result import FunctionActionResult from semantic_kernel.agents.open_ai.openai_assistant_agent import OpenAIAssistantAgent from semantic_kernel.agents.open_ai.run_polling_options import RunPollingOptions +from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior from semantic_kernel.contents.chat_message_content import ChatMessageContent from semantic_kernel.contents.file_reference_content import FileReferenceContent from semantic_kernel.contents.function_call_content import FunctionCallContent @@ -853,3 +854,327 @@ async def test_handle_streaming_requires_action_returns_none(): dummy_args, ) assert result is None + + +# region Security tests for tools override and function_choice_behavior + + +async def test_validate_function_choice_behavior_rejects_required(): + """Required FCB is not supported for agent invocations.""" + with pytest.raises(AgentInvokeException, match="not supported"): + AssistantThreadActions._validate_function_choice_behavior(FunctionChoiceBehavior.Required()) + + +async def test_validate_function_choice_behavior_accepts_auto(): + """Auto FCB should be accepted without error.""" + AssistantThreadActions._validate_function_choice_behavior(FunctionChoiceBehavior.Auto()) + + +async def test_validate_function_choice_behavior_rejects_none_invoke(): + """NoneInvoke FCB is not supported for agent invocations.""" + with pytest.raises(AgentInvokeException, match="not supported"): + AssistantThreadActions._validate_function_choice_behavior(FunctionChoiceBehavior.NoneInvoke()) + + +async def test_validate_function_choice_behavior_accepts_none(): + """None (no FCB) should be accepted.""" + AssistantThreadActions._validate_function_choice_behavior(None) + + +async def test_validate_function_choice_behavior_rejects_auto_invoke_false(): + """Auto with auto_invoke=False is not supported for agent invocations.""" + with pytest.raises(AgentInvokeException, match="auto_invoke"): + AssistantThreadActions._validate_function_choice_behavior(FunctionChoiceBehavior.Auto(auto_invoke=False)) + + +async def test_validate_function_choice_behavior_rejects_empty_filters(): + """Empty filters dict should be rejected.""" + fcb = FunctionChoiceBehavior.Auto() + fcb.filters = {} + with pytest.raises(AgentInvokeException, match="must not be empty"): + AssistantThreadActions._validate_function_choice_behavior(fcb) + + +async def test_validate_function_choice_behavior_rejects_unknown_filter_keys(): + """Unknown filter keys should be rejected.""" + fcb = FunctionChoiceBehavior.Auto() + # Bypass Pydantic validation to simulate a mistyped key reaching the validator + object.__setattr__(fcb, "filters", {"include_functions": ["foo"]}) + with pytest.raises(AgentInvokeException, match="Unknown filter key"): + AssistantThreadActions._validate_function_choice_behavior(fcb) + + +async def test_validate_function_choice_behavior_accepts_valid_filters(): + """Valid filter keys should be accepted.""" + AssistantThreadActions._validate_function_choice_behavior( + FunctionChoiceBehavior.Auto(filters={"included_functions": ["plugin-func"]}) + ) + + +async def test_get_tools_with_tools_override(): + """When tools_override is provided, it should replace agent.definition.tools.""" + agent = MagicMock(spec=OpenAIAssistantAgent) + agent.definition = MagicMock() + agent.definition.tools = [CodeInterpreterTool(type="code_interpreter")] + agent.kernel = MagicMock(spec=Kernel) + + kernel = MagicMock(spec=Kernel) + kernel.get_full_list_of_function_metadata.return_value = [] + + # Override with file_search only + override_tools = [FileSearchTool(type="file_search")] + tools = AssistantThreadActions._get_tools(agent=agent, kernel=kernel, tools_override=override_tools) + # Should contain file_search from override, not code_interpreter from agent + tool_types = [t.get("type") if isinstance(t, dict) else None for t in tools] + assert "file_search" in tool_types + # Agent's code_interpreter should NOT be in the result + assert "code_interpreter" not in tool_types + + +async def test_get_tools_with_fcb_filters(): + """When function_choice_behavior has filters, only matching functions should be included.""" + agent = MagicMock(spec=OpenAIAssistantAgent) + agent.definition = MagicMock() + agent.definition.tools = [] + agent.kernel = MagicMock(spec=Kernel) + + kernel = MagicMock(spec=Kernel) + + mock_metadata = MagicMock() + mock_metadata.fully_qualified_name = "Plugin-AllowedFunc" + mock_metadata.name = "AllowedFunc" + mock_metadata.plugin_name = "Plugin" + mock_metadata.description = "An allowed function" + mock_metadata.parameters = [] + mock_metadata.is_prompt = False + mock_metadata.return_parameter = MagicMock() + mock_metadata.return_parameter.description = "" + mock_metadata.return_parameter.type_ = "str" + mock_metadata.additional_properties = {} + + kernel.get_list_of_function_metadata.return_value = [mock_metadata] + + fcb = FunctionChoiceBehavior.Auto(filters={"included_functions": ["Plugin-AllowedFunc"]}) + AssistantThreadActions._get_tools(agent=agent, kernel=kernel, function_choice_behavior=fcb) + kernel.get_list_of_function_metadata.assert_called_once_with(fcb.filters) + + +async def test_get_tools_with_fcb_disable_kernel_functions(): + """When enable_kernel_functions=False, no kernel functions should be included.""" + agent = MagicMock(spec=OpenAIAssistantAgent) + agent.definition = MagicMock() + agent.definition.tools = [] + agent.kernel = MagicMock(spec=Kernel) + + kernel = MagicMock(spec=Kernel) + + fcb = FunctionChoiceBehavior.Auto(enable_kernel_functions=False) + AssistantThreadActions._get_tools(agent=agent, kernel=kernel, function_choice_behavior=fcb) + kernel.get_full_list_of_function_metadata.assert_not_called() + kernel.get_list_of_function_metadata.assert_not_called() + + +async def test_invoke_function_calls_passes_function_behavior(): + """_invoke_function_calls should pass function_behavior to kernel.invoke_function_call.""" + mock_kernel = AsyncMock(spec=Kernel) + mock_kernel.invoke_function_call.return_value = None + + fcc = FunctionCallContent(name="Plugin-Func", arguments={}, id="call1") + from semantic_kernel.contents.chat_history import ChatHistory + + chat_history = ChatHistory() + fcb = FunctionChoiceBehavior.Auto(filters={"included_functions": ["Plugin-Func"]}) + + await AssistantThreadActions._invoke_function_calls( + kernel=mock_kernel, + fccs=[fcc], + chat_history=chat_history, + arguments=KernelArguments(), + function_choice_behavior=fcb, + ) + + mock_kernel.invoke_function_call.assert_awaited_once() + call_kwargs = mock_kernel.invoke_function_call.call_args + assert call_kwargs.kwargs.get("function_behavior") is fcb + + +async def test_invoke_function_calls_passes_disabled_kernel_functions(): + """_invoke_function_calls should pass enable_kernel_functions=False FCB to kernel.""" + mock_kernel = AsyncMock(spec=Kernel) + mock_kernel.invoke_function_call.return_value = None + + fcc = FunctionCallContent(name="Plugin-Func", arguments={}, id="call1") + from semantic_kernel.contents.chat_history import ChatHistory + + chat_history = ChatHistory() + fcb = FunctionChoiceBehavior.Auto(enable_kernel_functions=False) + + await AssistantThreadActions._invoke_function_calls( + kernel=mock_kernel, + fccs=[fcc], + chat_history=chat_history, + arguments=KernelArguments(), + function_choice_behavior=fcb, + ) + + mock_kernel.invoke_function_call.assert_awaited_once() + call_kwargs = mock_kernel.invoke_function_call.call_args + passed_behavior = call_kwargs.kwargs.get("function_behavior") + assert passed_behavior is fcb + assert not passed_behavior.enable_kernel_functions + + +async def test_get_tools_uses_passed_kernel_not_agent_kernel(): + """_get_tools should use the passed kernel parameter, not agent.kernel.""" + agent = MagicMock(spec=OpenAIAssistantAgent) + agent.definition = MagicMock() + agent.definition.tools = [] + agent.kernel = MagicMock(spec=Kernel) + agent.kernel.get_full_list_of_function_metadata.return_value = ["should_not_be_used"] + + kernel = MagicMock(spec=Kernel) + kernel.get_full_list_of_function_metadata.return_value = [] + + AssistantThreadActions._get_tools(agent=agent, kernel=kernel) + # Should call the passed kernel, not agent.kernel + kernel.get_full_list_of_function_metadata.assert_called_once() + agent.kernel.get_full_list_of_function_metadata.assert_not_called() + + +async def test_invoke_function_calls_blocks_disallowed_function(): + """A real Kernel should block a function call not in the FCB allowlist. + + This verifies that the enforcement in kernel.invoke_function_call actually + rejects a disallowed function name when filters are provided, rather than + only asserting that the kwarg is forwarded. + """ + from semantic_kernel.contents.chat_history import ChatHistory + from semantic_kernel.functions.kernel_function_from_method import KernelFunctionFromMethod + + @kernel_function + def allowed_func() -> str: + return "allowed" + + @kernel_function + def disallowed_func() -> str: + return "disallowed" + + kernel = Kernel() + kernel.add_plugin( + KernelPlugin( + name="Plugin", + functions=[ + KernelFunctionFromMethod(method=allowed_func, plugin_name="Plugin"), + KernelFunctionFromMethod(method=disallowed_func, plugin_name="Plugin"), + ], + ) + ) + + fcb = FunctionChoiceBehavior.Auto(filters={"included_functions": ["Plugin-allowed_func"]}) + + # Call a function NOT in the allowlist + fcc = FunctionCallContent( + name="Plugin-disallowed_func", + plugin_name="Plugin", + function_name="disallowed_func", + arguments={}, + id="call1", + ) + chat_history = ChatHistory() + + result = await kernel.invoke_function_call( + function_call=fcc, + chat_history=chat_history, + function_behavior=fcb, + ) + # invoke_function_call catches the FunctionExecutionException and returns None, + # adding an error message to chat_history instead of raising. + assert result is None + assert len(chat_history.messages) == 1 + result_item = chat_history.messages[0].items[0] + assert "not part of the provided tools" in str(result_item.result) + + +async def test_invoke_function_calls_allows_permitted_function(): + """A real Kernel should allow a function call that IS in the FCB allowlist.""" + from semantic_kernel.contents.chat_history import ChatHistory + from semantic_kernel.functions.kernel_function_from_method import KernelFunctionFromMethod + + @kernel_function + def allowed_func() -> str: + return "ok" + + @kernel_function + def other_func() -> str: + return "other" + + kernel = Kernel() + kernel.add_plugin( + KernelPlugin( + name="Plugin", + functions=[ + KernelFunctionFromMethod(method=allowed_func, plugin_name="Plugin"), + KernelFunctionFromMethod(method=other_func, plugin_name="Plugin"), + ], + ) + ) + + fcb = FunctionChoiceBehavior.Auto(filters={"included_functions": ["Plugin-allowed_func"]}) + + fcc = FunctionCallContent( + name="Plugin-allowed_func", + plugin_name="Plugin", + function_name="allowed_func", + arguments={}, + id="call1", + ) + chat_history = ChatHistory() + + await kernel.invoke_function_call( + function_call=fcc, + chat_history=chat_history, + function_behavior=fcb, + ) + # Should succeed — the function result should be in chat_history + assert len(chat_history.messages) == 1 + result_item = chat_history.messages[0].items[0] + assert "ok" in str(result_item.result) + + +async def test_invoke_raises_for_non_auto_fcb(): + """Calling AssistantThreadActions.invoke() with a non-Auto FCB should raise before any API call.""" + agent = MagicMock(spec=OpenAIAssistantAgent) + agent.definition = MagicMock() + agent.definition.tools = [] + agent.kernel = Kernel() + agent.format_instructions = AsyncMock(return_value="") + + with pytest.raises(AgentInvokeException, match="not supported"): + async for _ in AssistantThreadActions.invoke( + agent=agent, + thread_id="thread123", + kernel=Kernel(), + function_choice_behavior=FunctionChoiceBehavior.Required(), + ): + pass + + +async def test_invoke_stream_raises_for_non_auto_fcb(): + """Calling AssistantThreadActions.invoke_stream() with a non-Auto FCB should raise before any API call.""" + agent = MagicMock(spec=OpenAIAssistantAgent) + agent.definition = MagicMock() + agent.definition.tools = [] + agent.kernel = Kernel() + agent.format_instructions = AsyncMock(return_value="") + + with pytest.raises(AgentInvokeException, match="not supported"): + async for _ in AssistantThreadActions.invoke_stream( + agent=agent, + thread_id="thread123", + kernel=Kernel(), + function_choice_behavior=FunctionChoiceBehavior.NoneInvoke(), + ): + pass + + +# endregion diff --git a/python/tests/unit/agents/openai_assistant/test_openai_assistant_agent.py b/python/tests/unit/agents/openai_assistant/test_openai_assistant_agent.py index 6423ebf39b..40c3e42145 100644 --- a/python/tests/unit/agents/openai_assistant/test_openai_assistant_agent.py +++ b/python/tests/unit/agents/openai_assistant/test_openai_assistant_agent.py @@ -10,6 +10,7 @@ from pydantic import BaseModel, ValidationError from semantic_kernel.agents import AgentRegistry, AgentResponseItem, OpenAIAssistantAgent from semantic_kernel.agents.open_ai.openai_assistant_agent import AssistantAgentThread from semantic_kernel.agents.open_ai.run_polling_options import RunPollingOptions +from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior from semantic_kernel.contents.chat_history import ChatHistory from semantic_kernel.contents.chat_message_content import ChatMessageContent from semantic_kernel.contents.function_call_content import FunctionCallContent @@ -497,3 +498,82 @@ name: ShouldFail """ with pytest.raises(AgentInitializationException, match="not registered"): await AgentRegistry.create_from_yaml(spec) + + +async def test_openai_assistant_agent_get_response_passes_function_choice_behavior(openai_client, assistant_definition): + agent = OpenAIAssistantAgent(client=openai_client, definition=assistant_definition) + thread = AsyncMock(spec=AssistantAgentThread) + fcb = FunctionChoiceBehavior.Auto() + captured_kwargs = {} + + async def fake_invoke(*args, **kwargs): + captured_kwargs.update(kwargs) + yield True, ChatMessageContent(role=AuthorRole.ASSISTANT, content="content") + + with patch( + "semantic_kernel.agents.open_ai.assistant_thread_actions.AssistantThreadActions.invoke", + side_effect=fake_invoke, + ): + await agent.get_response(messages="message", thread=thread, function_choice_behavior=fcb) + + assert captured_kwargs.get("function_choice_behavior") is fcb + + +async def test_openai_assistant_agent_invoke_passes_function_choice_behavior(openai_client, assistant_definition): + agent = OpenAIAssistantAgent(client=openai_client, definition=assistant_definition) + thread = AsyncMock(spec=AssistantAgentThread) + fcb = FunctionChoiceBehavior.Auto() + captured_kwargs = {} + + async def fake_invoke(*args, **kwargs): + captured_kwargs.update(kwargs) + yield True, ChatMessageContent(role=AuthorRole.ASSISTANT, content="content") + + with patch( + "semantic_kernel.agents.open_ai.assistant_thread_actions.AssistantThreadActions.invoke", + side_effect=fake_invoke, + ): + async for _ in agent.invoke(messages="message", thread=thread, function_choice_behavior=fcb): + pass + + assert captured_kwargs.get("function_choice_behavior") is fcb + + +async def test_openai_assistant_agent_invoke_stream_passes_function_choice_behavior( + openai_client, assistant_definition +): + agent = OpenAIAssistantAgent(client=openai_client, definition=assistant_definition) + thread = AsyncMock(spec=AssistantAgentThread) + fcb = FunctionChoiceBehavior.Auto() + captured_kwargs = {} + + async def fake_invoke(*args, **kwargs): + captured_kwargs.update(kwargs) + yield ChatMessageContent(role=AuthorRole.ASSISTANT, content="content") + + with patch( + "semantic_kernel.agents.open_ai.assistant_thread_actions.AssistantThreadActions.invoke_stream", + side_effect=fake_invoke, + ): + async for _ in agent.invoke_stream(messages="message", thread=thread, function_choice_behavior=fcb): + pass + + assert captured_kwargs.get("function_choice_behavior") is fcb + + +async def test_openai_assistant_agent_get_response_no_fcb_passes_none(openai_client, assistant_definition): + agent = OpenAIAssistantAgent(client=openai_client, definition=assistant_definition) + thread = AsyncMock(spec=AssistantAgentThread) + captured_kwargs = {} + + async def fake_invoke(*args, **kwargs): + captured_kwargs.update(kwargs) + yield True, ChatMessageContent(role=AuthorRole.ASSISTANT, content="content") + + with patch( + "semantic_kernel.agents.open_ai.assistant_thread_actions.AssistantThreadActions.invoke", + side_effect=fake_invoke, + ): + await agent.get_response(messages="message", thread=thread) + + assert captured_kwargs.get("function_choice_behavior") is None