agent/agent-framework
2
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: [BREAKING] Replace Hosted*Tool classes with tool methods (#3634)

Browse Source 
* Replace Hosted*Tool classes with client static factory methods

* fixed failing test

* mypy fix

* mypy fix 2

* declarative mypy fix

* addressed comments

* ToolProtocol removal

* fixed test

* agents mypy fix

* fix failing tests

* mypy fix

* addressed comments

* fixed tests

* addressed comments + added factory method overrides for azureai v2 client

* mypy fix

* added kwargs to azureai tool methods

* fixed in test

* _sessions fix

* test fix
This commit is contained in:
Giles Odigwe
2026-02-11 00:04:27 +00:00
committed by GitHub
parent d249473a6d
commit 7a88af0aef
133 changed files with 3018 additions and 2650 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/packages/azure-ai/agent_framework_azure_ai/_agent_provider.py
+11 -12
View File
@@ -12,7 +12,6 @@ from agent_framework import (
ContextProvider,
FunctionTool,
MiddlewareTypes,
ToolProtocol,
normalize_tools,
)
from agent_framework._mcp import MCPTool
@@ -169,10 +168,10 @@ class AzureAIAgentsProvider(Generic[OptionsCoT]):
model: str | None = None,
instructions: str | None = None,
description: str | None = None,
tools: ToolProtocol
tools: FunctionTool
| Callable[..., Any]
| MutableMapping[str, Any]
| Sequence[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]]
| Sequence[FunctionTool | Callable[..., Any] | MutableMapping[str, Any]]
| None = None,
default_options: OptionsCoT | None = None,
middleware: Sequence[MiddlewareTypes] | None = None,
@@ -266,10 +265,10 @@ class AzureAIAgentsProvider(Generic[OptionsCoT]):
self,
id: str,
*,
tools: ToolProtocol
tools: FunctionTool
| Callable[..., Any]
| MutableMapping[str, Any]
| Sequence[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]]
| Sequence[FunctionTool | Callable[..., Any] | MutableMapping[str, Any]]
| None = None,
default_options: OptionsCoT | None = None,
middleware: Sequence[MiddlewareTypes] | None = None,
@@ -322,10 +321,10 @@ class AzureAIAgentsProvider(Generic[OptionsCoT]):
def as_agent(
self,
agent: AzureAgent,
tools: ToolProtocol
tools: FunctionTool
| Callable[..., Any]
| MutableMapping[str, Any]
| Sequence[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]]
| Sequence[FunctionTool | Callable[..., Any] | MutableMapping[str, Any]]
| None = None,
default_options: OptionsCoT | None = None,
middleware: Sequence[MiddlewareTypes] | None = None,
@@ -379,7 +378,7 @@ class AzureAIAgentsProvider(Generic[OptionsCoT]):
def _to_chat_agent_from_agent(
self,
agent: AzureAgent,
provided_tools: Sequence[ToolProtocol | MutableMapping[str, Any]] | None = None,
provided_tools: Sequence[FunctionTool | MutableMapping[str, Any]] | None = None,
default_options: OptionsCoT | None = None,
middleware: Sequence[MiddlewareTypes] | None = None,
context_provider: ContextProvider | None = None,
@@ -422,8 +421,8 @@ class AzureAIAgentsProvider(Generic[OptionsCoT]):
def _merge_tools(
self,
agent_tools: Sequence[Any] | None,
provided_tools: Sequence[ToolProtocol | MutableMapping[str, Any]] | None,
) -> list[ToolProtocol | dict[str, Any]]:
provided_tools: Sequence[FunctionTool | MutableMapping[str, Any]] | None,
) -> list[FunctionTool | dict[str, Any]]:
"""Merge hosted tools from agent with user-provided function tools.
Args:
@@ -433,7 +432,7 @@ class AzureAIAgentsProvider(Generic[OptionsCoT]):
Returns:
Combined list of tools for the Agent.
"""
merged: list[ToolProtocol | dict[str, Any]] = []
merged: list[FunctionTool | dict[str, Any]] = []
# Convert hosted tools from agent definition
hosted_tools = from_azure_ai_agent_tools(agent_tools)
@@ -459,7 +458,7 @@ class AzureAIAgentsProvider(Generic[OptionsCoT]):
def _validate_function_tools(
self,
agent_tools: Sequence[Any] | None,
provided_tools: Sequence[ToolProtocol | MutableMapping[str, Any]] | None,
provided_tools: Sequence[FunctionTool | MutableMapping[str, Any]] | None,
) -> None:
"""Validate that required function tools are provided.
python/packages/azure-ai/agent_framework_azure_ai/_chat_client.py
+245 -110
View File
@@ -26,16 +26,11 @@ from agent_framework import (
FunctionInvocationConfiguration,
FunctionInvocationLayer,
FunctionTool,
HostedCodeInterpreterTool,
HostedFileSearchTool,
HostedMCPTool,
HostedWebSearchTool,
Message,
MiddlewareTypes,
ResponseStream,
Role,
TextSpanRegion,
ToolProtocol,
UsageDetails,
get_logger,
prepare_function_call_results,
@@ -55,7 +50,7 @@ from azure.ai.agents.models import (
AsyncAgentRunStream,
BingCustomSearchTool,
BingGroundingTool,
CodeInterpreterToolDefinition,
CodeInterpreterTool,
FileSearchTool,
FunctionName,
FunctionToolDefinition,
@@ -217,6 +212,198 @@ class AzureAIAgentClient(
OTEL_PROVIDER_NAME: ClassVar[str] = "azure.ai" # type: ignore[reportIncompatibleVariableOverride, misc]
# region Hosted Tool Factory Methods
@staticmethod
def get_code_interpreter_tool() -> CodeInterpreterTool:
"""Create a code interpreter tool configuration for Azure AI Agents.
Returns:
A CodeInterpreterTool instance ready to pass to ChatAgent.
Examples:
.. code-block:: python
from agent_framework.azure import AzureAIAgentClient
tool = AzureAIAgentClient.get_code_interpreter_tool()
agent = ChatAgent(client, tools=[tool])
"""
return CodeInterpreterTool()
@staticmethod
def get_file_search_tool(
*,
vector_store_ids: list[str],
) -> FileSearchTool:
"""Create a file search tool configuration for Azure AI Agents.
Keyword Args:
vector_store_ids: List of vector store IDs to search within.
Returns:
A FileSearchTool instance ready to pass to ChatAgent.
Examples:
.. code-block:: python
from agent_framework.azure import AzureAIAgentClient
tool = AzureAIAgentClient.get_file_search_tool(
vector_store_ids=["vs_abc123"],
)
agent = ChatAgent(client, tools=[tool])
"""
return FileSearchTool(vector_store_ids=vector_store_ids)
@staticmethod
def get_web_search_tool(
*,
bing_connection_id: str | None = None,
bing_custom_connection_id: str | None = None,
bing_custom_instance_id: str | None = None,
) -> BingGroundingTool | BingCustomSearchTool:
"""Create a web search tool configuration for Azure AI Agents.
For Azure AI Agents, web search uses Bing Grounding or Bing Custom Search.
If no arguments are provided, attempts to read from environment variables.
If no connection IDs are found, raises ValueError.
Keyword Args:
bing_connection_id: The Bing Grounding connection ID for standard web search.
Falls back to BING_CONNECTION_ID environment variable.
bing_custom_connection_id: The Bing Custom Search connection ID.
Falls back to BING_CUSTOM_CONNECTION_ID environment variable.
bing_custom_instance_id: The Bing Custom Search instance ID.
Falls back to BING_CUSTOM_INSTANCE_NAME environment variable.
Returns:
A BingGroundingTool or BingCustomSearchTool instance ready to pass to ChatAgent.
Examples:
.. code-block:: python
from agent_framework.azure import AzureAIAgentClient
# Bing Grounding (explicit)
tool = AzureAIAgentClient.get_web_search_tool(
bing_connection_id="conn_bing_123",
)
# Bing Grounding (from environment variable)
tool = AzureAIAgentClient.get_web_search_tool()
# Bing Custom Search (explicit)
tool = AzureAIAgentClient.get_web_search_tool(
bing_custom_connection_id="conn_custom_123",
bing_custom_instance_id="instance_456",
)
# Bing Custom Search (from environment variables)
# Set BING_CUSTOM_CONNECTION_ID and BING_CUSTOM_INSTANCE_NAME
tool = AzureAIAgentClient.get_web_search_tool()
agent = ChatAgent(client, tools=[tool])
"""
# Try explicit Bing Custom Search parameters first, then environment variables
resolved_custom_connection = bing_custom_connection_id or os.environ.get("BING_CUSTOM_CONNECTION_ID")
resolved_custom_instance = bing_custom_instance_id or os.environ.get("BING_CUSTOM_INSTANCE_NAME")
if resolved_custom_connection and resolved_custom_instance:
return BingCustomSearchTool(
connection_id=resolved_custom_connection,
instance_name=resolved_custom_instance,
)
# Try explicit Bing Grounding parameter first, then environment variable
resolved_connection_id = bing_connection_id or os.environ.get("BING_CONNECTION_ID")
if resolved_connection_id:
return BingGroundingTool(connection_id=resolved_connection_id)
# Azure AI Agents requires Bing connection for web search
raise ValueError(
"Azure AI Agents requires a Bing connection for web search. "
"Provide bing_connection_id (or set BING_CONNECTION_ID env var) for Bing Grounding, "
"or provide both bing_custom_connection_id and bing_custom_instance_id "
"(or set BING_CUSTOM_CONNECTION_ID and BING_CUSTOM_INSTANCE_NAME env vars) for Bing Custom Search."
)
@staticmethod
def get_mcp_tool(
*,
name: str,
url: str | None = None,
description: str | None = None,
approval_mode: str | dict[str, list[str]] | None = None,
allowed_tools: list[str] | None = None,
headers: dict[str, str] | None = None,
) -> McpTool:
"""Create a hosted MCP tool configuration for Azure AI Agents.
This configures an MCP (Model Context Protocol) server that will be called
by Azure AI's service. The tools from this MCP server are executed remotely
by Azure AI, not locally by your application.
Note:
For local MCP execution where your application calls the MCP server
directly, use the MCP client tools instead of this method.
Keyword Args:
name: A label/name for the MCP server.
url: The URL of the MCP server.
description: A description of what the MCP server provides.
approval_mode: Tool approval mode. Use "always_require" or "never_require" for all tools,
or provide a dict with "always_require_approval" and/or "never_require_approval"
keys mapping to lists of tool names.
allowed_tools: List of tool names that are allowed to be used from this MCP server.
headers: HTTP headers to include in requests to the MCP server.
Returns:
An McpTool instance ready to pass to ChatAgent.
Examples:
.. code-block:: python
from agent_framework.azure import AzureAIAgentClient
tool = AzureAIAgentClient.get_mcp_tool(
name="my_mcp",
url="https://mcp.example.com",
)
agent = ChatAgent(client, tools=[tool])
"""
mcp_tool = McpTool(
server_label=name.replace(" ", "_"),
server_url=url or "",
allowed_tools=list(allowed_tools) if allowed_tools else [],
)
# Set approval mode if provided
# The SDK's set_approval_mode() accepts dict at runtime even though type hints say str.
if approval_mode:
if isinstance(approval_mode, str):
if approval_mode == "never_require":
mcp_tool.set_approval_mode("never")
elif approval_mode == "always_require":
mcp_tool.set_approval_mode("always")
else:
mcp_tool.set_approval_mode(approval_mode)
elif isinstance(approval_mode, dict):
# Handle dict-based approval mode (per-tool approval settings)
if "never_require_approval" in approval_mode:
mcp_tool.set_approval_mode({"never": {"tool_names": approval_mode["never_require_approval"]}}) # type: ignore[arg-type]
elif "always_require_approval" in approval_mode:
mcp_tool.set_approval_mode({"always": {"tool_names": approval_mode["always_require_approval"]}}) # type: ignore[arg-type]
# Set headers if provided
if headers:
for key, value in headers.items():
mcp_tool.update_headers(key, value)
return mcp_tool
# endregion
def __init__(
self,
*,
@@ -1051,38 +1238,25 @@ class AzureAIAgentClient(
return tool_definitions
def _prepare_mcp_resources(self, tools: Sequence[ToolProtocol | MutableMapping[str, Any]]) -> list[dict[str, Any]]:
"""Prepare MCP tool resources for approval mode configuration."""
mcp_tools = [tool for tool in tools if isinstance(tool, HostedMCPTool)]
if not mcp_tools:
return []
def _prepare_mcp_resources(self, tools: Sequence[Any]) -> list[dict[str, Any]]:
"""Prepare MCP tool resources for approval mode configuration.
Extracts MCP resources from McpTool instances including server_label,
require_approval, and headers.
"""
mcp_resources: list[dict[str, Any]] = []
for mcp_tool in mcp_tools:
server_label = mcp_tool.name.replace(" ", "_")
mcp_resource: dict[str, Any] = {"server_label": server_label}
if mcp_tool.headers:
mcp_resource["headers"] = mcp_tool.headers
if mcp_tool.approval_mode is not None:
match mcp_tool.approval_mode:
case str():
# Map agent framework approval modes to Azure AI approval modes
approval_mode = "always" if mcp_tool.approval_mode == "always_require" else "never"
mcp_resource["require_approval"] = approval_mode
case _:
if "always_require_approval" in mcp_tool.approval_mode:
mcp_resource["require_approval"] = {
"always": mcp_tool.approval_mode["always_require_approval"]
}
elif "never_require_approval" in mcp_tool.approval_mode:
mcp_resource["require_approval"] = {
"never": mcp_tool.approval_mode["never_require_approval"]
}
mcp_resources.append(mcp_resource)
for tool in tools:
if isinstance(tool, McpTool):
# Use the resources property which includes all config (approval, headers)
tool_resources = tool.resources
if tool_resources and tool_resources.mcp:
for mcp_resource in tool_resources.mcp:
resource_dict: dict[str, Any] = {"server_label": mcp_resource.server_label}
if mcp_resource.require_approval:
resource_dict["require_approval"] = mcp_resource.require_approval
if mcp_resource.headers:
resource_dict["headers"] = mcp_resource.headers
mcp_resources.append(resource_dict)
return mcp_resources
def _prepare_messages(
@@ -1144,79 +1318,40 @@ class AzureAIAgentClient(
return additional_messages, instructions, required_action_results
async def _prepare_tools_for_azure_ai(
self, tools: Sequence[ToolProtocol | MutableMapping[str, Any]], run_options: dict[str, Any] | None = None
) -> list[ToolDefinition | dict[str, Any]]:
"""Prepare tool definitions for the Azure AI Agents API."""
tool_definitions: list[ToolDefinition | dict[str, Any]] = []
self, tools: Sequence[Any], run_options: dict[str, Any] | None = None
) -> list[Any]:
"""Prepare tool definitions for the Azure AI Agents API.
Converts FunctionTool to JSON schema format. SDK Tool wrappers with .definitions
are unpacked. All other tools (ToolDefinition, dict, etc.) pass through unchanged.
Args:
tools: Sequence of tools to prepare.
run_options: Optional run options dict that may be updated with tool_resources.
Returns:
List of tool definitions ready for the Azure AI API.
"""
tool_definitions: list[Any] = []
for tool in tools:
match tool:
case FunctionTool():
tool_definitions.append(tool.to_json_schema_spec()) # type: ignore[reportUnknownArgumentType]
case HostedWebSearchTool():
additional_props = tool.additional_properties or {}
config_args: dict[str, Any] = {}
if count := additional_props.get("count"):
config_args["count"] = count
if freshness := additional_props.get("freshness"):
config_args["freshness"] = freshness
if market := additional_props.get("market"):
config_args["market"] = market
if set_lang := additional_props.get("set_lang"):
config_args["set_lang"] = set_lang
# Bing Grounding
connection_id = additional_props.get("connection_id") or os.getenv("BING_CONNECTION_ID")
# Custom Bing Search
custom_connection_id = additional_props.get("custom_connection_id") or os.getenv(
"BING_CUSTOM_CONNECTION_ID"
)
custom_instance_name = additional_props.get("custom_instance_name") or os.getenv(
"BING_CUSTOM_INSTANCE_NAME"
)
bing_search: BingGroundingTool | BingCustomSearchTool | None = None
if (connection_id) and not custom_connection_id and not custom_instance_name:
if connection_id:
conn_id = connection_id
else:
raise ServiceInitializationError("Parameter connection_id is not provided.")
bing_search = BingGroundingTool(connection_id=conn_id, **config_args)
if custom_connection_id and custom_instance_name:
bing_search = BingCustomSearchTool(
connection_id=custom_connection_id,
instance_name=custom_instance_name,
**config_args,
)
if not bing_search:
raise ServiceInitializationError(
"Bing search tool requires either 'connection_id' for Bing Grounding "
"or both 'custom_connection_id' and 'custom_instance_name' for Custom Bing Search. "
"These can be provided via additional_properties or environment variables: "
"'BING_CONNECTION_ID', 'BING_CUSTOM_CONNECTION_ID', "
"'BING_CUSTOM_INSTANCE_NAME'"
)
tool_definitions.extend(bing_search.definitions)
case HostedCodeInterpreterTool():
tool_definitions.append(CodeInterpreterToolDefinition())
case HostedMCPTool():
mcp_tool = McpTool(
server_label=tool.name.replace(" ", "_"),
server_url=str(tool.url),
allowed_tools=list(tool.allowed_tools) if tool.allowed_tools else [],
)
tool_definitions.extend(mcp_tool.definitions)
case HostedFileSearchTool():
vector_stores = [inp for inp in tool.inputs or [] if inp.type == "hosted_vector_store"]
if vector_stores:
file_search = FileSearchTool(vector_store_ids=[vs.vector_store_id for vs in vector_stores]) # type: ignore[misc]
tool_definitions.extend(file_search.definitions)
# Set tool_resources for file search to work properly with Azure AI
if run_options is not None and "tool_resources" not in run_options:
run_options["tool_resources"] = file_search.resources
case ToolDefinition():
tool_definitions.append(tool)
case dict():
tool_definitions.append(tool)
case _:
raise ServiceInitializationError(f"Unsupported tool type: {type(tool)}")
if isinstance(tool, FunctionTool):
tool_definitions.append(tool.to_json_schema_spec())
elif hasattr(tool, "definitions") and not isinstance(tool, MutableMapping):
# SDK Tool wrappers (McpTool, FileSearchTool, BingGroundingTool, etc.)
tool_definitions.extend(tool.definitions)
# Handle tool resources (MCP resources handled separately by _prepare_mcp_resources)
if (
run_options is not None
and hasattr(tool, "resources")
and tool.resources
and "mcp" not in tool.resources
):
if "tool_resources" not in run_options:
run_options["tool_resources"] = {}
run_options["tool_resources"].update(tool.resources)
else:
# Pass through ToolDefinition, dict, and other types unchanged
tool_definitions.append(tool)
return tool_definitions
def _prepare_tool_outputs_for_azure_ai(
@@ -1293,10 +1428,10 @@ class AzureAIAgentClient(
name: str | None = None,
description: str | None = None,
instructions: str | None = None,
tools: ToolProtocol
tools: FunctionTool
| Callable[..., Any]
| MutableMapping[str, Any]
| Sequence[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]]
| Sequence[FunctionTool | Callable[..., Any] | MutableMapping[str, Any]]
| None = None,
default_options: AzureAIAgentOptionsT | Mapping[str, Any] | None = None,
chat_message_store_factory: Callable[[], ChatMessageStoreProtocol] | None = None,
python/packages/azure-ai/agent_framework_azure_ai/_client.py
+265 -28
View File
@@ -4,7 +4,7 @@ from __future__ import annotations
import sys
from collections.abc import Callable, Mapping, MutableMapping, Sequence
from typing import Any, ClassVar, Generic, TypedDict, TypeVar, cast
from typing import Any, ClassVar, Generic, Literal, TypedDict, TypeVar, cast
from agent_framework import (
AGENT_FRAMEWORK_USER_AGENT,
@@ -15,10 +15,9 @@ from agent_framework import (
ContextProvider,
FunctionInvocationConfiguration,
FunctionInvocationLayer,
HostedMCPTool,
FunctionTool,
Message,
MiddlewareTypes,
ToolProtocol,
get_logger,
)
from agent_framework.exceptions import ServiceInitializationError
@@ -26,12 +25,24 @@ from agent_framework.observability import ChatTelemetryLayer
from agent_framework.openai import OpenAIResponsesOptions
from agent_framework.openai._responses_client import RawOpenAIResponsesClient
from azure.ai.projects.aio import AIProjectClient
from azure.ai.projects.models import MCPTool, PromptAgentDefinition, PromptAgentDefinitionText, RaiConfig, Reasoning
from azure.ai.projects.models import (
ApproximateLocation,
CodeInterpreterTool,
CodeInterpreterToolAuto,
ImageGenTool,
MCPTool,
PromptAgentDefinition,
PromptAgentDefinitionText,
RaiConfig,
Reasoning,
WebSearchPreviewTool,
)
from azure.ai.projects.models import FileSearchTool as ProjectsFileSearchTool
from azure.core.credentials_async import AsyncTokenCredential
from azure.core.exceptions import ResourceNotFoundError
from pydantic import ValidationError
from ._shared import AzureAISettings, _extract_project_connection_id, create_text_format_config
from ._shared import AzureAISettings, create_text_format_config
if sys.version_info >= (3, 13):
from typing import TypeVar # type: ignore # pragma: no cover
@@ -526,37 +537,263 @@ class RawAzureAIClient(RawOpenAIResponsesClient[AzureAIClientOptionsT], Generic[
if description and not self.agent_description:
self.agent_description = description
# region Hosted Tool Factory Methods (Azure-specific overrides)
@staticmethod
def _prepare_mcp_tool(tool: HostedMCPTool) -> MCPTool: # type: ignore[override]
"""Get MCP tool from HostedMCPTool."""
mcp = MCPTool(server_label=tool.name.replace(" ", "_"), server_url=str(tool.url))
def get_code_interpreter_tool( # type: ignore[override]
*,
file_ids: list[str] | None = None,
container: Literal["auto"] | dict[str, Any] = "auto",
**kwargs: Any,
) -> CodeInterpreterTool:
"""Create a code interpreter tool configuration for Azure AI Projects.
if tool.description:
mcp["server_description"] = tool.description
Keyword Args:
file_ids: Optional list of file IDs to make available to the code interpreter.
container: Container configuration. Use "auto" for automatic container management.
Note: Custom container settings from this parameter are not used by Azure AI Projects;
use file_ids instead.
**kwargs: Additional arguments passed to the SDK CodeInterpreterTool constructor.
Returns:
A CodeInterpreterTool ready to pass to ChatAgent.
Examples:
.. code-block:: python
from agent_framework.azure import AzureAIClient
tool = AzureAIClient.get_code_interpreter_tool()
agent = ChatAgent(client, tools=[tool])
"""
# Extract file_ids from container if provided as dict and file_ids not explicitly set
if file_ids is None and isinstance(container, dict):
file_ids = container.get("file_ids")
tool_container = CodeInterpreterToolAuto(file_ids=file_ids if file_ids else None)
return CodeInterpreterTool(container=tool_container, **kwargs)
@staticmethod
def get_file_search_tool(
*,
vector_store_ids: list[str],
max_num_results: int | None = None,
ranking_options: dict[str, Any] | None = None,
filters: dict[str, Any] | None = None,
**kwargs: Any,
) -> ProjectsFileSearchTool:
"""Create a file search tool configuration for Azure AI Projects.
Keyword Args:
vector_store_ids: List of vector store IDs to search.
max_num_results: Maximum number of results to return (1-50).
ranking_options: Ranking options for search results.
filters: A filter to apply (ComparisonFilter or CompoundFilter).
**kwargs: Additional arguments passed to the SDK FileSearchTool constructor.
Returns:
A FileSearchTool ready to pass to ChatAgent.
Raises:
ValueError: If vector_store_ids is empty.
Examples:
.. code-block:: python
from agent_framework.azure import AzureAIClient
tool = AzureAIClient.get_file_search_tool(
vector_store_ids=["vs_abc123"],
)
agent = ChatAgent(client, tools=[tool])
"""
if not vector_store_ids:
raise ValueError("File search tool requires 'vector_store_ids' to be specified.")
return ProjectsFileSearchTool(
vector_store_ids=vector_store_ids,
max_num_results=max_num_results,
ranking_options=ranking_options, # type: ignore[arg-type]
filters=filters, # type: ignore[arg-type]
**kwargs,
)
@staticmethod
def get_web_search_tool( # type: ignore[override]
*,
user_location: dict[str, str] | None = None,
search_context_size: Literal["low", "medium", "high"] | None = None,
**kwargs: Any,
) -> WebSearchPreviewTool:
"""Create a web search preview tool configuration for Azure AI Projects.
Keyword Args:
user_location: Location context for search results. Dict with keys like
"city", "country", "region", "timezone".
search_context_size: Amount of context to include from search results.
One of "low", "medium", or "high". Defaults to "medium".
**kwargs: Additional arguments passed to the SDK WebSearchPreviewTool constructor.
Returns:
A WebSearchPreviewTool ready to pass to ChatAgent.
Examples:
.. code-block:: python
from agent_framework.azure import AzureAIClient
tool = AzureAIClient.get_web_search_tool()
agent = ChatAgent(client, tools=[tool])
# With location and context size
tool = AzureAIClient.get_web_search_tool(
user_location={"city": "Seattle", "country": "US"},
search_context_size="high",
)
"""
ws_tool = WebSearchPreviewTool(search_context_size=search_context_size, **kwargs)
if user_location:
ws_tool.user_location = ApproximateLocation(
city=user_location.get("city"),
country=user_location.get("country"),
region=user_location.get("region"),
timezone=user_location.get("timezone"),
)
return ws_tool
@staticmethod
def get_image_generation_tool( # type: ignore[override]
*,
model: Literal["gpt-image-1"] | str | None = None,
size: Literal["1024x1024", "1024x1536", "1536x1024", "auto"] | None = None,
output_format: Literal["png", "webp", "jpeg"] | None = None,
quality: Literal["low", "medium", "high", "auto"] | None = None,
background: Literal["transparent", "opaque", "auto"] | None = None,
partial_images: int | None = None,
moderation: Literal["auto", "low"] | None = None,
output_compression: int | None = None,
**kwargs: Any,
) -> ImageGenTool:
"""Create an image generation tool configuration for Azure AI Projects.
Keyword Args:
model: The model to use for image generation.
size: Output image size.
output_format: Output image format.
quality: Output image quality.
background: Background transparency setting.
partial_images: Number of partial images to return during generation.
moderation: Moderation level.
output_compression: Compression level.
**kwargs: Additional arguments passed to the SDK ImageGenTool constructor.
Returns:
An ImageGenTool ready to pass to ChatAgent.
Examples:
.. code-block:: python
from agent_framework.azure import AzureAIClient
tool = AzureAIClient.get_image_generation_tool()
agent = ChatAgent(client, tools=[tool])
"""
return ImageGenTool( # type: ignore[misc]
model=model, # type: ignore[arg-type]
size=size,
output_format=output_format,
quality=quality,
background=background,
partial_images=partial_images,
moderation=moderation,
output_compression=output_compression,
**kwargs,
)
@staticmethod
def get_mcp_tool(
*,
name: str,
url: str | None = None,
description: str | None = None,
approval_mode: Literal["always_require", "never_require"] | dict[str, list[str]] | None = None,
allowed_tools: list[str] | None = None,
headers: dict[str, str] | None = None,
project_connection_id: str | None = None,
**kwargs: Any,
) -> MCPTool:
"""Create a hosted MCP tool configuration for Azure AI.
This configures an MCP (Model Context Protocol) server that will be called
by Azure AI's service. The tools from this MCP server are executed remotely
by Azure AI, not locally by your application.
Note:
For local MCP execution where your application calls the MCP server
directly, use the MCP client tools instead of this method.
Keyword Args:
name: A label/name for the MCP server.
url: The URL of the MCP server. Required if project_connection_id is not provided.
description: A description of what the MCP server provides.
approval_mode: Tool approval mode. Use "always_require" or "never_require" for all tools,
or provide a dict with "always_require_approval" and/or "never_require_approval"
keys mapping to lists of tool names.
allowed_tools: List of tool names that are allowed to be used from this MCP server.
headers: HTTP headers to include in requests to the MCP server.
project_connection_id: Azure AI Foundry connection ID for managed MCP connections.
If provided, url and headers are not required.
**kwargs: Additional arguments passed to the SDK MCPTool constructor.
Returns:
An MCPTool configuration ready to pass to ChatAgent.
Examples:
.. code-block:: python
from agent_framework.azure import AzureAIClient
# With URL
tool = AzureAIClient.get_mcp_tool(
name="my_mcp",
url="https://mcp.example.com",
)
# With Azure AI Foundry connection
tool = AzureAIClient.get_mcp_tool(
name="github_mcp",
project_connection_id="conn_abc123",
description="GitHub MCP via Azure AI Foundry",
)
agent = ChatAgent(client, tools=[tool])
"""
mcp = MCPTool(server_label=name.replace(" ", "_"), server_url=url or "", **kwargs)
if description:
mcp["server_description"] = description
# Check for project_connection_id in additional_properties (for Azure AI Foundry connections)
project_connection_id = _extract_project_connection_id(tool.additional_properties)
if project_connection_id:
mcp["project_connection_id"] = project_connection_id
elif tool.headers:
# Only use headers if no project_connection_id is available
mcp["headers"] = tool.headers
elif headers:
mcp["headers"] = headers
if tool.allowed_tools:
mcp["allowed_tools"] = list(tool.allowed_tools)
if allowed_tools:
mcp["allowed_tools"] = allowed_tools
if tool.approval_mode:
match tool.approval_mode:
case str():
mcp["require_approval"] = "always" if tool.approval_mode == "always_require" else "never"
case _:
if always_require_approvals := tool.approval_mode.get("always_require_approval"):
mcp["require_approval"] = {"always": {"tool_names": list(always_require_approvals)}}
if never_require_approvals := tool.approval_mode.get("never_require_approval"):
mcp["require_approval"] = {"never": {"tool_names": list(never_require_approvals)}}
if approval_mode:
if isinstance(approval_mode, str):
mcp["require_approval"] = "always" if approval_mode == "always_require" else "never"
else:
if always_require := approval_mode.get("always_require_approval"):
mcp["require_approval"] = {"always": {"tool_names": always_require}}
if never_require := approval_mode.get("never_require_approval"):
mcp["require_approval"] = {"never": {"tool_names": never_require}}
return mcp
# endregion
@override
def as_agent(
self,
@@ -565,10 +802,10 @@ class RawAzureAIClient(RawOpenAIResponsesClient[AzureAIClientOptionsT], Generic[
name: str | None = None,
description: str | None = None,
instructions: str | None = None,
tools: ToolProtocol
tools: FunctionTool
| Callable[..., Any]
| MutableMapping[str, Any]
| Sequence[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]]
| Sequence[FunctionTool | Callable[..., Any] | MutableMapping[str, Any]]
| None = None,
default_options: AzureAIClientOptionsT | Mapping[str, Any] | None = None,
chat_message_store_factory: Callable[[], ChatMessageStoreProtocol] | None = None,
python/packages/azure-ai/agent_framework_azure_ai/_project_provider.py
+14 -15
View File
@@ -12,7 +12,6 @@ from agent_framework import (
ContextProvider,
FunctionTool,
MiddlewareTypes,
ToolProtocol,
get_logger,
normalize_tools,
)
@@ -162,10 +161,10 @@ class AzureAIProjectAgentProvider(Generic[OptionsCoT]):
model: str | None = None,
instructions: str | None = None,
description: str | None = None,
tools: ToolProtocol
tools: FunctionTool
| Callable[..., Any]
| MutableMapping[str, Any]
| Sequence[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]]
| Sequence[FunctionTool | Callable[..., Any] | MutableMapping[str, Any]]
| None = None,
default_options: OptionsCoT | None = None,
middleware: Sequence[MiddlewareTypes] | None = None,
@@ -221,7 +220,7 @@ class AzureAIProjectAgentProvider(Generic[OptionsCoT]):
# Normalize tools and separate MCP tools from other tools
normalized_tools = normalize_tools(tools)
mcp_tools: list[MCPTool] = []
non_mcp_tools: list[ToolProtocol | MutableMapping[str, Any]] = []
non_mcp_tools: list[FunctionTool | MutableMapping[str, Any]] = []
if normalized_tools:
for tool in normalized_tools:
@@ -239,7 +238,7 @@ class AzureAIProjectAgentProvider(Generic[OptionsCoT]):
mcp_discovered_functions.extend(mcp_tool.functions)
# Combine non-MCP tools with discovered MCP functions for Azure AI
all_tools_for_azure: list[ToolProtocol | MutableMapping[str, Any]] = list(non_mcp_tools)
all_tools_for_azure: list[FunctionTool | MutableMapping[str, Any]] = list(non_mcp_tools)
all_tools_for_azure.extend(mcp_discovered_functions)
if all_tools_for_azure:
@@ -264,10 +263,10 @@ class AzureAIProjectAgentProvider(Generic[OptionsCoT]):
*,
name: str | None = None,
reference: AgentReference | None = None,
tools: ToolProtocol
tools: FunctionTool
| Callable[..., Any]
| MutableMapping[str, Any]
| Sequence[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]]
| Sequence[FunctionTool | Callable[..., Any] | MutableMapping[str, Any]]
| None = None,
default_options: OptionsCoT | None = None,
middleware: Sequence[MiddlewareTypes] | None = None,
@@ -324,10 +323,10 @@ class AzureAIProjectAgentProvider(Generic[OptionsCoT]):
def as_agent(
self,
details: AgentVersionDetails,
tools: ToolProtocol
tools: FunctionTool
| Callable[..., Any]
| MutableMapping[str, Any]
| Sequence[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]]
| Sequence[FunctionTool | Callable[..., Any] | MutableMapping[str, Any]]
| None = None,
default_options: OptionsCoT | None = None,
middleware: Sequence[MiddlewareTypes] | None = None,
@@ -368,7 +367,7 @@ class AzureAIProjectAgentProvider(Generic[OptionsCoT]):
def _to_chat_agent_from_details(
self,
details: AgentVersionDetails,
provided_tools: Sequence[ToolProtocol | MutableMapping[str, Any]] | None = None,
provided_tools: Sequence[FunctionTool | MutableMapping[str, Any]] | None = None,
default_options: OptionsCoT | None = None,
middleware: Sequence[MiddlewareTypes] | None = None,
context_provider: ContextProvider | None = None,
@@ -416,8 +415,8 @@ class AzureAIProjectAgentProvider(Generic[OptionsCoT]):
def _merge_tools(
self,
definition_tools: Sequence[Any] | None,
provided_tools: Sequence[ToolProtocol | MutableMapping[str, Any]] | None,
) -> list[ToolProtocol | dict[str, Any]]:
provided_tools: Sequence[FunctionTool | MutableMapping[str, Any]] | None,
) -> list[FunctionTool | dict[str, Any]]:
"""Merge hosted tools from definition with user-provided function tools.
Args:
@@ -427,7 +426,7 @@ class AzureAIProjectAgentProvider(Generic[OptionsCoT]):
Returns:
Combined list of tools for the Agent.
"""
merged: list[ToolProtocol | dict[str, Any]] = []
merged: list[FunctionTool | dict[str, Any]] = []
# Convert hosted tools from definition (MCP, code interpreter, file search, web search)
# Function tools from the definition are skipped - we use user-provided implementations instead
@@ -451,10 +450,10 @@ class AzureAIProjectAgentProvider(Generic[OptionsCoT]):
def _validate_function_tools(
self,
agent_tools: Sequence[Any] | None,
provided_tools: ToolProtocol
provided_tools: FunctionTool
| Callable[..., Any]
| MutableMapping[str, Any]
| Sequence[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]]
| Sequence[FunctionTool | Callable[..., Any] | MutableMapping[str, Any]]
| None,
) -> None:
"""Validate that required function tools are provided."""
python/packages/azure-ai/agent_framework_azure_ai/_shared.py
+132 -277
View File
@@ -2,37 +2,21 @@
from __future__ import annotations
import os
from collections.abc import Mapping, MutableMapping, Sequence
from typing import Any, ClassVar, Literal, cast
from typing import Any, ClassVar, cast
from agent_framework import (
Content,
FunctionTool,
HostedCodeInterpreterTool,
HostedFileSearchTool,
HostedImageGenerationTool,
HostedMCPTool,
HostedWebSearchTool,
ToolProtocol,
get_logger,
)
from agent_framework._pydantic import AFBaseSettings
from agent_framework.exceptions import ServiceInitializationError, ServiceInvalidRequestError
from agent_framework.exceptions import ServiceInvalidRequestError
from azure.ai.agents.models import (
BingCustomSearchTool,
BingGroundingTool,
CodeInterpreterToolDefinition,
McpTool,
ToolDefinition,
)
from azure.ai.agents.models import FileSearchTool as AgentsFileSearchTool
from azure.ai.projects.models import (
ApproximateLocation,
CodeInterpreterTool,
CodeInterpreterToolAuto,
ImageGenTool,
ImageGenToolInputImageMask,
MCPTool,
ResponseTextFormatConfigurationJsonObject,
ResponseTextFormatConfigurationJsonSchema,
@@ -93,13 +77,13 @@ class AzureAISettings(AFBaseSettings):
def _extract_project_connection_id(additional_properties: dict[str, Any] | None) -> str | None:
"""Extract project_connection_id from HostedMCPTool additional_properties.
"""Extract project_connection_id from tool additional_properties.
Checks for both direct 'project_connection_id' key (programmatic usage)
and 'connection.name' structure (declarative/YAML usage).
Args:
additional_properties: The additional_properties dict from a HostedMCPTool.
additional_properties: The additional_properties dict from a tool.
Returns:
The project_connection_id if found, None otherwise.
@@ -124,11 +108,13 @@ def _extract_project_connection_id(additional_properties: dict[str, Any] | None)
def to_azure_ai_agent_tools(
tools: Sequence[ToolProtocol | MutableMapping[str, Any]] | None,
tools: Sequence[FunctionTool | MutableMapping[str, Any]] | None,
run_options: dict[str, Any] | None = None,
) -> list[ToolDefinition | dict[str, Any]]:
"""Convert Agent Framework tools to Azure AI V1 SDK tool definitions.
Handles FunctionTool instances and dict-based tools from static factory methods.
Args:
tools: Sequence of Agent Framework tools to convert.
run_options: Optional dict with run options.
@@ -144,91 +130,53 @@ def to_azure_ai_agent_tools(
tool_definitions: list[ToolDefinition | dict[str, Any]] = []
for tool in tools:
match tool:
case FunctionTool():
tool_definitions.append(tool.to_json_schema_spec()) # type: ignore[reportUnknownArgumentType]
case HostedWebSearchTool():
additional_props = tool.additional_properties or {}
config_args: dict[str, Any] = {}
if count := additional_props.get("count"):
config_args["count"] = count
if freshness := additional_props.get("freshness"):
config_args["freshness"] = freshness
if market := additional_props.get("market"):
config_args["market"] = market
if set_lang := additional_props.get("set_lang"):
config_args["set_lang"] = set_lang
# Bing Grounding
connection_id = additional_props.get("connection_id") or os.getenv("BING_CONNECTION_ID")
# Custom Bing Search
custom_connection_id = additional_props.get("custom_connection_id") or os.getenv(
"BING_CUSTOM_CONNECTION_ID"
)
custom_instance_name = additional_props.get("custom_instance_name") or os.getenv(
"BING_CUSTOM_INSTANCE_NAME"
)
bing_search: BingGroundingTool | BingCustomSearchTool | None = None
if connection_id and not custom_connection_id and not custom_instance_name:
bing_search = BingGroundingTool(connection_id=connection_id, **config_args)
if custom_connection_id and custom_instance_name:
bing_search = BingCustomSearchTool(
connection_id=custom_connection_id,
instance_name=custom_instance_name,
**config_args,
)
if not bing_search:
raise ServiceInitializationError(
"Bing search tool requires either 'connection_id' for Bing Grounding "
"or both 'custom_connection_id' and 'custom_instance_name' for Custom Bing Search. "
"These can be provided via additional_properties or environment variables: "
"'BING_CONNECTION_ID', 'BING_CUSTOM_CONNECTION_ID', 'BING_CUSTOM_INSTANCE_NAME'"
)
tool_definitions.extend(bing_search.definitions)
case HostedCodeInterpreterTool():
tool_definitions.append(CodeInterpreterToolDefinition())
case HostedMCPTool():
mcp_tool = McpTool(
server_label=tool.name.replace(" ", "_"),
server_url=str(tool.url),
allowed_tools=list(tool.allowed_tools) if tool.allowed_tools else [],
)
tool_definitions.extend(mcp_tool.definitions)
case HostedFileSearchTool():
vector_stores = [inp for inp in tool.inputs or [] if inp.type == "hosted_vector_store"]
if vector_stores:
file_search = AgentsFileSearchTool(vector_store_ids=[vs.vector_store_id for vs in vector_stores]) # type: ignore[misc]
tool_definitions.extend(file_search.definitions)
# Set tool_resources for file search to work properly with Azure AI
if run_options is not None and "tool_resources" not in run_options:
run_options["tool_resources"] = file_search.resources
case ToolDefinition():
tool_definitions.append(tool)
case dict():
tool_definitions.append(tool)
case _:
raise ServiceInitializationError(f"Unsupported tool type: {type(tool)}")
if isinstance(tool, FunctionTool):
tool_definitions.append(tool.to_json_schema_spec()) # type: ignore[reportUnknownArgumentType]
elif isinstance(tool, ToolDefinition):
# Pass through ToolDefinition subclasses unchanged (includes CodeInterpreterToolDefinition, etc.)
tool_definitions.append(tool)
elif hasattr(tool, "definitions") and not isinstance(tool, (dict, MutableMapping)):
# SDK Tool wrappers (McpTool, FileSearchTool, BingGroundingTool, etc.)
tool_definitions.extend(tool.definitions)
# Handle tool resources (MCP resources handled separately)
if (
run_options is not None
and hasattr(tool, "resources")
and tool.resources
and "mcp" not in tool.resources
):
if "tool_resources" not in run_options:
run_options["tool_resources"] = {}
run_options["tool_resources"].update(tool.resources)
elif isinstance(tool, (dict, MutableMapping)):
# Handle dict-based tools - pass through directly
tool_dict = tool if isinstance(tool, dict) else dict(tool)
tool_definitions.append(tool_dict)
else:
# Pass through other types unchanged
tool_definitions.append(tool)
return tool_definitions
def from_azure_ai_agent_tools(
tools: Sequence[ToolDefinition | dict[str, Any]] | None,
) -> list[ToolProtocol | dict[str, Any]]:
"""Convert Azure AI V1 SDK tool definitions to Agent Framework tools.
) -> list[dict[str, Any]]:
"""Convert Azure AI V1 SDK tool definitions to dict-based tools.
Args:
tools: Sequence of Azure AI V1 SDK tool definitions.
Returns:
List of Agent Framework tools.
List of dict-based tool definitions.
"""
if not tools:
return []
result: list[ToolProtocol | dict[str, Any]] = []
result: list[dict[str, Any]] = []
for tool in tools:
# Handle SDK objects
if isinstance(tool, CodeInterpreterToolDefinition):
result.append(HostedCodeInterpreterTool())
result.append({"type": "code_interpreter"})
elif isinstance(tool, dict):
# Handle dict format
converted = _convert_dict_tool(tool)
@@ -242,35 +190,38 @@ def from_azure_ai_agent_tools(
return result
def _convert_dict_tool(tool: dict[str, Any]) -> ToolProtocol | dict[str, Any] | None:
"""Convert a dict-format Azure AI tool to Agent Framework tool."""
def _convert_dict_tool(tool: dict[str, Any]) -> dict[str, Any] | None:
"""Convert a dict-format Azure AI tool to dict-based tool format."""
tool_type = tool.get("type")
if tool_type == "code_interpreter":
return HostedCodeInterpreterTool()
return {"type": "code_interpreter"}
if tool_type == "file_search":
file_search_config = tool.get("file_search", {})
vector_store_ids = file_search_config.get("vector_store_ids", [])
inputs = [Content.from_hosted_vector_store(vector_store_id=vs_id) for vs_id in vector_store_ids]
return HostedFileSearchTool(inputs=inputs if inputs else None) # type: ignore
return {"type": "file_search", "vector_store_ids": vector_store_ids}
if tool_type == "bing_grounding":
bing_config = tool.get("bing_grounding", {})
connection_id = bing_config.get("connection_id")
return HostedWebSearchTool(additional_properties={"connection_id": connection_id} if connection_id else None)
return {"type": "bing_grounding", "connection_id": connection_id} if connection_id else None
if tool_type == "bing_custom_search":
bing_config = tool.get("bing_custom_search", {})
return HostedWebSearchTool(
additional_properties={
"custom_connection_id": bing_config.get("connection_id"),
"custom_instance_name": bing_config.get("instance_name"),
connection_id = bing_config.get("connection_id")
instance_name = bing_config.get("instance_name")
# Only return if both required fields are present
if connection_id and instance_name:
return {
"type": "bing_custom_search",
"connection_id": connection_id,
"instance_name": instance_name,
}
)
return None
if tool_type == "mcp":
# Hosted MCP tools are defined on the Azure agent, no local handling needed
# MCP tools are defined on the Azure agent, no local handling needed
# Azure may not return full server_url, so skip conversion
return None
@@ -282,35 +233,38 @@ def _convert_dict_tool(tool: dict[str, Any]) -> ToolProtocol | dict[str, Any] |
return tool
def _convert_sdk_tool(tool: ToolDefinition) -> ToolProtocol | dict[str, Any] | None:
"""Convert an SDK-object Azure AI tool to Agent Framework tool."""
def _convert_sdk_tool(tool: ToolDefinition) -> dict[str, Any] | None:
"""Convert an SDK-object Azure AI tool to dict-based tool format."""
tool_type = getattr(tool, "type", None)
if tool_type == "code_interpreter":
return HostedCodeInterpreterTool()
return {"type": "code_interpreter"}
if tool_type == "file_search":
file_search_config = getattr(tool, "file_search", None)
vector_store_ids = getattr(file_search_config, "vector_store_ids", []) if file_search_config else []
inputs = [Content.from_hosted_vector_store(vector_store_id=vs_id) for vs_id in vector_store_ids]
return HostedFileSearchTool(inputs=inputs if inputs else None) # type: ignore
return {"type": "file_search", "vector_store_ids": vector_store_ids}
if tool_type == "bing_grounding":
bing_config = getattr(tool, "bing_grounding", None)
connection_id = getattr(bing_config, "connection_id", None) if bing_config else None
return HostedWebSearchTool(additional_properties={"connection_id": connection_id} if connection_id else None)
return {"type": "bing_grounding", "connection_id": connection_id} if connection_id else None
if tool_type == "bing_custom_search":
bing_config = getattr(tool, "bing_custom_search", None)
return HostedWebSearchTool(
additional_properties={
"custom_connection_id": getattr(bing_config, "connection_id", None) if bing_config else None,
"custom_instance_name": getattr(bing_config, "instance_name", None) if bing_config else None,
connection_id = getattr(bing_config, "connection_id", None) if bing_config else None
instance_name = getattr(bing_config, "instance_name", None) if bing_config else None
# Only return if both required fields are present
if connection_id and instance_name:
return {
"type": "bing_custom_search",
"connection_id": connection_id,
"instance_name": instance_name,
}
)
return None
if tool_type == "mcp":
# Hosted MCP tools are defined on the Azure agent, no local handling needed
# MCP tools are defined on the Azure agent, no local handling needed
# Azure may not return full server_url, so skip conversion
return None
@@ -324,18 +278,17 @@ def _convert_sdk_tool(tool: ToolDefinition) -> ToolProtocol | dict[str, Any] | N
return {"type": tool_type} if tool_type else {}
def from_azure_ai_tools(tools: Sequence[Tool | dict[str, Any]] | None) -> list[ToolProtocol | dict[str, Any]]:
"""Parses and converts a sequence of Azure AI tools into Agent Framework compatible tools.
def from_azure_ai_tools(tools: Sequence[Tool | dict[str, Any]] | None) -> list[dict[str, Any]]:
"""Parses and converts a sequence of Azure AI tools into dict-based tools.
Args:
tools: A sequence of tool objects or dictionaries
defining the tools to be parsed. Can be None.
Returns:
list[ToolProtocol | dict[str, Any]]: A list of converted tools compatible with the
Agent Framework.
list[dict[str, Any]]: A list of dict-based tool definitions.
"""
agent_tools: list[ToolProtocol | dict[str, Any]] = []
agent_tools: list[dict[str, Any]] = []
if not tools:
return agent_tools
for tool in tools:
@@ -345,81 +298,62 @@ def from_azure_ai_tools(tools: Sequence[Tool | dict[str, Any]] | None) -> list[T
if tool_type == "mcp":
mcp_tool = cast(MCPTool, tool_dict)
approval_mode: Literal["always_require", "never_require"] | dict[str, set[str]] | None = None
result: dict[str, Any] = {
"type": "mcp",
"server_label": mcp_tool.get("server_label", ""),
"server_url": mcp_tool.get("server_url", ""),
}
if description := mcp_tool.get("server_description"):
result["server_description"] = description
if headers := mcp_tool.get("headers"):
result["headers"] = headers
if allowed_tools := mcp_tool.get("allowed_tools"):
result["allowed_tools"] = allowed_tools
if require_approval := mcp_tool.get("require_approval"):
if require_approval == "always":
approval_mode = "always_require"
elif require_approval == "never":
approval_mode = "never_require"
elif isinstance(require_approval, dict):
approval_mode = {}
if "always" in require_approval:
approval_mode["always_require_approval"] = set(require_approval["always"].get("tool_names", [])) # type: ignore
if "never" in require_approval:
approval_mode["never_require_approval"] = set(require_approval["never"].get("tool_names", [])) # type: ignore
# Preserve project_connection_id in additional_properties
additional_props: dict[str, Any] | None = None
result["require_approval"] = require_approval
if project_connection_id := mcp_tool.get("project_connection_id"):
additional_props = {"connection": {"name": project_connection_id}}
agent_tools.append(
HostedMCPTool(
name=mcp_tool.get("server_label", "").replace("_", " "),
url=mcp_tool.get("server_url", ""),
description=mcp_tool.get("server_description"),
headers=mcp_tool.get("headers"),
allowed_tools=mcp_tool.get("allowed_tools"),
approval_mode=approval_mode, # type: ignore
additional_properties=additional_props,
)
)
result["project_connection_id"] = project_connection_id
agent_tools.append(result)
elif tool_type == "code_interpreter":
ci_tool = cast(CodeInterpreterTool, tool_dict)
container = ci_tool.get("container", {})
ci_inputs: list[Content] = []
result = {"type": "code_interpreter"}
if "file_ids" in container:
for file_id in container["file_ids"]:
ci_inputs.append(Content.from_hosted_file(file_id=file_id))
agent_tools.append(HostedCodeInterpreterTool(inputs=ci_inputs if ci_inputs else None)) # type: ignore
result["file_ids"] = container["file_ids"]
agent_tools.append(result)
elif tool_type == "file_search":
fs_tool = cast(ProjectsFileSearchTool, tool_dict)
fs_inputs: list[Content] = []
result = {"type": "file_search"}
if "vector_store_ids" in fs_tool:
for vs_id in fs_tool["vector_store_ids"]:
fs_inputs.append(Content.from_hosted_vector_store(vector_store_id=vs_id))
agent_tools.append(
HostedFileSearchTool(
inputs=fs_inputs if fs_inputs else None, # type: ignore
max_results=fs_tool.get("max_num_results"),
)
)
result["vector_store_ids"] = fs_tool["vector_store_ids"]
if max_results := fs_tool.get("max_num_results"):
result["max_num_results"] = max_results
agent_tools.append(result)
elif tool_type == "web_search_preview":
ws_tool = cast(WebSearchPreviewTool, tool_dict)
additional_properties: dict[str, Any] = {}
result = {"type": "web_search_preview"}
if user_location := ws_tool.get("user_location"):
additional_properties["user_location"] = {
result["user_location"] = {
"city": user_location.get("city"),
"country": user_location.get("country"),
"region": user_location.get("region"),
"timezone": user_location.get("timezone"),
}
agent_tools.append(HostedWebSearchTool(additional_properties=additional_properties))
agent_tools.append(result)
else:
agent_tools.append(tool_dict)
return agent_tools
def to_azure_ai_tools(
tools: Sequence[ToolProtocol | MutableMapping[str, Any]] | None,
tools: Sequence[FunctionTool | MutableMapping[str, Any] | Tool] | None,
) -> list[Tool | dict[str, Any]]:
"""Converts Agent Framework tools into Azure AI compatible tools.
Handles FunctionTool instances and passes through SDK Tool types directly.
Args:
tools: A sequence of Agent Framework tool objects or dictionaries
tools: A sequence of Agent Framework tool objects, SDK Tool types, or dictionaries
defining the tools to be converted. Can be None.
Returns:
@@ -430,133 +364,54 @@ def to_azure_ai_tools(
return azure_tools
for tool in tools:
if isinstance(tool, ToolProtocol):
match tool:
case HostedMCPTool():
azure_tools.append(_prepare_mcp_tool_for_azure_ai(tool))
case HostedCodeInterpreterTool():
file_ids: list[str] = []
if tool.inputs:
for tool_input in tool.inputs:
if tool_input.type == "hosted_file":
file_ids.append(tool_input.file_id) # type: ignore[misc, arg-type]
container = CodeInterpreterToolAuto(file_ids=file_ids if file_ids else None)
ci_tool: CodeInterpreterTool = CodeInterpreterTool(container=container)
azure_tools.append(ci_tool)
case FunctionTool():
params = tool.parameters()
params["additionalProperties"] = False
azure_tools.append(
AzureFunctionTool(
name=tool.name,
parameters=params,
strict=False,
description=tool.description,
)
)
case HostedFileSearchTool():
if not tool.inputs:
raise ValueError("HostedFileSearchTool requires inputs to be specified.")
vector_store_ids: list[str] = [
inp.vector_store_id # type: ignore[misc]
for inp in tool.inputs
if inp.type == "hosted_vector_store"
]
if not vector_store_ids:
raise ValueError(
"HostedFileSearchTool requires inputs to be of type `Content` with "
"type 'hosted_vector_store'."
)
fs_tool: ProjectsFileSearchTool = ProjectsFileSearchTool(vector_store_ids=vector_store_ids)
if tool.max_results:
fs_tool["max_num_results"] = tool.max_results
azure_tools.append(fs_tool)
case HostedWebSearchTool():
ws_tool: WebSearchPreviewTool = WebSearchPreviewTool()
if tool.additional_properties:
location: dict[str, str] | None = (
tool.additional_properties.get("user_location", None)
if tool.additional_properties
else None
)
if location:
ws_tool.user_location = ApproximateLocation(
city=location.get("city"),
country=location.get("country"),
region=location.get("region"),
timezone=location.get("timezone"),
)
azure_tools.append(ws_tool)
case HostedImageGenerationTool():
opts = tool.options or {}
addl = tool.additional_properties or {}
# Azure ImageGenTool requires the constant model "gpt-image-1"
ig_tool: ImageGenTool = ImageGenTool(
model=opts.get("model_id", "gpt-image-1"), # type: ignore
size=cast(
Literal["1024x1024", "1024x1536", "1536x1024", "auto"] | None, opts.get("image_size")
),
output_format=cast(Literal["png", "webp", "jpeg"] | None, opts.get("media_type")),
input_image_mask=(
ImageGenToolInputImageMask(
image_url=addl.get("input_image_mask", {}).get("image_url"),
file_id=addl.get("input_image_mask", {}).get("file_id"),
)
if isinstance(addl.get("input_image_mask"), dict)
else None
),
quality=cast(Literal["low", "medium", "high", "auto"] | None, addl.get("quality")),
background=cast(Literal["transparent", "opaque", "auto"] | None, addl.get("background")),
output_compression=cast(int | None, addl.get("output_compression")),
moderation=cast(Literal["auto", "low"] | None, addl.get("moderation")),
partial_images=opts.get("streaming_count"),
)
azure_tools.append(ig_tool)
case _:
logger.debug("Unsupported tool passed (type: %s)", type(tool))
if isinstance(tool, FunctionTool):
params = tool.parameters()
params["additionalProperties"] = False
azure_tools.append(
AzureFunctionTool(
name=tool.name,
parameters=params,
strict=False,
description=tool.description,
)
)
elif isinstance(tool, Tool):
# Pass through SDK Tool types directly (CodeInterpreterTool, FileSearchTool, etc.)
azure_tools.append(tool)
else:
# Handle raw dictionary tools
tool_dict = tool if isinstance(tool, dict) else dict(tool)
azure_tools.append(tool_dict)
# Pass through dict-based tools directly
azure_tools.append(dict(tool) if isinstance(tool, MutableMapping) else tool) # type: ignore[arg-type]
return azure_tools
def _prepare_mcp_tool_for_azure_ai(tool: HostedMCPTool) -> MCPTool:
"""Convert HostedMCPTool to Azure AI MCPTool format.
def _prepare_mcp_tool_dict_for_azure_ai(tool_dict: dict[str, Any]) -> MCPTool:
"""Convert dict-based MCP tool to Azure AI MCPTool format.
Args:
tool: The HostedMCPTool to convert.
tool_dict: The dict-based MCP tool configuration.
Returns:
MCPTool: The converted Azure AI MCPTool.
"""
mcp: MCPTool = MCPTool(server_label=tool.name.replace(" ", "_"), server_url=str(tool.url))
server_label = tool_dict.get("server_label", "")
server_url = tool_dict.get("server_url", "")
mcp: MCPTool = MCPTool(server_label=server_label, server_url=server_url)
if tool.description:
mcp["server_description"] = tool.description
if description := tool_dict.get("server_description"):
mcp["server_description"] = description
# Check for project_connection_id in additional_properties (for Azure AI Foundry connections)
project_connection_id = _extract_project_connection_id(tool.additional_properties)
if project_connection_id:
# Check for project_connection_id
if project_connection_id := tool_dict.get("project_connection_id"):
mcp["project_connection_id"] = project_connection_id
elif tool.headers:
# Only use headers if no project_connection_id is available
# Note: Azure AI Agent Service may reject headers with sensitive info
mcp["headers"] = tool.headers
elif headers := tool_dict.get("headers"):
mcp["headers"] = headers
if tool.allowed_tools:
mcp["allowed_tools"] = list(tool.allowed_tools)
if allowed_tools := tool_dict.get("allowed_tools"):
mcp["allowed_tools"] = list(allowed_tools)
if tool.approval_mode:
match tool.approval_mode:
case str():
mcp["require_approval"] = "always" if tool.approval_mode == "always_require" else "never"
case _:
if always_require_approvals := tool.approval_mode.get("always_require_approval"):
mcp["require_approval"] = {"always": {"tool_names": list(always_require_approvals)}}
if never_require_approvals := tool.approval_mode.get("never_require_approval"):
mcp["require_approval"] = {"never": {"tool_names": list(never_require_approvals)}}
if require_approval := tool_dict.get("require_approval"):
mcp["require_approval"] = require_approval
return mcp
python/packages/azure-ai/tests/test_agent_provider.py
+37 -43
View File
@@ -7,11 +7,6 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from agent_framework import (
Agent,
Content,
HostedCodeInterpreterTool,
HostedFileSearchTool,
HostedMCPTool,
HostedWebSearchTool,
tool,
)
from agent_framework.exceptions import ServiceInitializationError
@@ -25,6 +20,7 @@ from azure.identity.aio import AzureCliCredential
from pydantic import BaseModel
from agent_framework_azure_ai import (
AzureAIAgentClient,
AzureAIAgentsProvider,
AzureAISettings,
)
@@ -466,8 +462,9 @@ def test_as_agent_with_hosted_tools(
agent = provider.as_agent(mock_agent)
assert isinstance(agent, Agent)
# Should have HostedCodeInterpreterTool in the default_options tools
assert any(isinstance(t, HostedCodeInterpreterTool) for t in (agent.default_options.get("tools") or [])) # type: ignore
# Should have code_interpreter dict tool in the default_options tools
tools = agent.default_options.get("tools") or []
assert any(isinstance(t, dict) and t.get("type") == "code_interpreter" for t in tools)
def test_as_agent_with_dict_function_tools_validates(
@@ -571,8 +568,8 @@ def test_to_azure_ai_agent_tools_function() -> None:
def test_to_azure_ai_agent_tools_code_interpreter() -> None:
"""Test converting HostedCodeInterpreterTool."""
tool = HostedCodeInterpreterTool()
"""Test converting code_interpreter dict tool."""
tool = AzureAIAgentClient.get_code_interpreter_tool()
result = to_azure_ai_agent_tools([tool])
@@ -581,8 +578,8 @@ def test_to_azure_ai_agent_tools_code_interpreter() -> None:
def test_to_azure_ai_agent_tools_file_search() -> None:
"""Test converting HostedFileSearchTool with vector stores."""
tool = HostedFileSearchTool(inputs=[Content.from_hosted_vector_store(vector_store_id="vs-123")])
"""Test converting file_search dict tool with vector stores."""
tool = AzureAIAgentClient.get_file_search_tool(vector_store_ids=["vs-123"])
run_options: dict[str, Any] = {}
result = to_azure_ai_agent_tools([tool], run_options)
@@ -592,15 +589,14 @@ def test_to_azure_ai_agent_tools_file_search() -> None:
def test_to_azure_ai_agent_tools_web_search_bing_grounding(monkeypatch: Any) -> None:
"""Test converting HostedWebSearchTool for Bing Grounding."""
"""Test converting web_search dict tool for Bing Grounding."""
# Use a properly formatted connection ID as required by Azure SDK
valid_conn_id = (
"/subscriptions/test-sub/resourceGroups/test-rg/"
"providers/Microsoft.CognitiveServices/accounts/test-account/"
"projects/test-project/connections/test-connection"
)
monkeypatch.setenv("BING_CONNECTION_ID", valid_conn_id)
tool = HostedWebSearchTool()
tool = AzureAIAgentClient.get_web_search_tool(bing_connection_id=valid_conn_id)
result = to_azure_ai_agent_tools([tool])
@@ -608,10 +604,11 @@ def test_to_azure_ai_agent_tools_web_search_bing_grounding(monkeypatch: Any) ->
def test_to_azure_ai_agent_tools_web_search_custom(monkeypatch: Any) -> None:
"""Test converting HostedWebSearchTool for Custom Bing Search."""
monkeypatch.setenv("BING_CUSTOM_CONNECTION_ID", "custom-conn-id")
monkeypatch.setenv("BING_CUSTOM_INSTANCE_NAME", "my-instance")
tool = HostedWebSearchTool()
"""Test converting web_search dict tool for Custom Bing Search."""
tool = AzureAIAgentClient.get_web_search_tool(
bing_custom_connection_id="custom-conn-id",
bing_custom_instance_id="my-instance",
)
result = to_azure_ai_agent_tools([tool])
@@ -619,22 +616,23 @@ def test_to_azure_ai_agent_tools_web_search_custom(monkeypatch: Any) -> None:
def test_to_azure_ai_agent_tools_web_search_missing_config(monkeypatch: Any) -> None:
"""Test converting HostedWebSearchTool raises error when config is missing."""
"""Test converting web_search dict tool without bing config returns empty."""
monkeypatch.delenv("BING_CONNECTION_ID", raising=False)
monkeypatch.delenv("BING_CUSTOM_CONNECTION_ID", raising=False)
monkeypatch.delenv("BING_CUSTOM_INSTANCE_NAME", raising=False)
tool = HostedWebSearchTool()
tool = {"type": "web_search"}
with pytest.raises(ServiceInitializationError):
to_azure_ai_agent_tools([tool])
result = to_azure_ai_agent_tools([tool])
# web_search without bing connection is passed through as dict
assert len(result) == 1
def test_to_azure_ai_agent_tools_mcp() -> None:
"""Test converting HostedMCPTool."""
tool = HostedMCPTool(
"""Test converting MCP dict tool."""
tool = AzureAIAgentClient.get_mcp_tool(
name="my mcp server",
url="https://mcp.example.com",
allowed_tools=["tool1", "tool2"],
)
result = to_azure_ai_agent_tools([tool])
@@ -653,13 +651,15 @@ def test_to_azure_ai_agent_tools_dict_passthrough() -> None:
def test_to_azure_ai_agent_tools_unsupported_type() -> None:
"""Test that unsupported tool types raise error."""
"""Test that unsupported tool types pass through unchanged."""
class UnsupportedTool:
pass
with pytest.raises(ServiceInitializationError):
to_azure_ai_agent_tools([UnsupportedTool()]) # type: ignore
unsupported = UnsupportedTool()
result = to_azure_ai_agent_tools([unsupported]) # type: ignore
assert len(result) == 1
assert result[0] is unsupported # Passed through unchanged
# endregion
@@ -684,7 +684,7 @@ def test_from_azure_ai_agent_tools_code_interpreter() -> None:
result = from_azure_ai_agent_tools([tool])
assert len(result) == 1
assert isinstance(result[0], HostedCodeInterpreterTool)
assert result[0] == {"type": "code_interpreter"}
def test_from_azure_ai_agent_tools_code_interpreter_dict() -> None:
@@ -694,7 +694,7 @@ def test_from_azure_ai_agent_tools_code_interpreter_dict() -> None:
result = from_azure_ai_agent_tools([tool])
assert len(result) == 1
assert isinstance(result[0], HostedCodeInterpreterTool)
assert result[0] == {"type": "code_interpreter"}
def test_from_azure_ai_agent_tools_file_search_dict() -> None:
@@ -707,8 +707,8 @@ def test_from_azure_ai_agent_tools_file_search_dict() -> None:
result = from_azure_ai_agent_tools([tool])
assert len(result) == 1
assert isinstance(result[0], HostedFileSearchTool)
assert len(result[0].inputs or []) == 2
assert result[0]["type"] == "file_search"
assert result[0]["vector_store_ids"] == ["vs-123", "vs-456"]
def test_from_azure_ai_agent_tools_bing_grounding_dict() -> None:
@@ -721,12 +721,8 @@ def test_from_azure_ai_agent_tools_bing_grounding_dict() -> None:
result = from_azure_ai_agent_tools([tool])
assert len(result) == 1
assert isinstance(result[0], HostedWebSearchTool)
additional_properties = result[0].additional_properties
assert additional_properties
assert additional_properties.get("connection_id") == "conn-123"
assert result[0]["type"] == "bing_grounding"
assert result[0]["connection_id"] == "conn-123"
def test_from_azure_ai_agent_tools_bing_custom_search_dict() -> None:
@@ -742,11 +738,9 @@ def test_from_azure_ai_agent_tools_bing_custom_search_dict() -> None:
result = from_azure_ai_agent_tools([tool])
assert len(result) == 1
assert isinstance(result[0], HostedWebSearchTool)
additional_properties = result[0].additional_properties
assert additional_properties
assert additional_properties.get("custom_connection_id") == "custom-conn"
assert result[0]["type"] == "bing_custom_search"
assert result[0]["connection_id"] == "custom-conn"
assert result[0]["instance_name"] == "my-instance"
def test_from_azure_ai_agent_tools_mcp_dict() -> None:
python/packages/azure-ai/tests/test_azure_ai_agent_client.py
+94 -125
View File
@@ -16,10 +16,6 @@ from agent_framework import (
ChatResponse,
ChatResponseUpdate,
Content,
HostedCodeInterpreterTool,
HostedFileSearchTool,
HostedMCPTool,
HostedWebSearchTool,
Message,
SupportsChatGetResponse,
tool,
@@ -721,147 +717,130 @@ def test_azure_ai_chat_client_service_url_method(mock_agents_client: MagicMock)
async def test_azure_ai_chat_client_prepare_options_mcp_never_require(mock_agents_client: MagicMock) -> None:
"""Test _prepare_options with HostedMCPTool having never_require approval mode."""
"""Test _prepare_options with MCP dict tool having never_require approval mode."""
client = create_test_azure_ai_chat_client(mock_agents_client)
mcp_tool = HostedMCPTool(name="Test MCP Tool", url="https://example.com/mcp", approval_mode="never_require")
messages = [Message(role="user", text="Hello")]
chat_options: ChatOptions = {"tools": [mcp_tool], "tool_choice": "auto"}
with patch("agent_framework_azure_ai._shared.McpTool") as mock_mcp_tool_class:
mock_mcp_tool_instance = MagicMock()
mock_mcp_tool_instance.definitions = [{"type": "mcp", "name": "test_mcp"}]
mock_mcp_tool_class.return_value = mock_mcp_tool_instance
run_options, _ = await client._prepare_options(messages, chat_options) # type: ignore
# Verify tool_resources is created with correct MCP approval structure
assert "tool_resources" in run_options, (
f"Expected 'tool_resources' in run_options keys: {list(run_options.keys())}"
)
assert "mcp" in run_options["tool_resources"]
assert len(run_options["tool_resources"]["mcp"]) == 1
mcp_resource = run_options["tool_resources"]["mcp"][0]
assert mcp_resource["server_label"] == "Test_MCP_Tool"
assert mcp_resource["require_approval"] == "never"
async def test_azure_ai_chat_client_prepare_options_mcp_with_headers(mock_agents_client: MagicMock) -> None:
"""Test _prepare_options with HostedMCPTool having headers."""
client = create_test_azure_ai_chat_client(mock_agents_client)
# Test with headers
headers = {"Authorization": "Bearer DUMMY_TOKEN", "X-API-Key": "DUMMY_KEY"}
mcp_tool = HostedMCPTool(
name="Test MCP Tool", url="https://example.com/mcp", headers=headers, approval_mode="never_require"
# Create MCP tool with approval_mode parameter
mcp_tool = AzureAIAgentClient.get_mcp_tool(
name="Test MCP Tool", url="https://example.com/mcp", approval_mode="never_require"
)
messages = [Message(role="user", text="Hello")]
chat_options: ChatOptions = {"tools": [mcp_tool], "tool_choice": "auto"}
with patch("agent_framework_azure_ai._shared.McpTool") as mock_mcp_tool_class:
mock_mcp_tool_instance = MagicMock()
mock_mcp_tool_instance.definitions = [{"type": "mcp", "name": "test_mcp"}]
mock_mcp_tool_class.return_value = mock_mcp_tool_instance
run_options, _ = await client._prepare_options(messages, chat_options) # type: ignore
run_options, _ = await client._prepare_options(messages, chat_options) # type: ignore
# Verify tool_resources is created with correct MCP approval structure
assert "tool_resources" in run_options, f"Expected 'tool_resources' in run_options keys: {list(run_options.keys())}"
assert "mcp" in run_options["tool_resources"]
assert len(run_options["tool_resources"]["mcp"]) == 1
# Verify tool_resources is created with headers
assert "tool_resources" in run_options
assert "mcp" in run_options["tool_resources"]
assert len(run_options["tool_resources"]["mcp"]) == 1
mcp_resource = run_options["tool_resources"]["mcp"][0]
assert mcp_resource["server_label"] == "Test_MCP_Tool"
assert mcp_resource["require_approval"] == "never"
mcp_resource = run_options["tool_resources"]["mcp"][0]
assert mcp_resource["server_label"] == "Test_MCP_Tool"
assert mcp_resource["require_approval"] == "never"
assert mcp_resource["headers"] == headers
async def test_azure_ai_chat_client_prepare_options_mcp_with_headers(mock_agents_client: MagicMock) -> None:
"""Test _prepare_options with MCP dict tool having headers."""
client = create_test_azure_ai_chat_client(mock_agents_client)
# Test with headers - create MCP tool with all options
headers = {"Authorization": "Bearer DUMMY_TOKEN", "X-API-Key": "DUMMY_KEY"}
mcp_tool = AzureAIAgentClient.get_mcp_tool(
name="Test MCP Tool",
url="https://example.com/mcp",
headers=headers,
approval_mode="never_require",
)
messages = [Message(role="user", text="Hello")]
chat_options: ChatOptions = {"tools": [mcp_tool], "tool_choice": "auto"}
run_options, _ = await client._prepare_options(messages, chat_options) # type: ignore
# Verify tool_resources is created with headers
assert "tool_resources" in run_options
assert "mcp" in run_options["tool_resources"]
assert len(run_options["tool_resources"]["mcp"]) == 1
mcp_resource = run_options["tool_resources"]["mcp"][0]
assert mcp_resource["server_label"] == "Test_MCP_Tool"
assert mcp_resource["require_approval"] == "never"
assert mcp_resource["headers"] == headers
async def test_azure_ai_chat_client_prepare_tools_for_azure_ai_web_search_bing_grounding(
mock_agents_client: MagicMock,
) -> None:
"""Test _prepare_tools_for_azure_ai with HostedWebSearchTool using Bing Grounding."""
"""Test _prepare_tools_for_azure_ai with BingGroundingTool from get_web_search_tool()."""
client = create_test_azure_ai_chat_client(mock_agents_client, agent_id="test-agent")
web_search_tool = HostedWebSearchTool(
additional_properties={
"connection_id": "test-connection-id",
"count": 5,
"freshness": "Day",
"market": "en-US",
"set_lang": "en",
}
)
# Mock BingGroundingTool
# Mock BingGroundingTool to avoid SDK validation of connection ID
with patch("agent_framework_azure_ai._chat_client.BingGroundingTool") as mock_bing_grounding:
mock_bing_tool = MagicMock()
mock_bing_tool.definitions = [{"type": "bing_grounding"}]
mock_bing_grounding.return_value = mock_bing_tool
# get_web_search_tool now returns a BingGroundingTool directly
web_search_tool = client.get_web_search_tool(bing_connection_id="test-connection-id")
# Verify the factory method created the tool with correct args
mock_bing_grounding.assert_called_once_with(connection_id="test-connection-id")
result = await client._prepare_tools_for_azure_ai([web_search_tool]) # type: ignore
# BingGroundingTool.definitions should be extended into result
assert len(result) == 1
assert result[0] == {"type": "bing_grounding"}
call_args = mock_bing_grounding.call_args[1]
assert call_args["count"] == 5
assert call_args["freshness"] == "Day"
assert call_args["market"] == "en-US"
assert call_args["set_lang"] == "en"
assert "connection_id" in call_args
async def test_azure_ai_chat_client_prepare_tools_for_azure_ai_web_search_bing_grounding_with_connection_id(
mock_agents_client: MagicMock,
) -> None:
"""Test _prepare_tools_... with HostedWebSearchTool using Bing Grounding with connection_id (no HTTP call)."""
"""Test _prepare_tools_for_azure_ai with BingGroundingTool using explicit connection_id."""
client = create_test_azure_ai_chat_client(mock_agents_client, agent_id="test-agent")
web_search_tool = HostedWebSearchTool(
additional_properties={
"connection_id": "direct-connection-id",
"count": 3,
}
)
# Mock BingGroundingTool
# Mock BingGroundingTool to avoid SDK validation of connection ID
with patch("agent_framework_azure_ai._chat_client.BingGroundingTool") as mock_bing_grounding:
mock_bing_tool = MagicMock()
mock_bing_tool.definitions = [{"type": "bing_grounding"}]
mock_bing_grounding.return_value = mock_bing_tool
web_search_tool = client.get_web_search_tool(bing_connection_id="direct-connection-id")
mock_bing_grounding.assert_called_once_with(connection_id="direct-connection-id")
result = await client._prepare_tools_for_azure_ai([web_search_tool]) # type: ignore
assert len(result) == 1
assert result[0] == {"type": "bing_grounding"}
mock_bing_grounding.assert_called_once_with(connection_id="direct-connection-id", count=3)
async def test_azure_ai_chat_client_prepare_tools_for_azure_ai_web_search_custom_bing(
mock_agents_client: MagicMock,
) -> None:
"""Test _prepare_tools_for_azure_ai with HostedWebSearchTool using Custom Bing Search."""
"""Test _prepare_tools_for_azure_ai with BingCustomSearchTool from get_web_search_tool()."""
client = create_test_azure_ai_chat_client(mock_agents_client, agent_id="test-agent")
web_search_tool = HostedWebSearchTool(
additional_properties={
"custom_connection_id": "custom-connection-id",
"custom_instance_name": "custom-instance",
"count": 10,
}
)
# Mock BingCustomSearchTool
# Mock BingCustomSearchTool to avoid SDK validation
with patch("agent_framework_azure_ai._chat_client.BingCustomSearchTool") as mock_custom_bing:
mock_custom_tool = MagicMock()
mock_custom_tool.definitions = [{"type": "bing_custom_search"}]
mock_custom_bing.return_value = mock_custom_tool
web_search_tool = client.get_web_search_tool(
bing_custom_connection_id="custom-connection-id",
bing_custom_instance_id="custom-instance",
)
mock_custom_bing.assert_called_once_with(
connection_id="custom-connection-id",
instance_name="custom-instance",
)
result = await client._prepare_tools_for_azure_ai([web_search_tool]) # type: ignore
assert len(result) == 1
@@ -871,27 +850,19 @@ async def test_azure_ai_chat_client_prepare_tools_for_azure_ai_web_search_custom
async def test_azure_ai_chat_client_prepare_tools_for_azure_ai_file_search_with_vector_stores(
mock_agents_client: MagicMock,
) -> None:
"""Test _prepare_tools_for_azure_ai with HostedFileSearchTool using vector stores."""
"""Test _prepare_tools_for_azure_ai with FileSearchTool from get_file_search_tool()."""
client = create_test_azure_ai_chat_client(mock_agents_client, agent_id="test-agent")
vector_store_input = Content.from_hosted_vector_store(vector_store_id="vs-123")
file_search_tool = HostedFileSearchTool(inputs=[vector_store_input])
# get_file_search_tool() now returns a FileSearchTool instance directly
file_search_tool = client.get_file_search_tool(vector_store_ids=["vs-123"])
# Mock FileSearchTool
with patch("agent_framework_azure_ai._chat_client.FileSearchTool") as mock_file_search:
mock_file_tool = MagicMock()
mock_file_tool.definitions = [{"type": "file_search"}]
mock_file_tool.resources = {"vector_store_ids": ["vs-123"]}
mock_file_search.return_value = mock_file_tool
run_options: dict[str, Any] = {}
result = await client._prepare_tools_for_azure_ai([file_search_tool], run_options) # type: ignore
run_options = {}
result = await client._prepare_tools_for_azure_ai([file_search_tool], run_options) # type: ignore
assert len(result) == 1
assert result[0] == {"type": "file_search"}
assert run_options["tool_resources"] == {"vector_store_ids": ["vs-123"]}
mock_file_search.assert_called_once_with(vector_store_ids=["vs-123"])
assert len(result) == 1
assert result[0] == {"type": "file_search"}
assert run_options["tool_resources"] == {"file_search": {"vector_store_ids": ["vs-123"]}}
async def test_azure_ai_chat_client_create_agent_stream_submit_tool_approvals(
@@ -1615,7 +1586,7 @@ async def test_azure_ai_chat_client_agent_code_interpreter():
async with Agent(
client=AzureAIAgentClient(credential=AzureCliCredential()),
instructions="You are a helpful assistant that can write and execute Python code.",
tools=[HostedCodeInterpreterTool()],
tools=[AzureAIAgentClient.get_code_interpreter_tool()],
) as agent:
# Request code execution
response = await agent.run("Write Python code to calculate the factorial of 5 and show the result.")
@@ -1645,9 +1616,7 @@ async def test_azure_ai_chat_client_agent_file_search():
)
# 2. Create file search tool with uploaded resources
file_search_tool = HostedFileSearchTool(
inputs=[Content.from_hosted_vector_store(vector_store_id=vector_store.id)]
)
file_search_tool = AzureAIAgentClient.get_file_search_tool(vector_store_ids=[vector_store.id])
async with Agent(
client=client,
@@ -1679,9 +1648,9 @@ async def test_azure_ai_chat_client_agent_file_search():
@skip_if_azure_ai_integration_tests_disabled
async def test_azure_ai_chat_client_agent_hosted_mcp_tool() -> None:
"""Integration test for HostedMCPTool with Azure AI Agent using Microsoft Learn MCP."""
"""Integration test for MCP tool with Azure AI Agent using Microsoft Learn MCP."""
mcp_tool = HostedMCPTool(
mcp_tool = AzureAIAgentClient.get_mcp_tool(
name="Microsoft Learn MCP",
url="https://learn.microsoft.com/api/mcp",
description="A Microsoft Learn MCP server for documentation questions",
@@ -2066,11 +2035,11 @@ def test_azure_ai_chat_client_prepare_mcp_resources_with_dict_approval_mode(
"""Test _prepare_mcp_resources with dict-based approval mode (always_require_approval)."""
client = create_test_azure_ai_chat_client(mock_agents_client)
# MCP tool with dict-based approval mode
mcp_tool = HostedMCPTool(
# MCP tool with dict-based approval mode - use approval_mode parameter
mcp_tool = AzureAIAgentClient.get_mcp_tool(
name="Test MCP",
url="https://example.com/mcp",
approval_mode={"always_require_approval": {"tool1", "tool2"}},
approval_mode={"always_require_approval": ["tool1", "tool2"]},
)
result = client._prepare_mcp_resources([mcp_tool]) # type: ignore
@@ -2078,7 +2047,6 @@ def test_azure_ai_chat_client_prepare_mcp_resources_with_dict_approval_mode(
assert len(result) == 1
assert result[0]["server_label"] == "Test_MCP"
assert "require_approval" in result[0]
assert result[0]["require_approval"] == {"always": {"tool1", "tool2"}}
def test_azure_ai_chat_client_prepare_mcp_resources_with_never_require_dict(
@@ -2087,17 +2055,17 @@ def test_azure_ai_chat_client_prepare_mcp_resources_with_never_require_dict(
"""Test _prepare_mcp_resources with dict-based approval mode (never_require_approval)."""
client = create_test_azure_ai_chat_client(mock_agents_client)
# MCP tool with never_require_approval dict
mcp_tool = HostedMCPTool(
# MCP tool with never require approval - use approval_mode parameter
mcp_tool = AzureAIAgentClient.get_mcp_tool(
name="Test MCP",
url="https://example.com/mcp",
approval_mode={"never_require_approval": {"safe_tool"}},
approval_mode={"never_require_approval": ["safe_tool"]},
)
result = client._prepare_mcp_resources([mcp_tool]) # type: ignore
assert len(result) == 1
assert result[0]["require_approval"] == {"never": {"safe_tool"}}
assert "require_approval" in result[0]
def test_azure_ai_chat_client_prepare_messages_with_function_result(
@@ -2140,13 +2108,12 @@ def test_azure_ai_chat_client_prepare_messages_with_raw_content_block(
async def test_azure_ai_chat_client_prepare_tools_for_azure_ai_mcp_tool(
mock_agents_client: MagicMock,
) -> None:
"""Test _prepare_tools_for_azure_ai with HostedMCPTool."""
"""Test _prepare_tools_for_azure_ai with MCP dict tool."""
client = create_test_azure_ai_chat_client(mock_agents_client, agent_id="test-agent")
mcp_tool = HostedMCPTool(
mcp_tool = AzureAIAgentClient.get_mcp_tool(
name="Test MCP Server",
url="https://example.com/mcp",
allowed_tools=["tool1", "tool2"],
)
tool_definitions = await client._prepare_tools_for_azure_ai([mcp_tool]) # type: ignore
@@ -2191,14 +2158,16 @@ async def test_azure_ai_chat_client_prepare_tools_for_azure_ai_dict_passthrough(
async def test_azure_ai_chat_client_prepare_tools_for_azure_ai_unsupported_type(
mock_agents_client: MagicMock,
) -> None:
"""Test _prepare_tools_for_azure_ai raises error for unsupported tool type."""
"""Test _prepare_tools_for_azure_ai passes through unsupported tool types."""
client = create_test_azure_ai_chat_client(mock_agents_client, agent_id="test-agent")
# Pass an unsupported tool type
# Pass an unsupported tool type - it should be passed through unchanged
class UnsupportedTool:
pass
unsupported_tool = UnsupportedTool()
with pytest.raises(ServiceInitializationError, match="Unsupported tool type"):
await client._prepare_tools_for_azure_ai([unsupported_tool]) # type: ignore
# Unsupported tools are now passed through unchanged (server will reject if invalid)
tool_definitions = await client._prepare_tools_for_azure_ai([unsupported_tool]) # type: ignore
assert len(tool_definitions) == 1
assert tool_definitions[0] is unsupported_tool
python/packages/azure-ai/tests/test_azure_ai_client.py
+140 -165
View File
@@ -16,10 +16,6 @@ from agent_framework import (
ChatOptions,
ChatResponse,
Content,
HostedCodeInterpreterTool,
HostedFileSearchTool,
HostedMCPTool,
HostedWebSearchTool,
Message,
SupportsChatGetResponse,
tool,
@@ -31,6 +27,7 @@ from azure.ai.projects.models import (
CodeInterpreterTool,
CodeInterpreterToolAuto,
FileSearchTool,
ImageGenTool,
MCPTool,
ResponseTextFormatConfigurationJsonSchema,
WebSearchPreviewTool,
@@ -1100,178 +1097,50 @@ def test_get_conversation_id_with_parsed_response_no_conversation() -> None:
assert result == "resp_parsed_12345"
def test_prepare_mcp_tool_basic() -> None:
"""Test _prepare_mcp_tool with basic HostedMCPTool."""
mcp_tool = HostedMCPTool(
name="Test MCP Server",
url="https://example.com/mcp",
)
result = AzureAIClient._prepare_mcp_tool(mcp_tool) # type: ignore
assert result["server_label"] == "Test_MCP_Server"
assert result["server_url"] == "https://example.com/mcp"
# region MCP Tool Dict Tests
# These tests verify that dict-based MCP tools are processed correctly by from_azure_ai_tools
def test_prepare_mcp_tool_with_description() -> None:
"""Test _prepare_mcp_tool with description."""
mcp_tool = HostedMCPTool(
name="Test MCP",
url="https://example.com/mcp",
description="A test MCP server",
)
result = AzureAIClient._prepare_mcp_tool(mcp_tool) # type: ignore
assert result["server_description"] == "A test MCP server"
def test_prepare_mcp_tool_with_project_connection_id() -> None:
"""Test _prepare_mcp_tool with project_connection_id in additional_properties."""
mcp_tool = HostedMCPTool(
name="Test MCP",
url="https://example.com/mcp",
additional_properties={"project_connection_id": "conn-123"},
)
result = AzureAIClient._prepare_mcp_tool(mcp_tool) # type: ignore
assert result["project_connection_id"] == "conn-123"
assert "headers" not in result # headers should not be set when project_connection_id is present
def test_prepare_mcp_tool_with_headers() -> None:
"""Test _prepare_mcp_tool with headers (no project_connection_id)."""
mcp_tool = HostedMCPTool(
name="Test MCP",
url="https://example.com/mcp",
headers={"Authorization": "Bearer token123"},
)
result = AzureAIClient._prepare_mcp_tool(mcp_tool) # type: ignore
assert result["headers"] == {"Authorization": "Bearer token123"}
def test_prepare_mcp_tool_with_allowed_tools() -> None:
"""Test _prepare_mcp_tool with allowed_tools."""
mcp_tool = HostedMCPTool(
name="Test MCP",
url="https://example.com/mcp",
allowed_tools=["tool1", "tool2"],
)
result = AzureAIClient._prepare_mcp_tool(mcp_tool) # type: ignore
assert set(result["allowed_tools"]) == {"tool1", "tool2"}
def test_prepare_mcp_tool_with_approval_mode_always_require() -> None:
"""Test _prepare_mcp_tool with string approval_mode 'always_require'."""
mcp_tool = HostedMCPTool(
name="Test MCP",
url="https://example.com/mcp",
approval_mode="always_require",
)
result = AzureAIClient._prepare_mcp_tool(mcp_tool) # type: ignore
assert result["require_approval"] == "always"
def test_prepare_mcp_tool_with_approval_mode_never_require() -> None:
"""Test _prepare_mcp_tool with string approval_mode 'never_require'."""
mcp_tool = HostedMCPTool(
name="Test MCP",
url="https://example.com/mcp",
approval_mode="never_require",
)
result = AzureAIClient._prepare_mcp_tool(mcp_tool) # type: ignore
assert result["require_approval"] == "never"
def test_prepare_mcp_tool_with_dict_approval_mode_always() -> None:
"""Test _prepare_mcp_tool with dict approval_mode containing always_require_approval."""
mcp_tool = HostedMCPTool(
name="Test MCP",
url="https://example.com/mcp",
approval_mode={"always_require_approval": {"dangerous_tool", "risky_tool"}},
)
result = AzureAIClient._prepare_mcp_tool(mcp_tool) # type: ignore
assert "require_approval" in result
assert "always" in result["require_approval"]
assert set(result["require_approval"]["always"]["tool_names"]) == {"dangerous_tool", "risky_tool"}
def test_prepare_mcp_tool_with_dict_approval_mode_never() -> None:
"""Test _prepare_mcp_tool with dict approval_mode containing never_require_approval."""
mcp_tool = HostedMCPTool(
name="Test MCP",
url="https://example.com/mcp",
approval_mode={"never_require_approval": {"safe_tool"}},
)
result = AzureAIClient._prepare_mcp_tool(mcp_tool) # type: ignore
assert "require_approval" in result
assert "never" in result["require_approval"]
assert set(result["require_approval"]["never"]["tool_names"]) == {"safe_tool"}
def test_from_azure_ai_tools() -> None:
"""Test from_azure_ai_tools."""
# Test MCP tool
def test_from_azure_ai_tools_mcp() -> None:
"""Test from_azure_ai_tools with MCP tool."""
mcp_tool = MCPTool(server_label="test_server", server_url="http://localhost:8080")
parsed_tools = from_azure_ai_tools([mcp_tool])
assert len(parsed_tools) == 1
assert isinstance(parsed_tools[0], HostedMCPTool)
assert parsed_tools[0].name == "test server"
assert str(parsed_tools[0].url).rstrip("/") == "http://localhost:8080"
assert parsed_tools[0]["type"] == "mcp"
assert parsed_tools[0]["server_label"] == "test_server"
assert parsed_tools[0]["server_url"] == "http://localhost:8080"
# Test Code Interpreter tool
def test_from_azure_ai_tools_code_interpreter() -> None:
"""Test from_azure_ai_tools with Code Interpreter tool."""
ci_tool = CodeInterpreterTool(container=CodeInterpreterToolAuto(file_ids=["file-1"]))
parsed_tools = from_azure_ai_tools([ci_tool])
assert len(parsed_tools) == 1
assert isinstance(parsed_tools[0], HostedCodeInterpreterTool)
assert parsed_tools[0].inputs is not None
assert len(parsed_tools[0].inputs) == 1
assert parsed_tools[0]["type"] == "code_interpreter"
tool_input = parsed_tools[0].inputs[0]
assert tool_input and tool_input.type == "hosted_file" and tool_input.file_id == "file-1"
# Test File Search tool
def test_from_azure_ai_tools_file_search() -> None:
"""Test from_azure_ai_tools with File Search tool."""
fs_tool = FileSearchTool(vector_store_ids=["vs-1"], max_num_results=5)
parsed_tools = from_azure_ai_tools([fs_tool])
assert len(parsed_tools) == 1
assert isinstance(parsed_tools[0], HostedFileSearchTool)
assert parsed_tools[0].inputs is not None
assert len(parsed_tools[0].inputs) == 1
assert parsed_tools[0]["type"] == "file_search"
assert parsed_tools[0]["vector_store_ids"] == ["vs-1"]
assert parsed_tools[0]["max_num_results"] == 5
tool_input = parsed_tools[0].inputs[0]
assert tool_input and tool_input.type == "hosted_vector_store" and tool_input.vector_store_id == "vs-1"
assert parsed_tools[0].max_results == 5
# Test Web Search tool
def test_from_azure_ai_tools_web_search() -> None:
"""Test from_azure_ai_tools with Web Search tool."""
ws_tool = WebSearchPreviewTool(
user_location=ApproximateLocation(city="Seattle", country="US", region="WA", timezone="PST")
)
parsed_tools = from_azure_ai_tools([ws_tool])
assert len(parsed_tools) == 1
assert isinstance(parsed_tools[0], HostedWebSearchTool)
assert parsed_tools[0].additional_properties
assert parsed_tools[0]["type"] == "web_search_preview"
assert parsed_tools[0]["user_location"]["city"] == "Seattle"
user_location = parsed_tools[0].additional_properties["user_location"]
assert user_location["city"] == "Seattle"
assert user_location["country"] == "US"
assert user_location["region"] == "WA"
assert user_location["timezone"] == "PST"
# endregion
# region Integration Tests
@@ -1535,7 +1404,7 @@ async def test_integration_web_search() -> None:
"messages": "Who are the main characters of Kpop Demon Hunters? Do a web search to find the answer.",
"options": {
"tool_choice": "auto",
"tools": [HostedWebSearchTool()],
"tools": [client.get_web_search_tool()],
},
}
if streaming:
@@ -1550,17 +1419,11 @@ async def test_integration_web_search() -> None:
assert "Zoey" in response.text
# Test that the client will use the web search tool with location
additional_properties = {
"user_location": {
"country": "US",
"city": "Seattle",
}
}
content = {
"messages": "What is the current weather? Do not ask for my current location.",
"options": {
"tool_choice": "auto",
"tools": [HostedWebSearchTool(additional_properties=additional_properties)],
"tools": [client.get_web_search_tool(user_location={"country": "US", "city": "Seattle"})],
},
}
if streaming:
@@ -1573,14 +1436,14 @@ async def test_integration_web_search() -> None:
@pytest.mark.flaky
@skip_if_azure_ai_integration_tests_disabled
async def test_integration_agent_hosted_mcp_tool() -> None:
"""Integration test for HostedMCPTool with Azure Response Agent using Microsoft Learn MCP."""
"""Integration test for MCP tool with Azure Response Agent using Microsoft Learn MCP."""
async with temporary_chat_client(agent_name="af-int-test-mcp") as client:
response = await client.get_response(
"How to create an Azure storage account using az cli?",
options={
# this needs to be high enough to handle the full MCP tool response.
"max_tokens": 5000,
"tools": HostedMCPTool(
"tools": client.get_mcp_tool(
name="Microsoft Learn MCP",
url="https://learn.microsoft.com/api/mcp",
description="A Microsoft Learn MCP server for documentation questions",
@@ -1597,12 +1460,12 @@ async def test_integration_agent_hosted_mcp_tool() -> None:
@pytest.mark.flaky
@skip_if_azure_ai_integration_tests_disabled
async def test_integration_agent_hosted_code_interpreter_tool():
"""Test Azure Responses Client agent with HostedCodeInterpreterTool through AzureAIClient."""
"""Test Azure Responses Client agent with code interpreter tool through AzureAIClient."""
async with temporary_chat_client(agent_name="af-int-test-code-interpreter") as client:
response = await client.get_response(
"Calculate the sum of numbers from 1 to 10 using Python code.",
options={
"tools": [HostedCodeInterpreterTool()],
"tools": [client.get_code_interpreter_tool()],
},
)
# Should contain calculation result (sum of 1-10 = 55) or code execution content
@@ -1651,3 +1514,115 @@ async def test_integration_agent_existing_thread():
assert isinstance(second_response, AgentResponse)
assert second_response.text is not None
assert "photography" in second_response.text.lower()
# region Factory Method Tests
def test_get_code_interpreter_tool_basic() -> None:
"""Test get_code_interpreter_tool returns CodeInterpreterTool."""
tool = AzureAIClient.get_code_interpreter_tool()
assert isinstance(tool, CodeInterpreterTool)
def test_get_code_interpreter_tool_with_file_ids() -> None:
"""Test get_code_interpreter_tool with file_ids."""
tool = AzureAIClient.get_code_interpreter_tool(file_ids=["file-123", "file-456"])
assert isinstance(tool, CodeInterpreterTool)
assert tool["container"]["file_ids"] == ["file-123", "file-456"]
def test_get_file_search_tool_basic() -> None:
"""Test get_file_search_tool returns FileSearchTool."""
tool = AzureAIClient.get_file_search_tool(vector_store_ids=["vs-123"])
assert isinstance(tool, FileSearchTool)
assert tool["vector_store_ids"] == ["vs-123"]
def test_get_file_search_tool_with_options() -> None:
"""Test get_file_search_tool with max_num_results."""
tool = AzureAIClient.get_file_search_tool(
vector_store_ids=["vs-123"],
max_num_results=10,
)
assert isinstance(tool, FileSearchTool)
assert tool["max_num_results"] == 10
def test_get_file_search_tool_requires_vector_store_ids() -> None:
"""Test get_file_search_tool raises ValueError when vector_store_ids is empty."""
with pytest.raises(ValueError, match="vector_store_ids"):
AzureAIClient.get_file_search_tool(vector_store_ids=[])
def test_get_web_search_tool_basic() -> None:
"""Test get_web_search_tool returns WebSearchPreviewTool."""
tool = AzureAIClient.get_web_search_tool()
assert isinstance(tool, WebSearchPreviewTool)
def test_get_web_search_tool_with_location() -> None:
"""Test get_web_search_tool with user_location."""
tool = AzureAIClient.get_web_search_tool(
user_location={"city": "Seattle", "country": "US"},
)
assert isinstance(tool, WebSearchPreviewTool)
assert tool.user_location is not None
assert tool.user_location.city == "Seattle"
assert tool.user_location.country == "US"
def test_get_web_search_tool_with_search_context_size() -> None:
"""Test get_web_search_tool with search_context_size."""
tool = AzureAIClient.get_web_search_tool(search_context_size="high")
assert isinstance(tool, WebSearchPreviewTool)
assert tool.search_context_size == "high"
def test_get_mcp_tool_basic() -> None:
"""Test get_mcp_tool returns MCPTool."""
tool = AzureAIClient.get_mcp_tool(name="test_mcp", url="https://example.com")
assert isinstance(tool, MCPTool)
assert tool["server_label"] == "test_mcp"
assert tool["server_url"] == "https://example.com"
def test_get_mcp_tool_with_description() -> None:
"""Test get_mcp_tool with description."""
tool = AzureAIClient.get_mcp_tool(
name="test_mcp",
url="https://example.com",
description="Test MCP server",
)
assert tool["server_description"] == "Test MCP server"
def test_get_mcp_tool_with_project_connection_id() -> None:
"""Test get_mcp_tool with project_connection_id."""
tool = AzureAIClient.get_mcp_tool(
name="test_mcp",
project_connection_id="conn-123",
)
assert tool["project_connection_id"] == "conn-123"
def test_get_image_generation_tool_basic() -> None:
"""Test get_image_generation_tool returns ImageGenTool."""
tool = AzureAIClient.get_image_generation_tool()
assert isinstance(tool, ImageGenTool)
def test_get_image_generation_tool_with_options() -> None:
"""Test get_image_generation_tool with various options."""
tool = AzureAIClient.get_image_generation_tool(
size="1024x1024",
quality="high",
output_format="png",
)
assert isinstance(tool, ImageGenTool)
assert tool["size"] == "1024x1024"
assert tool["quality"] == "high"
assert tool["output_format"] == "png"
# endregion
python/packages/azure-ai/tests/test_provider.py
+4 -6
View File
@@ -440,19 +440,17 @@ def test_provider_merge_tools_skips_function_tool_dicts(mock_project_client: Mag
# Call _merge_tools with user-provided function implementation
merged = provider._merge_tools(definition_tools, [mock_ai_function]) # type: ignore
# Should have 2 items: the converted HostedMCPTool and the user-provided FunctionTool
# Should have 2 items: the converted MCP dict and the user-provided FunctionTool
assert len(merged) == 2
# Check that the function tool dict was NOT included (it was skipped)
function_dicts = [t for t in merged if isinstance(t, dict) and t.get("type") == "function"]
assert len(function_dicts) == 0
# Check that the MCP tool was converted to HostedMCPTool
from agent_framework import HostedMCPTool
mcp_tools = [t for t in merged if isinstance(t, HostedMCPTool)]
# Check that the MCP tool was converted to dict
mcp_tools = [t for t in merged if isinstance(t, dict) and t.get("type") == "mcp"]
assert len(mcp_tools) == 1
assert mcp_tools[0].name == "my mcp" # server_label with _ replaced by space
assert mcp_tools[0]["server_label"] == "my_mcp"
# Check that the user-provided FunctionTool was included
ai_functions = [t for t in merged if isinstance(t, FunctionTool)]
python/packages/azure-ai/tests/test_shared.py
+102 -82
View File
@@ -5,29 +5,26 @@ from unittest.mock import MagicMock, patch
import pytest
from agent_framework import (
Content,
FunctionTool,
HostedCodeInterpreterTool,
HostedFileSearchTool,
HostedImageGenerationTool,
HostedMCPTool,
HostedWebSearchTool,
)
from agent_framework.exceptions import ServiceInitializationError, ServiceInvalidRequestError
from agent_framework.exceptions import ServiceInvalidRequestError
from azure.ai.agents.models import CodeInterpreterToolDefinition
from pydantic import BaseModel
from agent_framework_azure_ai import AzureAIAgentClient
from agent_framework_azure_ai._shared import (
_convert_response_format, # type: ignore
_convert_sdk_tool, # type: ignore
_extract_project_connection_id, # type: ignore
_prepare_mcp_tool_for_azure_ai, # type: ignore
create_text_format_config,
from_azure_ai_agent_tools,
from_azure_ai_tools,
to_azure_ai_agent_tools,
to_azure_ai_tools,
)
from agent_framework_azure_ai._shared import (
_prepare_mcp_tool_dict_for_azure_ai as _prepare_mcp_tool_for_azure_ai, # type: ignore
)
def test_extract_project_connection_id_direct() -> None:
@@ -69,16 +66,15 @@ def test_to_azure_ai_agent_tools_function_tool() -> None:
def test_to_azure_ai_agent_tools_code_interpreter() -> None:
"""Test converting HostedCodeInterpreterTool."""
tool = HostedCodeInterpreterTool()
"""Test converting code_interpreter dict tool."""
tool = AzureAIAgentClient.get_code_interpreter_tool()
result = to_azure_ai_agent_tools([tool])
assert len(result) == 1
assert isinstance(result[0], CodeInterpreterToolDefinition)
def test_to_azure_ai_agent_tools_web_search_missing_connection() -> None:
"""Test HostedWebSearchTool raises without connection info."""
tool = HostedWebSearchTool()
"""Test web search tool raises without connection info."""
# Clear any environment variables that could provide connection info
with patch.dict(
os.environ,
@@ -90,8 +86,9 @@ def test_to_azure_ai_agent_tools_web_search_missing_connection() -> None:
for key in ["BING_CONNECTION_ID", "BING_CUSTOM_CONNECTION_ID", "BING_CUSTOM_INSTANCE_NAME"]:
env_backup[key] = os.environ.pop(key, None)
try:
with pytest.raises(ServiceInitializationError, match="Bing search tool requires"):
to_azure_ai_agent_tools([tool])
# get_web_search_tool now raises ValueError when no connection info is available
with pytest.raises(ValueError, match="Azure AI Agents requires a Bing connection"):
AzureAIAgentClient.get_web_search_tool()
finally:
# Restore environment
for key, value in env_backup.items():
@@ -107,13 +104,15 @@ def test_to_azure_ai_agent_tools_dict_passthrough() -> None:
def test_to_azure_ai_agent_tools_unsupported_type() -> None:
"""Test unsupported tool type raises error."""
"""Test unsupported tool type passes through unchanged."""
class UnsupportedTool:
pass
with pytest.raises(ServiceInitializationError, match="Unsupported tool type"):
to_azure_ai_agent_tools([UnsupportedTool()]) # type: ignore
unsupported = UnsupportedTool()
result = to_azure_ai_agent_tools([unsupported]) # type: ignore
assert len(result) == 1
assert result[0] is unsupported # Passed through unchanged
def test_from_azure_ai_agent_tools_empty() -> None:
@@ -127,7 +126,7 @@ def test_from_azure_ai_agent_tools_code_interpreter() -> None:
tool = CodeInterpreterToolDefinition()
result = from_azure_ai_agent_tools([tool])
assert len(result) == 1
assert isinstance(result[0], HostedCodeInterpreterTool)
assert result[0] == {"type": "code_interpreter"}
def test_convert_sdk_tool_code_interpreter() -> None:
@@ -135,7 +134,7 @@ def test_convert_sdk_tool_code_interpreter() -> None:
tool = MagicMock()
tool.type = "code_interpreter"
result = _convert_sdk_tool(tool)
assert isinstance(result, HostedCodeInterpreterTool)
assert result == {"type": "code_interpreter"}
def test_convert_sdk_tool_function_returns_none() -> None:
@@ -161,8 +160,8 @@ def test_convert_sdk_tool_file_search() -> None:
tool.file_search = MagicMock()
tool.file_search.vector_store_ids = ["vs-1", "vs-2"]
result = _convert_sdk_tool(tool)
assert isinstance(result, HostedFileSearchTool)
assert len(result.inputs) == 2 # type: ignore
assert result["type"] == "file_search"
assert result["vector_store_ids"] == ["vs-1", "vs-2"]
def test_convert_sdk_tool_bing_grounding() -> None:
@@ -172,8 +171,8 @@ def test_convert_sdk_tool_bing_grounding() -> None:
tool.bing_grounding = MagicMock()
tool.bing_grounding.connection_id = "conn-123"
result = _convert_sdk_tool(tool)
assert isinstance(result, HostedWebSearchTool)
assert result.additional_properties["connection_id"] == "conn-123" # type: ignore
assert result["type"] == "bing_grounding"
assert result["connection_id"] == "conn-123"
def test_convert_sdk_tool_bing_custom_search() -> None:
@@ -184,9 +183,9 @@ def test_convert_sdk_tool_bing_custom_search() -> None:
tool.bing_custom_search.connection_id = "conn-123"
tool.bing_custom_search.instance_name = "my-instance"
result = _convert_sdk_tool(tool)
assert isinstance(result, HostedWebSearchTool)
assert result.additional_properties["custom_connection_id"] == "conn-123" # type: ignore
assert result.additional_properties["custom_instance_name"] == "my-instance" # type: ignore
assert result["type"] == "bing_custom_search"
assert result["connection_id"] == "conn-123"
assert result["instance_name"] == "my-instance"
def test_to_azure_ai_tools_empty() -> None:
@@ -196,14 +195,14 @@ def test_to_azure_ai_tools_empty() -> None:
def test_to_azure_ai_tools_code_interpreter_with_file_ids() -> None:
"""Test converting HostedCodeInterpreterTool with file inputs."""
tool = HostedCodeInterpreterTool(
inputs=[Content.from_hosted_file(file_id="file-123")] # type: ignore
)
"""Test converting code_interpreter dict tool with file inputs."""
tool = {
"type": "code_interpreter",
"file_ids": ["file-123"],
}
result = to_azure_ai_tools([tool])
assert len(result) == 1
assert result[0]["type"] == "code_interpreter"
assert result[0]["container"]["file_ids"] == ["file-123"]
def test_to_azure_ai_tools_function_tool() -> None:
@@ -221,11 +220,12 @@ def test_to_azure_ai_tools_function_tool() -> None:
def test_to_azure_ai_tools_file_search() -> None:
"""Test converting HostedFileSearchTool."""
tool = HostedFileSearchTool(
inputs=[Content.from_hosted_vector_store(vector_store_id="vs-123")], # type: ignore
max_results=10,
)
"""Test converting file_search dict tool."""
tool = {
"type": "file_search",
"vector_store_ids": ["vs-123"],
"max_num_results": 10,
}
result = to_azure_ai_tools([tool])
assert len(result) == 1
assert result[0]["type"] == "file_search"
@@ -234,28 +234,29 @@ def test_to_azure_ai_tools_file_search() -> None:
def test_to_azure_ai_tools_web_search_with_location() -> None:
"""Test converting HostedWebSearchTool with user location."""
tool = HostedWebSearchTool(
additional_properties={
"user_location": {
"city": "Seattle",
"country": "US",
"region": "WA",
"timezone": "PST",
}
}
)
"""Test converting web_search dict tool with user location."""
tool = {
"type": "web_search_preview",
"user_location": {
"city": "Seattle",
"country": "US",
"region": "WA",
"timezone": "PST",
},
}
result = to_azure_ai_tools([tool])
assert len(result) == 1
assert result[0]["type"] == "web_search_preview"
def test_to_azure_ai_tools_image_generation() -> None:
"""Test converting HostedImageGenerationTool."""
tool = HostedImageGenerationTool(
options={"model_id": "gpt-image-1", "image_size": "1024x1024"},
additional_properties={"quality": "high"},
)
"""Test converting image_generation dict tool."""
tool = {
"type": "image_generation",
"model": "gpt-image-1",
"size": "1024x1024",
"quality": "high",
}
result = to_azure_ai_tools([tool])
assert len(result) == 1
assert result[0]["type"] == "image_generation"
@@ -264,7 +265,7 @@ def test_to_azure_ai_tools_image_generation() -> None:
def test_prepare_mcp_tool_basic() -> None:
"""Test basic MCP tool conversion."""
tool = HostedMCPTool(name="my tool", url="http://localhost:8080")
tool = {"type": "mcp", "server_label": "my_tool", "server_url": "http://localhost:8080"}
result = _prepare_mcp_tool_for_azure_ai(tool)
assert result["server_label"] == "my_tool"
assert "http://localhost:8080" in result["server_url"]
@@ -272,26 +273,37 @@ def test_prepare_mcp_tool_basic() -> None:
def test_prepare_mcp_tool_with_description() -> None:
"""Test MCP tool with description."""
tool = HostedMCPTool(name="my tool", url="http://localhost:8080", description="My MCP server")
tool = {
"type": "mcp",
"server_label": "my_tool",
"server_url": "http://localhost:8080",
"server_description": "My MCP server",
}
result = _prepare_mcp_tool_for_azure_ai(tool)
assert result["server_description"] == "My MCP server"
def test_prepare_mcp_tool_with_headers() -> None:
"""Test MCP tool with headers (no project_connection_id)."""
tool = HostedMCPTool(name="my tool", url="http://localhost:8080", headers={"X-Api-Key": "secret"})
tool = {
"type": "mcp",
"server_label": "my_tool",
"server_url": "http://localhost:8080",
"headers": {"X-Api-Key": "secret"},
}
result = _prepare_mcp_tool_for_azure_ai(tool)
assert result["headers"] == {"X-Api-Key": "secret"}
def test_prepare_mcp_tool_project_connection_takes_precedence() -> None:
"""Test project_connection_id takes precedence over headers."""
tool = HostedMCPTool(
name="my tool",
url="http://localhost:8080",
headers={"X-Api-Key": "secret"},
additional_properties={"project_connection_id": "my-conn"},
)
tool = {
"type": "mcp",
"server_label": "my_tool",
"server_url": "http://localhost:8080",
"headers": {"X-Api-Key": "secret"},
"project_connection_id": "my-conn",
}
result = _prepare_mcp_tool_for_azure_ai(tool)
assert result["project_connection_id"] == "my-conn"
assert "headers" not in result
@@ -299,30 +311,38 @@ def test_prepare_mcp_tool_project_connection_takes_precedence() -> None:
def test_prepare_mcp_tool_approval_mode_always() -> None:
"""Test MCP tool with always_require approval mode."""
tool = HostedMCPTool(name="my tool", url="http://localhost:8080", approval_mode="always_require")
tool = {
"type": "mcp",
"server_label": "my_tool",
"server_url": "http://localhost:8080",
"require_approval": "always",
}
result = _prepare_mcp_tool_for_azure_ai(tool)
assert result["require_approval"] == "always"
def test_prepare_mcp_tool_approval_mode_never() -> None:
"""Test MCP tool with never_require approval mode."""
tool = HostedMCPTool(name="my tool", url="http://localhost:8080", approval_mode="never_require")
tool = {
"type": "mcp",
"server_label": "my_tool",
"server_url": "http://localhost:8080",
"require_approval": "never",
}
result = _prepare_mcp_tool_for_azure_ai(tool)
assert result["require_approval"] == "never"
def test_prepare_mcp_tool_approval_mode_dict() -> None:
"""Test MCP tool with dict approval mode."""
tool = HostedMCPTool(
name="my tool",
url="http://localhost:8080",
approval_mode={
"always_require_approval": {"sensitive_tool"},
"never_require_approval": {"safe_tool"},
},
)
tool = {
"type": "mcp",
"server_label": "my_tool",
"server_url": "http://localhost:8080",
"require_approval": {"always": {"tool_names": ["sensitive_tool", "dangerous_tool"]}},
}
result = _prepare_mcp_tool_for_azure_ai(tool)
# The last assignment wins in the current implementation
# The approval mode is passed through
assert "require_approval" in result
@@ -385,7 +405,7 @@ def test_convert_response_format_json_schema_missing_schema_raises() -> None:
def test_from_azure_ai_tools_mcp_approval_mode_always() -> None:
"""Test from_azure_ai_tools converts MCP require_approval='always' to approval_mode."""
"""Test from_azure_ai_tools converts MCP require_approval='always' to dict."""
tools = [
{
"type": "mcp",
@@ -396,12 +416,12 @@ def test_from_azure_ai_tools_mcp_approval_mode_always() -> None:
]
result = from_azure_ai_tools(tools)
assert len(result) == 1
assert isinstance(result[0], HostedMCPTool)
assert result[0].approval_mode == "always_require"
assert result[0]["type"] == "mcp"
assert result[0]["require_approval"] == "always"
def test_from_azure_ai_tools_mcp_approval_mode_never() -> None:
"""Test from_azure_ai_tools converts MCP require_approval='never' to approval_mode."""
"""Test from_azure_ai_tools converts MCP require_approval='never' to dict."""
tools = [
{
"type": "mcp",
@@ -412,8 +432,8 @@ def test_from_azure_ai_tools_mcp_approval_mode_never() -> None:
]
result = from_azure_ai_tools(tools)
assert len(result) == 1
assert isinstance(result[0], HostedMCPTool)
assert result[0].approval_mode == "never_require"
assert result[0]["type"] == "mcp"
assert result[0]["require_approval"] == "never"
def test_from_azure_ai_tools_mcp_approval_mode_dict_always() -> None:
@@ -428,8 +448,8 @@ def test_from_azure_ai_tools_mcp_approval_mode_dict_always() -> None:
]
result = from_azure_ai_tools(tools)
assert len(result) == 1
assert isinstance(result[0], HostedMCPTool)
assert result[0].approval_mode == {"always_require_approval": {"sensitive_tool", "dangerous_tool"}}
assert result[0]["type"] == "mcp"
assert result[0]["require_approval"] == {"always": {"tool_names": ["sensitive_tool", "dangerous_tool"]}}
def test_from_azure_ai_tools_mcp_approval_mode_dict_never() -> None:
@@ -444,5 +464,5 @@ def test_from_azure_ai_tools_mcp_approval_mode_dict_never() -> None:
]
result = from_azure_ai_tools(tools)
assert len(result) == 1
assert isinstance(result[0], HostedMCPTool)
assert result[0].approval_mode == {"never_require_approval": {"safe_tool"}}
assert result[0]["type"] == "mcp"
assert result[0]["require_approval"] == {"never": {"tool_names": ["safe_tool"]}}