Python: Create/Get Agent API for Azure V1 (#3192)

* Added provider implementation for Azure AI V1

* Small fixes

* Fixed OpenAPI example

* Fixed local MCP example

* Fixed hosted MCP example

* Fixed file search sample

* Small fixes

* Resolved comments

* Doc updates
This commit is contained in:
Dmytro Struk
2026-01-15 14:19:03 -08:00
committed by GitHub
Unverified
parent 6e9420f614
commit 48d124efbe
32 changed files with 2119 additions and 640 deletions
@@ -2,9 +2,10 @@
import importlib.metadata
from ._agent_provider import AzureAIAgentsProvider
from ._chat_client import AzureAIAgentClient, AzureAIAgentOptions
from ._client import AzureAIClient
from ._provider import AzureAIProjectAgentProvider
from ._project_provider import AzureAIProjectAgentProvider
from ._shared import AzureAISettings
try:
@@ -15,6 +16,7 @@ except importlib.metadata.PackageNotFoundError:
__all__ = [
"AzureAIAgentClient",
"AzureAIAgentOptions",
"AzureAIAgentsProvider",
"AzureAIClient",
"AzureAIProjectAgentProvider",
"AzureAISettings",
@@ -0,0 +1,519 @@
# Copyright (c) Microsoft. All rights reserved.
import sys
from collections.abc import Callable, MutableMapping, Sequence
from typing import TYPE_CHECKING, Any, Generic, TypedDict, cast
from agent_framework import (
AGENT_FRAMEWORK_USER_AGENT,
AIFunction,
ChatAgent,
ContextProvider,
Middleware,
ToolProtocol,
normalize_tools,
)
from agent_framework._mcp import MCPTool
from agent_framework.exceptions import ServiceInitializationError
from azure.ai.agents.aio import AgentsClient
from azure.ai.agents.models import Agent, ResponseFormatJsonSchema, ResponseFormatJsonSchemaType
from azure.core.credentials_async import AsyncTokenCredential
from pydantic import BaseModel, ValidationError
from ._chat_client import AzureAIAgentClient
from ._shared import AzureAISettings, from_azure_ai_agent_tools, to_azure_ai_agent_tools
if TYPE_CHECKING:
from ._chat_client import AzureAIAgentOptions
if sys.version_info >= (3, 13):
from typing import Self, TypeVar # pragma: no cover
else:
from typing_extensions import Self, TypeVar # pragma: no cover
# Type variable for options - allows typed ChatAgent[TOptions] returns
# Default matches AzureAIAgentClient's default options type
TOptions_co = TypeVar(
"TOptions_co",
bound=TypedDict, # type: ignore[valid-type]
default="AzureAIAgentOptions",
covariant=True,
)
class AzureAIAgentsProvider(Generic[TOptions_co]):
"""Provider for Azure AI Agent Service V1 (Persistent Agents API).
This provider enables creating, retrieving, and wrapping Azure AI agents as ChatAgent
instances. It manages the underlying AgentsClient lifecycle and provides a high-level
interface for agent operations.
The provider can be initialized with either:
- An existing AgentsClient instance
- Azure credentials and endpoint for automatic client creation
Examples:
Using credentials (auto-creates client):
.. code-block:: python
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
async with (
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="MyAgent",
instructions="You are a helpful assistant.",
)
result = await agent.run("Hello!")
Using existing AgentsClient:
.. code-block:: python
from agent_framework.azure import AzureAIAgentsProvider
from azure.ai.agents.aio import AgentsClient
async with AgentsClient(endpoint=endpoint, credential=credential) as client:
provider = AzureAIAgentsProvider(agents_client=client)
agent = await provider.create_agent(name="MyAgent", instructions="...")
"""
def __init__(
self,
agents_client: AgentsClient | None = None,
*,
project_endpoint: str | None = None,
credential: AsyncTokenCredential | None = None,
env_file_path: str | None = None,
env_file_encoding: str | None = None,
) -> None:
"""Initialize the Azure AI Agents Provider.
Args:
agents_client: An existing AgentsClient to use. If provided, the provider
will not manage its lifecycle.
Keyword Args:
project_endpoint: The Azure AI Project endpoint URL.
Can also be set via AZURE_AI_PROJECT_ENDPOINT environment variable.
credential: Azure async credential for authentication.
Required if agents_client is not provided.
env_file_path: Path to .env file for loading settings.
env_file_encoding: Encoding of the .env file.
Raises:
ServiceInitializationError: If required parameters are missing or invalid.
"""
try:
self._settings = AzureAISettings(
project_endpoint=project_endpoint,
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
self._should_close_client = False
if agents_client is not None:
self._agents_client = agents_client
else:
if not self._settings.project_endpoint:
raise ServiceInitializationError(
"Azure AI project endpoint is required. Provide 'project_endpoint' parameter "
"or set 'AZURE_AI_PROJECT_ENDPOINT' environment variable."
)
if not credential:
raise ServiceInitializationError("Azure credential is required when agents_client is not provided.")
self._agents_client = AgentsClient(
endpoint=self._settings.project_endpoint,
credential=credential,
user_agent=AGENT_FRAMEWORK_USER_AGENT,
)
self._should_close_client = True
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 AgentsClient if it was created by this provider.
"""
if self._should_close_client:
await self._agents_client.close()
async def create_agent(
self,
name: str,
*,
model: str | None = None,
instructions: str | None = None,
description: str | 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 ChatAgent.
This method creates a persistent agent on the Azure AI service with the specified
configuration and returns a local ChatAgent instance for interaction.
Args:
name: The name for the agent.
Keyword Args:
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's behavior.
description: A description of the agent's purpose.
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 model deployment name is not available.
Examples:
.. code-block:: python
agent = await provider.create_agent(
name="WeatherAgent",
instructions="You are a helpful weather assistant.",
tools=get_weather,
)
"""
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."
)
# Extract response_format from default_options if present
opts = dict(default_options) if default_options else {}
response_format = opts.get("response_format")
args: dict[str, Any] = {
"model": resolved_model,
"name": name,
}
if description:
args["description"] = description
if instructions:
args["instructions"] = instructions
# Handle response format
if response_format and isinstance(response_format, type) and issubclass(response_format, BaseModel):
args["response_format"] = self._create_response_format_config(response_format)
# Normalize and convert tools
# Local MCP tools (MCPTool) are handled by ChatAgent at runtime, not stored on the Azure agent
normalized_tools = normalize_tools(tools)
if normalized_tools:
# Only convert non-MCP tools to Azure AI format
non_mcp_tools = [t for t in normalized_tools if not isinstance(t, MCPTool)]
if non_mcp_tools:
# Pass run_options to capture tool_resources (e.g., for file search vector stores)
run_options: dict[str, Any] = {}
args["tools"] = to_azure_ai_agent_tools(non_mcp_tools, run_options)
if "tool_resources" in run_options:
args["tool_resources"] = run_options["tool_resources"]
# Create the agent on the service
created_agent = await self._agents_client.create_agent(**args)
# Create ChatAgent wrapper
return self._to_chat_agent_from_agent(
created_agent,
normalized_tools,
default_options=default_options,
middleware=middleware,
context_provider=context_provider,
)
async def get_agent(
self,
id: str,
*,
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 service and return a ChatAgent.
This method fetches an agent by ID from the Azure AI service
and returns a local ChatAgent instance for interaction.
Args:
id: The ID of the agent to retrieve from the service.
Keyword Args:
tools: Tools to make available to the agent. Required if the agent
has function tools that need implementations.
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:
ServiceInitializationError: If required function tools are not provided.
Examples:
.. code-block:: python
agent = await provider.get_agent("agent-123")
# With function tools
agent = await provider.get_agent("agent-123", tools=my_function)
"""
agent = await self._agents_client.get_agent(id)
# Validate function tools
normalized_tools = normalize_tools(tools)
self._validate_function_tools(agent.tools, normalized_tools)
return self._to_chat_agent_from_agent(
agent,
normalized_tools,
default_options=default_options,
middleware=middleware,
context_provider=context_provider,
)
def as_agent(
self,
agent: Agent,
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 existing Agent SDK object as a ChatAgent without making HTTP calls.
Use this method when you already have an Agent object from a previous
SDK operation and want to use it with the Agent Framework.
Args:
agent: The Agent object to wrap.
tools: Tools to make available to the agent. Required if the agent
has function tools that need implementations.
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.
Raises:
ServiceInitializationError: If required function tools are not provided.
Examples:
.. code-block:: python
# Create agent directly with SDK
sdk_agent = await agents_client.create_agent(
model="gpt-4",
name="MyAgent",
instructions="...",
)
# Wrap as ChatAgent
chat_agent = provider.as_agent(sdk_agent)
"""
# Validate function tools
normalized_tools = normalize_tools(tools)
self._validate_function_tools(agent.tools, normalized_tools)
return self._to_chat_agent_from_agent(
agent,
normalized_tools,
default_options=default_options,
middleware=middleware,
context_provider=context_provider,
)
def _to_chat_agent_from_agent(
self,
agent: Agent,
provided_tools: Sequence[ToolProtocol | 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 Agent SDK object.
Args:
agent: The Agent SDK object.
provided_tools: User-provided tools (including function implementations).
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.
"""
# Create the underlying client
client = AzureAIAgentClient(
agents_client=self._agents_client,
agent_id=agent.id,
agent_name=agent.name,
agent_description=agent.description,
should_cleanup_agent=False, # Provider manages agent lifecycle
)
# Merge tools: convert agent's hosted tools + user-provided function tools
merged_tools = self._merge_tools(agent.tools, provided_tools)
return ChatAgent( # type: ignore[return-value]
chat_client=client,
id=agent.id,
name=agent.name,
description=agent.description,
instructions=agent.instructions,
model_id=agent.model,
tools=merged_tools,
default_options=default_options, # type: ignore[arg-type]
middleware=middleware,
context_provider=context_provider,
)
def _merge_tools(
self,
agent_tools: Sequence[Any] | None,
provided_tools: Sequence[ToolProtocol | MutableMapping[str, Any]] | None,
) -> list[ToolProtocol | dict[str, Any]]:
"""Merge hosted tools from agent with user-provided function tools.
Args:
agent_tools: Tools from the agent definition (Azure AI format).
provided_tools: User-provided tools (Agent Framework format).
Returns:
Combined list of tools for the ChatAgent.
"""
merged: list[ToolProtocol | dict[str, Any]] = []
# Convert hosted tools from agent definition
hosted_tools = from_azure_ai_agent_tools(agent_tools)
for hosted_tool in hosted_tools:
# Skip function tool dicts - they don't have implementations
# Skip OpenAPI tool dicts - they're defined on the agent, not needed at runtime
if isinstance(hosted_tool, dict):
tool_type = hosted_tool.get("type")
if tool_type == "function" or tool_type == "openapi":
continue
merged.append(hosted_tool)
# Add user-provided function tools and MCP tools
if provided_tools:
for provided_tool in provided_tools:
# AIFunction - has implementation for function calling
# MCPTool - ChatAgent handles MCP connection and tool discovery at runtime
if isinstance(provided_tool, (AIFunction, MCPTool)):
merged.append(provided_tool) # type: ignore[reportUnknownArgumentType]
return merged
def _validate_function_tools(
self,
agent_tools: Sequence[Any] | None,
provided_tools: Sequence[ToolProtocol | MutableMapping[str, Any]] | None,
) -> None:
"""Validate that required function tools are provided.
Raises:
ServiceInitializationError: If agent has function tools but user
didn't provide implementations.
"""
if not agent_tools:
return
# Get function tool names from agent definition
function_tool_names: set[str] = set()
for tool in agent_tools:
if isinstance(tool, dict):
tool_dict = cast(dict[str, Any], tool)
if tool_dict.get("type") == "function":
func_def = cast(dict[str, Any], tool_dict.get("function", {}))
name = func_def.get("name")
if isinstance(name, str):
function_tool_names.add(name)
elif hasattr(tool, "type") and tool.type == "function":
func_attr = getattr(tool, "function", None)
if func_attr and hasattr(func_attr, "name"):
function_tool_names.add(str(func_attr.name))
if not function_tool_names:
return
# Get provided function names
provided_names: set[str] = set()
if provided_tools:
for tool in provided_tools:
if isinstance(tool, AIFunction):
provided_names.add(tool.name)
# Check for missing implementations
missing = function_tool_names - provided_names
if missing:
raise ServiceInitializationError(
f"Agent has function tools that require implementations: {missing}. "
"Provide these functions via the 'tools' parameter."
)
def _create_response_format_config(
self,
response_format: type[BaseModel],
) -> ResponseFormatJsonSchemaType:
"""Create response format configuration for Azure AI.
Args:
response_format: Pydantic model for structured output.
Returns:
Azure AI response format configuration.
"""
return ResponseFormatJsonSchemaType(
json_schema=ResponseFormatJsonSchema(
name=response_format.__name__,
schema=response_format.model_json_schema(),
)
)
@@ -2,7 +2,6 @@
import ast
import json
import os
import re
import sys
from collections.abc import AsyncIterable, Mapping, MutableMapping, MutableSequence, Sequence
@@ -10,7 +9,6 @@ from typing import Any, ClassVar, Generic, TypedDict
from agent_framework import (
AGENT_FRAMEWORK_USER_AGENT,
AIFunction,
BaseChatClient,
ChatMessage,
ChatOptions,
@@ -23,12 +21,8 @@ from agent_framework import (
FunctionApprovalResponseContent,
FunctionCallContent,
FunctionResultContent,
HostedCodeInterpreterTool,
HostedFileContent,
HostedFileSearchTool,
HostedMCPTool,
HostedVectorStoreContent,
HostedWebSearchTool,
Role,
TextContent,
TextSpanRegion,
@@ -52,14 +46,9 @@ from azure.ai.agents.models import (
AgentStreamEvent,
AsyncAgentEventHandler,
AsyncAgentRunStream,
BingCustomSearchTool,
BingGroundingTool,
CodeInterpreterToolDefinition,
FileSearchTool,
FunctionName,
FunctionToolDefinition,
ListSortOrder,
McpTool,
MessageDeltaChunk,
MessageDeltaTextContent,
MessageDeltaTextFileCitationAnnotation,
@@ -91,7 +80,7 @@ from azure.ai.agents.models import (
from azure.core.credentials_async import AsyncTokenCredential
from pydantic import BaseModel, ValidationError
from ._shared import AzureAISettings
from ._shared import AzureAISettings, to_azure_ai_agent_tools
if sys.version_info >= (3, 13):
from typing import TypeVar # type: ignore # pragma: no cover
@@ -1007,7 +996,7 @@ class AzureAIAgentClient(BaseChatClient[TAzureAIAgentOptions], Generic[TAzureAIA
tool_choice = options.get("tool_choice")
tools = options.get("tools")
if tool_choice is not None and tool_choice != "none" and tools:
tool_definitions.extend(await self._prepare_tools_for_azure_ai(tools, run_options))
tool_definitions.extend(to_azure_ai_agent_tools(tools, run_options))
# Handle MCP tool resources
mcp_resources = self._prepare_mcp_resources(tools)
@@ -1106,82 +1095,6 @@ class AzureAIAgentClient(BaseChatClient[TAzureAIAgentOptions], Generic[TAzureAIA
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]] = []
for tool in tools:
match tool:
case AIFunction():
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 isinstance(inp, HostedVectorStoreContent)]
if vector_stores:
file_search = FileSearchTool(vector_store_ids=[vs.vector_store_id for vs in vector_stores])
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)}")
return tool_definitions
def _prepare_tool_outputs_for_azure_ai(
self,
required_action_results: list[FunctionResultContent | FunctionApprovalResponseContent] | None,
@@ -24,7 +24,7 @@ from azure.ai.projects.models import (
PromptAgentDefinitionText,
)
from azure.core.credentials_async import AsyncTokenCredential
from pydantic import BaseModel, ValidationError
from pydantic import ValidationError
from ._client import AzureAIClient
from ._shared import AzureAISettings, create_text_format_config, from_azure_ai_tools, to_azure_ai_tools
@@ -156,7 +156,6 @@ class AzureAIProjectAgentProvider(Generic[TOptions_co]):
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]
@@ -174,8 +173,6 @@ class AzureAIProjectAgentProvider(Generic[TOptions_co]):
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.
@@ -196,12 +193,18 @@ class AzureAIProjectAgentProvider(Generic[TOptions_co]):
"or set 'AZURE_AI_MODEL_DEPLOYMENT_NAME' environment variable."
)
# Extract response_format from default_options if present
opts = dict(default_options) if default_options else {}
response_format = opts.get("response_format")
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))
if response_format and isinstance(response_format, (type, dict)):
args["text"] = PromptAgentDefinitionText(
format=create_text_format_config(response_format) # type: ignore[arg-type]
)
# Normalize tools once and reuse for both Azure AI API and ChatAgent
normalized_tools = normalize_tools(tools)
@@ -217,7 +220,6 @@ class AzureAIProjectAgentProvider(Generic[TOptions_co]):
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,
@@ -333,7 +335,6 @@ class AzureAIProjectAgentProvider(Generic[TOptions_co]):
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,
@@ -344,8 +345,6 @@ class AzureAIProjectAgentProvider(Generic[TOptions_co]):
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.
@@ -374,7 +373,6 @@ class AzureAIProjectAgentProvider(Generic[TOptions_co]):
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,
@@ -1,5 +1,6 @@
# Copyright (c) Microsoft. All rights reserved.
import os
from collections.abc import Mapping, MutableMapping, Sequence
from typing import Any, ClassVar, Literal, cast
@@ -16,12 +17,19 @@ from agent_framework import (
get_logger,
)
from agent_framework._pydantic import AFBaseSettings
from agent_framework.exceptions import ServiceInvalidRequestError
from agent_framework.exceptions import ServiceInitializationError, 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,
FileSearchTool,
FunctionTool,
MCPTool,
ResponseTextFormatConfigurationJsonObject,
@@ -30,6 +38,9 @@ from azure.ai.projects.models import (
Tool,
WebSearchPreviewTool,
)
from azure.ai.projects.models import (
FileSearchTool as ProjectsFileSearchTool,
)
from pydantic import BaseModel
logger = get_logger("agent_framework.azure")
@@ -76,6 +87,207 @@ class AzureAISettings(AFBaseSettings):
model_deployment_name: str | None = None
def to_azure_ai_agent_tools(
tools: Sequence[ToolProtocol | 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.
Args:
tools: Sequence of Agent Framework tools to convert.
run_options: Optional dict with run options.
Returns:
List of Azure AI V1 SDK tool definitions.
Raises:
ServiceInitializationError: If tool configuration is invalid.
"""
if not tools:
return []
tool_definitions: list[ToolDefinition | dict[str, Any]] = []
for tool in tools:
match tool:
case AIFunction():
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 isinstance(inp, HostedVectorStoreContent)]
if vector_stores:
file_search = AgentsFileSearchTool(vector_store_ids=[vs.vector_store_id for vs in vector_stores])
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)}")
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.
Args:
tools: Sequence of Azure AI V1 SDK tool definitions.
Returns:
List of Agent Framework tools.
"""
if not tools:
return []
result: list[ToolProtocol | dict[str, Any]] = []
for tool in tools:
# Handle SDK objects
if isinstance(tool, CodeInterpreterToolDefinition):
result.append(HostedCodeInterpreterTool())
elif isinstance(tool, dict):
# Handle dict format
converted = _convert_dict_tool(tool)
if converted is not None:
result.append(converted)
elif hasattr(tool, "type"):
# Handle other SDK objects by type
converted = _convert_sdk_tool(tool)
if converted is not None:
result.append(converted)
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."""
tool_type = tool.get("type")
if tool_type == "code_interpreter":
return HostedCodeInterpreterTool()
if tool_type == "file_search":
file_search_config = tool.get("file_search", {})
vector_store_ids = file_search_config.get("vector_store_ids", [])
inputs = [HostedVectorStoreContent(vector_store_id=vs_id) for vs_id in vector_store_ids]
return HostedFileSearchTool(inputs=inputs if inputs else None) # type: ignore
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)
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"),
}
)
if tool_type == "mcp":
# Hosted MCP tools are defined on the Azure agent, no local handling needed
# Azure may not return full server_url, so skip conversion
return None
if tool_type == "function":
# Function tools are returned as dicts - users must provide implementations
return tool
# Unknown tool type - pass through
return tool
def _convert_sdk_tool(tool: ToolDefinition) -> ToolProtocol | dict[str, Any] | None:
"""Convert an SDK-object Azure AI tool to Agent Framework tool."""
tool_type = getattr(tool, "type", None)
if tool_type == "code_interpreter":
return HostedCodeInterpreterTool()
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 = [HostedVectorStoreContent(vector_store_id=vs_id) for vs_id in vector_store_ids]
return HostedFileSearchTool(inputs=inputs if inputs else None) # type: ignore
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)
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,
}
)
if tool_type == "mcp":
# Hosted MCP tools are defined on the Azure agent, no local handling needed
# Azure may not return full server_url, so skip conversion
return None
if tool_type == "function":
# Function tools from SDK don't have implementations - skip
return None
# Unknown tool type - convert to dict if possible
if hasattr(tool, "as_dict"):
return tool.as_dict() # type: ignore[union-attr]
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.
@@ -130,7 +342,7 @@ def from_azure_ai_tools(tools: Sequence[Tool | dict[str, Any]] | None) -> list[T
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_tool = cast(ProjectsFileSearchTool, tool_dict)
fs_inputs: list[Contents] = []
if "vector_store_ids" in fs_tool:
for vs_id in fs_tool["vector_store_ids"]:
@@ -210,7 +422,7 @@ def to_azure_ai_tools(
raise ValueError(
"HostedFileSearchTool requires inputs to be of type `HostedVectorStoreContent`."
)
fs_tool: FileSearchTool = FileSearchTool(vector_store_ids=vector_store_ids)
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)
@@ -0,0 +1,803 @@
# Copyright (c) Microsoft. All rights reserved.
import os
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from agent_framework import (
ChatAgent,
HostedCodeInterpreterTool,
HostedFileSearchTool,
HostedMCPTool,
HostedVectorStoreContent,
HostedWebSearchTool,
ai_function,
)
from agent_framework.exceptions import ServiceInitializationError
from azure.ai.agents.models import (
Agent,
CodeInterpreterToolDefinition,
)
from pydantic import BaseModel
from agent_framework_azure_ai import (
AzureAIAgentsProvider,
AzureAISettings,
)
from agent_framework_azure_ai._shared import (
from_azure_ai_agent_tools,
to_azure_ai_agent_tools,
)
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/"),
reason="No real AZURE_AI_PROJECT_ENDPOINT provided; skipping integration tests."
if os.getenv("RUN_INTEGRATION_TESTS", "false").lower() == "true"
else "Integration tests are disabled.",
)
# region Provider Initialization Tests
def test_provider_init_with_agents_client(mock_agents_client: MagicMock) -> None:
"""Test AzureAIAgentsProvider initialization with existing AgentsClient."""
provider = AzureAIAgentsProvider(agents_client=mock_agents_client)
assert provider._agents_client is mock_agents_client # type: ignore
assert provider._should_close_client is False # type: ignore
def test_provider_init_with_credential(
azure_ai_unit_test_env: dict[str, str],
mock_azure_credential: MagicMock,
) -> None:
"""Test AzureAIAgentsProvider initialization with credential."""
with patch("agent_framework_azure_ai._agent_provider.AgentsClient") as mock_client_class:
mock_client_instance = MagicMock()
mock_client_class.return_value = mock_client_instance
provider = AzureAIAgentsProvider(credential=mock_azure_credential)
mock_client_class.assert_called_once()
assert provider._agents_client is mock_client_instance # type: ignore
assert provider._should_close_client is True # type: ignore
def test_provider_init_with_explicit_endpoint(mock_azure_credential: MagicMock) -> None:
"""Test AzureAIAgentsProvider initialization with explicit endpoint."""
with patch("agent_framework_azure_ai._agent_provider.AgentsClient") as mock_client_class:
mock_client_instance = MagicMock()
mock_client_class.return_value = mock_client_instance
provider = AzureAIAgentsProvider(
project_endpoint="https://custom-endpoint.com/",
credential=mock_azure_credential,
)
mock_client_class.assert_called_once()
call_kwargs = mock_client_class.call_args.kwargs
assert call_kwargs["endpoint"] == "https://custom-endpoint.com/"
assert provider._should_close_client is True # type: ignore
def test_provider_init_missing_endpoint_raises(
mock_azure_credential: MagicMock,
) -> None:
"""Test AzureAIAgentsProvider raises error when endpoint is missing."""
# Mock AzureAISettings to return None for project_endpoint
with patch("agent_framework_azure_ai._agent_provider.AzureAISettings") as mock_settings_class:
mock_settings = MagicMock()
mock_settings.project_endpoint = None
mock_settings.model_deployment_name = "test-model"
mock_settings_class.return_value = mock_settings
with pytest.raises(ServiceInitializationError) as exc_info:
AzureAIAgentsProvider(credential=mock_azure_credential)
assert "project endpoint is required" in str(exc_info.value).lower()
def test_provider_init_missing_credential_raises(azure_ai_unit_test_env: dict[str, str]) -> None:
"""Test AzureAIAgentsProvider raises error when credential is missing."""
with pytest.raises(ServiceInitializationError) as exc_info:
AzureAIAgentsProvider()
assert "credential is required" in str(exc_info.value).lower()
# endregion
# region Context Manager Tests
async def test_provider_context_manager_closes_client(mock_agents_client: MagicMock) -> None:
"""Test that context manager closes client when it was created by provider."""
with patch("agent_framework_azure_ai._agent_provider.AgentsClient") as mock_client_class:
mock_client_instance = AsyncMock()
mock_client_class.return_value = mock_client_instance
with patch.object(AzureAIAgentsProvider, "__init__", lambda self: None): # type: ignore
provider = AzureAIAgentsProvider.__new__(AzureAIAgentsProvider)
provider._agents_client = mock_client_instance # type: ignore
provider._should_close_client = True # type: ignore
provider._settings = AzureAISettings(project_endpoint="https://test.com") # type: ignore
async with provider:
pass
mock_client_instance.close.assert_called_once()
async def test_provider_context_manager_does_not_close_external_client(mock_agents_client: MagicMock) -> None:
"""Test that context manager does not close externally provided client."""
mock_agents_client.close = AsyncMock()
provider = AzureAIAgentsProvider(agents_client=mock_agents_client)
async with provider:
pass
mock_agents_client.close.assert_not_called()
# endregion
# region create_agent Tests
async def test_create_agent_basic(
azure_ai_unit_test_env: dict[str, str],
mock_agents_client: MagicMock,
) -> None:
"""Test creating a basic agent."""
mock_agent = MagicMock(spec=Agent)
mock_agent.id = "test-agent-id"
mock_agent.name = "TestAgent"
mock_agent.description = "A test agent"
mock_agent.instructions = "Be helpful"
mock_agent.model = "gpt-4"
mock_agent.temperature = 0.7
mock_agent.top_p = 0.9
mock_agent.tools = []
mock_agents_client.create_agent = AsyncMock(return_value=mock_agent)
provider = AzureAIAgentsProvider(agents_client=mock_agents_client)
agent = await provider.create_agent(
name="TestAgent",
instructions="Be helpful",
description="A test agent",
)
assert isinstance(agent, ChatAgent)
assert agent.name == "TestAgent"
assert agent.id == "test-agent-id"
mock_agents_client.create_agent.assert_called_once()
async def test_create_agent_with_model(
azure_ai_unit_test_env: dict[str, str],
mock_agents_client: MagicMock,
) -> None:
"""Test creating an agent with explicit model."""
mock_agent = MagicMock(spec=Agent)
mock_agent.id = "test-agent-id"
mock_agent.name = "TestAgent"
mock_agent.description = None
mock_agent.instructions = None
mock_agent.model = "custom-model"
mock_agent.temperature = None
mock_agent.top_p = None
mock_agent.tools = []
mock_agents_client.create_agent = AsyncMock(return_value=mock_agent)
provider = AzureAIAgentsProvider(agents_client=mock_agents_client)
await provider.create_agent(name="TestAgent", model="custom-model")
call_kwargs = mock_agents_client.create_agent.call_args.kwargs
assert call_kwargs["model"] == "custom-model"
async def test_create_agent_with_tools(
azure_ai_unit_test_env: dict[str, str],
mock_agents_client: MagicMock,
) -> None:
"""Test creating an agent with tools."""
mock_agent = MagicMock(spec=Agent)
mock_agent.id = "test-agent-id"
mock_agent.name = "TestAgent"
mock_agent.description = None
mock_agent.instructions = None
mock_agent.model = "gpt-4"
mock_agent.temperature = None
mock_agent.top_p = None
mock_agent.tools = []
mock_agents_client.create_agent = AsyncMock(return_value=mock_agent)
provider = AzureAIAgentsProvider(agents_client=mock_agents_client)
@ai_function
def get_weather(city: str) -> str:
"""Get weather for a city."""
return f"Weather in {city}"
await provider.create_agent(name="TestAgent", tools=get_weather)
call_kwargs = mock_agents_client.create_agent.call_args.kwargs
assert "tools" in call_kwargs
assert len(call_kwargs["tools"]) > 0
async def test_create_agent_with_response_format(
azure_ai_unit_test_env: dict[str, str],
mock_agents_client: MagicMock,
) -> None:
"""Test creating an agent with structured response format via default_options."""
class WeatherResponse(BaseModel):
temperature: float
description: str
mock_agent = MagicMock(spec=Agent)
mock_agent.id = "test-agent-id"
mock_agent.name = "TestAgent"
mock_agent.description = None
mock_agent.instructions = None
mock_agent.model = "gpt-4"
mock_agent.temperature = None
mock_agent.top_p = None
mock_agent.tools = []
mock_agents_client.create_agent = AsyncMock(return_value=mock_agent)
provider = AzureAIAgentsProvider(agents_client=mock_agents_client)
await provider.create_agent(
name="TestAgent",
default_options={"response_format": WeatherResponse},
)
call_kwargs = mock_agents_client.create_agent.call_args.kwargs
assert "response_format" in call_kwargs
async def test_create_agent_missing_model_raises(
mock_agents_client: MagicMock,
) -> None:
"""Test that create_agent raises error when model is not specified."""
# Create provider with mocked settings that has no model
with patch("agent_framework_azure_ai._agent_provider.AzureAISettings") as mock_settings_class:
mock_settings = MagicMock()
mock_settings.project_endpoint = "https://test.com"
mock_settings.model_deployment_name = None # No model configured
mock_settings_class.return_value = mock_settings
provider = AzureAIAgentsProvider(agents_client=mock_agents_client)
with pytest.raises(ServiceInitializationError) as exc_info:
await provider.create_agent(name="TestAgent")
assert "model deployment name is required" in str(exc_info.value).lower()
# endregion
# region get_agent Tests
async def test_get_agent_by_id(
azure_ai_unit_test_env: dict[str, str],
mock_agents_client: MagicMock,
) -> None:
"""Test getting an agent by ID."""
mock_agent = MagicMock(spec=Agent)
mock_agent.id = "existing-agent-id"
mock_agent.name = "ExistingAgent"
mock_agent.description = "An existing agent"
mock_agent.instructions = "Be helpful"
mock_agent.model = "gpt-4"
mock_agent.temperature = 0.7
mock_agent.top_p = 0.9
mock_agent.tools = []
mock_agents_client.get_agent = AsyncMock(return_value=mock_agent)
provider = AzureAIAgentsProvider(agents_client=mock_agents_client)
agent = await provider.get_agent("existing-agent-id")
assert isinstance(agent, ChatAgent)
assert agent.id == "existing-agent-id"
mock_agents_client.get_agent.assert_called_once_with("existing-agent-id")
async def test_get_agent_with_function_tools(
azure_ai_unit_test_env: dict[str, str],
mock_agents_client: MagicMock,
) -> None:
"""Test getting an agent that has function tools requires tool implementations."""
mock_function_tool = MagicMock()
mock_function_tool.type = "function"
mock_function_tool.function = MagicMock()
mock_function_tool.function.name = "get_weather"
mock_agent = MagicMock(spec=Agent)
mock_agent.id = "agent-with-tools"
mock_agent.name = "AgentWithTools"
mock_agent.description = None
mock_agent.instructions = None
mock_agent.model = "gpt-4"
mock_agent.temperature = None
mock_agent.top_p = None
mock_agent.tools = [mock_function_tool]
mock_agents_client.get_agent = AsyncMock(return_value=mock_agent)
provider = AzureAIAgentsProvider(agents_client=mock_agents_client)
with pytest.raises(ServiceInitializationError) as exc_info:
await provider.get_agent("agent-with-tools")
assert "get_weather" in str(exc_info.value)
async def test_get_agent_with_provided_function_tools(
azure_ai_unit_test_env: dict[str, str],
mock_agents_client: MagicMock,
) -> None:
"""Test getting an agent with function tools when implementations are provided."""
mock_function_tool = MagicMock()
mock_function_tool.type = "function"
mock_function_tool.function = MagicMock()
mock_function_tool.function.name = "get_weather"
mock_agent = MagicMock(spec=Agent)
mock_agent.id = "agent-with-tools"
mock_agent.name = "AgentWithTools"
mock_agent.description = None
mock_agent.instructions = None
mock_agent.model = "gpt-4"
mock_agent.temperature = None
mock_agent.top_p = None
mock_agent.tools = [mock_function_tool]
mock_agents_client.get_agent = AsyncMock(return_value=mock_agent)
@ai_function
def get_weather(city: str) -> str:
"""Get weather for a city."""
return f"Weather in {city}"
provider = AzureAIAgentsProvider(agents_client=mock_agents_client)
agent = await provider.get_agent("agent-with-tools", tools=get_weather)
assert isinstance(agent, ChatAgent)
assert agent.id == "agent-with-tools"
# endregion
# region as_agent Tests
def test_as_agent_wraps_without_http(
azure_ai_unit_test_env: dict[str, str],
mock_agents_client: MagicMock,
) -> None:
"""Test as_agent wraps Agent object without making HTTP calls."""
mock_agent = MagicMock(spec=Agent)
mock_agent.id = "wrap-agent-id"
mock_agent.name = "WrapAgent"
mock_agent.description = "Wrapped agent"
mock_agent.instructions = "Be helpful"
mock_agent.model = "gpt-4"
mock_agent.temperature = 0.5
mock_agent.top_p = 0.8
mock_agent.tools = []
provider = AzureAIAgentsProvider(agents_client=mock_agents_client)
agent = provider.as_agent(mock_agent)
assert isinstance(agent, ChatAgent)
assert agent.id == "wrap-agent-id"
assert agent.name == "WrapAgent"
# Ensure no HTTP calls were made
mock_agents_client.get_agent.assert_not_called()
mock_agents_client.create_agent.assert_not_called()
def test_as_agent_with_function_tools_validates(
azure_ai_unit_test_env: dict[str, str],
mock_agents_client: MagicMock,
) -> None:
"""Test as_agent validates that function tool implementations are provided."""
mock_function_tool = MagicMock()
mock_function_tool.type = "function"
mock_function_tool.function = MagicMock()
mock_function_tool.function.name = "my_function"
mock_agent = MagicMock(spec=Agent)
mock_agent.id = "agent-id"
mock_agent.name = "Agent"
mock_agent.description = None
mock_agent.instructions = None
mock_agent.model = "gpt-4"
mock_agent.temperature = None
mock_agent.top_p = None
mock_agent.tools = [mock_function_tool]
provider = AzureAIAgentsProvider(agents_client=mock_agents_client)
with pytest.raises(ServiceInitializationError) as exc_info:
provider.as_agent(mock_agent)
assert "my_function" in str(exc_info.value)
def test_as_agent_with_hosted_tools(
azure_ai_unit_test_env: dict[str, str],
mock_agents_client: MagicMock,
) -> None:
"""Test as_agent handles hosted tools correctly."""
mock_code_interpreter = MagicMock()
mock_code_interpreter.type = "code_interpreter"
mock_agent = MagicMock(spec=Agent)
mock_agent.id = "agent-id"
mock_agent.name = "Agent"
mock_agent.description = None
mock_agent.instructions = None
mock_agent.model = "gpt-4"
mock_agent.temperature = None
mock_agent.top_p = None
mock_agent.tools = [mock_code_interpreter]
provider = AzureAIAgentsProvider(agents_client=mock_agents_client)
agent = provider.as_agent(mock_agent)
assert isinstance(agent, ChatAgent)
# Should have HostedCodeInterpreterTool in the default_options tools
assert any(isinstance(t, HostedCodeInterpreterTool) for t in (agent.default_options.get("tools") or []))
# endregion
# region Tool Conversion Tests - to_azure_ai_agent_tools
def test_to_azure_ai_agent_tools_empty() -> None:
"""Test converting empty tools list."""
result = to_azure_ai_agent_tools(None)
assert result == []
result = to_azure_ai_agent_tools([])
assert result == []
def test_to_azure_ai_agent_tools_function() -> None:
"""Test converting AIFunction to Azure tool definition."""
@ai_function
def get_weather(city: str) -> str:
"""Get weather for a city."""
return f"Weather in {city}"
result = to_azure_ai_agent_tools([get_weather])
assert len(result) == 1
assert result[0]["type"] == "function"
assert result[0]["function"]["name"] == "get_weather"
def test_to_azure_ai_agent_tools_code_interpreter() -> None:
"""Test converting HostedCodeInterpreterTool."""
tool = HostedCodeInterpreterTool()
result = to_azure_ai_agent_tools([tool])
assert len(result) == 1
assert isinstance(result[0], CodeInterpreterToolDefinition)
def test_to_azure_ai_agent_tools_file_search() -> None:
"""Test converting HostedFileSearchTool with vector stores."""
tool = HostedFileSearchTool(inputs=[HostedVectorStoreContent(vector_store_id="vs-123")])
run_options: dict[str, Any] = {}
result = to_azure_ai_agent_tools([tool], run_options)
assert len(result) == 1
assert "tool_resources" in run_options
def test_to_azure_ai_agent_tools_web_search_bing_grounding(monkeypatch: Any) -> None:
"""Test converting HostedWebSearchTool 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()
result = to_azure_ai_agent_tools([tool])
assert len(result) > 0
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()
result = to_azure_ai_agent_tools([tool])
assert len(result) > 0
def test_to_azure_ai_agent_tools_web_search_missing_config(monkeypatch: Any) -> None:
"""Test converting HostedWebSearchTool raises error when config is missing."""
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()
with pytest.raises(ServiceInitializationError):
to_azure_ai_agent_tools([tool])
def test_to_azure_ai_agent_tools_mcp() -> None:
"""Test converting HostedMCPTool."""
tool = HostedMCPTool(
name="my mcp server",
url="https://mcp.example.com",
allowed_tools=["tool1", "tool2"],
)
result = to_azure_ai_agent_tools([tool])
assert len(result) > 0
def test_to_azure_ai_agent_tools_dict_passthrough() -> None:
"""Test that dict tools are passed through."""
tool = {"type": "custom_tool", "config": {"key": "value"}}
result = to_azure_ai_agent_tools([tool])
assert len(result) == 1
assert result[0] == tool
def test_to_azure_ai_agent_tools_unsupported_type() -> None:
"""Test that unsupported tool types raise error."""
class UnsupportedTool:
pass
with pytest.raises(ServiceInitializationError):
to_azure_ai_agent_tools([UnsupportedTool()]) # type: ignore
# endregion
# region Tool Conversion Tests - from_azure_ai_agent_tools
def test_from_azure_ai_agent_tools_empty() -> None:
"""Test converting empty tools list."""
result = from_azure_ai_agent_tools(None)
assert result == []
result = from_azure_ai_agent_tools([])
assert result == []
def test_from_azure_ai_agent_tools_code_interpreter() -> None:
"""Test converting CodeInterpreterToolDefinition."""
tool = CodeInterpreterToolDefinition()
result = from_azure_ai_agent_tools([tool])
assert len(result) == 1
assert isinstance(result[0], HostedCodeInterpreterTool)
def test_from_azure_ai_agent_tools_code_interpreter_dict() -> None:
"""Test converting code_interpreter dict."""
tool = {"type": "code_interpreter"}
result = from_azure_ai_agent_tools([tool])
assert len(result) == 1
assert isinstance(result[0], HostedCodeInterpreterTool)
def test_from_azure_ai_agent_tools_file_search_dict() -> None:
"""Test converting file_search dict with vector store IDs."""
tool = {
"type": "file_search",
"file_search": {"vector_store_ids": ["vs-123", "vs-456"]},
}
result = from_azure_ai_agent_tools([tool])
assert len(result) == 1
assert isinstance(result[0], HostedFileSearchTool)
assert len(result[0].inputs or []) == 2
def test_from_azure_ai_agent_tools_bing_grounding_dict() -> None:
"""Test converting bing_grounding dict."""
tool = {
"type": "bing_grounding",
"bing_grounding": {"connection_id": "conn-123"},
}
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"
def test_from_azure_ai_agent_tools_bing_custom_search_dict() -> None:
"""Test converting bing_custom_search dict."""
tool = {
"type": "bing_custom_search",
"bing_custom_search": {
"connection_id": "custom-conn",
"instance_name": "my-instance",
},
}
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"
def test_from_azure_ai_agent_tools_mcp_dict() -> None:
"""Test that mcp dict is skipped (hosted on Azure, no local handling needed)."""
tool = {
"type": "mcp",
"mcp": {
"server_label": "my_server",
"server_url": "https://mcp.example.com",
"allowed_tools": ["tool1"],
},
}
result = from_azure_ai_agent_tools([tool])
# MCP tools are hosted on Azure agent, skipped in conversion
assert len(result) == 0
def test_from_azure_ai_agent_tools_function_dict() -> None:
"""Test converting function tool dict (returned as-is)."""
tool: dict[str, Any] = {
"type": "function",
"function": {
"name": "get_weather",
"description": "Get weather",
"parameters": {},
},
}
result = from_azure_ai_agent_tools([tool])
assert len(result) == 1
assert result[0] == tool
def test_from_azure_ai_agent_tools_unknown_dict() -> None:
"""Test converting unknown tool type dict."""
tool = {"type": "unknown_tool", "config": "value"}
result = from_azure_ai_agent_tools([tool])
assert len(result) == 1
assert result[0] == tool
# endregion
# region Integration Tests
@skip_if_azure_ai_integration_tests_disabled
async def test_integration_create_agent() -> None:
"""Integration test: Create an agent using the provider."""
from azure.identity.aio import AzureCliCredential
async with (
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="IntegrationTestAgent",
instructions="You are a helpful assistant for testing.",
)
try:
assert isinstance(agent, ChatAgent)
assert agent.name == "IntegrationTestAgent"
assert agent.id is not None
finally:
# Cleanup: delete the agent
if agent.id:
await provider._agents_client.delete_agent(agent.id) # type: ignore
@skip_if_azure_ai_integration_tests_disabled
async def test_integration_get_agent() -> None:
"""Integration test: Get an existing agent using the provider."""
from azure.identity.aio import AzureCliCredential
async with (
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
# First create an agent
created = await provider._agents_client.create_agent( # type: ignore
model=os.getenv("AZURE_AI_MODEL_DEPLOYMENT_NAME", "gpt-4o"),
name="GetAgentTest",
instructions="Test agent",
)
try:
# Then get it using the provider
agent = await provider.get_agent(created.id)
assert isinstance(agent, ChatAgent)
assert agent.id == created.id
finally:
await provider._agents_client.delete_agent(created.id) # type: ignore
@skip_if_azure_ai_integration_tests_disabled
async def test_integration_create_and_run() -> None:
"""Integration test: Create an agent and run a conversation."""
from azure.identity.aio import AzureCliCredential
async with (
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="RunTestAgent",
instructions="You are a helpful assistant. Always respond with 'Hello!' to any greeting.",
)
try:
result = await agent.run("Hi there!")
assert result is not None
assert len(result.messages) > 0
finally:
if agent.id:
await provider._agents_client.delete_agent(agent.id) # type: ignore
# endregion
@@ -11,7 +11,6 @@ from agent_framework import (
AgentResponse,
AgentResponseUpdate,
AgentThread,
AIFunction,
ChatAgent,
ChatClientProtocol,
ChatMessage,
@@ -28,7 +27,6 @@ from agent_framework import (
HostedFileSearchTool,
HostedMCPTool,
HostedVectorStoreContent,
HostedWebSearchTool,
Role,
TextContent,
UriContent,
@@ -38,7 +36,6 @@ from agent_framework.exceptions import ServiceInitializationError
from azure.ai.agents.models import (
AgentsNamedToolChoice,
AgentsNamedToolChoiceType,
CodeInterpreterToolDefinition,
FileInfo,
MessageDeltaChunk,
MessageDeltaTextContent,
@@ -672,60 +669,6 @@ def test_azure_ai_chat_client_service_url_method(mock_agents_client: MagicMock)
assert url == "https://test-endpoint.com/"
async def test_azure_ai_chat_client_prepare_tools_for_azure_ai_ai_function(mock_agents_client: MagicMock) -> None:
"""Test _prepare_tools_for_azure_ai with AIFunction tool."""
chat_client = create_test_azure_ai_chat_client(mock_agents_client, agent_id="test-agent")
# Create a mock AIFunction
mock_ai_function = MagicMock(spec=AIFunction)
mock_ai_function.to_json_schema_spec.return_value = {"type": "function", "function": {"name": "test_function"}}
result = await chat_client._prepare_tools_for_azure_ai([mock_ai_function]) # type: ignore
assert len(result) == 1
assert result[0] == {"type": "function", "function": {"name": "test_function"}}
mock_ai_function.to_json_schema_spec.assert_called_once()
async def test_azure_ai_chat_client_prepare_tools_for_azure_ai_code_interpreter(mock_agents_client: MagicMock) -> None:
"""Test _prepare_tools_for_azure_ai with HostedCodeInterpreterTool."""
chat_client = create_test_azure_ai_chat_client(mock_agents_client, agent_id="test-agent")
code_interpreter_tool = HostedCodeInterpreterTool()
result = await chat_client._prepare_tools_for_azure_ai([code_interpreter_tool]) # type: ignore
assert len(result) == 1
assert isinstance(result[0], CodeInterpreterToolDefinition)
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."""
chat_client = create_test_azure_ai_chat_client(mock_agents_client, agent_id="test-agent")
mcp_tool = HostedMCPTool(name="Test MCP Tool", url="https://example.com/mcp", allowed_tools=["tool1", "tool2"])
# Mock McpTool to have a definitions attribute
with patch("agent_framework_azure_ai._chat_client.McpTool") as mock_mcp_tool_class:
mock_mcp_tool = MagicMock()
mock_mcp_tool.definitions = [{"type": "mcp", "name": "test_mcp"}]
mock_mcp_tool_class.return_value = mock_mcp_tool
result = await chat_client._prepare_tools_for_azure_ai([mcp_tool]) # type: ignore
assert len(result) == 1
assert result[0] == {"type": "mcp", "name": "test_mcp"}
# Check that the call was made (order of allowed_tools may vary)
mock_mcp_tool_class.assert_called_once()
call_args = mock_mcp_tool_class.call_args[1]
assert call_args["server_label"] == "Test_MCP_Tool"
assert call_args["server_url"] == "https://example.com/mcp"
assert set(call_args["allowed_tools"]) == {"tool1", "tool2"}
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."""
chat_client = create_test_azure_ai_chat_client(mock_agents_client)
@@ -735,8 +678,7 @@ async def test_azure_ai_chat_client_prepare_options_mcp_never_require(mock_agent
messages = [ChatMessage(role=Role.USER, text="Hello")]
chat_options: ChatOptions = {"tools": [mcp_tool], "tool_choice": "auto"}
with patch("agent_framework_azure_ai._chat_client.McpTool") as mock_mcp_tool_class:
# Mock _prepare_tools_for_azure_ai to avoid actual tool preparation
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
@@ -768,8 +710,7 @@ async def test_azure_ai_chat_client_prepare_options_mcp_with_headers(mock_agents
messages = [ChatMessage(role=Role.USER, text="Hello")]
chat_options: ChatOptions = {"tools": [mcp_tool], "tool_choice": "auto"}
with patch("agent_framework_azure_ai._chat_client.McpTool") as mock_mcp_tool_class:
# Mock _prepare_tools_for_azure_ai to avoid actual tool preparation
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
@@ -787,121 +728,6 @@ async def test_azure_ai_chat_client_prepare_options_mcp_with_headers(mock_agents
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."""
chat_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
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
result = await chat_client._prepare_tools_for_azure_ai([web_search_tool]) # type: ignore
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)."""
chat_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
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
result = await chat_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."""
chat_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
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
result = await chat_client._prepare_tools_for_azure_ai([web_search_tool]) # type: ignore
assert len(result) == 1
assert result[0] == {"type": "bing_custom_search"}
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."""
chat_client = create_test_azure_ai_chat_client(mock_agents_client, agent_id="test-agent")
vector_store_input = HostedVectorStoreContent(vector_store_id="vs-123")
file_search_tool = HostedFileSearchTool(inputs=[vector_store_input])
# 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 = {}
result = await chat_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"])
async def test_azure_ai_chat_client_create_agent_stream_submit_tool_approvals(
mock_agents_client: MagicMock,
) -> None:
@@ -943,28 +769,6 @@ async def test_azure_ai_chat_client_create_agent_stream_submit_tool_approvals(
assert call_args["tool_approvals"][0].approve is True
async def test_azure_ai_chat_client_prepare_tools_for_azure_ai_dict_tool(mock_agents_client: MagicMock) -> None:
"""Test _prepare_tools_for_azure_ai with dictionary tool definition."""
chat_client = create_test_azure_ai_chat_client(mock_agents_client, agent_id="test-agent")
dict_tool = {"type": "custom_tool", "config": {"param": "value"}}
result = await chat_client._prepare_tools_for_azure_ai([dict_tool]) # type: ignore
assert len(result) == 1
assert result[0] == dict_tool
async def test_azure_ai_chat_client_prepare_tools_for_azure_ai_unsupported_tool(mock_agents_client: MagicMock) -> None:
"""Test _prepare_tools_for_azure_ai with unsupported tool type."""
chat_client = create_test_azure_ai_chat_client(mock_agents_client, agent_id="test-agent")
unsupported_tool = "not_a_tool"
with pytest.raises(ServiceInitializationError, match="Unsupported tool type: <class 'str'>"):
await chat_client._prepare_tools_for_azure_ai([unsupported_tool]) # type: ignore
async def test_azure_ai_chat_client_get_active_thread_run_with_active_run(mock_agents_client: MagicMock) -> None:
"""Test _get_active_thread_run when there's an active run."""
@@ -86,7 +86,7 @@ def test_provider_init_with_credential_and_endpoint(
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:
with patch("agent_framework_azure_ai._project_provider.AIProjectClient") as mock_ai_project_client:
mock_client = MagicMock()
mock_ai_project_client.return_value = mock_client
@@ -104,7 +104,7 @@ def test_provider_init_with_credential_and_endpoint(
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:
with patch("agent_framework_azure_ai._project_provider.AzureAISettings") as mock_settings:
mock_settings.return_value.project_endpoint = None
mock_settings.return_value.model_deployment_name = "test-model"
@@ -127,7 +127,7 @@ async def test_provider_create_agent(
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:
with patch("agent_framework_azure_ai._project_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"]
@@ -165,7 +165,7 @@ async def test_provider_create_agent_with_env_model(
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:
with patch("agent_framework_azure_ai._project_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"]
@@ -197,7 +197,7 @@ async def test_provider_create_agent_with_env_model(
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:
with patch("agent_framework_azure_ai._project_provider.AzureAISettings") as mock_settings:
mock_settings.return_value.project_endpoint = "https://test.com"
mock_settings.return_value.model_deployment_name = None
@@ -326,12 +326,12 @@ def test_provider_as_agent(mock_project_client: MagicMock) -> None:
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:
with patch("agent_framework_azure_ai._project_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:
with patch("agent_framework_azure_ai._project_provider.AzureAISettings") as mock_settings:
mock_settings.return_value.project_endpoint = "https://test.com"
mock_settings.return_value.model_deployment_name = "test-model"
@@ -355,12 +355,12 @@ async def test_provider_context_manager_with_provided_client(mock_project_client
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:
with patch("agent_framework_azure_ai._project_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:
with patch("agent_framework_azure_ai._project_provider.AzureAISettings") as mock_settings:
mock_settings.return_value.project_endpoint = "https://test.com"
mock_settings.return_value.model_deployment_name = "test-model"
@@ -14,6 +14,7 @@ _IMPORTS: dict[str, tuple[str, str]] = {
"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"),
"AzureAIAgentsProvider": ("agent_framework_azure_ai", "agent-framework-azure-ai"),
"AzureOpenAIAssistantsClient": ("agent_framework.azure._assistants_client", "agent-framework-core"),
"AzureOpenAIAssistantsOptions": ("agent_framework.azure._assistants_client", "agent-framework-core"),
"AzureOpenAIChatClient": ("agent_framework.azure._chat_client", "agent-framework-core"),
@@ -1,6 +1,12 @@
# Copyright (c) Microsoft. All rights reserved.
from agent_framework_azure_ai import AzureAIAgentClient, AzureAIClient, AzureAIProjectAgentProvider, AzureAISettings
from agent_framework_azure_ai import (
AzureAIAgentClient,
AzureAIAgentsProvider,
AzureAIClient,
AzureAIProjectAgentProvider,
AzureAISettings,
)
from agent_framework_azure_ai_search import AzureAISearchContextProvider, AzureAISearchSettings
from agent_framework_azurefunctions import (
AgentCallbackContext,
@@ -20,6 +26,7 @@ __all__ = [
"AgentFunctionApp",
"AgentResponseCallbackProtocol",
"AzureAIAgentClient",
"AzureAIAgentsProvider",
"AzureAIClient",
"AzureAIProjectAgentProvider",
"AzureAISearchContextProvider",
@@ -34,7 +34,7 @@ async def main() -> None:
name="ProductMarketerAgent",
instructions="Return launch briefs as structured JSON.",
# Specify type to use as response
options={"response_format": ReleaseBrief},
default_options={"response_format": ReleaseBrief},
)
query = "Draft a launch brief for the Contoso Note app."
@@ -1,27 +1,53 @@
# Azure AI Agent Examples
This folder contains examples demonstrating different ways to create and use agents with the Azure AI chat client from the `agent_framework.azure` package. These examples use the `AzureAIAgentClient` with the `azure-ai-agents` 1.x (V1) API surface. For updated V2 (`azure-ai-projects` 2.x) samples, see the [Azure AI V2 examples folder](../azure_ai/).
This folder contains examples demonstrating different ways to create and use agents with Azure AI using the `AzureAIAgentsProvider` from the `agent_framework.azure` package. These examples use the `azure-ai-agents` 1.x (V1) API surface. For updated V2 (`azure-ai-projects` 2.x) samples, see the [Azure AI V2 examples folder](../azure_ai/).
## Provider Pattern
All examples in this folder use the `AzureAIAgentsProvider` class which provides a high-level interface for agent operations:
- **`create_agent()`** - Create a new agent on the Azure AI service
- **`get_agent()`** - Retrieve an existing agent by ID or from a pre-fetched Agent object
- **`as_agent()`** - Wrap an SDK Agent object as a ChatAgent without HTTP calls
```python
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
async with (
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="MyAgent",
instructions="You are a helpful assistant.",
tools=my_function,
)
result = await agent.run("Hello!")
```
## Examples
| File | Description |
|------|-------------|
| [`azure_ai_basic.py`](azure_ai_basic.py) | The simplest way to create an agent using `ChatAgent` with `AzureAIAgentClient`. It automatically handles all configuration using environment variables. |
| [`azure_ai_provider_methods.py`](azure_ai_provider_methods.py) | Comprehensive example demonstrating all `AzureAIAgentsProvider` methods: `create_agent()`, `get_agent()`, `as_agent()`, and managing multiple agents from a single provider. |
| [`azure_ai_basic.py`](azure_ai_basic.py) | The simplest way to create an agent using `AzureAIAgentsProvider`. It automatically handles all configuration using environment variables. Shows both streaming and non-streaming responses. |
| [`azure_ai_with_bing_custom_search.py`](azure_ai_with_bing_custom_search.py) | Shows how to use Bing Custom Search with Azure AI agents to find real-time information from the web using custom search configurations. Demonstrates how to set up and use HostedWebSearchTool with custom search instances. |
| [`azure_ai_with_bing_grounding.py`](azure_ai_with_bing_grounding.py) | Shows how to use Bing Grounding search with Azure AI agents to find real-time information from the web. Demonstrates web search capabilities with proper source citations and comprehensive error handling. |
| [`azure_ai_with_bing_grounding_citations.py`](azure_ai_with_bing_grounding_citations.py) | Demonstrates how to extract and display citations from Bing Grounding search responses. Shows how to collect citation annotations (title, URL, snippet) during streaming responses, enabling users to verify sources and access referenced content. |
| [`azure_ai_with_code_interpreter_file_generation.py`](azure_ai_with_code_interpreter_file_generation.py) | Shows how to retrieve file IDs from code interpreter generated files using both streaming and non-streaming approaches. |
| [`azure_ai_with_code_interpreter.py`](azure_ai_with_code_interpreter.py) | Shows how to use the HostedCodeInterpreterTool with Azure AI agents to write and execute Python code. Includes helper methods for accessing code interpreter data from response chunks. |
| [`azure_ai_with_existing_agent.py`](azure_ai_with_existing_agent.py) | Shows how to work with a pre-existing agent by providing the agent ID to the Azure AI chat client. This example also demonstrates proper cleanup of manually created agents. |
| [`azure_ai_with_existing_thread.py`](azure_ai_with_existing_thread.py) | Shows how to work with a pre-existing thread by providing the thread ID to the Azure AI chat client. This example also demonstrates proper cleanup of manually created threads. |
| [`azure_ai_with_explicit_settings.py`](azure_ai_with_explicit_settings.py) | Shows how to create an agent with explicitly configured `AzureAIAgentClient` settings, including project endpoint, model deployment, credentials, and agent name. |
| [`azure_ai_with_azure_ai_search.py`](azure_ai_with_azure_ai_search.py) | Demonstrates how to use Azure AI Search with Azure AI agents to search through indexed data. Shows how to configure search parameters, query types, and integrate with existing search indexes. |
| [`azure_ai_with_file_search.py`](azure_ai_with_file_search.py) | Demonstrates how to use the HostedFileSearchTool with Azure AI agents to search through uploaded documents. Shows file upload, vector store creation, and querying document content. Includes both streaming and non-streaming examples. |
| [`azure_ai_with_existing_agent.py`](azure_ai_with_existing_agent.py) | Shows how to work with an existing SDK Agent object using `provider.as_agent()`. This wraps the agent without making HTTP calls. |
| [`azure_ai_with_existing_thread.py`](azure_ai_with_existing_thread.py) | Shows how to work with a pre-existing thread by providing the thread ID. Demonstrates proper cleanup of manually created threads. |
| [`azure_ai_with_explicit_settings.py`](azure_ai_with_explicit_settings.py) | Shows how to create an agent with explicitly configured provider settings, including project endpoint and model deployment name. |
| [`azure_ai_with_azure_ai_search.py`](azure_ai_with_azure_ai_search.py) | Demonstrates how to use Azure AI Search with Azure AI agents. Shows how to create an agent with search tools using the SDK directly and wrap it with `provider.get_agent()`. |
| [`azure_ai_with_file_search.py`](azure_ai_with_file_search.py) | Demonstrates how to use the HostedFileSearchTool with Azure AI agents to search through uploaded documents. Shows file upload, vector store creation, and querying document content. |
| [`azure_ai_with_function_tools.py`](azure_ai_with_function_tools.py) | Demonstrates how to use function tools with agents. Shows both agent-level tools (defined when creating the agent) and query-level tools (provided with specific queries). |
| [`azure_ai_with_hosted_mcp.py`](azure_ai_with_hosted_mcp.py) | Shows how to integrate Azure AI agents with hosted Model Context Protocol (MCP) servers for enhanced functionality and tool integration. Demonstrates remote MCP server connections and tool discovery. |
| [`azure_ai_with_local_mcp.py`](azure_ai_with_local_mcp.py) | Shows how to integrate Azure AI agents with local Model Context Protocol (MCP) servers for enhanced functionality and tool integration. Demonstrates both agent-level and run-level tool configuration. |
| [`azure_ai_with_multiple_tools.py`](azure_ai_with_multiple_tools.py) | Demonstrates how to use multiple tools together with Azure AI agents, including web search, MCP servers, and function tools. Shows coordinated multi-tool interactions and approval workflows. |
| [`azure_ai_with_openapi_tools.py`](azure_ai_with_openapi_tools.py) | Demonstrates how to use OpenAPI tools with Azure AI agents to integrate external REST APIs. Shows OpenAPI specification loading, anonymous authentication, thread context management, and coordinated multi-API conversations using weather and countries APIs. |
| [`azure_ai_with_openapi_tools.py`](azure_ai_with_openapi_tools.py) | Demonstrates how to use OpenAPI tools with Azure AI agents to integrate external REST APIs. Shows OpenAPI specification loading, anonymous authentication, thread context management, and coordinated multi-API conversations. |
| [`azure_ai_with_response_format.py`](azure_ai_with_response_format.py) | Demonstrates how to use structured outputs with Azure AI agents using Pydantic models. |
| [`azure_ai_with_thread.py`](azure_ai_with_thread.py) | Demonstrates thread management with Azure AI agents, including automatic thread creation for stateless conversations and explicit thread management for maintaining conversation context across multiple interactions. |
## Environment Variables
@@ -4,14 +4,14 @@ import asyncio
from random import randint
from typing import Annotated
from agent_framework.azure import AzureAIAgentClient
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
from pydantic import Field
"""
Azure AI Agent Basic Example
This sample demonstrates basic usage of AzureAIAgentClient to create agents with automatic
This sample demonstrates basic usage of AzureAIAgentsProvider to create agents with automatic
lifecycle management. Shows both streaming and non-streaming responses with function tools.
"""
@@ -28,18 +28,17 @@ async def non_streaming_example() -> None:
"""Example of non-streaming response (get the complete result at once)."""
print("=== Non-streaming Response Example ===")
# Since no Agent ID is provided, the agent will be automatically created
# and deleted after getting a response
# For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
# authentication option.
async with (
AzureCliCredential() as credential,
AzureAIAgentClient(credential=credential).create_agent(
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="WeatherAgent",
instructions="You are a helpful weather agent.",
tools=get_weather,
) as agent,
):
)
query = "What's the weather like in Seattle?"
print(f"User: {query}")
result = await agent.run(query)
@@ -50,18 +49,17 @@ async def streaming_example() -> None:
"""Example of streaming response (get results as they are generated)."""
print("=== Streaming Response Example ===")
# Since no Agent ID is provided, the agent will be automatically created
# and deleted after getting a response
# For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
# authentication option.
async with (
AzureCliCredential() as credential,
AzureAIAgentClient(credential=credential).create_agent(
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="WeatherAgent",
instructions="You are a helpful weather agent.",
tools=get_weather,
) as agent,
):
)
query = "What's the weather like in Portland?"
print(f"User: {query}")
print("Agent: ", end="", flush=True)
@@ -0,0 +1,142 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from random import randint
from typing import Annotated
from agent_framework.azure import AzureAIAgentsProvider
from azure.ai.agents.aio import AgentsClient
from azure.identity.aio import AzureCliCredential
from pydantic import Field
"""
Azure AI Agent Provider Methods Example
This sample demonstrates the methods available on the AzureAIAgentsProvider class:
- create_agent(): Create a new agent on the service
- get_agent(): Retrieve an existing agent by ID
- as_agent(): Wrap an SDK Agent object without making HTTP calls
"""
def get_weather(
location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
"""Get the weather for a given location."""
conditions = ["sunny", "cloudy", "rainy", "stormy"]
return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."
async def create_agent_example() -> None:
"""Create a new agent using provider.create_agent()."""
print("\n--- create_agent() ---")
async with (
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="WeatherAgent",
instructions="You are a helpful weather assistant.",
tools=get_weather,
)
print(f"Created: {agent.name} (ID: {agent.id})")
result = await agent.run("What's the weather in Seattle?")
print(f"Response: {result}")
async def get_agent_example() -> None:
"""Retrieve an existing agent by ID using provider.get_agent()."""
print("\n--- get_agent() ---")
async with (
AzureCliCredential() as credential,
AgentsClient(endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"], credential=credential) as agents_client,
AzureAIAgentsProvider(agents_client=agents_client) as provider,
):
# Create an agent directly with SDK (simulating pre-existing agent)
sdk_agent = await agents_client.create_agent(
model=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
name="ExistingAgent",
instructions="You always respond with 'Hello!'",
)
try:
# Retrieve using provider
agent = await provider.get_agent(sdk_agent.id)
print(f"Retrieved: {agent.name} (ID: {agent.id})")
result = await agent.run("Hi there!")
print(f"Response: {result}")
finally:
await agents_client.delete_agent(sdk_agent.id)
async def as_agent_example() -> None:
"""Wrap an SDK Agent object using provider.as_agent()."""
print("\n--- as_agent() ---")
async with (
AzureCliCredential() as credential,
AgentsClient(endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"], credential=credential) as agents_client,
AzureAIAgentsProvider(agents_client=agents_client) as provider,
):
# Create agent using SDK
sdk_agent = await agents_client.create_agent(
model=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
name="WrappedAgent",
instructions="You respond with poetry.",
)
try:
# Wrap synchronously (no HTTP call)
agent = provider.as_agent(sdk_agent)
print(f"Wrapped: {agent.name} (ID: {agent.id})")
result = await agent.run("Tell me about the sunset.")
print(f"Response: {result}")
finally:
await agents_client.delete_agent(sdk_agent.id)
async def multiple_agents_example() -> None:
"""Create and manage multiple agents with a single provider."""
print("\n--- Multiple Agents ---")
async with (
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
weather_agent = await provider.create_agent(
name="WeatherSpecialist",
instructions="You are a weather specialist.",
tools=get_weather,
)
greeter_agent = await provider.create_agent(
name="GreeterAgent",
instructions="You are a friendly greeter.",
)
print(f"Created: {weather_agent.name}, {greeter_agent.name}")
greeting = await greeter_agent.run("Hello!")
print(f"Greeter: {greeting}")
weather = await weather_agent.run("What's the weather in Tokyo?")
print(f"Weather: {weather}")
async def main() -> None:
print("Azure AI Agent Provider Methods")
await create_agent_example()
await get_agent_example()
await as_agent_example()
await multiple_agents_example()
if __name__ == "__main__":
asyncio.run(main())
@@ -3,8 +3,8 @@
import asyncio
import os
from agent_framework import ChatAgent, CitationAnnotation
from agent_framework.azure import AzureAIAgentClient
from agent_framework import CitationAnnotation
from agent_framework.azure import AzureAIAgentsProvider
from azure.ai.agents.aio import AgentsClient
from azure.ai.projects.aio import AIProjectClient
from azure.ai.projects.models import ConnectionType
@@ -41,6 +41,7 @@ async def main() -> None:
AzureCliCredential() as credential,
AIProjectClient(endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"], credential=credential) as project_client,
AgentsClient(endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"], credential=credential) as agents_client,
AzureAIAgentsProvider(agents_client=agents_client) as provider,
):
ai_search_conn_id = ""
async for connection in project_client.connections.list():
@@ -48,7 +49,8 @@ async def main() -> None:
ai_search_conn_id = connection.id
break
# 1. Create Azure AI agent with the search tool
# 1. Create Azure AI agent with the search tool using SDK directly
# (Azure AI Search tool requires special tool_resources configuration)
azure_ai_agent = await agents_client.create_agent(
model=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
name="HotelSearchAgent",
@@ -70,47 +72,42 @@ async def main() -> None:
},
)
# 2. Create chat client with the existing agent
chat_client = AzureAIAgentClient(agents_client=agents_client, agent_id=azure_ai_agent.id)
try:
async with ChatAgent(
chat_client=chat_client,
# Additional instructions for this specific conversation
instructions=("You are a helpful agent that uses the search tool and index to find hotel information."),
) as agent:
print("This agent uses raw Azure AI Search tool to search hotel data.\n")
# 2. Use provider.as_agent() to wrap the existing agent
agent = provider.as_agent(agent=azure_ai_agent)
# 3. Simulate conversation with the agent
user_input = (
"Use Azure AI search knowledge tool to find detailed information about a winter hotel."
" Use the search tool and index." # You can modify prompt to force tool usage
)
print(f"User: {user_input}")
print("Agent: ", end="", flush=True)
print("This agent uses raw Azure AI Search tool to search hotel data.\n")
# Stream the response and collect citations
citations: list[CitationAnnotation] = []
async for chunk in agent.run_stream(user_input):
if chunk.text:
print(chunk.text, end="", flush=True)
# 3. Simulate conversation with the agent
user_input = (
"Use Azure AI search knowledge tool to find detailed information about a winter hotel."
" Use the search tool and index." # You can modify prompt to force tool usage
)
print(f"User: {user_input}")
print("Agent: ", end="", flush=True)
# Collect citations from Azure AI Search responses
for content in getattr(chunk, "contents", []):
annotations = getattr(content, "annotations", [])
if annotations:
citations.extend(annotations)
# Stream the response and collect citations
citations: list[CitationAnnotation] = []
async for chunk in agent.run_stream(user_input):
if chunk.text:
print(chunk.text, end="", flush=True)
print()
# Collect citations from Azure AI Search responses
for content in getattr(chunk, "contents", []):
annotations = getattr(content, "annotations", [])
if annotations:
citations.extend(annotations)
# Display collected citation
if citations:
print("\n\nCitation:")
for i, citation in enumerate(citations, 1):
print(f"[{i}] {citation.url}")
print()
print("\n" + "=" * 50 + "\n")
print("Hotel search conversation completed!")
# Display collected citation
if citations:
print("\n\nCitation:")
for i, citation in enumerate(citations, 1):
print(f"[{i}] {citation.url}")
print("\n" + "=" * 50 + "\n")
print("Hotel search conversation completed!")
finally:
# Clean up the agent manually
@@ -2,8 +2,8 @@
import asyncio
from agent_framework import ChatAgent, HostedWebSearchTool
from agent_framework.azure import AzureAIAgentClient
from agent_framework import HostedWebSearchTool
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
"""
@@ -37,19 +37,20 @@ async def main() -> None:
description="Search the web for current information using Bing Custom Search",
)
# 2. Use AzureAIAgentClient as async context manager for automatic cleanup
# 2. Use AzureAIAgentsProvider for agent creation and management
async with (
AzureAIAgentClient(credential=AzureCliCredential()) as client,
ChatAgent(
chat_client=client,
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="BingSearchAgent",
instructions=(
"You are a helpful agent that can use Bing Custom Search tools to assist users. "
"Use the available Bing Custom Search tools to answer questions and perform tasks."
),
tools=bing_search_tool,
) as agent,
):
)
# 3. Demonstrate agent capabilities with bing custom search
print("=== Azure AI Agent with Bing Custom Search ===\n")
@@ -2,8 +2,8 @@
import asyncio
from agent_framework import ChatAgent, HostedWebSearchTool
from agent_framework_azure_ai import AzureAIAgentClient
from agent_framework import HostedWebSearchTool
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
"""
@@ -32,11 +32,12 @@ async def main() -> None:
description="Search the web for current information using Bing",
)
# 2. Use AzureAIAgentClient as async context manager for automatic cleanup
# 2. Use AzureAIAgentsProvider for agent creation and management
async with (
AzureAIAgentClient(credential=AzureCliCredential()) as client,
ChatAgent(
chat_client=client,
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="BingSearchAgent",
instructions=(
"You are a helpful assistant that can search the web for current information. "
@@ -44,9 +45,9 @@ async def main() -> None:
"well-sourced answers. Always cite your sources when possible."
),
tools=bing_search_tool,
) as agent,
):
# 4. Demonstrate agent capabilities with web search
)
# 3. Demonstrate agent capabilities with web search
print("=== Azure AI Agent with Bing Grounding Search ===\n")
user_input = "What is the most popular programming language?"
@@ -2,8 +2,8 @@
import asyncio
from agent_framework import ChatAgent, CitationAnnotation, HostedWebSearchTool
from agent_framework.azure import AzureAIAgentClient
from agent_framework import CitationAnnotation, HostedWebSearchTool
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
"""
@@ -34,11 +34,12 @@ async def main() -> None:
description="Search the web for current information using Bing",
)
# 2. Use AzureAIAgentClient as async context manager for automatic cleanup
# 2. Use AzureAIAgentsProvider for agent creation and management
async with (
AzureAIAgentClient(credential=AzureCliCredential()) as client,
ChatAgent(
chat_client=client,
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="BingSearchAgent",
instructions=(
"You are a helpful assistant that can search the web for current information. "
@@ -46,8 +47,8 @@ async def main() -> None:
"well-sourced answers. Always cite your sources when possible."
),
tools=bing_search_tool,
) as agent,
):
)
# 3. Demonstrate agent capabilities with web search
print("=== Azure AI Agent with Bing Grounding Search ===\n")
@@ -3,7 +3,7 @@
import asyncio
from agent_framework import AgentResponse, ChatResponseUpdate, HostedCodeInterpreterTool
from agent_framework.azure import AzureAIAgentClient
from agent_framework.azure import AzureAIAgentsProvider
from azure.ai.agents.models import (
RunStepDeltaCodeInterpreterDetailItemObject,
)
@@ -39,16 +39,16 @@ async def main() -> None:
# authentication option.
async with (
AzureCliCredential() as credential,
AzureAIAgentClient(credential=credential) as chat_client,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = chat_client.create_agent(
agent = await provider.create_agent(
name="CodingAgent",
instructions=("You are a helpful assistant that can write and execute Python code to solve problems."),
tools=HostedCodeInterpreterTool(),
)
query = "Generate the factorial of 100 using python code, show the code and execute it."
print(f"User: {query}")
response = await AgentResponse.from_agent_response_generator(agent.run_stream(query))
response = await agent.run(query)
print(f"Agent: {response}")
# To review the code interpreter outputs, you can access
# them from the response raw_representations, just uncomment the next line:
@@ -1,15 +1,21 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from agent_framework import AgentResponseUpdate, ChatAgent, HostedCodeInterpreterTool, HostedFileContent
from agent_framework.azure import AzureAIAgentClient
from agent_framework import (
AgentResponseUpdate,
HostedCodeInterpreterTool,
HostedFileContent,
)
from agent_framework.azure import AzureAIAgentsProvider
from azure.ai.agents.aio import AgentsClient
from azure.identity.aio import AzureCliCredential
"""
Azure AI Agent Code Interpreter File Generation Example
This sample demonstrates using HostedCodeInterpreterTool with AzureAIAgentClient
This sample demonstrates using HostedCodeInterpreterTool with AzureAIAgentsProvider
to generate a text file and then retrieve it.
The test flow:
@@ -23,79 +29,77 @@ The test flow:
async def main() -> None:
"""Test file generation and retrieval with code interpreter."""
async with AzureCliCredential() as credential:
client = AzureAIAgentClient(credential=credential)
async with (
AzureCliCredential() as credential,
AgentsClient(endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"], credential=credential) as agents_client,
AzureAIAgentsProvider(agents_client=agents_client) as provider,
):
agent = await provider.create_agent(
name="CodeInterpreterAgent",
instructions=(
"You are a Python code execution assistant. "
"ALWAYS use the code interpreter tool to execute Python code when asked to create files. "
"Write actual Python code to create files, do not just describe what you would do."
),
tools=[HostedCodeInterpreterTool()],
)
try:
async with ChatAgent(
chat_client=client,
instructions=(
"You are a Python code execution assistant. "
"ALWAYS use the code interpreter tool to execute Python code when asked to create files. "
"Write actual Python code to create files, do not just describe what you would do."
),
tools=[HostedCodeInterpreterTool()],
) as agent:
# Be very explicit about wanting code execution and a download link
query = (
"Use the code interpreter to execute this Python code and then provide me "
"with a download link for the generated file:\n"
"```python\n"
"with open('/mnt/data/sample.txt', 'w') as f:\n"
" f.write('Hello, World! This is a test file.')\n"
"'/mnt/data/sample.txt'\n" # Return the path so it becomes downloadable
"```"
)
print(f"User: {query}\n")
print("=" * 60)
# Be very explicit about wanting code execution and a download link
query = (
"Use the code interpreter to execute this Python code and then provide me "
"with a download link for the generated file:\n"
"```python\n"
"with open('/mnt/data/sample.txt', 'w') as f:\n"
" f.write('Hello, World! This is a test file.')\n"
"'/mnt/data/sample.txt'\n" # Return the path so it becomes downloadable
"```"
)
print(f"User: {query}\n")
print("=" * 60)
# Collect file_ids from the response
file_ids: list[str] = []
# Collect file_ids from the response
file_ids: list[str] = []
async for chunk in agent.run_stream(query):
if not isinstance(chunk, AgentResponseUpdate):
continue
async for chunk in agent.run_stream(query):
if not isinstance(chunk, AgentResponseUpdate):
continue
for content in chunk.contents:
if content.type == "text":
print(content.text, end="", flush=True)
elif content.type == "hosted_file":
if isinstance(content, HostedFileContent):
file_ids.append(content.file_id)
print(f"\n[File generated: {content.file_id}]")
for content in chunk.contents:
if content.type == "text":
print(content.text, end="", flush=True)
elif content.type == "hosted_file" and isinstance(content, HostedFileContent):
file_ids.append(content.file_id)
print(f"\n[File generated: {content.file_id}]")
print("\n" + "=" * 60)
print("\n" + "=" * 60)
# Attempt to retrieve discovered files
if file_ids:
print(f"\nAttempting to retrieve {len(file_ids)} file(s):")
for file_id in file_ids:
try:
file_info = await client.agents_client.files.get(file_id)
print(f" File {file_id}: Retrieved successfully")
print(f" Filename: {file_info.filename}")
print(f" Purpose: {file_info.purpose}")
print(f" Bytes: {file_info.bytes}")
except Exception as e:
print(f" File {file_id}: FAILED to retrieve - {e}")
else:
print("No file IDs were captured from the response.")
# List all files to see if any exist
print("\nListing all files in the agent service:")
# Attempt to retrieve discovered files
if file_ids:
print(f"\nAttempting to retrieve {len(file_ids)} file(s):")
for file_id in file_ids:
try:
files_list = await client.agents_client.files.list()
count = 0
for file_info in files_list.data:
count += 1
print(f" - {file_info.id}: {file_info.filename} ({file_info.purpose})")
if count == 0:
print(" No files found.")
file_info = await agents_client.files.get(file_id)
print(f" File {file_id}: Retrieved successfully")
print(f" Filename: {file_info.filename}")
print(f" Purpose: {file_info.purpose}")
print(f" Bytes: {file_info.bytes}")
except Exception as e:
print(f" Failed to list files: {e}")
print(f" File {file_id}: FAILED to retrieve - {e}")
else:
print("No file IDs were captured from the response.")
finally:
await client.close()
# List all files to see if any exist
print("\nListing all files in the agent service:")
try:
files_list = await agents_client.files.list()
count = 0
for file_info in files_list.data:
count += 1
print(f" - {file_info.id}: {file_info.filename} ({file_info.purpose})")
if count == 0:
print(" No files found.")
except Exception as e:
print(f" Failed to list files: {e}")
if __name__ == "__main__":
@@ -3,8 +3,7 @@
import asyncio
import os
from agent_framework import ChatAgent
from agent_framework.azure import AzureAIAgentClient
from agent_framework.azure import AzureAIAgentsProvider
from azure.ai.agents.aio import AgentsClient
from azure.identity.aio import AzureCliCredential
@@ -17,37 +16,29 @@ agent IDs, showing agent reuse patterns for production scenarios.
async def main() -> None:
print("=== Azure AI Chat Client with Existing Agent ===")
print("=== Azure AI Agent with Existing Agent ===")
# Create the client
# Create the client and provider
async with (
AzureCliCredential() as credential,
AgentsClient(endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"], credential=credential) as agents_client,
AzureAIAgentsProvider(agents_client=agents_client) as provider,
):
# Create an agent on the service with default instructions
# These instructions will persist on created agent for every run.
azure_ai_agent = await agents_client.create_agent(
model=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
# Create remote agent with default instructions
# These instructions will persist on created agent for every run.
instructions="End each response with [END].",
)
chat_client = AzureAIAgentClient(agents_client=agents_client, agent_id=azure_ai_agent.id)
try:
async with ChatAgent(
chat_client=chat_client,
# Instructions here are applicable only to this ChatAgent instance
# These instructions will be combined with instructions on existing remote agent.
# The final instructions during the execution will look like:
# "'End each response with [END]. Respond with 'Hello World' only'"
instructions="Respond with 'Hello World' only",
) as agent:
query = "How are you?"
print(f"User: {query}")
result = await agent.run(query)
# Based on local and remote instructions, the result will be
# 'Hello World [END]'.
print(f"Agent: {result}\n")
# Wrap existing agent instance using provider.as_agent()
agent = provider.as_agent(azure_ai_agent)
query = "How are you?"
print(f"User: {query}")
result = await agent.run(query)
print(f"Agent: {result}\n")
finally:
# Clean up the agent manually
await agents_client.delete_agent(azure_ai_agent.id)
@@ -5,8 +5,7 @@ import os
from random import randint
from typing import Annotated
from agent_framework import ChatAgent
from agent_framework.azure import AzureAIAgentClient
from agent_framework.azure import AzureAIAgentsProvider
from azure.ai.agents.aio import AgentsClient
from azure.identity.aio import AzureCliCredential
from pydantic import Field
@@ -28,28 +27,29 @@ def get_weather(
async def main() -> None:
print("=== Azure AI Chat Client with Existing Thread ===")
print("=== Azure AI Agent with Existing Thread ===")
# Create the client
# Create the client and provider
async with (
AzureCliCredential() as credential,
AgentsClient(endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"], credential=credential) as agents_client,
AzureAIAgentsProvider(agents_client=agents_client) as provider,
):
# Create an thread that will persist
# Create a thread that will persist
created_thread = await agents_client.threads.create()
try:
async with ChatAgent(
# passing in the client is optional here, so if you take the agent_id from the portal
# you can use it directly without the two lines above.
chat_client=AzureAIAgentClient(agents_client=agents_client),
# Create agent using provider
agent = await provider.create_agent(
name="WeatherAgent",
instructions="You are a helpful weather agent.",
tools=get_weather,
) as agent:
thread = agent.get_new_thread(service_thread_id=created_thread.id)
assert thread.is_initialized
result = await agent.run("What's the weather like in Tokyo?", thread=thread)
print(f"Result: {result}\n")
)
thread = agent.get_new_thread(service_thread_id=created_thread.id)
assert thread.is_initialized
result = await agent.run("What's the weather like in Tokyo?", thread=thread)
print(f"Result: {result}\n")
finally:
# Clean up the thread manually
await agents_client.threads.delete(created_thread.id)
@@ -5,8 +5,7 @@ import os
from random import randint
from typing import Annotated
from agent_framework import ChatAgent
from agent_framework.azure import AzureAIAgentClient
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
from pydantic import Field
@@ -27,26 +26,23 @@ def get_weather(
async def main() -> None:
print("=== Azure AI Chat Client with Explicit Settings ===")
print("=== Azure AI Agent with Explicit Settings ===")
# Since no Agent ID is provided, the agent will be automatically created
# and deleted after getting a response
# For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
# authentication option.
async with (
AzureCliCredential() as credential,
ChatAgent(
chat_client=AzureAIAgentClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
model_deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=credential,
agent_name="WeatherAgent",
should_cleanup_agent=True, # Set to False if you want to disable automatic agent cleanup
),
AzureAIAgentsProvider(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
credential=credential,
) as provider,
):
agent = await provider.create_agent(
name="WeatherAgent",
model=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
instructions="You are a helpful weather agent.",
tools=get_weather,
) as agent,
):
)
result = await agent.run("What's the weather like in New York?")
print(f"Result: {result}\n")
@@ -1,10 +1,12 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from pathlib import Path
from agent_framework import ChatAgent, HostedFileSearchTool, HostedVectorStoreContent
from agent_framework.azure import AzureAIAgentClient
from agent_framework import HostedFileSearchTool, HostedVectorStoreContent
from agent_framework.azure import AzureAIAgentsProvider
from azure.ai.agents.aio import AgentsClient
from azure.ai.agents.models import FileInfo, VectorStore
from azure.identity.aio import AzureCliCredential
@@ -24,67 +26,54 @@ USER_INPUTS = [
async def main() -> None:
"""Main function demonstrating Azure AI agent with file search capabilities."""
client = AzureAIAgentClient(credential=AzureCliCredential())
file: FileInfo | None = None
vector_store: VectorStore | None = None
try:
# 1. Upload file and create vector store
pdf_file_path = Path(__file__).parent.parent / "resources" / "employees.pdf"
print(f"Uploading file from: {pdf_file_path}")
async with (
AzureCliCredential() as credential,
AgentsClient(endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"], credential=credential) as agents_client,
AzureAIAgentsProvider(agents_client=agents_client) as provider,
):
try:
# 1. Upload file and create vector store
pdf_file_path = Path(__file__).parent.parent / "resources" / "employees.pdf"
print(f"Uploading file from: {pdf_file_path}")
file = await client.agents_client.files.upload_and_poll(file_path=str(pdf_file_path), purpose="assistants")
print(f"Uploaded file, file ID: {file.id}")
file = await agents_client.files.upload_and_poll(file_path=str(pdf_file_path), purpose="assistants")
print(f"Uploaded file, file ID: {file.id}")
vector_store = await client.agents_client.vector_stores.create_and_poll(
file_ids=[file.id], name="my_vectorstore"
)
print(f"Created vector store, vector store ID: {vector_store.id}")
vector_store = await agents_client.vector_stores.create_and_poll(file_ids=[file.id], name="my_vectorstore")
print(f"Created vector store, vector store ID: {vector_store.id}")
# 2. Create file search tool with uploaded resources
file_search_tool = HostedFileSearchTool(inputs=[HostedVectorStoreContent(vector_store_id=vector_store.id)])
# 2. Create file search tool with uploaded resources
file_search_tool = HostedFileSearchTool(inputs=[HostedVectorStoreContent(vector_store_id=vector_store.id)])
# 3. Create an agent with file search capabilities
agent = await provider.create_agent(
name="EmployeeSearchAgent",
instructions=(
"You are a helpful assistant that can search through uploaded employee files "
"to answer questions about employees."
),
tools=file_search_tool,
)
# 3. Create an agent with file search capabilities
# The tool_resources are automatically extracted from HostedFileSearchTool
async with ChatAgent(
chat_client=client,
name="EmployeeSearchAgent",
instructions=(
"You are a helpful assistant that can search through uploaded employee files "
"to answer questions about employees."
),
tools=file_search_tool,
) as agent:
# 4. Simulate conversation with the agent
for user_input in USER_INPUTS:
print(f"# User: '{user_input}'")
response = await agent.run(user_input)
print(f"# Agent: {response.text}")
finally:
# 5. Cleanup: Delete the vector store and file
try:
if vector_store:
await client.agents_client.vector_stores.delete(vector_store.id)
await agents_client.vector_stores.delete(vector_store.id)
if file:
await client.agents_client.files.delete(file.id)
await agents_client.files.delete(file.id)
except Exception:
# Ignore cleanup errors to avoid masking issues
pass
finally:
# 6. Cleanup: Delete the vector store and file in case of earlier failure to prevent orphaned resources.
# Refreshing the client is required since chat agent closes it
client = AzureAIAgentClient(credential=AzureCliCredential())
try:
if vector_store:
await client.agents_client.vector_stores.delete(vector_store.id)
if file:
await client.agents_client.files.delete(file.id)
except Exception:
# Ignore cleanup errors to avoid masking issues
pass
finally:
await client.close()
if __name__ == "__main__":
@@ -5,8 +5,7 @@ from datetime import datetime, timezone
from random import randint
from typing import Annotated
from agent_framework import ChatAgent
from agent_framework.azure import AzureAIAgentClient
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
from pydantic import Field
@@ -42,12 +41,14 @@ async def tools_on_agent_level() -> None:
# authentication option.
async with (
AzureCliCredential() as credential,
ChatAgent(
chat_client=AzureAIAgentClient(credential=credential),
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="AssistantAgent",
instructions="You are a helpful assistant that can provide weather and time information.",
tools=[get_weather, get_time], # Tools defined at agent creation
) as agent,
):
)
# First query - agent can use weather tool
query1 = "What's the weather like in New York?"
print(f"User: {query1}")
@@ -76,12 +77,14 @@ async def tools_on_run_level() -> None:
# authentication option.
async with (
AzureCliCredential() as credential,
ChatAgent(
chat_client=AzureAIAgentClient(credential=credential),
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="AssistantAgent",
instructions="You are a helpful assistant.",
# No tools defined here
) as agent,
):
)
# First query with weather tool
query1 = "What's the weather like in Seattle?"
print(f"User: {query1}")
@@ -110,12 +113,14 @@ async def mixed_tools_example() -> None:
# authentication option.
async with (
AzureCliCredential() as credential,
ChatAgent(
chat_client=AzureAIAgentClient(credential=credential),
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="AssistantAgent",
instructions="You are a comprehensive assistant that can help with various information requests.",
tools=[get_weather], # Base tool available for all queries
) as agent,
):
)
# Query using both agent tool and additional run-method tools
query = "What's the weather in Denver and what's the current UTC time?"
print(f"User: {query}")
@@ -4,7 +4,7 @@ import asyncio
from typing import Any
from agent_framework import AgentProtocol, AgentResponse, AgentThread, HostedMCPTool
from agent_framework.azure import AzureAIAgentClient
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
"""
@@ -42,9 +42,9 @@ async def main() -> None:
"""Example showing Hosted MCP tools for a Azure AI Agent."""
async with (
AzureCliCredential() as credential,
AzureAIAgentClient(credential=credential) as chat_client,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = chat_client.create_agent(
agent = await provider.create_agent(
name="DocsAgent",
instructions="You are a helpful assistant that can help with microsoft documentation questions.",
tools=HostedMCPTool(
@@ -2,8 +2,8 @@
import asyncio
from agent_framework import ChatAgent, MCPStreamableHTTPTool
from agent_framework.azure import AzureAIAgentClient
from agent_framework import MCPStreamableHTTPTool
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
"""
@@ -27,12 +27,12 @@ async def mcp_tools_on_run_level() -> None:
name="Microsoft Learn MCP",
url="https://learn.microsoft.com/api/mcp",
) as mcp_server,
ChatAgent(
chat_client=AzureAIAgentClient(credential=credential),
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="DocsAgent",
instructions="You are a helpful assistant that can help with microsoft documentation questions.",
) as agent,
):
)
# First query
query1 = "How to create an Azure storage account using az cli?"
print(f"User: {query1}")
@@ -47,34 +47,37 @@ async def mcp_tools_on_run_level() -> None:
async def mcp_tools_on_agent_level() -> None:
"""Example showing tools defined when creating the agent."""
"""Example showing local MCP tools passed when creating the agent."""
print("=== Tools Defined on Agent Level ===")
# Tools are provided when creating the agent
# The agent can use these tools for any query during its lifetime
# The agent will connect to the MCP server through its context manager.
# The ChatAgent will connect to the MCP server through its context manager
# and discover tools at runtime
async with (
AzureCliCredential() as credential,
AzureAIAgentClient(credential=credential).create_agent(
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="DocsAgent",
instructions="You are a helpful assistant that can help with microsoft documentation questions.",
tools=MCPStreamableHTTPTool( # Tools defined at agent creation
tools=MCPStreamableHTTPTool(
name="Microsoft Learn MCP",
url="https://learn.microsoft.com/api/mcp",
),
) as agent,
):
# First query
query1 = "How to create an Azure storage account using az cli?"
print(f"User: {query1}")
result1 = await agent.run(query1)
print(f"{agent.name}: {result1}\n")
print("\n=======================================\n")
# Second query
query2 = "What is Microsoft Agent Framework?"
print(f"User: {query2}")
result2 = await agent.run(query2)
print(f"{agent.name}: {result2}\n")
)
# Use agent as context manager to connect MCP tools
async with agent:
# First query
query1 = "How to create an Azure storage account using az cli?"
print(f"User: {query1}")
result1 = await agent.run(query1)
print(f"{agent.name}: {result1}\n")
print("\n=======================================\n")
# Second query
query2 = "What is Microsoft Agent Framework?"
print(f"User: {query2}")
result2 = await agent.run(query2)
print(f"{agent.name}: {result2}\n")
async def main() -> None:
@@ -10,7 +10,7 @@ from agent_framework import (
HostedMCPTool,
HostedWebSearchTool,
)
from agent_framework.azure import AzureAIAgentClient
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
"""
@@ -67,9 +67,9 @@ async def main() -> None:
"""Example showing Hosted MCP tools for a Azure AI Agent."""
async with (
AzureCliCredential() as credential,
AzureAIAgentClient(credential=credential) as chat_client,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = chat_client.create_agent(
agent = await provider.create_agent(
name="DocsAgent",
instructions="You are a helpful assistant that can help with microsoft documentation questions.",
tools=[
@@ -5,8 +5,7 @@ import json
from pathlib import Path
from typing import Any
from agent_framework import ChatAgent
from agent_framework_azure_ai import AzureAIAgentClient
from agent_framework.azure import AzureAIAgentsProvider
from azure.ai.agents.models import OpenApiAnonymousAuthDetails, OpenApiTool
from azure.identity.aio import AzureCliCredential
@@ -40,8 +39,11 @@ async def main() -> None:
# 1. Load OpenAPI specifications (synchronous operation)
weather_openapi_spec, countries_openapi_spec = load_openapi_specs()
# 2. Use AzureAIAgentClient as async context manager for automatic cleanup
async with AzureAIAgentClient(credential=AzureCliCredential()) as client:
# 2. Use AzureAIAgentsProvider for agent creation and management
async with (
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
# 3. Create OpenAPI tools using Azure AI's OpenApiTool
auth = OpenApiAnonymousAuthDetails()
@@ -62,8 +64,7 @@ async def main() -> None:
# 4. Create an agent with OpenAPI tools
# Note: We need to pass the Azure AI native OpenApiTool definitions directly
# since the agent framework doesn't have a HostedOpenApiTool wrapper yet
async with ChatAgent(
chat_client=client,
agent = await provider.create_agent(
name="OpenAPIAgent",
instructions=(
"You are a helpful assistant that can search for country information "
@@ -73,18 +74,19 @@ async def main() -> None:
),
# Pass the raw tool definitions from Azure AI's OpenApiTool
tools=[*openapi_countries.definitions, *openapi_weather.definitions],
) as agent:
# 5. Simulate conversation with the agent maintaining thread context
print("=== Azure AI Agent with OpenAPI Tools ===\n")
)
# Create a thread to maintain conversation context across multiple runs
thread = agent.get_new_thread()
# 5. Simulate conversation with the agent maintaining thread context
print("=== Azure AI Agent with OpenAPI Tools ===\n")
for user_input in USER_INPUTS:
print(f"User: {user_input}")
# Pass the thread to maintain context across multiple agent.run() calls
response = await agent.run(user_input, thread=thread)
print(f"Agent: {response.text}\n")
# Create a thread to maintain conversation context across multiple runs
thread = agent.get_new_thread()
for user_input in USER_INPUTS:
print(f"User: {user_input}")
# Pass the thread to maintain context across multiple agent.run() calls
response = await agent.run(user_input, thread=thread)
print(f"Agent: {response.text}\n")
if __name__ == "__main__":
@@ -0,0 +1,55 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
from pydantic import BaseModel, ConfigDict
"""
Azure AI Agent Provider Response Format Example
This sample demonstrates using AzureAIAgentsProvider with default_options
containing response_format for structured outputs.
"""
class WeatherInfo(BaseModel):
"""Structured weather information."""
location: str
temperature: int
conditions: str
recommendation: str
model_config = ConfigDict(extra="forbid")
async def main() -> None:
"""Example of using default_options with response_format in AzureAIAgentsProvider."""
async with (
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="WeatherReporter",
instructions="You provide weather reports in structured JSON format.",
default_options={"response_format": WeatherInfo},
)
query = "What's the weather like in Paris today?"
print(f"User: {query}")
result = await agent.run(query)
if isinstance(result.value, WeatherInfo):
weather = result.value
print("Agent:")
print(f"Location: {weather.location}")
print(f"Temperature: {weather.temperature}")
print(f"Conditions: {weather.conditions}")
print(f"Recommendation: {weather.recommendation}")
if __name__ == "__main__":
asyncio.run(main())
@@ -4,8 +4,8 @@ import asyncio
from random import randint
from typing import Annotated
from agent_framework import AgentThread, ChatAgent
from agent_framework.azure import AzureAIAgentClient
from agent_framework import AgentThread
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
from pydantic import Field
@@ -33,12 +33,14 @@ async def example_with_automatic_thread_creation() -> None:
# authentication option.
async with (
AzureCliCredential() as credential,
ChatAgent(
chat_client=AzureAIAgentClient(credential=credential),
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="WeatherAgent",
instructions="You are a helpful weather agent.",
tools=get_weather,
) as agent,
):
)
# First conversation - no thread provided, will be created automatically
first_query = "What's the weather like in Seattle?"
print(f"User: {first_query}")
@@ -62,12 +64,14 @@ async def example_with_thread_persistence() -> None:
# authentication option.
async with (
AzureCliCredential() as credential,
ChatAgent(
chat_client=AzureAIAgentClient(credential=credential),
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="WeatherAgent",
instructions="You are a helpful weather agent.",
tools=get_weather,
) as agent,
):
)
# Create a new thread that will be reused
thread = agent.get_new_thread()
@@ -103,12 +107,14 @@ async def example_with_existing_thread_id() -> None:
# authentication option.
async with (
AzureCliCredential() as credential,
ChatAgent(
chat_client=AzureAIAgentClient(credential=credential),
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="WeatherAgent",
instructions="You are a helpful weather agent.",
tools=get_weather,
) as agent,
):
)
# Start a conversation and get the thread ID
thread = agent.get_new_thread()
first_query = "What's the weather in Paris?"
@@ -123,15 +129,17 @@ async def example_with_existing_thread_id() -> None:
if existing_thread_id:
print("\n--- Continuing with the same thread ID in a new agent instance ---")
# Create a new agent instance but use the existing thread ID
# Create a new provider and agent but use the existing thread ID
async with (
AzureCliCredential() as credential,
ChatAgent(
chat_client=AzureAIAgentClient(thread_id=existing_thread_id, credential=credential),
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="WeatherAgent",
instructions="You are a helpful weather agent.",
tools=get_weather,
) as agent,
):
)
# Create a thread with the existing ID
thread = AgentThread(service_thread_id=existing_thread_id)