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: Create/Get Agent API for Azure V2 (#3059)

Browse Source 
* Added get_agent method to Azure AI V2

* Small fixes

* Small fix

* Removed AzureAIAgentProvider

* Added create_agent method

* Small fixes

* Fixed code interpreter tool mapping

* Added agent provider for V2 client

* Updated response format handling

* Added provider example

* Fixed errors

* Update python/samples/getting_started/agents/azure_ai/README.md

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Small fix

* Updates from merge

* Resolved comments

* Resolved comments

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Dmytro Struk
2026-01-14 11:35:01 -08:00
committed by GitHub
Unverified
parent f56808b279
commit 99c5718696
35 changed files with 1903 additions and 336 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/packages/azure-ai/agent_framework_azure_ai/__init__.py
+2
View File
@@ -4,6 +4,7 @@ import importlib.metadata
from ._chat_client import AzureAIAgentClient, AzureAIAgentOptions
from ._client import AzureAIClient
from ._provider import AzureAIProjectAgentProvider
from ._shared import AzureAISettings
try:
@@ -15,6 +16,7 @@ __all__ = [
"AzureAIAgentClient",
"AzureAIAgentOptions",
"AzureAIClient",
"AzureAIProjectAgentProvider",
"AzureAISettings",
"__version__",
]
python/packages/azure-ai/agent_framework_azure_ai/_client.py
+5 -49
View File
@@ -2,7 +2,7 @@
import sys
from collections.abc import Mapping, MutableSequence
from typing import TYPE_CHECKING, Any, ClassVar, Generic, TypedDict, cast
from typing import TYPE_CHECKING, Any, ClassVar, Generic, TypedDict, TypeVar, cast
from agent_framework import (
AGENT_FRAMEWORK_USER_AGENT,
@@ -13,7 +13,7 @@ from agent_framework import (
use_chat_middleware,
use_function_invocation,
)
from agent_framework.exceptions import ServiceInitializationError, ServiceInvalidRequestError
from agent_framework.exceptions import ServiceInitializationError
from agent_framework.observability import use_instrumentation
from agent_framework.openai._responses_client import OpenAIBaseResponsesClient
from azure.ai.projects.aio import AIProjectClient
@@ -21,15 +21,12 @@ from azure.ai.projects.models import (
MCPTool,
PromptAgentDefinition,
PromptAgentDefinitionText,
ResponseTextFormatConfigurationJsonObject,
ResponseTextFormatConfigurationJsonSchema,
ResponseTextFormatConfigurationText,
)
from azure.core.credentials_async import AsyncTokenCredential
from azure.core.exceptions import ResourceNotFoundError
from pydantic import BaseModel, ValidationError
from pydantic import ValidationError
from ._shared import AzureAISettings
from ._shared import AzureAISettings, create_text_format_config
if TYPE_CHECKING:
from agent_framework.openai import OpenAIResponsesOptions
@@ -286,47 +283,6 @@ class AzureAIClient(OpenAIBaseResponsesClient[TAzureAIClientOptions], Generic[TA
"""Close the project_client."""
await self._close_client_if_needed()
def _create_text_format_config(
self, response_format: type[BaseModel] | Mapping[str, Any]
) -> (
ResponseTextFormatConfigurationJsonSchema
| ResponseTextFormatConfigurationJsonObject
| ResponseTextFormatConfigurationText
):
"""Convert response_format into Azure text format configuration."""
if isinstance(response_format, type) and issubclass(response_format, BaseModel):
schema = response_format.model_json_schema()
# Ensure additionalProperties is explicitly false to satisfy Azure validation
if isinstance(schema, dict):
schema.setdefault("additionalProperties", False)
return ResponseTextFormatConfigurationJsonSchema(
name=response_format.__name__,
schema=schema,
)
if isinstance(response_format, Mapping):
format_config = self._convert_response_format(response_format)
format_type = format_config.get("type")
if format_type == "json_schema":
# Ensure schema includes additionalProperties=False to satisfy Azure validation
schema = dict(format_config.get("schema", {})) # type: ignore[assignment]
schema.setdefault("additionalProperties", False)
config_kwargs: dict[str, Any] = {
"name": format_config.get("name") or "response",
"schema": schema,
}
if "strict" in format_config:
config_kwargs["strict"] = format_config["strict"]
if "description" in format_config:
config_kwargs["description"] = format_config["description"]
return ResponseTextFormatConfigurationJsonSchema(**config_kwargs)
if format_type == "json_object":
return ResponseTextFormatConfigurationJsonObject()
if format_type == "text":
return ResponseTextFormatConfigurationText()
raise ServiceInvalidRequestError("response_format must be a Pydantic model or mapping.")
async def _get_agent_reference_or_create(
self,
run_options: dict[str, Any],
@@ -380,7 +336,7 @@ class AzureAIClient(OpenAIBaseResponsesClient[TAzureAIClientOptions], Generic[TA
# response_format is accessed from chat_options or additional_properties
# since the base class excludes it from run_options
if chat_options and (response_format := chat_options.get("response_format")):
args["text"] = PromptAgentDefinitionText(format=self._create_text_format_config(response_format))
args["text"] = PromptAgentDefinitionText(format=create_text_format_config(response_format))
# Combine instructions from messages and options
combined_instructions = [
python/packages/azure-ai/agent_framework_azure_ai/_provider.py
+455
View File
@@ -0,0 +1,455 @@
# Copyright (c) Microsoft. All rights reserved.
import sys
from collections.abc import Callable, MutableMapping, Sequence
from typing import TYPE_CHECKING, Any, Generic, TypedDict
from agent_framework import (
AGENT_FRAMEWORK_USER_AGENT,
AIFunction,
ChatAgent,
ContextProvider,
Middleware,
ToolProtocol,
get_logger,
normalize_tools,
)
from agent_framework.exceptions import ServiceInitializationError
from azure.ai.projects.aio import AIProjectClient
from azure.ai.projects.models import (
AgentReference,
AgentVersionDetails,
FunctionTool,
PromptAgentDefinition,
PromptAgentDefinitionText,
)
from azure.core.credentials_async import AsyncTokenCredential
from pydantic import BaseModel, ValidationError
from ._client import AzureAIClient
from ._shared import AzureAISettings, create_text_format_config, from_azure_ai_tools, to_azure_ai_tools
if TYPE_CHECKING:
from agent_framework.openai import OpenAIResponsesOptions
if sys.version_info >= (3, 13):
from typing import Self, TypeVar # pragma: no cover
else:
from typing_extensions import Self, TypeVar # pragma: no cover
logger = get_logger("agent_framework.azure")
# Type variable for options - allows typed ChatAgent[TOptions] returns
# Default matches AzureAIClient's default options type
TOptions_co = TypeVar(
"TOptions_co",
bound=TypedDict, # type: ignore[valid-type]
default="OpenAIResponsesOptions",
covariant=True,
)
class AzureAIProjectAgentProvider(Generic[TOptions_co]):
"""Provider for Azure AI Agent Service (Responses API).
This provider allows you to create, retrieve, and manage Azure AI agents
using the AIProjectClient from the Azure AI Projects SDK.
Examples:
Using with explicit AIProjectClient:
.. code-block:: python
from agent_framework.azure import AzureAIProjectAgentProvider
from azure.ai.projects.aio import AIProjectClient
from azure.identity.aio import DefaultAzureCredential
async with AIProjectClient(endpoint, credential) as client:
provider = AzureAIProjectAgentProvider(client)
agent = await provider.create_agent(
name="MyAgent",
model="gpt-4",
instructions="You are a helpful assistant.",
)
response = await agent.run("Hello!")
Using with credential and endpoint (auto-creates client):
.. code-block:: python
from agent_framework.azure import AzureAIProjectAgentProvider
from azure.identity.aio import DefaultAzureCredential
async with AzureAIProjectAgentProvider(credential=credential) as provider:
agent = await provider.create_agent(
name="MyAgent",
model="gpt-4",
instructions="You are a helpful assistant.",
)
response = await agent.run("Hello!")
"""
def __init__(
self,
project_client: AIProjectClient | None = None,
*,
project_endpoint: str | None = None,
model: str | None = None,
credential: AsyncTokenCredential | None = None,
env_file_path: str | None = None,
env_file_encoding: str | None = None,
) -> None:
"""Initialize an Azure AI Project Agent Provider.
Args:
project_client: An existing AIProjectClient to use. If not provided, one will be created.
project_endpoint: The Azure AI Project endpoint URL.
Can also be set via environment variable AZURE_AI_PROJECT_ENDPOINT.
Ignored when a project_client is passed.
model: The default model deployment name to use for agent creation.
Can also be set via environment variable AZURE_AI_MODEL_DEPLOYMENT_NAME.
credential: Azure async credential to use for authentication.
Required when project_client is not provided.
env_file_path: Path to environment file for loading settings.
env_file_encoding: Encoding of the environment file.
Raises:
ServiceInitializationError: If required parameters are missing or invalid.
"""
try:
self._settings = AzureAISettings(
project_endpoint=project_endpoint,
model_deployment_name=model,
env_file_path=env_file_path,
env_file_encoding=env_file_encoding,
)
except ValidationError as ex:
raise ServiceInitializationError("Failed to create Azure AI settings.", ex) from ex
# Track whether we should close client connection
self._should_close_client = False
if project_client is None:
if not self._settings.project_endpoint:
raise ServiceInitializationError(
"Azure AI project endpoint is required. Set via 'project_endpoint' parameter "
"or 'AZURE_AI_PROJECT_ENDPOINT' environment variable."
)
if not credential:
raise ServiceInitializationError("Azure credential is required when project_client is not provided.")
project_client = AIProjectClient(
endpoint=self._settings.project_endpoint,
credential=credential,
user_agent=AGENT_FRAMEWORK_USER_AGENT,
)
self._should_close_client = True
self._project_client = project_client
async def create_agent(
self,
name: str,
model: str | None = None,
instructions: str | None = None,
description: str | None = None,
response_format: type[BaseModel] | MutableMapping[str, Any] | None = None,
tools: ToolProtocol
| Callable[..., Any]
| MutableMapping[str, Any]
| Sequence[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]]
| None = None,
default_options: TOptions_co | None = None,
middleware: Sequence[Middleware] | None = None,
context_provider: ContextProvider | None = None,
) -> "ChatAgent[TOptions_co]":
"""Create a new agent on the Azure AI service and return a local ChatAgent wrapper.
Args:
name: The name of the agent to create.
model: The model deployment name to use. Falls back to AZURE_AI_MODEL_DEPLOYMENT_NAME
environment variable if not provided.
instructions: Instructions for the agent.
description: A description of the agent.
response_format: The format of the response. Can be a Pydantic model for structured
output, or a dict with JSON schema configuration.
tools: Tools to make available to the agent.
default_options: A TypedDict containing default chat options for the agent.
These options are applied to every run unless overridden.
middleware: List of middleware to intercept agent and function invocations.
context_provider: Context provider to include during agent invocation.
Returns:
ChatAgent: A ChatAgent instance configured with the created agent.
Raises:
ServiceInitializationError: If required parameters are missing.
"""
# Resolve model from parameter or environment variable
resolved_model = model or self._settings.model_deployment_name
if not resolved_model:
raise ServiceInitializationError(
"Model deployment name is required. Provide 'model' parameter "
"or set 'AZURE_AI_MODEL_DEPLOYMENT_NAME' environment variable."
)
args: dict[str, Any] = {"model": resolved_model}
if instructions:
args["instructions"] = instructions
if response_format:
args["text"] = PromptAgentDefinitionText(format=create_text_format_config(response_format))
# Normalize tools once and reuse for both Azure AI API and ChatAgent
normalized_tools = normalize_tools(tools)
if normalized_tools:
args["tools"] = to_azure_ai_tools(normalized_tools)
created_agent = await self._project_client.agents.create_version(
agent_name=name,
definition=PromptAgentDefinition(**args),
description=description,
)
return self._to_chat_agent_from_details(
created_agent,
normalized_tools,
response_format=response_format,
default_options=default_options,
middleware=middleware,
context_provider=context_provider,
)
async def get_agent(
self,
*,
name: str | None = None,
reference: AgentReference | None = None,
tools: ToolProtocol
| Callable[..., Any]
| MutableMapping[str, Any]
| Sequence[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]]
| None = None,
default_options: TOptions_co | None = None,
middleware: Sequence[Middleware] | None = None,
context_provider: ContextProvider | None = None,
) -> "ChatAgent[TOptions_co]":
"""Retrieve an existing agent from the Azure AI service and return a local ChatAgent wrapper.
You must provide either name or reference. Use `as_agent()` if you already have
AgentVersionDetails and want to avoid an async call.
Args:
name: The name of the agent to retrieve (fetches latest version).
reference: Reference containing the agent's name and optionally a specific version.
tools: Tools to make available to the agent. Required if the agent has function tools.
default_options: A TypedDict containing default chat options for the agent.
These options are applied to every run unless overridden.
middleware: List of middleware to intercept agent and function invocations.
context_provider: Context provider to include during agent invocation.
Returns:
ChatAgent: A ChatAgent instance configured with the retrieved agent.
Raises:
ValueError: If no identifier is provided or required tools are missing.
"""
existing_agent: AgentVersionDetails
if reference and reference.version:
# Fetch specific version
existing_agent = await self._project_client.agents.get_version(
agent_name=reference.name, agent_version=reference.version
)
elif agent_name := (reference.name if reference else name):
# Fetch latest version
details = await self._project_client.agents.get(agent_name=agent_name)
existing_agent = details.versions.latest
else:
raise ValueError("Either name or reference must be provided to get an agent.")
if not isinstance(existing_agent.definition, PromptAgentDefinition):
raise ValueError("Agent definition must be PromptAgentDefinition to get a ChatAgent.")
# Validate that required function tools are provided
self._validate_function_tools(existing_agent.definition.tools, tools)
return self._to_chat_agent_from_details(
existing_agent,
normalize_tools(tools),
default_options=default_options,
middleware=middleware,
context_provider=context_provider,
)
def as_agent(
self,
details: AgentVersionDetails,
tools: ToolProtocol
| Callable[..., Any]
| MutableMapping[str, Any]
| Sequence[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]]
| None = None,
default_options: TOptions_co | None = None,
middleware: Sequence[Middleware] | None = None,
context_provider: ContextProvider | None = None,
) -> "ChatAgent[TOptions_co]":
"""Wrap an SDK agent version object into a ChatAgent without making HTTP calls.
Use this when you already have an AgentVersionDetails from a previous API call.
Args:
details: The AgentVersionDetails to wrap.
tools: Tools to make available to the agent. Required if the agent has function tools.
default_options: A TypedDict containing default chat options for the agent.
These options are applied to every run unless overridden.
middleware: List of middleware to intercept agent and function invocations.
context_provider: Context provider to include during agent invocation.
Returns:
ChatAgent: A ChatAgent instance configured with the agent version.
Raises:
ValueError: If the agent definition is not a PromptAgentDefinition or required tools are missing.
"""
if not isinstance(details.definition, PromptAgentDefinition):
raise ValueError("Agent definition must be PromptAgentDefinition to create a ChatAgent.")
# Validate that required function tools are provided
self._validate_function_tools(details.definition.tools, tools)
return self._to_chat_agent_from_details(
details,
normalize_tools(tools),
default_options=default_options,
middleware=middleware,
context_provider=context_provider,
)
def _to_chat_agent_from_details(
self,
details: AgentVersionDetails,
provided_tools: Sequence[ToolProtocol | MutableMapping[str, Any]] | None = None,
response_format: type[BaseModel] | MutableMapping[str, Any] | None = None,
default_options: TOptions_co | None = None,
middleware: Sequence[Middleware] | None = None,
context_provider: ContextProvider | None = None,
) -> "ChatAgent[TOptions_co]":
"""Create a ChatAgent from an AgentVersionDetails.
Args:
details: The AgentVersionDetails containing the agent definition.
provided_tools: User-provided tools (including function implementations).
These are merged with hosted tools from the definition.
response_format: The response format. Can be a Pydantic model for structured
output parsing, or a dict with JSON schema for service-side formatting.
default_options: A TypedDict containing default chat options for the agent.
These options are applied to every run unless overridden.
middleware: List of middleware to intercept agent and function invocations.
context_provider: Context provider to include during agent invocation.
"""
if not isinstance(details.definition, PromptAgentDefinition):
raise ValueError("Agent definition must be PromptAgentDefinition to get a ChatAgent.")
client = AzureAIClient(
project_client=self._project_client,
agent_name=details.name,
agent_version=details.version,
agent_description=details.description,
)
# Merge tools: hosted tools from definition + user-provided function tools
# from_azure_ai_tools converts hosted tools (MCP, code interpreter, file search, web search)
# but function tools need the actual implementations from provided_tools
merged_tools = self._merge_tools(details.definition.tools, provided_tools)
return ChatAgent( # type: ignore[return-value]
chat_client=client,
id=details.id,
name=details.name,
description=details.description,
instructions=details.definition.instructions,
model_id=details.definition.model,
tools=merged_tools,
response_format=response_format,
default_options=default_options, # type: ignore[arg-type]
middleware=middleware,
context_provider=context_provider,
)
def _merge_tools(
self,
definition_tools: Sequence[Any] | None,
provided_tools: Sequence[ToolProtocol | MutableMapping[str, Any]] | None,
) -> list[ToolProtocol | dict[str, Any]]:
"""Merge hosted tools from definition with user-provided function tools.
Args:
definition_tools: Tools from the agent definition (Azure AI format).
provided_tools: User-provided tools (Agent Framework format), including function implementations.
Returns:
Combined list of tools for the ChatAgent.
"""
merged: list[ToolProtocol | 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
hosted_tools = from_azure_ai_tools(definition_tools)
for hosted_tool in hosted_tools:
# Skip function tool dicts - they don't have implementations
if isinstance(hosted_tool, dict) and hosted_tool.get("type") == "function":
continue
merged.append(hosted_tool)
# Add user-provided function tools (these have the actual implementations)
if provided_tools:
for provided_tool in provided_tools:
if isinstance(provided_tool, AIFunction):
merged.append(provided_tool) # type: ignore[reportUnknownArgumentType]
return merged
def _validate_function_tools(
self,
agent_tools: Sequence[Any] | None,
provided_tools: ToolProtocol
| Callable[..., Any]
| MutableMapping[str, Any]
| Sequence[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]]
| None,
) -> None:
"""Validate that required function tools are provided."""
# Normalize and validate function tools
normalized_tools = normalize_tools(provided_tools)
tool_names = {tool.name for tool in normalized_tools if isinstance(tool, AIFunction)}
# If function tools exist in agent definition but were not provided,
# we need to raise an error, as it won't be possible to invoke the function.
missing_tools = [
tool.name for tool in (agent_tools or []) if isinstance(tool, FunctionTool) and tool.name not in tool_names
]
if missing_tools:
raise ValueError(
f"The following prompt agent definition required tools were not provided: {', '.join(missing_tools)}"
)
async def __aenter__(self) -> Self:
"""Async context manager entry."""
return self
async def __aexit__(self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any) -> None:
"""Async context manager exit."""
await self.close()
async def close(self) -> None:
"""Close the provider and release resources.
Only closes the underlying AIProjectClient if it was created by this provider.
"""
if self._should_close_client:
await self._project_client.close()
python/packages/azure-ai/agent_framework_azure_ai/_shared.py
+307 -1
View File
@@ -1,8 +1,38 @@
# Copyright (c) Microsoft. All rights reserved.
from typing import ClassVar
from collections.abc import Mapping, MutableMapping, Sequence
from typing import Any, ClassVar, Literal, cast
from agent_framework import (
AIFunction,
Contents,
HostedCodeInterpreterTool,
HostedFileContent,
HostedFileSearchTool,
HostedMCPTool,
HostedVectorStoreContent,
HostedWebSearchTool,
ToolProtocol,
get_logger,
)
from agent_framework._pydantic import AFBaseSettings
from agent_framework.exceptions import ServiceInvalidRequestError
from azure.ai.projects.models import (
ApproximateLocation,
CodeInterpreterTool,
CodeInterpreterToolAuto,
FileSearchTool,
FunctionTool,
MCPTool,
ResponseTextFormatConfigurationJsonObject,
ResponseTextFormatConfigurationJsonSchema,
ResponseTextFormatConfigurationText,
Tool,
WebSearchPreviewTool,
)
from pydantic import BaseModel
logger = get_logger("agent_framework.azure")
class AzureAISettings(AFBaseSettings):
@@ -44,3 +74,279 @@ class AzureAISettings(AFBaseSettings):
project_endpoint: str | None = None
model_deployment_name: str | None = None
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.
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.
"""
agent_tools: list[ToolProtocol | dict[str, Any]] = []
if not tools:
return agent_tools
for tool in tools:
# Handle raw dictionary tools
tool_dict = tool if isinstance(tool, dict) else dict(tool)
tool_type = tool_dict.get("type")
if tool_type == "mcp":
mcp_tool = cast(MCPTool, tool_dict)
approval_mode: Literal["always_require", "never_require"] | dict[str, set[str]] | None = None
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
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
)
)
elif tool_type == "code_interpreter":
ci_tool = cast(CodeInterpreterTool, tool_dict)
container = ci_tool.get("container", {})
ci_inputs: list[Contents] = []
if "file_ids" in container:
for file_id in container["file_ids"]:
ci_inputs.append(HostedFileContent(file_id=file_id))
agent_tools.append(HostedCodeInterpreterTool(inputs=ci_inputs if ci_inputs else None)) # type: ignore
elif tool_type == "file_search":
fs_tool = cast(FileSearchTool, tool_dict)
fs_inputs: list[Contents] = []
if "vector_store_ids" in fs_tool:
for vs_id in fs_tool["vector_store_ids"]:
fs_inputs.append(HostedVectorStoreContent(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"),
)
)
elif tool_type == "web_search_preview":
ws_tool = cast(WebSearchPreviewTool, tool_dict)
additional_properties: dict[str, Any] = {}
if user_location := ws_tool.get("user_location"):
additional_properties["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))
else:
agent_tools.append(tool_dict)
return agent_tools
def to_azure_ai_tools(
tools: Sequence[ToolProtocol | MutableMapping[str, Any]] | None,
) -> list[Tool | dict[str, Any]]:
"""Converts Agent Framework tools into Azure AI compatible tools.
Args:
tools: A sequence of Agent Framework tool objects or dictionaries
defining the tools to be converted. Can be None.
Returns:
list[Tool | dict[str, Any]]: A list of converted tools compatible with Azure AI.
"""
azure_tools: list[Tool | dict[str, Any]] = []
if not 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 isinstance(tool_input, HostedFileContent):
file_ids.append(tool_input.file_id)
container = CodeInterpreterToolAuto(file_ids=file_ids if file_ids else None)
ci_tool: CodeInterpreterTool = CodeInterpreterTool(container=container)
azure_tools.append(ci_tool)
case AIFunction():
params = tool.parameters()
params["additionalProperties"] = False
azure_tools.append(
FunctionTool(
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 for inp in tool.inputs if isinstance(inp, HostedVectorStoreContent)
]
if not vector_store_ids:
raise ValueError(
"HostedFileSearchTool requires inputs to be of type `HostedVectorStoreContent`."
)
fs_tool: FileSearchTool = FileSearchTool(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 _:
logger.debug("Unsupported tool passed (type: %s)", type(tool))
else:
# Handle raw dictionary tools
tool_dict = tool if isinstance(tool, dict) else dict(tool)
azure_tools.append(tool_dict)
return azure_tools
def _prepare_mcp_tool_for_azure_ai(tool: HostedMCPTool) -> MCPTool:
"""Convert HostedMCPTool to Azure AI MCPTool format.
Args:
tool: The HostedMCPTool to convert.
Returns:
MCPTool: The converted Azure AI MCPTool.
"""
mcp: MCPTool = MCPTool(server_label=tool.name.replace(" ", "_"), server_url=str(tool.url))
if tool.description:
mcp["server_description"] = tool.description
if tool.headers:
mcp["headers"] = tool.headers
if tool.allowed_tools:
mcp["allowed_tools"] = list(tool.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)}}
return mcp
def create_text_format_config(
response_format: type[BaseModel] | Mapping[str, Any],
) -> (
ResponseTextFormatConfigurationJsonSchema
| ResponseTextFormatConfigurationJsonObject
| ResponseTextFormatConfigurationText
):
"""Convert response_format into Azure text format configuration."""
if isinstance(response_format, type) and issubclass(response_format, BaseModel):
schema = response_format.model_json_schema()
# Ensure additionalProperties is explicitly false to satisfy Azure validation
if isinstance(schema, dict):
schema.setdefault("additionalProperties", False)
return ResponseTextFormatConfigurationJsonSchema(
name=response_format.__name__,
schema=schema,
)
if isinstance(response_format, Mapping):
format_config = _convert_response_format(response_format)
format_type = format_config.get("type")
if format_type == "json_schema":
# Ensure schema includes additionalProperties=False to satisfy Azure validation
schema = dict(format_config.get("schema", {})) # type: ignore[assignment]
schema.setdefault("additionalProperties", False)
config_kwargs: dict[str, Any] = {
"name": format_config.get("name") or "response",
"schema": schema,
}
if "strict" in format_config:
config_kwargs["strict"] = format_config["strict"]
if "description" in format_config:
config_kwargs["description"] = format_config["description"]
return ResponseTextFormatConfigurationJsonSchema(**config_kwargs)
if format_type == "json_object":
return ResponseTextFormatConfigurationJsonObject()
if format_type == "text":
return ResponseTextFormatConfigurationText()
raise ServiceInvalidRequestError("response_format must be a Pydantic model or mapping.")
def _convert_response_format(response_format: Mapping[str, Any]) -> dict[str, Any]:
"""Convert Chat style response_format into Responses text format config."""
if "format" in response_format and isinstance(response_format["format"], Mapping):
return dict(cast("Mapping[str, Any]", response_format["format"]))
format_type = response_format.get("type")
if format_type == "json_schema":
schema_section = response_format.get("json_schema", response_format)
if not isinstance(schema_section, Mapping):
raise ServiceInvalidRequestError("json_schema response_format must be a mapping.")
schema_section_typed = cast("Mapping[str, Any]", schema_section)
schema: Any = schema_section_typed.get("schema")
if schema is None:
raise ServiceInvalidRequestError("json_schema response_format requires a schema.")
name: str = str(
schema_section_typed.get("name")
or schema_section_typed.get("title")
or (cast("Mapping[str, Any]", schema).get("title") if isinstance(schema, Mapping) else None)
or "response"
)
format_config: dict[str, Any] = {
"type": "json_schema",
"name": name,
"schema": schema,
}
if "strict" in schema_section:
format_config["strict"] = schema_section["strict"]
if "description" in schema_section and schema_section["description"] is not None:
format_config["description"] = schema_section["description"]
return format_config
if format_type in {"json_object", "text"}:
return {"type": format_type}
raise ServiceInvalidRequestError("Unsupported response_format provided for Azure AI client.")
python/packages/azure-ai/tests/test_azure_ai_client.py
+63
View File
@@ -17,7 +17,10 @@ from agent_framework import (
ChatOptions,
ChatResponse,
HostedCodeInterpreterTool,
HostedFileContent,
HostedFileSearchTool,
HostedMCPTool,
HostedVectorStoreContent,
HostedWebSearchTool,
Role,
TextContent,
@@ -25,7 +28,13 @@ from agent_framework import (
from agent_framework.exceptions import ServiceInitializationError
from azure.ai.projects.aio import AIProjectClient
from azure.ai.projects.models import (
ApproximateLocation,
CodeInterpreterTool,
CodeInterpreterToolAuto,
FileSearchTool,
MCPTool,
ResponseTextFormatConfigurationJsonSchema,
WebSearchPreviewTool,
)
from azure.identity.aio import AzureCliCredential
from openai.types.responses.parsed_response import ParsedResponse
@@ -34,6 +43,7 @@ from pydantic import BaseModel, ConfigDict, Field, ValidationError
from pytest import fixture, param
from agent_framework_azure_ai import AzureAIClient, AzureAISettings
from agent_framework_azure_ai._shared import from_azure_ai_tools
skip_if_azure_ai_integration_tests_disabled = pytest.mark.skipif(
os.getenv("RUN_INTEGRATION_TESTS", "false").lower() != "true"
@@ -962,6 +972,58 @@ def test_get_conversation_id_with_parsed_response_no_conversation() -> None:
assert result == "resp_parsed_12345"
def test_from_azure_ai_tools() -> None:
"""Test from_azure_ai_tools."""
# Test 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"
# Test 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
tool_input = parsed_tools[0].inputs[0]
assert tool_input and isinstance(tool_input, HostedFileContent) and tool_input.file_id == "file-1"
# Test 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
tool_input = parsed_tools[0].inputs[0]
assert tool_input and isinstance(tool_input, HostedVectorStoreContent) and tool_input.vector_store_id == "vs-1"
assert parsed_tools[0].max_results == 5
# Test 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
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"
# region Integration Tests
@@ -993,6 +1055,7 @@ async def client() -> AsyncGenerator[AzureAIClient, None]:
agent_name=agent_name,
)
try:
assert client.function_invocation_configuration
client.function_invocation_configuration.max_iterations = 1
yield client
finally:
python/packages/azure-ai/tests/test_provider.py
+408
View File
@@ -0,0 +1,408 @@
# Copyright (c) Microsoft. All rights reserved.
import os
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from agent_framework import ChatAgent
from agent_framework.exceptions import ServiceInitializationError
from azure.ai.projects.aio import AIProjectClient
from azure.ai.projects.models import (
AgentReference,
AgentVersionDetails,
FunctionTool,
PromptAgentDefinition,
)
from azure.identity.aio import AzureCliCredential
from agent_framework_azure_ai import AzureAIProjectAgentProvider
skip_if_azure_ai_integration_tests_disabled = pytest.mark.skipif(
os.getenv("RUN_INTEGRATION_TESTS", "false").lower() != "true"
or os.getenv("AZURE_AI_PROJECT_ENDPOINT", "") in ("", "https://test-project.cognitiveservices.azure.com/")
or os.getenv("AZURE_AI_MODEL_DEPLOYMENT_NAME", "") == "",
reason=(
"No real AZURE_AI_PROJECT_ENDPOINT or AZURE_AI_MODEL_DEPLOYMENT_NAME provided; skipping integration tests."
if os.getenv("RUN_INTEGRATION_TESTS", "false").lower() == "true"
else "Integration tests are disabled."
),
)
@pytest.fixture
def mock_project_client() -> MagicMock:
"""Fixture that provides a mock AIProjectClient."""
mock_client = MagicMock()
# Mock agents property
mock_client.agents = MagicMock()
mock_client.agents.create_version = AsyncMock()
# Mock conversations property
mock_client.conversations = MagicMock()
mock_client.conversations.create = AsyncMock()
# Mock telemetry property
mock_client.telemetry = MagicMock()
mock_client.telemetry.get_application_insights_connection_string = AsyncMock()
# Mock get_openai_client method
mock_client.get_openai_client = AsyncMock()
# Mock close method
mock_client.close = AsyncMock()
return mock_client
@pytest.fixture
def mock_azure_credential() -> MagicMock:
"""Fixture that provides a mock Azure credential."""
return MagicMock()
@pytest.fixture
def azure_ai_unit_test_env(monkeypatch: pytest.MonkeyPatch) -> dict[str, str]:
"""Fixture that sets up Azure AI environment variables for unit testing."""
env_vars = {
"AZURE_AI_PROJECT_ENDPOINT": "https://test-project.cognitiveservices.azure.com/",
"AZURE_AI_MODEL_DEPLOYMENT_NAME": "test-model-deployment",
}
for key, value in env_vars.items():
monkeypatch.setenv(key, value)
return env_vars
def test_provider_init_with_project_client(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider initialization with existing project_client."""
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
assert provider._project_client is mock_project_client # type: ignore
assert not provider._should_close_client # type: ignore
def test_provider_init_with_credential_and_endpoint(
azure_ai_unit_test_env: dict[str, str],
mock_azure_credential: MagicMock,
) -> None:
"""Test AzureAIProjectAgentProvider initialization with credential and endpoint."""
with patch("agent_framework_azure_ai._provider.AIProjectClient") as mock_ai_project_client:
mock_client = MagicMock()
mock_ai_project_client.return_value = mock_client
provider = AzureAIProjectAgentProvider(
project_endpoint=azure_ai_unit_test_env["AZURE_AI_PROJECT_ENDPOINT"],
credential=mock_azure_credential,
)
assert provider._project_client is mock_client # type: ignore
assert provider._should_close_client # type: ignore
# Verify AIProjectClient was called with correct parameters
mock_ai_project_client.assert_called_once()
def test_provider_init_missing_endpoint() -> None:
"""Test AzureAIProjectAgentProvider initialization when endpoint is missing."""
with patch("agent_framework_azure_ai._provider.AzureAISettings") as mock_settings:
mock_settings.return_value.project_endpoint = None
mock_settings.return_value.model_deployment_name = "test-model"
with pytest.raises(ServiceInitializationError, match="Azure AI project endpoint is required"):
AzureAIProjectAgentProvider(credential=MagicMock())
def test_provider_init_missing_credential(azure_ai_unit_test_env: dict[str, str]) -> None:
"""Test AzureAIProjectAgentProvider initialization when credential is missing."""
with pytest.raises(
ServiceInitializationError, match="Azure credential is required when project_client is not provided"
):
AzureAIProjectAgentProvider(
project_endpoint=azure_ai_unit_test_env["AZURE_AI_PROJECT_ENDPOINT"],
)
async def test_provider_create_agent(
mock_project_client: MagicMock,
azure_ai_unit_test_env: dict[str, str],
) -> None:
"""Test AzureAIProjectAgentProvider.create_agent method."""
with patch("agent_framework_azure_ai._provider.AzureAISettings") as mock_settings:
mock_settings.return_value.project_endpoint = azure_ai_unit_test_env["AZURE_AI_PROJECT_ENDPOINT"]
mock_settings.return_value.model_deployment_name = azure_ai_unit_test_env["AZURE_AI_MODEL_DEPLOYMENT_NAME"]
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
# Mock agent creation response
mock_agent_version = MagicMock(spec=AgentVersionDetails)
mock_agent_version.id = "agent-id"
mock_agent_version.name = "test-agent"
mock_agent_version.version = "1.0"
mock_agent_version.description = "Test Agent"
mock_agent_version.definition = MagicMock(spec=PromptAgentDefinition)
mock_agent_version.definition.model = "gpt-4"
mock_agent_version.definition.instructions = "Test instructions"
mock_agent_version.definition.temperature = 0.7
mock_agent_version.definition.top_p = 0.9
mock_agent_version.definition.tools = []
mock_project_client.agents.create_version = AsyncMock(return_value=mock_agent_version)
agent = await provider.create_agent(
name="test-agent",
model="gpt-4",
instructions="Test instructions",
description="Test Agent",
)
assert isinstance(agent, ChatAgent)
assert agent.name == "test-agent"
mock_project_client.agents.create_version.assert_called_once()
async def test_provider_create_agent_with_env_model(
mock_project_client: MagicMock,
azure_ai_unit_test_env: dict[str, str],
) -> None:
"""Test AzureAIProjectAgentProvider.create_agent uses model from env var."""
with patch("agent_framework_azure_ai._provider.AzureAISettings") as mock_settings:
mock_settings.return_value.project_endpoint = azure_ai_unit_test_env["AZURE_AI_PROJECT_ENDPOINT"]
mock_settings.return_value.model_deployment_name = azure_ai_unit_test_env["AZURE_AI_MODEL_DEPLOYMENT_NAME"]
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
# Mock agent creation response
mock_agent_version = MagicMock(spec=AgentVersionDetails)
mock_agent_version.id = "agent-id"
mock_agent_version.name = "test-agent"
mock_agent_version.version = "1.0"
mock_agent_version.description = None
mock_agent_version.definition = MagicMock(spec=PromptAgentDefinition)
mock_agent_version.definition.model = azure_ai_unit_test_env["AZURE_AI_MODEL_DEPLOYMENT_NAME"]
mock_agent_version.definition.instructions = None
mock_agent_version.definition.temperature = None
mock_agent_version.definition.top_p = None
mock_agent_version.definition.tools = []
mock_project_client.agents.create_version = AsyncMock(return_value=mock_agent_version)
# Call without model parameter - should use env var
agent = await provider.create_agent(name="test-agent")
assert isinstance(agent, ChatAgent)
# Verify the model from env var was used
call_args = mock_project_client.agents.create_version.call_args
assert call_args[1]["definition"].model == azure_ai_unit_test_env["AZURE_AI_MODEL_DEPLOYMENT_NAME"]
async def test_provider_create_agent_missing_model(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider.create_agent raises when model is missing."""
with patch("agent_framework_azure_ai._provider.AzureAISettings") as mock_settings:
mock_settings.return_value.project_endpoint = "https://test.com"
mock_settings.return_value.model_deployment_name = None
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
with pytest.raises(ServiceInitializationError, match="Model deployment name is required"):
await provider.create_agent(name="test-agent")
async def test_provider_get_agent_with_name(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider.get_agent with name parameter."""
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
# Mock agent response
mock_agent_version = MagicMock(spec=AgentVersionDetails)
mock_agent_version.id = "agent-id"
mock_agent_version.name = "test-agent"
mock_agent_version.version = "1.0"
mock_agent_version.description = "Test Agent"
mock_agent_version.definition = MagicMock(spec=PromptAgentDefinition)
mock_agent_version.definition.model = "gpt-4"
mock_agent_version.definition.instructions = "Test instructions"
mock_agent_version.definition.temperature = None
mock_agent_version.definition.top_p = None
mock_agent_version.definition.tools = []
mock_agent_object = MagicMock()
mock_agent_object.versions.latest = mock_agent_version
mock_project_client.agents = AsyncMock()
mock_project_client.agents.get.return_value = mock_agent_object
agent = await provider.get_agent(name="test-agent")
assert isinstance(agent, ChatAgent)
assert agent.name == "test-agent"
mock_project_client.agents.get.assert_called_with(agent_name="test-agent")
async def test_provider_get_agent_with_reference(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider.get_agent with reference parameter."""
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
# Mock agent response
mock_agent_version = MagicMock(spec=AgentVersionDetails)
mock_agent_version.id = "agent-id"
mock_agent_version.name = "test-agent"
mock_agent_version.version = "1.0"
mock_agent_version.description = "Test Agent"
mock_agent_version.definition = MagicMock(spec=PromptAgentDefinition)
mock_agent_version.definition.model = "gpt-4"
mock_agent_version.definition.instructions = "Test instructions"
mock_agent_version.definition.temperature = None
mock_agent_version.definition.top_p = None
mock_agent_version.definition.tools = []
mock_project_client.agents = AsyncMock()
mock_project_client.agents.get_version.return_value = mock_agent_version
agent_reference = AgentReference(name="test-agent", version="1.0")
agent = await provider.get_agent(reference=agent_reference)
assert isinstance(agent, ChatAgent)
assert agent.name == "test-agent"
mock_project_client.agents.get_version.assert_called_with(agent_name="test-agent", agent_version="1.0")
async def test_provider_get_agent_missing_parameters(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider.get_agent raises when no identifier provided."""
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
with pytest.raises(ValueError, match="Either name or reference must be provided"):
await provider.get_agent()
async def test_provider_get_agent_missing_function_tools(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider.get_agent raises when required tools are missing."""
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
# Mock agent with function tools
mock_agent_version = MagicMock(spec=AgentVersionDetails)
mock_agent_version.id = "agent-id"
mock_agent_version.name = "test-agent"
mock_agent_version.version = "1.0"
mock_agent_version.description = None
mock_agent_version.definition = MagicMock(spec=PromptAgentDefinition)
mock_agent_version.definition.tools = [
FunctionTool(name="test_tool", parameters=[], strict=True, description="Test tool")
]
mock_agent_object = MagicMock()
mock_agent_object.versions.latest = mock_agent_version
mock_project_client.agents = AsyncMock()
mock_project_client.agents.get.return_value = mock_agent_object
with pytest.raises(
ValueError, match="The following prompt agent definition required tools were not provided: test_tool"
):
await provider.get_agent(name="test-agent")
def test_provider_as_agent(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider.as_agent method."""
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
# Create mock agent version
mock_agent_version = MagicMock(spec=AgentVersionDetails)
mock_agent_version.id = "agent-id"
mock_agent_version.name = "test-agent"
mock_agent_version.version = "1.0"
mock_agent_version.description = "Test Agent"
mock_agent_version.definition = MagicMock(spec=PromptAgentDefinition)
mock_agent_version.definition.model = "gpt-4"
mock_agent_version.definition.instructions = "Test instructions"
mock_agent_version.definition.temperature = 0.7
mock_agent_version.definition.top_p = 0.9
mock_agent_version.definition.tools = []
agent = provider.as_agent(mock_agent_version)
assert isinstance(agent, ChatAgent)
assert agent.name == "test-agent"
assert agent.description == "Test Agent"
async def test_provider_context_manager(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider async context manager."""
with patch("agent_framework_azure_ai._provider.AIProjectClient") as mock_ai_project_client:
mock_client = MagicMock()
mock_client.close = AsyncMock()
mock_ai_project_client.return_value = mock_client
with patch("agent_framework_azure_ai._provider.AzureAISettings") as mock_settings:
mock_settings.return_value.project_endpoint = "https://test.com"
mock_settings.return_value.model_deployment_name = "test-model"
async with AzureAIProjectAgentProvider(credential=MagicMock()) as provider:
assert provider._project_client is mock_client # type: ignore
# Should call close after exiting context
mock_client.close.assert_called_once()
async def test_provider_context_manager_with_provided_client(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider context manager doesn't close provided client."""
mock_project_client.close = AsyncMock()
async with AzureAIProjectAgentProvider(project_client=mock_project_client) as provider:
assert provider._project_client is mock_project_client # type: ignore
# Should NOT call close when client was provided
mock_project_client.close.assert_not_called()
async def test_provider_close_method(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider.close method."""
with patch("agent_framework_azure_ai._provider.AIProjectClient") as mock_ai_project_client:
mock_client = MagicMock()
mock_client.close = AsyncMock()
mock_ai_project_client.return_value = mock_client
with patch("agent_framework_azure_ai._provider.AzureAISettings") as mock_settings:
mock_settings.return_value.project_endpoint = "https://test.com"
mock_settings.return_value.model_deployment_name = "test-model"
provider = AzureAIProjectAgentProvider(credential=MagicMock())
await provider.close()
mock_client.close.assert_called_once()
@pytest.mark.flaky
@skip_if_azure_ai_integration_tests_disabled
async def test_provider_create_and_get_agent_integration() -> None:
"""Integration test for provider create_agent and get_agent."""
endpoint = os.environ["AZURE_AI_PROJECT_ENDPOINT"]
model = os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"]
async with (
AzureCliCredential() as credential,
AIProjectClient(endpoint=endpoint, credential=credential) as project_client,
):
provider = AzureAIProjectAgentProvider(project_client=project_client)
try:
# Create agent
agent = await provider.create_agent(
name="ProviderTestAgent",
model=model,
instructions="You are a helpful assistant. Always respond with 'Hello from provider!'",
)
assert isinstance(agent, ChatAgent)
assert agent.name == "ProviderTestAgent"
# Run the agent
response = await agent.run("Hi!")
assert response.text is not None
assert len(response.text) > 0
# Get the same agent
retrieved_agent = await provider.get_agent(name="ProviderTestAgent")
assert retrieved_agent.name == "ProviderTestAgent"
finally:
# Cleanup
await project_client.agents.delete(agent_name="ProviderTestAgent")
python/packages/core/agent_framework/_types.py
+62 -13
View File
@@ -66,6 +66,7 @@ __all__ = [
"UsageContent",
"UsageDetails",
"merge_chat_options",
"normalize_tools",
"prepare_function_call_results",
"prepend_instructions_to_messages",
"validate_chat_options",
@@ -3490,6 +3491,60 @@ async def validate_chat_options(options: dict[str, Any]) -> dict[str, Any]:
return result
def normalize_tools(
tools: (
ToolProtocol
| Callable[..., Any]
| MutableMapping[str, Any]
| Sequence[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]]
| None
),
) -> list[ToolProtocol | MutableMapping[str, Any]]:
"""Normalize tools into a list.
Converts callables to AIFunction objects and ensures all tools are either
ToolProtocol instances or MutableMappings.
Args:
tools: Tools to normalize - can be a single tool, callable, or sequence.
Returns:
Normalized list of tools.
Examples:
.. code-block:: python
from agent_framework import normalize_tools, ai_function
@ai_function
def my_tool(x: int) -> int:
return x * 2
# Single tool
tools = normalize_tools(my_tool)
# List of tools
tools = normalize_tools([my_tool, another_tool])
"""
final_tools: list[ToolProtocol | MutableMapping[str, Any]] = []
if not tools:
return final_tools
if not isinstance(tools, Sequence) or isinstance(tools, (str, MutableMapping)):
# Single tool (not a sequence, or is a mapping which shouldn't be treated as sequence)
if not isinstance(tools, (ToolProtocol, MutableMapping)):
return [ai_function(tools)]
return [tools]
for tool in tools:
if isinstance(tool, (ToolProtocol, MutableMapping)):
final_tools.append(tool)
else:
# Convert callable to AIFunction
final_tools.append(ai_function(tool))
return final_tools
async def validate_tools(
tools: (
ToolProtocol
@@ -3528,16 +3583,12 @@ async def validate_tools(
# List of tools
tools = await validate_tools([my_tool, another_tool])
"""
# Sequence of tools - convert callables and expand MCP tools
# Use normalize_tools for common sync logic (converts callables to AIFunction)
normalized = normalize_tools(tools)
# Handle MCP tool expansion (async-only)
final_tools: list[ToolProtocol | MutableMapping[str, Any]] = []
if not tools:
return final_tools
if not isinstance(tools, Sequence) or isinstance(tools, (str, MutableMapping)):
# Single tool (not a sequence, or is a mapping which shouldn't be treated as sequence)
if not isinstance(tools, (ToolProtocol, MutableMapping)):
return [ai_function(tools)]
return [tools]
for tool in tools:
for tool in normalized:
# Import MCPTool here to avoid circular imports
from ._mcp import MCPTool
@@ -3546,11 +3597,9 @@ async def validate_tools(
if not tool.is_connected:
await tool.connect()
final_tools.extend(tool.functions) # type: ignore
elif isinstance(tool, (ToolProtocol, MutableMapping)):
final_tools.append(tool)
else:
# Convert callable to AIFunction
final_tools.append(ai_function(tool))
final_tools.append(tool)
return final_tools
python/packages/core/agent_framework/azure/__init__.py
+1
View File
@@ -10,6 +10,7 @@ _IMPORTS: dict[str, tuple[str, str]] = {
"AzureAIAgentClient": ("agent_framework_azure_ai", "agent-framework-azure-ai"),
"AzureAIAgentOptions": ("agent_framework_azure_ai", "agent-framework-azure-ai"),
"AzureAIClient": ("agent_framework_azure_ai", "agent-framework-azure-ai"),
"AzureAIProjectAgentProvider": ("agent_framework_azure_ai", "agent-framework-azure-ai"),
"AzureAISearchContextProvider": ("agent_framework_azure_ai_search", "agent-framework-azure-ai-search"),
"AzureAISearchSettings": ("agent_framework_azure_ai_search", "agent-framework-azure-ai-search"),
"AzureAISettings": ("agent_framework_azure_ai", "agent-framework-azure-ai"),
python/packages/core/agent_framework/azure/__init__.pyi
+2 -1
View File
@@ -1,6 +1,6 @@
# Copyright (c) Microsoft. All rights reserved.
from agent_framework_azure_ai import AzureAIAgentClient, AzureAIClient, AzureAISettings
from agent_framework_azure_ai import AzureAIAgentClient, AzureAIClient, AzureAIProjectAgentProvider, AzureAISettings
from agent_framework_azure_ai_search import AzureAISearchContextProvider, AzureAISearchSettings
from agent_framework_azurefunctions import (
AgentCallbackContext,
@@ -21,6 +21,7 @@ __all__ = [
"AgentResponseCallbackProtocol",
"AzureAIAgentClient",
"AzureAIClient",
"AzureAIProjectAgentProvider",
"AzureAISearchContextProvider",
"AzureAISearchSettings",
"AzureAISettings",