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

Python: [BREAKING] Remove deprecated Python OpenAI/Azure AI surfaces (#4990)

Browse Source 
* [BREAKING] Remove deprecated Python OpenAI/Azure AI surfaces

Also clean up follow-on docs, environment guidance, package metadata, and lab test stability.

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

* Fix deleted semantic-kernel sample links

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

* Address PR review feedback

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

* improve foundry language

* Fix A2A Foundry sample regression

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

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Eduard van Valkenburg
2026-03-31 20:36:21 +00:00
committed by GitHub
co-authored by Copilot
parent a5eacbbe65
commit 3a49b1d6dd
144 changed files with 669 additions and 18739 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/packages/azure-ai/tests/azure_openai/conftest.py
-61
View File
@@ -1,61 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
from typing import Any
from agent_framework import Message
from pytest import fixture
# region: Connector Settings fixtures
@fixture
def exclude_list(request: Any) -> list[str]:
"""Fixture that returns a list of environment variables to exclude."""
return request.param if hasattr(request, "param") else []
@fixture
def override_env_param_dict(request: Any) -> dict[str, str]:
"""Fixture that returns a dict of environment variables to override."""
return request.param if hasattr(request, "param") else {}
# These two fixtures are used for multiple things, also non-connector tests
@fixture()
def azure_openai_unit_test_env(monkeypatch, exclude_list, override_env_param_dict): # type: ignore
"""Fixture to set environment variables for AzureOpenAISettings."""
if exclude_list is None:
exclude_list = []
if override_env_param_dict is None:
override_env_param_dict = {}
env_vars = {
"AZURE_OPENAI_ENDPOINT": "https://test-endpoint.com",
"AZURE_OPENAI_CHAT_DEPLOYMENT_NAME": "test_chat_deployment",
"AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME": "test_chat_deployment",
"AZURE_OPENAI_TEXT_DEPLOYMENT_NAME": "test_text_deployment",
"AZURE_OPENAI_EMBEDDING_DEPLOYMENT_NAME": "test_embedding_deployment",
"AZURE_OPENAI_TEXT_TO_IMAGE_DEPLOYMENT_NAME": "test_text_to_image_deployment",
"AZURE_OPENAI_AUDIO_TO_TEXT_DEPLOYMENT_NAME": "test_audio_to_text_deployment",
"AZURE_OPENAI_TEXT_TO_AUDIO_DEPLOYMENT_NAME": "test_text_to_audio_deployment",
"AZURE_OPENAI_REALTIME_DEPLOYMENT_NAME": "test_realtime_deployment",
"AZURE_OPENAI_API_KEY": "test_api_key",
"AZURE_OPENAI_API_VERSION": "2023-03-15-preview",
"AZURE_OPENAI_BASE_URL": "https://test_text_deployment.test-base-url.com",
"AZURE_OPENAI_TOKEN_ENDPOINT": "https://test-token-endpoint.com",
}
env_vars.update(override_env_param_dict) # type: ignore
for key, value in env_vars.items():
if key in exclude_list:
monkeypatch.delenv(key, raising=False) # type: ignore
continue
monkeypatch.setenv(key, value) # type: ignore
return env_vars
@fixture(scope="function")
def chat_history() -> list[Message]:
return []
python/packages/azure-ai/tests/azure_openai/test_azure_assistants_client.py
-409
View File
@@ -1,409 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
from typing import Annotated
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from agent_framework import (
SupportsChatGetResponse,
tool,
)
from agent_framework._settings import SecretString
from agent_framework.azure import AzureOpenAIAssistantsClient
from pydantic import Field
def create_test_azure_assistants_client(
mock_async_azure_openai: MagicMock,
deployment_name: str | None = None,
assistant_id: str | None = None,
assistant_name: str | None = None,
thread_id: str | None = None,
should_delete_assistant: bool = False,
) -> AzureOpenAIAssistantsClient:
"""Helper function to create AzureOpenAIAssistantsClient instances for testing."""
client = AzureOpenAIAssistantsClient(
deployment_name=deployment_name or "test_chat_deployment",
assistant_id=assistant_id,
assistant_name=assistant_name,
thread_id=thread_id,
api_key="test-api-key",
endpoint="https://test-endpoint.com",
async_client=mock_async_azure_openai,
)
# Set the _should_delete_assistant flag directly if needed
if should_delete_assistant:
object.__setattr__(client, "_should_delete_assistant", True)
return client
@pytest.fixture
def mock_async_azure_openai() -> MagicMock:
"""Mock AsyncAzureOpenAI client."""
mock_client = MagicMock()
# Mock beta.assistants
mock_client.beta.assistants.create = AsyncMock(return_value=MagicMock(id="test-assistant-id"))
mock_client.beta.assistants.delete = AsyncMock()
# Mock beta.threads
mock_client.beta.threads.create = AsyncMock(return_value=MagicMock(id="test-thread-id"))
mock_client.beta.threads.delete = AsyncMock()
# Mock beta.threads.runs
mock_client.beta.threads.runs.create = AsyncMock(return_value=MagicMock(id="test-run-id"))
mock_client.beta.threads.runs.retrieve = AsyncMock()
mock_client.beta.threads.runs.submit_tool_outputs = AsyncMock()
# Mock beta.threads.messages
mock_client.beta.threads.messages.create = AsyncMock()
mock_client.beta.threads.messages.list = AsyncMock(return_value=MagicMock(data=[]))
return mock_client
def test_azure_assistants_client_init_with_client(mock_async_azure_openai: MagicMock) -> None:
"""Test AzureOpenAIAssistantsClient initialization with existing client."""
client = create_test_azure_assistants_client(
mock_async_azure_openai,
deployment_name="test_chat_deployment",
assistant_id="existing-assistant-id",
thread_id="test-thread-id",
)
assert client.client is mock_async_azure_openai
assert client.model == "test_chat_deployment"
assert client.assistant_id == "existing-assistant-id"
assert client.thread_id == "test-thread-id"
assert not client._should_delete_assistant # type: ignore
assert isinstance(client, SupportsChatGetResponse)
def test_azure_assistants_client_init_auto_create_client(
azure_openai_unit_test_env: dict[str, str],
mock_async_azure_openai: MagicMock,
) -> None:
"""Test AzureOpenAIAssistantsClient initialization with auto-created client."""
client = AzureOpenAIAssistantsClient(
deployment_name=azure_openai_unit_test_env["AZURE_OPENAI_CHAT_DEPLOYMENT_NAME"],
assistant_name="TestAssistant",
api_key=azure_openai_unit_test_env["AZURE_OPENAI_API_KEY"],
endpoint=azure_openai_unit_test_env["AZURE_OPENAI_ENDPOINT"],
async_client=mock_async_azure_openai,
)
assert client.client is mock_async_azure_openai
assert client.model == azure_openai_unit_test_env["AZURE_OPENAI_CHAT_DEPLOYMENT_NAME"]
assert client.assistant_id is None
assert client.assistant_name == "TestAssistant"
assert not client._should_delete_assistant # type: ignore
def test_azure_assistants_client_init_validation_fail() -> None:
"""Test AzureOpenAIAssistantsClient initialization with validation failure."""
with pytest.raises(ValueError):
# Force failure by providing invalid deployment name type - this should cause validation to fail
AzureOpenAIAssistantsClient(deployment_name=123, api_key="valid-key") # type: ignore
@pytest.mark.parametrize("exclude_list", [["AZURE_OPENAI_CHAT_DEPLOYMENT_NAME"]], indirect=True)
def test_azure_assistants_client_init_missing_deployment_name(azure_openai_unit_test_env: dict[str, str]) -> None:
"""Test AzureOpenAIAssistantsClient initialization with missing deployment name."""
with pytest.raises(ValueError):
AzureOpenAIAssistantsClient(api_key=azure_openai_unit_test_env.get("AZURE_OPENAI_API_KEY", "test-key"))
def test_azure_assistants_client_init_with_default_headers(azure_openai_unit_test_env: dict[str, str]) -> None:
"""Test AzureOpenAIAssistantsClient initialization with default headers."""
default_headers = {"X-Unit-Test": "test-guid"}
client = AzureOpenAIAssistantsClient(
deployment_name="test_chat_deployment",
api_key=azure_openai_unit_test_env["AZURE_OPENAI_API_KEY"],
endpoint=azure_openai_unit_test_env["AZURE_OPENAI_ENDPOINT"],
default_headers=default_headers,
)
assert client.model == "test_chat_deployment"
assert isinstance(client, SupportsChatGetResponse)
# Assert that the default header we added is present in the client's default headers
for key, value in default_headers.items():
assert key in client.client.default_headers
assert client.client.default_headers[key] == value
async def test_azure_assistants_client_get_assistant_id_or_create_existing_assistant(
mock_async_azure_openai: MagicMock,
) -> None:
"""Test _get_assistant_id_or_create when assistant_id is already provided."""
client = create_test_azure_assistants_client(mock_async_azure_openai, assistant_id="existing-assistant-id")
assistant_id = await client._get_assistant_id_or_create() # type: ignore
assert assistant_id == "existing-assistant-id"
assert not client._should_delete_assistant # type: ignore
mock_async_azure_openai.beta.assistants.create.assert_not_called()
async def test_azure_assistants_client_get_assistant_id_or_create_create_new(
mock_async_azure_openai: MagicMock,
) -> None:
"""Test _get_assistant_id_or_create when creating a new assistant."""
client = create_test_azure_assistants_client(
mock_async_azure_openai, deployment_name="test_chat_deployment", assistant_name="TestAssistant"
)
assistant_id = await client._get_assistant_id_or_create() # type: ignore
assert assistant_id == "test-assistant-id"
assert client._should_delete_assistant # type: ignore
mock_async_azure_openai.beta.assistants.create.assert_called_once()
async def test_azure_assistants_client_aclose_should_not_delete(
mock_async_azure_openai: MagicMock,
) -> None:
"""Test close when assistant should not be deleted."""
client = create_test_azure_assistants_client(
mock_async_azure_openai, assistant_id="assistant-to-keep", should_delete_assistant=False
)
await client.close() # type: ignore
# Verify assistant deletion was not called
mock_async_azure_openai.beta.assistants.delete.assert_not_called()
assert not client._should_delete_assistant # type: ignore
async def test_azure_assistants_client_aclose_should_delete(mock_async_azure_openai: MagicMock) -> None:
"""Test close method calls cleanup."""
client = create_test_azure_assistants_client(
mock_async_azure_openai, assistant_id="assistant-to-delete", should_delete_assistant=True
)
await client.close()
# Verify assistant deletion was called
mock_async_azure_openai.beta.assistants.delete.assert_called_once_with("assistant-to-delete")
assert not client._should_delete_assistant # type: ignore
async def test_azure_assistants_client_async_context_manager(mock_async_azure_openai: MagicMock) -> None:
"""Test async context manager functionality."""
client = create_test_azure_assistants_client(
mock_async_azure_openai, assistant_id="assistant-to-delete", should_delete_assistant=True
)
# Test context manager
async with client:
pass # Just test that we can enter and exit
# Verify cleanup was called on exit
mock_async_azure_openai.beta.assistants.delete.assert_called_once_with("assistant-to-delete")
def test_azure_assistants_client_serialize(azure_openai_unit_test_env: dict[str, str]) -> None:
"""Test serialization of AzureOpenAIAssistantsClient."""
default_headers = {"X-Unit-Test": "test-guid"}
# Test basic initialization and to_dict
client = AzureOpenAIAssistantsClient(
deployment_name="test_chat_deployment",
assistant_id="test-assistant-id",
assistant_name="TestAssistant",
thread_id="test-thread-id",
api_key=azure_openai_unit_test_env["AZURE_OPENAI_API_KEY"],
endpoint=azure_openai_unit_test_env["AZURE_OPENAI_ENDPOINT"],
default_headers=default_headers,
)
dumped_settings = client.to_dict()
assert dumped_settings["model"] == "test_chat_deployment"
assert dumped_settings["assistant_id"] == "test-assistant-id"
assert dumped_settings["assistant_name"] == "TestAssistant"
assert dumped_settings["thread_id"] == "test-thread-id"
# Assert that the default header we added is present in the dumped_settings default headers
for key, value in default_headers.items():
assert key in dumped_settings["default_headers"]
assert dumped_settings["default_headers"][key] == value
# Assert that the 'User-Agent' header is not present in the dumped_settings default headers
assert "User-Agent" not in dumped_settings["default_headers"]
@tool(approval_mode="never_require")
def get_weather(
location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
"""Get the weather for a given location."""
return f"The weather in {location} is sunny with a high of 25°C."
def test_azure_assistants_client_entra_id_authentication() -> None:
"""Test credential authentication path with sync credential."""
mock_credential = MagicMock()
mock_provider = MagicMock(return_value="token-string")
with (
patch("agent_framework_azure_ai._deprecated_azure_openai.load_settings") as mock_load_settings,
patch(
"agent_framework_azure_ai._deprecated_azure_openai.resolve_credential_to_token_provider",
return_value=mock_provider,
) as mock_resolve,
patch("agent_framework_azure_ai._deprecated_azure_openai.AsyncAzureOpenAI") as mock_azure_client,
patch("agent_framework.openai.OpenAIAssistantsClient.__init__", return_value=None),
):
mock_load_settings.return_value = {
"chat_deployment_name": "test-deployment",
"responses_deployment_name": None,
"api_key": None,
"token_endpoint": "https://cognitiveservices.azure.com/.default",
"api_version": "2024-05-01-preview",
"endpoint": "https://test-endpoint.openai.azure.com",
"base_url": None,
}
client = AzureOpenAIAssistantsClient(
deployment_name="test-deployment",
endpoint="https://test-endpoint.openai.azure.com",
credential=mock_credential,
token_endpoint="https://cognitiveservices.azure.com/.default",
)
# Verify credential was resolved to a token provider
mock_resolve.assert_called_once_with(mock_credential, "https://cognitiveservices.azure.com/.default")
# Verify client was created with the token provider
mock_azure_client.assert_called_once()
call_args = mock_azure_client.call_args[1]
assert call_args["azure_ad_token_provider"] is mock_provider
assert client is not None
assert isinstance(client, AzureOpenAIAssistantsClient)
def test_azure_assistants_client_no_authentication_error() -> None:
"""Test authentication validation error when no auth provided."""
with patch("agent_framework_azure_ai._deprecated_azure_openai.load_settings") as mock_load_settings:
mock_load_settings.return_value = {
"chat_deployment_name": "test-deployment",
"responses_deployment_name": None,
"api_key": None,
"token_endpoint": None,
"api_version": "2024-05-01-preview",
"endpoint": "https://test-endpoint.openai.azure.com",
"base_url": None,
}
# Test missing authentication raises error
with pytest.raises(ValueError, match="api_key, credential, or a client"):
AzureOpenAIAssistantsClient(
deployment_name="test-deployment",
endpoint="https://test-endpoint.openai.azure.com",
# No authentication provided at all
)
def test_azure_assistants_client_callable_credential() -> None:
"""Test callable token provider as credential."""
mock_provider = MagicMock(return_value="my-token")
with (
patch("agent_framework_azure_ai._deprecated_azure_openai.load_settings") as mock_load_settings,
patch(
"agent_framework_azure_ai._deprecated_azure_openai.resolve_credential_to_token_provider",
return_value=mock_provider,
),
patch("agent_framework_azure_ai._deprecated_azure_openai.AsyncAzureOpenAI") as mock_azure_client,
patch("agent_framework.openai.OpenAIAssistantsClient.__init__", return_value=None),
):
mock_load_settings.return_value = {
"chat_deployment_name": "test-deployment",
"responses_deployment_name": None,
"api_key": None,
"token_endpoint": "https://cognitiveservices.azure.com/.default",
"api_version": "2024-05-01-preview",
"endpoint": "https://test-endpoint.openai.azure.com",
"base_url": None,
}
client = AzureOpenAIAssistantsClient(
deployment_name="test-deployment",
endpoint="https://test-endpoint.openai.azure.com",
credential=mock_provider,
token_endpoint="https://cognitiveservices.azure.com/.default",
)
# Verify client was created with the token provider
mock_azure_client.assert_called_once()
call_args = mock_azure_client.call_args[1]
assert call_args["azure_ad_token_provider"] is mock_provider
assert client is not None
assert isinstance(client, AzureOpenAIAssistantsClient)
def test_azure_assistants_client_base_url_configuration() -> None:
"""Test base_url client parameter path."""
with (
patch("agent_framework_azure_ai._deprecated_azure_openai.load_settings") as mock_load_settings,
patch("agent_framework_azure_ai._deprecated_azure_openai.AsyncAzureOpenAI") as mock_azure_client,
patch("agent_framework.openai.OpenAIAssistantsClient.__init__", return_value=None),
):
mock_load_settings.return_value = {
"chat_deployment_name": "test-deployment",
"responses_deployment_name": None,
"api_key": SecretString("test-api-key"),
"token_endpoint": None,
"api_version": "2024-05-01-preview",
"endpoint": None,
"base_url": "https://custom-base-url.com",
}
client = AzureOpenAIAssistantsClient(
deployment_name="test-deployment", api_key="test-api-key", base_url="https://custom-base-url.com"
)
# base_url path
mock_azure_client.assert_called_once()
call_args = mock_azure_client.call_args[1]
assert call_args["base_url"] == "https://custom-base-url.com"
assert "azure_endpoint" not in call_args
assert client is not None
assert isinstance(client, AzureOpenAIAssistantsClient)
def test_azure_assistants_client_azure_endpoint_configuration() -> None:
"""Test azure_endpoint client parameter path."""
with (
patch("agent_framework_azure_ai._deprecated_azure_openai.load_settings") as mock_load_settings,
patch("agent_framework_azure_ai._deprecated_azure_openai.AsyncAzureOpenAI") as mock_azure_client,
patch("agent_framework.openai.OpenAIAssistantsClient.__init__", return_value=None),
):
mock_load_settings.return_value = {
"chat_deployment_name": "test-deployment",
"responses_deployment_name": None,
"api_key": SecretString("test-api-key"),
"token_endpoint": None,
"api_version": "2024-05-01-preview",
"endpoint": "https://test-endpoint.openai.azure.com",
"base_url": None,
}
client = AzureOpenAIAssistantsClient(
deployment_name="test-deployment",
api_key="test-api-key",
endpoint="https://test-endpoint.openai.azure.com",
)
# azure_endpoint path
mock_azure_client.assert_called_once()
call_args = mock_azure_client.call_args[1]
assert call_args["azure_endpoint"] == "https://test-endpoint.openai.azure.com"
assert "base_url" not in call_args
assert client is not None
assert isinstance(client, AzureOpenAIAssistantsClient)
python/packages/azure-ai/tests/azure_openai/test_azure_chat_client.py
-1102
View File
File diff suppressed because it is too large Load Diff
python/packages/azure-ai/tests/azure_openai/test_azure_embedding_client.py
-219
View File
@@ -1,219 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
from __future__ import annotations
import os
from functools import wraps
from typing import Any
from unittest.mock import AsyncMock, MagicMock
import pytest
from agent_framework.azure import AzureOpenAIEmbeddingClient
from agent_framework.openai import OpenAIEmbeddingOptions
from azure.identity.aio import AzureCliCredential
from openai.types import CreateEmbeddingResponse
from openai.types import Embedding as OpenAIEmbedding
from openai.types.create_embedding_response import Usage
pytestmark = pytest.mark.filterwarnings("ignore:AzureOpenAIEmbeddingClient is deprecated\\..*:DeprecationWarning")
def _make_openai_response(
embeddings: list[list[float]],
model: str = "text-embedding-3-small",
prompt_tokens: int = 5,
total_tokens: int = 5,
) -> CreateEmbeddingResponse:
"""Helper to create a mock OpenAI embeddings response."""
data = [OpenAIEmbedding(embedding=emb, index=i, object="embedding") for i, emb in enumerate(embeddings)]
return CreateEmbeddingResponse(
data=data,
model=model,
object="list",
usage=Usage(prompt_tokens=prompt_tokens, total_tokens=total_tokens),
)
@pytest.fixture
def azure_embedding_unit_test_env(monkeypatch: pytest.MonkeyPatch) -> None:
"""Clear ambient Azure OpenAI embedding env vars for deterministic unit tests."""
for key in (
"AZURE_OPENAI_ENDPOINT",
"AZURE_OPENAI_API_KEY",
"AZURE_OPENAI_EMBEDDING_DEPLOYMENT_NAME",
"AZURE_OPENAI_BASE_URL",
"AZURE_OPENAI_TOKEN_ENDPOINT",
):
monkeypatch.delenv(key, raising=False)
def test_azure_construction_with_deployment_name(azure_embedding_unit_test_env: None) -> None:
client = AzureOpenAIEmbeddingClient(
deployment_name="text-embedding-3-small",
api_key="test-key",
endpoint="https://test.openai.azure.com/",
)
assert client.model == "text-embedding-3-small"
def test_azure_construction_with_existing_client(azure_embedding_unit_test_env: None) -> None:
mock_client = MagicMock()
client = AzureOpenAIEmbeddingClient(
deployment_name="my-deployment",
async_client=mock_client,
)
assert client.model == "my-deployment"
assert client.client is mock_client
def test_azure_construction_missing_deployment_name_raises(azure_embedding_unit_test_env: None) -> None:
with pytest.raises(ValueError, match="deployment name is required"):
AzureOpenAIEmbeddingClient(
api_key="test-key",
endpoint="https://test.openai.azure.com/",
)
def test_azure_construction_missing_credentials_raises(azure_embedding_unit_test_env: None) -> None:
with pytest.raises(ValueError, match="api_key, credential, or a client"):
AzureOpenAIEmbeddingClient(
deployment_name="test",
endpoint="https://test.openai.azure.com/",
)
async def test_azure_get_embeddings(azure_embedding_unit_test_env: None) -> None:
mock_response = _make_openai_response(
embeddings=[[0.1, 0.2]],
)
mock_async_client = MagicMock()
mock_async_client.embeddings = MagicMock()
mock_async_client.embeddings.create = AsyncMock(return_value=mock_response)
client = AzureOpenAIEmbeddingClient(
deployment_name="text-embedding-3-small",
async_client=mock_async_client,
)
result = await client.get_embeddings(["hello"])
assert len(result) == 1
assert result[0].vector == [0.1, 0.2]
def test_azure_otel_provider_name(azure_embedding_unit_test_env: None) -> None:
mock_client = MagicMock()
client = AzureOpenAIEmbeddingClient(
deployment_name="test",
async_client=mock_client,
)
assert client.OTEL_PROVIDER_NAME == "azure.ai.openai"
skip_if_azure_openai_integration_tests_disabled = pytest.mark.skipif(
os.getenv("AZURE_OPENAI_ENDPOINT", "") in ("", "https://test-endpoint.com")
or (
os.getenv("AZURE_OPENAI_EMBEDDING_DEPLOYMENT_NAME", "") == ""
and os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME", "") == ""
),
reason="No Azure OpenAI endpoint or embedding deployment provided; skipping integration tests.",
)
def _with_azure_openai_debug() -> Any:
def decorator(func: Any) -> Any:
@wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> Any:
try:
return await func(*args, **kwargs)
except Exception as exc:
model = os.getenv("AZURE_OPENAI_EMBEDDING_DEPLOYMENT_NAME") or os.getenv(
"AZURE_OPENAI_DEPLOYMENT_NAME", "<unset>"
)
api_version = os.getenv("AZURE_OPENAI_API_VERSION", "<unset>")
endpoint = os.getenv("AZURE_OPENAI_ENDPOINT", "<unset>")
debug_message = f"Azure OpenAI debug: endpoint={endpoint}, model={model}, api_version={api_version}"
if hasattr(exc, "add_note"):
exc.add_note(debug_message)
elif exc.args:
exc.args = (f"{exc.args[0]}\n{debug_message}", *exc.args[1:])
else:
exc.args = (debug_message,)
raise
return wrapper
return decorator
def _get_azure_embedding_deployment_name() -> str:
return os.getenv("AZURE_OPENAI_EMBEDDING_DEPLOYMENT_NAME") or os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"]
def _create_azure_openai_embedding_client(
*,
api_key: str | None = None,
credential: AzureCliCredential | None = None,
) -> AzureOpenAIEmbeddingClient:
resolved_api_key = (
api_key if api_key is not None else None if credential is not None else os.getenv("AZURE_OPENAI_API_KEY")
)
return AzureOpenAIEmbeddingClient(
deployment_name=_get_azure_embedding_deployment_name(),
api_key=resolved_api_key,
endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
credential=credential,
)
@pytest.mark.flaky
@pytest.mark.integration
@skip_if_azure_openai_integration_tests_disabled
@_with_azure_openai_debug()
async def test_integration_azure_openai_get_embeddings() -> None:
"""End-to-end test of Azure OpenAI embedding generation."""
async with AzureCliCredential() as credential:
client = _create_azure_openai_embedding_client(credential=credential)
result = await client.get_embeddings(["hello world"])
assert len(result) == 1
assert isinstance(result[0].vector, list)
assert len(result[0].vector) > 0
assert all(isinstance(v, float) for v in result[0].vector)
assert result[0].model_id is not None
assert result.usage is not None
assert result.usage["input_token_count"] > 0
@pytest.mark.flaky
@pytest.mark.integration
@skip_if_azure_openai_integration_tests_disabled
@_with_azure_openai_debug()
async def test_integration_azure_openai_get_embeddings_multiple() -> None:
"""Test Azure OpenAI embedding generation for multiple inputs."""
async with AzureCliCredential() as credential:
client = _create_azure_openai_embedding_client(credential=credential)
result = await client.get_embeddings(["hello", "world", "test"])
assert len(result) == 3
dims = [len(e.vector) for e in result]
assert all(d == dims[0] for d in dims)
@pytest.mark.flaky
@pytest.mark.integration
@skip_if_azure_openai_integration_tests_disabled
@_with_azure_openai_debug()
async def test_integration_azure_openai_get_embeddings_with_dimensions() -> None:
"""Test Azure OpenAI embedding generation with custom dimensions."""
async with AzureCliCredential() as credential:
client = _create_azure_openai_embedding_client(credential=credential)
options: OpenAIEmbeddingOptions = {"dimensions": 256}
result = await client.get_embeddings(["hello world"], options=options)
assert len(result) == 1
assert len(result[0].vector) == 256
python/packages/azure-ai/tests/azure_openai/test_azure_responses_client.py
-542
View File
@@ -1,542 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
import json
import logging
import os
from functools import wraps
from pathlib import Path
from typing import Annotated, Any
import pytest
from agent_framework import (
Agent,
AgentResponse,
ChatResponse,
Content,
Message,
SupportsChatGetResponse,
tool,
)
from agent_framework.azure import AzureOpenAIResponsesClient
from azure.identity import AzureCliCredential
from pydantic import BaseModel
from pytest import param
pytestmark = pytest.mark.filterwarnings("ignore:AzureOpenAIResponsesClient is deprecated\\..*:DeprecationWarning")
skip_if_azure_integration_tests_disabled = pytest.mark.skipif(
os.getenv("AZURE_OPENAI_ENDPOINT", "") in ("", "https://test-endpoint.com"),
reason="No real AZURE_OPENAI_ENDPOINT provided; skipping integration tests.",
)
def _with_azure_openai_debug() -> Any:
def decorator(func: Any) -> Any:
@wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> Any:
try:
return await func(*args, **kwargs)
except Exception as exc:
model = os.getenv("AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME") or os.getenv(
"AZURE_OPENAI_DEPLOYMENT_NAME", "<unset>"
)
api_version = os.getenv("AZURE_OPENAI_API_VERSION", "<unset>")
endpoint = os.getenv("AZURE_OPENAI_ENDPOINT", "<unset>")
debug_message = f"Azure OpenAI debug: endpoint={endpoint}, model={model}, api_version={api_version}"
if hasattr(exc, "add_note"):
exc.add_note(debug_message)
elif exc.args:
exc.args = (f"{exc.args[0]}\n{debug_message}", *exc.args[1:])
else:
exc.args = (debug_message,)
raise
return wrapper
return decorator
logger = logging.getLogger(__name__)
class OutputStruct(BaseModel):
"""A structured output for testing purposes."""
location: str
weather: str
@tool(approval_mode="never_require")
async def get_weather(location: Annotated[str, "The location as a city name"]) -> str:
"""Get the current weather in a given location."""
# Implementation of the tool to get weather
return f"The weather in {location} is sunny and 72°F."
async def create_vector_store(
client: AzureOpenAIResponsesClient,
) -> tuple[str, Content]:
"""Create a vector store with sample documents for testing."""
file = await client.client.files.create(
file=("todays_weather.txt", b"The weather today is sunny with a high of 75F."),
purpose="assistants",
)
vector_store = await client.client.vector_stores.create(
name="knowledge_base",
expires_after={"anchor": "last_active_at", "days": 1},
)
result = await client.client.vector_stores.files.create_and_poll(vector_store_id=vector_store.id, file_id=file.id)
if result.last_error is not None:
raise Exception(f"Vector store file processing failed with status: {result.last_error.message}")
return file.id, Content.from_hosted_vector_store(vector_store_id=vector_store.id)
async def delete_vector_store(client: AzureOpenAIResponsesClient, file_id: str, vector_store_id: str) -> None:
"""Delete the vector store after tests."""
await client.client.vector_stores.delete(vector_store_id=vector_store_id)
await client.client.files.delete(file_id=file_id)
def test_init(azure_openai_unit_test_env: dict[str, str]) -> None:
# Test successful initialization
azure_responses_client = AzureOpenAIResponsesClient(credential=AzureCliCredential())
assert azure_responses_client.model == azure_openai_unit_test_env["AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME"]
assert isinstance(azure_responses_client, SupportsChatGetResponse)
def test_init_validation_fail() -> None:
# Test successful initialization
with pytest.raises(ValueError):
AzureOpenAIResponsesClient(api_key="34523", deployment_name={"test": "dict"}) # type: ignore
def test_init_model_id_constructor(azure_openai_unit_test_env: dict[str, str]) -> None:
# Test successful initialization
model_id = "test_model_id"
azure_responses_client = AzureOpenAIResponsesClient(deployment_name=model_id)
assert azure_responses_client.model == model_id
assert isinstance(azure_responses_client, SupportsChatGetResponse)
def test_init_model_id_kwarg(azure_openai_unit_test_env: dict[str, str]) -> None:
"""Test that model_id kwarg correctly sets the deployment name (issue #4299)."""
azure_responses_client = AzureOpenAIResponsesClient(model_id="gpt-4o")
assert azure_responses_client.model == "gpt-4o"
assert isinstance(azure_responses_client, SupportsChatGetResponse)
def test_init_model_id_kwarg_does_not_override_deployment_name(
azure_openai_unit_test_env: dict[str, str],
) -> None:
"""Test that deployment_name takes precedence over model_id kwarg (issue #4299)."""
azure_responses_client = AzureOpenAIResponsesClient(deployment_name="my-deployment", model_id="gpt-4o")
assert azure_responses_client.model == "my-deployment"
assert isinstance(azure_responses_client, SupportsChatGetResponse)
def test_init_model_id_kwarg_none(azure_openai_unit_test_env: dict[str, str]) -> None:
"""Test that model_id=None does not override the env-var deployment name."""
azure_responses_client = AzureOpenAIResponsesClient(model_id=None)
assert azure_responses_client.model == azure_openai_unit_test_env["AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME"]
def test_init_with_default_header(azure_openai_unit_test_env: dict[str, str]) -> None:
default_headers = {"X-Unit-Test": "test-guid"}
# Test successful initialization
azure_responses_client = AzureOpenAIResponsesClient(
default_headers=default_headers,
)
assert azure_responses_client.model == azure_openai_unit_test_env["AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME"]
assert isinstance(azure_responses_client, SupportsChatGetResponse)
# Assert that the default header we added is present in the client's default headers
for key, value in default_headers.items():
assert key in azure_responses_client.client.default_headers
assert azure_responses_client.client.default_headers[key] == value
@pytest.mark.parametrize("exclude_list", [["AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME"]], indirect=True)
def test_init_with_empty_model_id(azure_openai_unit_test_env: dict[str, str]) -> None:
with pytest.raises(ValueError):
AzureOpenAIResponsesClient()
def test_serialize(azure_openai_unit_test_env: dict[str, str]) -> None:
default_headers = {"X-Unit-Test": "test-guid"}
settings = {
"deployment_name": azure_openai_unit_test_env["AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME"],
"api_key": azure_openai_unit_test_env["AZURE_OPENAI_API_KEY"],
"default_headers": default_headers,
}
azure_responses_client = AzureOpenAIResponsesClient.from_dict(settings)
dumped_settings = azure_responses_client.to_dict()
assert dumped_settings["deployment_name"] == azure_openai_unit_test_env["AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME"]
assert "api_key" not in dumped_settings
# Assert that the default header we added is present in the dumped_settings default headers
for key, value in default_headers.items():
assert key in dumped_settings["default_headers"]
assert dumped_settings["default_headers"][key] == value
# Assert that the 'User-Agent' header is not present in the dumped_settings default headers
assert "User-Agent" not in dumped_settings["default_headers"]
# region Integration Tests
@pytest.mark.flaky
@pytest.mark.integration
@skip_if_azure_integration_tests_disabled
@pytest.mark.parametrize(
"option_name,option_value,needs_validation",
[
# Simple ChatOptions - just verify they don't fail
param("max_tokens", 500, False, id="max_tokens"),
param("seed", 123, False, id="seed"),
param("user", "test-user-id", False, id="user"),
param("metadata", {"test_key": "test_value"}, False, id="metadata"),
param("frequency_penalty", 0.5, False, id="frequency_penalty"),
param("presence_penalty", 0.3, False, id="presence_penalty"),
param("stop", ["END"], False, id="stop"),
param("allow_multiple_tool_calls", True, False, id="allow_multiple_tool_calls"),
param("tool_choice", "none", True, id="tool_choice_none"),
# OpenAIResponsesOptions - just verify they don't fail
param("safety_identifier", "user-hash-abc123", False, id="safety_identifier"),
param("truncation", "auto", False, id="truncation"),
param("prompt_cache_key", "test-cache-key", False, id="prompt_cache_key"),
param("max_tool_calls", 3, False, id="max_tool_calls"),
# Complex options requiring output validation
param("tools", [get_weather], True, id="tools_function"),
param("tool_choice", "auto", True, id="tool_choice_auto"),
param(
"tool_choice",
{"mode": "required", "required_function_name": "get_weather"},
True,
id="tool_choice_required",
),
param("response_format", OutputStruct, True, id="response_format_pydantic"),
param(
"response_format",
{
"type": "json_schema",
"json_schema": {
"name": "WeatherDigest",
"strict": True,
"schema": {
"title": "WeatherDigest",
"type": "object",
"properties": {
"location": {"type": "string"},
"conditions": {"type": "string"},
"temperature_c": {"type": "number"},
"advisory": {"type": "string"},
},
"required": [
"location",
"conditions",
"temperature_c",
"advisory",
],
"additionalProperties": False,
},
},
},
True,
id="response_format_runtime_json_schema",
),
],
)
@_with_azure_openai_debug()
async def test_integration_options(
option_name: str,
option_value: Any,
needs_validation: bool,
) -> None:
"""Parametrized test covering all ChatOptions and OpenAIResponsesOptions.
Tests both streaming and non-streaming modes for each option to ensure
they don't cause failures. Options marked with needs_validation also
check that the feature actually works correctly.
"""
client = AzureOpenAIResponsesClient(credential=AzureCliCredential())
# Need at least 2 iterations for tool_choice tests: one to get function call, one to get final response
client.function_invocation_configuration["max_iterations"] = 2
# Prepare test message
if option_name == "tools" or option_name == "tool_choice":
# Use weather-related prompt for tool tests
messages = [Message(role="user", text="What is the weather in Seattle?")]
elif option_name == "response_format":
# Use prompt that works well with structured output
messages = [
Message(role="user", text="The weather in Seattle is sunny"),
Message(role="user", text="What is the weather in Seattle?"),
]
else:
# Generic prompt for simple options
messages = [Message(role="user", text="Say 'Hello World' briefly.")]
# Build options dict
options: dict[str, Any] = {option_name: option_value}
# Add tools if testing tool_choice to avoid errors
if option_name == "tool_choice":
options["tools"] = [get_weather]
# Test streaming mode
response = await client.get_response(messages=messages, stream=True, options=options).get_final_response()
assert response is not None
assert isinstance(response, ChatResponse)
assert response.text is not None, f"No text in response for option '{option_name}'"
assert len(response.text) > 0, f"Empty response for option '{option_name}'"
# Validate based on option type
if needs_validation:
if option_name == "tools" or option_name == "tool_choice":
# Should have called the weather function
text = response.text.lower()
assert "sunny" in text or "seattle" in text, f"Tool not invoked for {option_name}"
elif option_name == "response_format":
if option_value == OutputStruct:
# Should have structured output
assert response.value is not None, "No structured output"
assert isinstance(response.value, OutputStruct)
assert "seattle" in response.value.location.lower()
else:
# Runtime JSON schema
assert response.value is None, "No structured output, can't parse any json."
response_value = json.loads(response.text)
assert isinstance(response_value, dict)
assert "location" in response_value
assert "seattle" in response_value["location"].lower()
@pytest.mark.flaky
@pytest.mark.integration
@skip_if_azure_integration_tests_disabled
@_with_azure_openai_debug()
async def test_integration_web_search() -> None:
client = AzureOpenAIResponsesClient(credential=AzureCliCredential())
response = await client.get_response(
messages=[
Message(
role="user",
text="What is the current weather? Do not ask for my current location.",
)
],
options={
"tools": [
AzureOpenAIResponsesClient.get_web_search_tool(user_location={"country": "US", "city": "Seattle"})
]
},
stream=True,
).get_final_response()
assert response.text is not None
@pytest.mark.flaky
@pytest.mark.integration
@skip_if_azure_integration_tests_disabled
@_with_azure_openai_debug()
async def test_integration_client_file_search() -> None:
"""Test Azure responses client with file search tool."""
azure_responses_client = AzureOpenAIResponsesClient(credential=AzureCliCredential())
file_id, vector_store = await create_vector_store(azure_responses_client)
try:
# Test that the client will use the file search tool
response = await azure_responses_client.get_response(
messages=[
Message(
role="user",
text="What is the weather today? Do a file search to find the answer.",
)
],
options={
"tools": [
AzureOpenAIResponsesClient.get_file_search_tool(vector_store_ids=[vector_store.vector_store_id])
],
"tool_choice": "auto",
},
)
assert "sunny" in response.text.lower()
assert "75" in response.text
finally:
await delete_vector_store(azure_responses_client, file_id, vector_store.vector_store_id)
@pytest.mark.flaky
@pytest.mark.integration
@skip_if_azure_integration_tests_disabled
@_with_azure_openai_debug()
async def test_integration_client_file_search_streaming() -> None:
"""Test Azure responses client with file search tool and streaming."""
azure_responses_client = AzureOpenAIResponsesClient(credential=AzureCliCredential())
file_id, vector_store = await create_vector_store(azure_responses_client)
# Test that the client will use the file search tool
try:
response_stream = azure_responses_client.get_response(
messages=[
Message(
role="user",
text="What is the weather today? Do a file search to find the answer.",
)
],
stream=True,
options={
"tools": [
AzureOpenAIResponsesClient.get_file_search_tool(vector_store_ids=[vector_store.vector_store_id])
],
"tool_choice": "auto",
},
)
full_response = await response_stream.get_final_response()
assert "sunny" in full_response.text.lower()
assert "75" in full_response.text
finally:
await delete_vector_store(azure_responses_client, file_id, vector_store.vector_store_id)
@pytest.mark.flaky
@pytest.mark.integration
@skip_if_azure_integration_tests_disabled
@_with_azure_openai_debug()
async def test_integration_client_agent_hosted_mcp_tool() -> None:
"""Integration test for MCP tool with Azure Response Agent using Microsoft Learn MCP."""
client = AzureOpenAIResponsesClient(credential=AzureCliCredential())
response = await client.get_response(
messages=[Message(role="user", text="How to create an Azure storage account using az cli?")],
options={
# this needs to be high enough to handle the full MCP tool response.
"max_tokens": 5000,
"tools": AzureOpenAIResponsesClient.get_mcp_tool(
name="Microsoft Learn MCP",
url="https://learn.microsoft.com/api/mcp",
),
},
)
assert isinstance(response, ChatResponse)
# MCP server may return empty response intermittently - skip test rather than fail
if not response.text:
pytest.skip("MCP server returned empty response - service-side issue")
# Should contain Azure-related content since it's asking about Azure CLI
assert any(term in response.text.lower() for term in ["azure", "storage", "account", "cli"])
@pytest.mark.flaky
@pytest.mark.integration
@skip_if_azure_integration_tests_disabled
@_with_azure_openai_debug()
async def test_integration_client_agent_hosted_code_interpreter_tool():
"""Test Azure Responses Client agent with code interpreter tool."""
client = AzureOpenAIResponsesClient(credential=AzureCliCredential())
response = await client.get_response(
messages=[
Message(
role="user",
text="Calculate the sum of numbers from 1 to 10 using Python code.",
)
],
options={
"tools": [AzureOpenAIResponsesClient.get_code_interpreter_tool()],
},
)
# Should contain calculation result (sum of 1-10 = 55) or code execution content
contains_relevant_content = any(
term in response.text.lower() for term in ["55", "sum", "code", "python", "calculate", "10"]
)
assert contains_relevant_content or len(response.text.strip()) > 10
@pytest.mark.flaky
@pytest.mark.integration
@skip_if_azure_integration_tests_disabled
@_with_azure_openai_debug()
async def test_integration_client_agent_existing_session():
"""Test Azure Responses Client agent with existing session to continue conversations across agent instances."""
# First conversation - capture the session
preserved_session = None
async with Agent(
client=AzureOpenAIResponsesClient(credential=AzureCliCredential()),
instructions="You are a helpful assistant with good memory.",
) as first_agent:
# Start a conversation and capture the session
session = first_agent.create_session()
first_response = await first_agent.run(
"My hobby is photography. Remember this.", session=session, options={"store": True}
)
assert isinstance(first_response, AgentResponse)
assert first_response.text is not None
# Preserve the session for reuse
preserved_session = session
# Second conversation - reuse the session in a new agent instance
if preserved_session:
async with Agent(
client=AzureOpenAIResponsesClient(credential=AzureCliCredential()),
instructions="You are a helpful assistant with good memory.",
) as second_agent:
# Reuse the preserved session
second_response = await second_agent.run(
"What is my hobby?", session=preserved_session, options={"store": True}
)
assert isinstance(second_response, AgentResponse)
assert second_response.text is not None
assert "photography" in second_response.text.lower()
@pytest.mark.flaky
@pytest.mark.integration
@skip_if_azure_integration_tests_disabled
@_with_azure_openai_debug()
async def test_azure_openai_responses_client_tool_rich_content_image() -> None:
"""Test that Azure OpenAI Responses client can handle tool results containing images."""
image_path = Path(__file__).parent.parent / "assets" / "sample_image.jpg"
image_bytes = image_path.read_bytes()
@tool(approval_mode="never_require")
def get_test_image() -> Content:
"""Return a test image for analysis."""
return Content.from_data(data=image_bytes, media_type="image/jpeg")
client = AzureOpenAIResponsesClient(credential=AzureCliCredential())
client.function_invocation_configuration["max_iterations"] = 2
for streaming in [False, True]:
messages = [
Message(
role="user",
text="Call the get_test_image tool and describe what you see.",
)
]
options: dict[str, Any] = {"tools": [get_test_image], "tool_choice": "auto"}
if streaming:
response = await client.get_response(messages=messages, stream=True, options=options).get_final_response()
else:
response = await client.get_response(messages=messages, options=options)
assert response is not None
assert isinstance(response, ChatResponse)
assert response.text is not None
assert len(response.text) > 0
# sample_image.jpg contains a photo of a house; the model should mention it.
assert "house" in response.text.lower(), f"Model did not describe the house image. Response: {response.text}"
python/packages/azure-ai/tests/azure_openai/test_azure_responses_client_foundry.py
-131
View File
@@ -1,131 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
import warnings
from unittest.mock import MagicMock
import pytest
from agent_framework import SupportsChatGetResponse
warnings.filterwarnings(
"ignore",
message=r"RawAzureAIClient is deprecated\..*",
category=DeprecationWarning,
)
from agent_framework.azure import AzureOpenAIResponsesClient # noqa: E402
from azure.identity import AzureCliCredential # noqa: E402
pytestmark = pytest.mark.filterwarnings("ignore:AzureOpenAIResponsesClient is deprecated\\..*:DeprecationWarning")
def test_init_with_project_client(azure_openai_unit_test_env: dict[str, str]) -> None:
"""Test initialization with an existing AIProjectClient."""
from unittest.mock import patch
from openai import AsyncOpenAI
# Create a mock AIProjectClient that returns a mock AsyncOpenAI client
mock_openai_client = MagicMock(spec=AsyncOpenAI)
mock_openai_client.default_headers = {}
mock_project_client = MagicMock()
mock_project_client.get_openai_client.return_value = mock_openai_client
with patch(
"agent_framework_azure_ai._deprecated_azure_openai.AzureOpenAIResponsesClient._create_client_from_project",
return_value=mock_openai_client,
):
azure_responses_client = AzureOpenAIResponsesClient(
project_client=mock_project_client,
deployment_name="gpt-4o",
)
assert azure_responses_client.model == "gpt-4o"
assert azure_responses_client.client is mock_openai_client
assert isinstance(azure_responses_client, SupportsChatGetResponse)
def test_init_with_project_endpoint(azure_openai_unit_test_env: dict[str, str]) -> None:
"""Test initialization with a project endpoint and credential."""
from unittest.mock import patch
from openai import AsyncOpenAI
mock_openai_client = MagicMock(spec=AsyncOpenAI)
mock_openai_client.default_headers = {}
with patch(
"agent_framework_azure_ai._deprecated_azure_openai.AzureOpenAIResponsesClient._create_client_from_project",
return_value=mock_openai_client,
):
azure_responses_client = AzureOpenAIResponsesClient(
project_endpoint="https://test-project.services.ai.azure.com",
deployment_name="gpt-4o",
credential=AzureCliCredential(),
)
assert azure_responses_client.model == "gpt-4o"
assert azure_responses_client.client is mock_openai_client
assert isinstance(azure_responses_client, SupportsChatGetResponse)
def test_create_client_from_project_with_project_client() -> None:
"""Test _create_client_from_project with an existing project client."""
from openai import AsyncOpenAI
mock_openai_client = MagicMock(spec=AsyncOpenAI)
mock_project_client = MagicMock()
mock_project_client.get_openai_client.return_value = mock_openai_client
result = AzureOpenAIResponsesClient._create_client_from_project(
project_client=mock_project_client,
project_endpoint=None,
credential=None,
)
assert result is mock_openai_client
mock_project_client.get_openai_client.assert_called_once()
def test_create_client_from_project_with_endpoint() -> None:
"""Test _create_client_from_project with a project endpoint."""
from unittest.mock import patch
from openai import AsyncOpenAI
mock_openai_client = MagicMock(spec=AsyncOpenAI)
mock_credential = MagicMock()
with patch("agent_framework_azure_ai._deprecated_azure_openai.AIProjectClient") as MockAIProjectClient:
mock_instance = MockAIProjectClient.return_value
mock_instance.get_openai_client.return_value = mock_openai_client
result = AzureOpenAIResponsesClient._create_client_from_project(
project_client=None,
project_endpoint="https://test-project.services.ai.azure.com",
credential=mock_credential,
)
assert result is mock_openai_client
MockAIProjectClient.assert_called_once()
mock_instance.get_openai_client.assert_called_once()
def test_create_client_from_project_missing_endpoint() -> None:
"""Test _create_client_from_project raises error when endpoint is missing."""
with pytest.raises(ValueError, match="project endpoint is required"):
AzureOpenAIResponsesClient._create_client_from_project(
project_client=None,
project_endpoint=None,
credential=MagicMock(),
)
def test_create_client_from_project_missing_credential() -> None:
"""Test _create_client_from_project raises error when credential is missing."""
with pytest.raises(ValueError, match="credential is required"):
AzureOpenAIResponsesClient._create_client_from_project(
project_client=None,
project_endpoint="https://test-project.services.ai.azure.com",
credential=None,
)
python/packages/azure-ai/tests/test_agent_provider.py
-773
View File
@@ -1,773 +0,0 @@
# 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 (
Agent,
tool,
)
from azure.ai.agents.models import (
Agent as AzureAgent,
)
from azure.ai.agents.models import (
CodeInterpreterToolDefinition,
)
from pydantic import BaseModel
from agent_framework_azure_ai import (
AzureAIAgentClient,
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("AZURE_AI_PROJECT_ENDPOINT", "") in ("", "https://test-project.cognitiveservices.azure.com/"),
reason="No real AZURE_AI_PROJECT_ENDPOINT provided; skipping integration tests.",
)
# 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 load_settings to return a dict with None for project_endpoint
with patch("agent_framework_azure_ai._agent_provider.load_settings") as mock_load_settings:
mock_load_settings.return_value = {"project_endpoint": None, "model_deployment_name": "test-model"}
with pytest.raises(ValueError) 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(ValueError) 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=AzureAgent)
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, Agent)
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=AzureAgent)
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=AzureAgent)
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)
@tool(approval_mode="never_require")
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=AzureAgent)
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.load_settings") as mock_load_settings:
mock_load_settings.return_value = {"project_endpoint": "https://test.com", "model_deployment_name": None}
provider = AzureAIAgentsProvider(agents_client=mock_agents_client)
with pytest.raises(ValueError) 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=AzureAgent)
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, Agent)
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=AzureAgent)
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(ValueError) 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=AzureAgent)
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)
@tool(approval_mode="never_require")
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, Agent)
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=AzureAgent)
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, Agent)
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=AzureAgent)
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(ValueError) 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 excludes hosted tools from local tools (they stay on the server agent)."""
mock_code_interpreter = MagicMock()
mock_code_interpreter.type = "code_interpreter"
mock_agent = MagicMock(spec=AzureAgent)
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, Agent)
# Hosted tools (code_interpreter, file_search, etc.) are already on the server agent
# and should NOT be in local tools to avoid re-sending them at run time
tools = agent.default_options.get("tools") or []
assert not any(isinstance(t, dict) and t.get("type") == "code_interpreter" for t in tools)
def test_as_agent_with_dict_function_tools_validates(
azure_ai_unit_test_env: dict[str, str],
mock_agents_client: MagicMock,
) -> None:
"""Test as_agent validates dict-format function tools require implementations."""
# Dict-based function tool (as returned by some Azure AI SDK operations)
dict_function_tool = { # type: ignore
"type": "function",
"function": {
"name": "dict_based_function",
"description": "A function defined as dict",
"parameters": {"type": "object", "properties": {}},
},
}
mock_agent = MagicMock(spec=AzureAgent)
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 = [dict_function_tool]
provider = AzureAIAgentsProvider(agents_client=mock_agents_client)
with pytest.raises(ValueError) as exc_info:
provider.as_agent(mock_agent)
assert "dict_based_function" in str(exc_info.value)
def test_as_agent_with_dict_function_tools_provided(
azure_ai_unit_test_env: dict[str, str],
mock_agents_client: MagicMock,
) -> None:
"""Test as_agent succeeds when dict-format function tools have implementations provided."""
dict_function_tool = { # type: ignore
"type": "function",
"function": {
"name": "dict_based_function",
"description": "A function defined as dict",
"parameters": {"type": "object", "properties": {}},
},
}
mock_agent = MagicMock(spec=AzureAgent)
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 = [dict_function_tool]
@tool
def dict_based_function() -> str:
"""A function implementation."""
return "result"
provider = AzureAIAgentsProvider(agents_client=mock_agents_client)
agent = provider.as_agent(mock_agent, tools=dict_based_function)
assert isinstance(agent, Agent)
assert agent.id == "agent-id"
# 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 FunctionTool to Azure tool definition."""
@tool(approval_mode="never_require")
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 code_interpreter dict tool."""
tool = AzureAIAgentClient.get_code_interpreter_tool()
result = to_azure_ai_agent_tools([tool])
assert len(result) == 1
assert isinstance(result[0], CodeInterpreterToolDefinition)
def test_to_azure_ai_agent_tools_file_search() -> None:
"""Test converting file_search dict tool with vector stores."""
tool = AzureAIAgentClient.get_file_search_tool(vector_store_ids=["vs-123"])
run_options: dict[str, Any] = {}
result = to_azure_ai_agent_tools([tool], run_options)
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 web_search dict tool for Bing Grounding."""
# Use a properly formatted connection ID as required by Azure SDK
valid_conn_id = (
"/subscriptions/test-sub/resourceGroups/test-rg/"
"providers/Microsoft.CognitiveServices/accounts/test-account/"
"projects/test-project/connections/test-connection"
)
tool = AzureAIAgentClient.get_web_search_tool(bing_connection_id=valid_conn_id)
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 web_search dict tool for Custom Bing Search."""
tool = AzureAIAgentClient.get_web_search_tool(
bing_custom_connection_id="custom-conn-id",
bing_custom_instance_id="my-instance",
)
result = to_azure_ai_agent_tools([tool])
assert len(result) > 0
def test_to_azure_ai_agent_tools_web_search_missing_config(monkeypatch: Any) -> None:
"""Test converting web_search dict tool without bing config returns empty."""
monkeypatch.delenv("BING_CONNECTION_ID", raising=False)
monkeypatch.delenv("BING_CUSTOM_CONNECTION_ID", raising=False)
monkeypatch.delenv("BING_CUSTOM_INSTANCE_NAME", raising=False)
tool = {"type": "web_search"}
result = to_azure_ai_agent_tools([tool])
# web_search without bing connection is passed through as dict
assert len(result) == 1
def test_to_azure_ai_agent_tools_mcp() -> None:
"""Test converting MCP dict tool."""
tool = AzureAIAgentClient.get_mcp_tool(
name="my mcp server",
url="https://mcp.example.com",
)
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 pass through unchanged."""
class UnsupportedTool:
pass
unsupported = UnsupportedTool()
result = to_azure_ai_agent_tools([unsupported]) # type: ignore
assert len(result) == 1
assert result[0] is unsupported # Passed through unchanged
# 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 result[0] == {"type": "code_interpreter"}
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 result[0] == {"type": "code_interpreter"}
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 result[0]["type"] == "file_search"
assert result[0]["vector_store_ids"] == ["vs-123", "vs-456"]
def test_from_azure_ai_agent_tools_bing_grounding_dict() -> None:
"""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 result[0]["type"] == "bing_grounding"
assert result[0]["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 result[0]["type"] == "bing_custom_search"
assert result[0]["connection_id"] == "custom-conn"
assert result[0]["instance_name"] == "my-instance"
def test_from_azure_ai_agent_tools_mcp_dict() -> None:
"""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
python/packages/azure-ai/tests/test_azure_ai_agent_client.py
-1941
View File
File diff suppressed because it is too large Load Diff
python/packages/azure-ai/tests/test_azure_ai_client.py
-1965
View File
File diff suppressed because it is too large Load Diff
python/packages/azure-ai/tests/test_provider.py
-682
View File
@@ -1,682 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from agent_framework import Agent, FunctionTool
from agent_framework._mcp import MCPTool
from azure.ai.projects.models import (
AgentVersionDetails,
PromptAgentDefinition,
)
from azure.ai.projects.models import (
FunctionTool as AzureFunctionTool,
)
from agent_framework_azure_ai import AzureAIProjectAgentProvider
@pytest.fixture
def mock_project_client() -> MagicMock:
"""Fixture that provides a mock AIProjectClient."""
mock_client = MagicMock()
# Mock agents property
mock_client.agents = MagicMock()
mock_client.agents.create_version = AsyncMock()
# Mock conversations property
mock_client.conversations = MagicMock()
mock_client.conversations.create = AsyncMock()
# Mock telemetry property
mock_client.telemetry = MagicMock()
mock_client.telemetry.get_application_insights_connection_string = AsyncMock()
# AIProjectClient.get_openai_client() is a sync accessor, even on the aio client.
mock_client.get_openai_client = MagicMock(return_value=MagicMock())
# Mock close method
mock_client.close = AsyncMock()
return mock_client
@pytest.fixture
def mock_azure_credential() -> MagicMock:
"""Fixture that provides a mock Azure credential."""
return MagicMock()
@pytest.fixture
def azure_ai_unit_test_env(monkeypatch: pytest.MonkeyPatch) -> dict[str, str]:
"""Fixture that sets up Azure AI environment variables for unit testing."""
env_vars = {
"AZURE_AI_PROJECT_ENDPOINT": "https://test-project.cognitiveservices.azure.com/",
"AZURE_AI_MODEL_DEPLOYMENT_NAME": "test-model-deployment",
}
for key, value in env_vars.items():
monkeypatch.setenv(key, value)
return env_vars
def test_provider_init_with_project_client(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider initialization with existing project_client."""
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
assert provider._project_client is mock_project_client # type: ignore
assert not provider._should_close_client # type: ignore
def test_provider_init_with_credential_and_endpoint(
azure_ai_unit_test_env: dict[str, str],
mock_azure_credential: MagicMock,
) -> None:
"""Test AzureAIProjectAgentProvider initialization with credential and endpoint."""
with patch("agent_framework_azure_ai._project_provider.AIProjectClient") as mock_ai_project_client:
mock_client = MagicMock()
mock_ai_project_client.return_value = mock_client
provider = AzureAIProjectAgentProvider(
project_endpoint=azure_ai_unit_test_env["AZURE_AI_PROJECT_ENDPOINT"],
credential=mock_azure_credential,
)
assert provider._project_client is mock_client # type: ignore
assert provider._should_close_client # type: ignore
# Verify AIProjectClient was called with correct parameters
mock_ai_project_client.assert_called_once()
def test_provider_init_missing_endpoint() -> None:
"""Test AzureAIProjectAgentProvider initialization when endpoint is missing."""
with patch("agent_framework_azure_ai._project_provider.load_settings") as mock_load_settings:
mock_load_settings.return_value = {"project_endpoint": None, "model_deployment_name": "test-model"}
with pytest.raises(ValueError, match="Azure AI project endpoint is required"):
AzureAIProjectAgentProvider(credential=MagicMock())
def test_provider_init_missing_credential(azure_ai_unit_test_env: dict[str, str]) -> None:
"""Test AzureAIProjectAgentProvider initialization when credential is missing."""
with pytest.raises(ValueError, match="Azure credential is required when project_client is not provided"):
AzureAIProjectAgentProvider(
project_endpoint=azure_ai_unit_test_env["AZURE_AI_PROJECT_ENDPOINT"],
)
async def test_provider_create_agent(
mock_project_client: MagicMock,
azure_ai_unit_test_env: dict[str, str],
) -> None:
"""Test AzureAIProjectAgentProvider.create_agent method."""
with patch("agent_framework_azure_ai._project_provider.load_settings") as mock_load_settings:
mock_load_settings.return_value = {
"project_endpoint": azure_ai_unit_test_env["AZURE_AI_PROJECT_ENDPOINT"],
"model_deployment_name": azure_ai_unit_test_env["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
}
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
# Mock agent creation response
mock_agent_version = MagicMock(spec=AgentVersionDetails)
mock_agent_version.id = "agent-id"
mock_agent_version.name = "test-agent"
mock_agent_version.version = "1.0"
mock_agent_version.description = "Test Agent"
mock_agent_version.definition = MagicMock(spec=PromptAgentDefinition)
mock_agent_version.definition.model = "gpt-4"
mock_agent_version.definition.instructions = "Test instructions"
mock_agent_version.definition.temperature = 0.7
mock_agent_version.definition.top_p = 0.9
mock_agent_version.definition.tools = []
mock_project_client.agents.create_version = AsyncMock(return_value=mock_agent_version)
agent = await provider.create_agent(
name="test-agent",
model="gpt-4",
instructions="Test instructions",
description="Test Agent",
)
assert isinstance(agent, Agent)
assert agent.name == "test-agent"
mock_project_client.agents.create_version.assert_called_once()
async def test_provider_create_agent_with_env_model(
mock_project_client: MagicMock,
azure_ai_unit_test_env: dict[str, str],
) -> None:
"""Test AzureAIProjectAgentProvider.create_agent uses model from env var."""
with patch("agent_framework_azure_ai._project_provider.load_settings") as mock_load_settings:
mock_load_settings.return_value = {
"project_endpoint": azure_ai_unit_test_env["AZURE_AI_PROJECT_ENDPOINT"],
"model_deployment_name": azure_ai_unit_test_env["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
}
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
# Mock agent creation response
mock_agent_version = MagicMock(spec=AgentVersionDetails)
mock_agent_version.id = "agent-id"
mock_agent_version.name = "test-agent"
mock_agent_version.version = "1.0"
mock_agent_version.description = None
mock_agent_version.definition = MagicMock(spec=PromptAgentDefinition)
mock_agent_version.definition.model = azure_ai_unit_test_env["AZURE_AI_MODEL_DEPLOYMENT_NAME"]
mock_agent_version.definition.instructions = None
mock_agent_version.definition.temperature = None
mock_agent_version.definition.top_p = None
mock_agent_version.definition.tools = []
mock_project_client.agents.create_version = AsyncMock(return_value=mock_agent_version)
# Call without model parameter - should use env var
agent = await provider.create_agent(name="test-agent")
assert isinstance(agent, Agent)
# Verify the model from env var was used
call_args = mock_project_client.agents.create_version.call_args
assert call_args[1]["definition"].model == azure_ai_unit_test_env["AZURE_AI_MODEL_DEPLOYMENT_NAME"]
async def test_provider_create_agent_missing_model(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider.create_agent raises when model is missing."""
with patch("agent_framework_azure_ai._project_provider.load_settings") as mock_load_settings:
mock_load_settings.return_value = {"project_endpoint": "https://test.com", "model_deployment_name": None}
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
with pytest.raises(ValueError, match="Model deployment name is required"):
await provider.create_agent(name="test-agent")
async def test_provider_create_agent_with_rai_config(
mock_project_client: MagicMock,
azure_ai_unit_test_env: dict[str, str],
) -> None:
"""Test AzureAIProjectAgentProvider.create_agent passes rai_config from default_options."""
with patch("agent_framework_azure_ai._project_provider.load_settings") as mock_load_settings:
mock_load_settings.return_value = {
"project_endpoint": azure_ai_unit_test_env["AZURE_AI_PROJECT_ENDPOINT"],
"model_deployment_name": azure_ai_unit_test_env["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
}
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
# Mock agent creation response
mock_agent_version = MagicMock(spec=AgentVersionDetails)
mock_agent_version.id = "agent-id"
mock_agent_version.name = "test-agent"
mock_agent_version.version = "1.0"
mock_agent_version.description = None
mock_agent_version.definition = MagicMock(spec=PromptAgentDefinition)
mock_agent_version.definition.model = "gpt-4"
mock_agent_version.definition.instructions = None
mock_agent_version.definition.temperature = None
mock_agent_version.definition.top_p = None
mock_agent_version.definition.tools = []
mock_project_client.agents.create_version = AsyncMock(return_value=mock_agent_version)
# Create a mock RaiConfig-like object
mock_rai_config = MagicMock()
mock_rai_config.rai_policy_name = "policy-name"
# Call create_agent with rai_config in default_options
await provider.create_agent(
name="test-agent",
model="gpt-4",
default_options={"rai_config": mock_rai_config},
)
# Verify rai_config was passed to PromptAgentDefinition
call_args = mock_project_client.agents.create_version.call_args
definition = call_args[1]["definition"]
assert definition.rai_config is mock_rai_config
async def test_provider_create_agent_with_reasoning(
mock_project_client: MagicMock,
azure_ai_unit_test_env: dict[str, str],
) -> None:
"""Test AzureAIProjectAgentProvider.create_agent passes reasoning from default_options."""
with patch("agent_framework_azure_ai._project_provider.load_settings") as mock_load_settings:
mock_load_settings.return_value = {
"project_endpoint": azure_ai_unit_test_env["AZURE_AI_PROJECT_ENDPOINT"],
"model_deployment_name": azure_ai_unit_test_env["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
}
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
# Mock agent creation response
mock_agent_version = MagicMock(spec=AgentVersionDetails)
mock_agent_version.id = "agent-id"
mock_agent_version.name = "test-agent"
mock_agent_version.version = "1.0"
mock_agent_version.description = None
mock_agent_version.definition = MagicMock(spec=PromptAgentDefinition)
mock_agent_version.definition.model = "gpt-5.2"
mock_agent_version.definition.instructions = None
mock_agent_version.definition.temperature = None
mock_agent_version.definition.top_p = None
mock_agent_version.definition.tools = []
mock_project_client.agents.create_version = AsyncMock(return_value=mock_agent_version)
# Create a mock Reasoning-like object
mock_reasoning = MagicMock()
mock_reasoning.effort = "medium"
mock_reasoning.summary = "concise"
# Call create_agent with reasoning in default_options
await provider.create_agent(
name="test-agent",
model="gpt-5.2",
default_options={"reasoning": mock_reasoning},
)
# Verify reasoning was passed to PromptAgentDefinition
call_args = mock_project_client.agents.create_version.call_args
definition = call_args[1]["definition"]
assert definition.reasoning is mock_reasoning
async def test_provider_get_agent_with_name(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider.get_agent with name parameter."""
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
# Mock agent response
mock_agent_version = MagicMock(spec=AgentVersionDetails)
mock_agent_version.id = "agent-id"
mock_agent_version.name = "test-agent"
mock_agent_version.version = "1.0"
mock_agent_version.description = "Test Agent"
mock_agent_version.definition = MagicMock(spec=PromptAgentDefinition)
mock_agent_version.definition.model = "gpt-4"
mock_agent_version.definition.instructions = "Test instructions"
mock_agent_version.definition.temperature = None
mock_agent_version.definition.top_p = None
mock_agent_version.definition.tools = []
mock_agent_object = MagicMock()
mock_agent_object.versions.latest = mock_agent_version
mock_project_client.agents = AsyncMock()
mock_project_client.agents.get.return_value = mock_agent_object
agent = await provider.get_agent(name="test-agent")
assert isinstance(agent, Agent)
assert agent.name == "test-agent"
mock_project_client.agents.get.assert_called_with(agent_name="test-agent")
async def test_provider_get_agent_with_reference(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider.get_agent with reference parameter."""
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
# Mock agent response
mock_agent_version = MagicMock(spec=AgentVersionDetails)
mock_agent_version.id = "agent-id"
mock_agent_version.name = "test-agent"
mock_agent_version.version = "1.0"
mock_agent_version.description = "Test Agent"
mock_agent_version.definition = MagicMock(spec=PromptAgentDefinition)
mock_agent_version.definition.model = "gpt-4"
mock_agent_version.definition.instructions = "Test instructions"
mock_agent_version.definition.temperature = None
mock_agent_version.definition.top_p = None
mock_agent_version.definition.tools = []
mock_project_client.agents = AsyncMock()
mock_project_client.agents.get_version.return_value = mock_agent_version
agent_reference = {"name": "test-agent", "version": "1.0"}
agent = await provider.get_agent(reference=agent_reference)
assert isinstance(agent, Agent)
assert agent.name == "test-agent"
mock_project_client.agents.get_version.assert_called_with(agent_name="test-agent", agent_version="1.0")
async def test_provider_get_agent_missing_parameters(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider.get_agent raises when no identifier provided."""
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
with pytest.raises(ValueError, match="Either name or reference must be provided"):
await provider.get_agent()
async def test_provider_get_agent_missing_function_tools(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider.get_agent raises when required tools are missing."""
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
# Mock agent with function tools
mock_agent_version = MagicMock(spec=AgentVersionDetails)
mock_agent_version.id = "agent-id"
mock_agent_version.name = "test-agent"
mock_agent_version.version = "1.0"
mock_agent_version.description = None
mock_agent_version.definition = MagicMock(spec=PromptAgentDefinition)
mock_agent_version.definition.tools = [
AzureFunctionTool(name="test_tool", parameters=[], strict=True, description="Test tool")
]
mock_agent_object = MagicMock()
mock_agent_object.versions.latest = mock_agent_version
mock_project_client.agents = AsyncMock()
mock_project_client.agents.get.return_value = mock_agent_object
with pytest.raises(
ValueError, match="The following prompt agent definition required tools were not provided: test_tool"
):
await provider.get_agent(name="test-agent")
def test_provider_as_agent(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider.as_agent method."""
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
# Create mock agent version
mock_agent_version = MagicMock(spec=AgentVersionDetails)
mock_agent_version.id = "agent-id"
mock_agent_version.name = "test-agent"
mock_agent_version.version = "1.0"
mock_agent_version.description = "Test Agent"
mock_agent_version.definition = MagicMock(spec=PromptAgentDefinition)
mock_agent_version.definition.model = "gpt-4"
mock_agent_version.definition.instructions = "Test instructions"
mock_agent_version.definition.temperature = 0.7
mock_agent_version.definition.top_p = 0.9
mock_agent_version.definition.tools = []
with patch("agent_framework_azure_ai._project_provider.AzureAIClient") as mock_azure_ai_client:
agent = provider.as_agent(mock_agent_version)
assert isinstance(agent, Agent)
assert agent.name == "test-agent"
assert agent.description == "Test Agent"
# Verify AzureAIClient was called with correct parameters
mock_azure_ai_client.assert_called_once()
call_kwargs = mock_azure_ai_client.call_args[1]
assert call_kwargs["project_client"] is mock_project_client
assert call_kwargs["agent_name"] == "test-agent"
assert call_kwargs["agent_version"] == "1.0"
assert call_kwargs["agent_description"] == "Test Agent"
assert call_kwargs["model_deployment_name"] == "gpt-4"
def test_provider_merge_tools_skips_function_tool_dicts(mock_project_client: MagicMock) -> None:
"""Test that _merge_tools skips function tool dicts but keeps other hosted tools."""
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
# Create a mock FunctionTool to provide as implementation
mock_ai_function = create_mock_ai_function("my_function", "My function description")
# Definition tools include a function tool (dict) and an MCP tool
definition_tools = [
{"type": "function", "name": "my_function", "parameters": {}}, # Should be skipped
{"type": "mcp", "server_label": "my_mcp", "server_url": "http://localhost:8080"}, # Should be converted
]
# Call _merge_tools with user-provided function implementation
merged = provider._merge_tools(definition_tools, [mock_ai_function]) # type: ignore
# Should have 2 items: the converted MCP dict and the user-provided FunctionTool
assert len(merged) == 2
# Check that the function tool dict was NOT included (it was skipped)
function_dicts = [t for t in merged if isinstance(t, dict) and t.get("type") == "function"]
assert len(function_dicts) == 0
# Check that the MCP tool was converted to dict
mcp_tools = [t for t in merged if isinstance(t, dict) and t.get("type") == "mcp"]
assert len(mcp_tools) == 1
assert mcp_tools[0]["server_label"] == "my_mcp"
# Check that the user-provided FunctionTool was included
ai_functions = [t for t in merged if isinstance(t, FunctionTool)]
assert len(ai_functions) == 1
assert ai_functions[0].name == "my_function"
async def test_provider_context_manager(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider async context manager."""
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._project_provider.load_settings") as mock_load_settings:
mock_load_settings.return_value = {
"project_endpoint": "https://test.com",
"model_deployment_name": "test-model",
}
async with AzureAIProjectAgentProvider(credential=MagicMock()) as provider:
assert provider._project_client is mock_client # type: ignore
# Should call close after exiting context
mock_client.close.assert_called_once()
async def test_provider_context_manager_with_provided_client(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider context manager doesn't close provided client."""
mock_project_client.close = AsyncMock()
async with AzureAIProjectAgentProvider(project_client=mock_project_client) as provider:
assert provider._project_client is mock_project_client # type: ignore
# Should NOT call close when client was provided
mock_project_client.close.assert_not_called()
async def test_provider_close_method(mock_project_client: MagicMock) -> None:
"""Test AzureAIProjectAgentProvider.close method."""
with patch("agent_framework_azure_ai._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._project_provider.load_settings") as mock_load_settings:
mock_load_settings.return_value = {
"project_endpoint": "https://test.com",
"model_deployment_name": "test-model",
}
provider = AzureAIProjectAgentProvider(credential=MagicMock())
await provider.close()
mock_client.close.assert_called_once()
def test_create_text_format_config_sets_strict_for_pydantic_models() -> None:
"""Test that create_text_format_config sets strict=True for Pydantic models."""
from pydantic import BaseModel
from agent_framework_azure_ai._shared import create_text_format_config
class TestSchema(BaseModel):
subject: str
summary: str
result = create_text_format_config(TestSchema)
# Verify strict=True is set
assert result["strict"] is True
assert result["name"] == "TestSchema"
assert "schema" in result
class MockMCPTool(MCPTool): # pyright: ignore[reportGeneralTypeIssues]
"""A mock MCPTool subclass for testing that passes isinstance checks.
Note: This intentionally does NOT call super().__init__() because MCPTool's
constructor requires MCP server connection parameters that aren't needed for
unit testing. We only need isinstance(obj, MCPTool) to return True.
"""
def __init__(self, functions: list[FunctionTool] | None = None) -> None:
self.name = "MockMCPTool"
self.description = "A mock MCP tool for testing"
self.is_connected = False
self._mock_functions = functions or []
self._connect_called = False
@property
def functions(self) -> list[FunctionTool]:
return self._mock_functions
async def connect(self, *, reset: bool = False) -> None:
self._connect_called = True
self.is_connected = True
@pytest.fixture
def mock_mcp_tool() -> MockMCPTool:
"""Fixture that provides a mock MCPTool."""
mock_functions = [
create_mock_ai_function("mcp_function_1", "First MCP function"),
create_mock_ai_function("mcp_function_2", "Second MCP function"),
]
return MockMCPTool(functions=mock_functions)
def create_mock_ai_function(name: str, description: str = "A mock function") -> FunctionTool:
"""Create a real FunctionTool for testing."""
def mock_func(arg: str) -> str:
return f"Result from {name}: {arg}"
return FunctionTool(func=mock_func, name=name, description=description, approval_mode="never_require")
async def test_provider_create_agent_with_mcp_tool(
mock_project_client: MagicMock,
azure_ai_unit_test_env: dict[str, str],
mock_mcp_tool: "MockMCPTool",
) -> None:
"""Test that create_agent connects MCP tools and passes discovered functions to Azure AI."""
# Patch normalize_tools to return tools as-is in a list (avoids callable check)
def mock_normalize_tools(tools):
if tools is None:
return []
if isinstance(tools, list):
return tools
return [tools]
with (
patch("agent_framework_azure_ai._project_provider.load_settings") as mock_load_settings,
patch("agent_framework_azure_ai._project_provider.to_azure_ai_tools") as mock_to_azure_tools,
patch("agent_framework_azure_ai._project_provider.normalize_tools", side_effect=mock_normalize_tools),
):
mock_load_settings.return_value = {
"project_endpoint": azure_ai_unit_test_env["AZURE_AI_PROJECT_ENDPOINT"],
"model_deployment_name": azure_ai_unit_test_env["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
}
mock_to_azure_tools.return_value = [{"type": "function", "name": "mcp_function_1"}]
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
# Mock agent creation response
mock_agent_version = MagicMock(spec=AgentVersionDetails)
mock_agent_version.id = "agent-id"
mock_agent_version.name = "test-agent"
mock_agent_version.version = "1.0"
mock_agent_version.description = "Test Agent"
mock_agent_version.definition = MagicMock(spec=PromptAgentDefinition)
mock_agent_version.definition.model = "gpt-4"
mock_agent_version.definition.instructions = "Test instructions"
mock_agent_version.definition.tools = []
mock_project_client.agents.create_version = AsyncMock(return_value=mock_agent_version)
# Call create_agent with MCP tool
await provider.create_agent(
name="test-agent",
model="gpt-4",
instructions="Test instructions",
tools=mock_mcp_tool,
)
# Verify MCP tool was connected
assert mock_mcp_tool._connect_called is True
assert mock_mcp_tool.is_connected is True
# Verify to_azure_ai_tools was called with the discovered MCP functions
mock_to_azure_tools.assert_called_once()
tools_passed = mock_to_azure_tools.call_args[0][0]
assert len(tools_passed) == 2
assert tools_passed[0].name == "mcp_function_1"
assert tools_passed[1].name == "mcp_function_2"
async def test_provider_create_agent_with_mcp_and_regular_tools(
mock_project_client: MagicMock,
azure_ai_unit_test_env: dict[str, str],
mock_mcp_tool: "MockMCPTool",
) -> None:
"""Test that create_agent handles both MCP tools and regular FunctionTools."""
# Create a regular FunctionTool
regular_function = create_mock_ai_function("regular_function", "A regular function")
# Patch normalize_tools to return tools as-is in a list (avoids callable check)
def mock_normalize_tools(tools):
if tools is None:
return []
if isinstance(tools, list):
return tools
return [tools]
with (
patch("agent_framework_azure_ai._project_provider.load_settings") as mock_load_settings,
patch("agent_framework_azure_ai._project_provider.to_azure_ai_tools") as mock_to_azure_tools,
patch("agent_framework_azure_ai._project_provider.normalize_tools", side_effect=mock_normalize_tools),
):
mock_load_settings.return_value = {
"project_endpoint": azure_ai_unit_test_env["AZURE_AI_PROJECT_ENDPOINT"],
"model_deployment_name": azure_ai_unit_test_env["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
}
mock_to_azure_tools.return_value = []
provider = AzureAIProjectAgentProvider(project_client=mock_project_client)
# Mock agent creation response
mock_agent_version = MagicMock(spec=AgentVersionDetails)
mock_agent_version.id = "agent-id"
mock_agent_version.name = "test-agent"
mock_agent_version.version = "1.0"
mock_agent_version.description = None
mock_agent_version.definition = MagicMock(spec=PromptAgentDefinition)
mock_agent_version.definition.model = "gpt-4"
mock_agent_version.definition.instructions = None
mock_agent_version.definition.tools = []
mock_project_client.agents.create_version = AsyncMock(return_value=mock_agent_version)
# Pass both MCP tool and regular function
await provider.create_agent(
name="test-agent",
model="gpt-4",
tools=[mock_mcp_tool, regular_function],
)
# Verify to_azure_ai_tools was called with:
# - The regular FunctionTool (1)
# - The 2 discovered MCP functions
mock_to_azure_tools.assert_called_once()
tools_passed = mock_to_azure_tools.call_args[0][0]
assert len(tools_passed) == 3 # 1 regular + 2 MCP functions
# Verify the regular function is in the list
tool_names = [t.name for t in tools_passed]
assert "regular_function" in tool_names
assert "mcp_function_1" in tool_names
assert "mcp_function_2" in tool_names
python/packages/azure-ai/tests/test_shared.py
-494
View File
@@ -1,494 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
import os
from unittest.mock import MagicMock, patch
import pytest
from agent_framework import (
FunctionTool,
)
from agent_framework.exceptions import IntegrationInvalidRequestException
from azure.ai.agents.models import CodeInterpreterToolDefinition
from pydantic import BaseModel
from agent_framework_azure_ai import AzureAIAgentClient
from agent_framework_azure_ai._shared import (
_convert_response_format, # type: ignore
_convert_sdk_tool, # type: ignore
_extract_project_connection_id, # type: ignore
create_text_format_config,
from_azure_ai_agent_tools,
from_azure_ai_tools,
to_azure_ai_agent_tools,
to_azure_ai_tools,
)
from agent_framework_azure_ai._shared import (
_prepare_mcp_tool_dict_for_azure_ai as _prepare_mcp_tool_for_azure_ai, # type: ignore
)
def test_extract_project_connection_id_direct() -> None:
"""Test extracting project_connection_id from direct key."""
result = _extract_project_connection_id({"project_connection_id": "my-connection"})
assert result == "my-connection"
def test_extract_project_connection_id_from_connection_name() -> None:
"""Test extracting project_connection_id from connection.name structure."""
result = _extract_project_connection_id({"connection": {"name": "my-connection"}})
assert result == "my-connection"
def test_extract_project_connection_id_none() -> None:
"""Test returns None when no connection info."""
assert _extract_project_connection_id(None) is None
assert _extract_project_connection_id({}) is None
def test_to_azure_ai_agent_tools_empty() -> None:
"""Test converting empty/None tools list."""
assert to_azure_ai_agent_tools(None) == []
assert to_azure_ai_agent_tools([]) == []
def test_to_azure_ai_agent_tools_function_tool() -> None:
"""Test converting FunctionTool to tool definition."""
def my_func(arg: str) -> str:
"""My function."""
return arg
func_tool = FunctionTool(func=my_func, name="my_func", description="My function.") # type: ignore
result = to_azure_ai_agent_tools([func_tool]) # type: ignore
assert len(result) == 1
assert result[0]["type"] == "function"
assert result[0]["function"]["name"] == "my_func"
def test_to_azure_ai_agent_tools_code_interpreter() -> None:
"""Test converting code_interpreter dict tool."""
tool = AzureAIAgentClient.get_code_interpreter_tool()
result = to_azure_ai_agent_tools([tool])
assert len(result) == 1
assert isinstance(result[0], CodeInterpreterToolDefinition)
def test_to_azure_ai_agent_tools_web_search_missing_connection() -> None:
"""Test web search tool raises without connection info."""
# Clear any environment variables that could provide connection info
with patch.dict(
os.environ,
{"BING_CONNECTION_ID": "", "BING_CUSTOM_CONNECTION_ID": "", "BING_CUSTOM_INSTANCE_NAME": ""},
clear=False,
):
# Also need to unset the keys if they exist
env_backup = {}
for key in ["BING_CONNECTION_ID", "BING_CUSTOM_CONNECTION_ID", "BING_CUSTOM_INSTANCE_NAME"]:
env_backup[key] = os.environ.pop(key, None)
try:
# get_web_search_tool now raises ValueError when no connection info is available
with pytest.raises(ValueError, match="Azure AI Agents requires a Bing connection"):
AzureAIAgentClient.get_web_search_tool()
finally:
# Restore environment
for key, value in env_backup.items():
if value is not None:
os.environ[key] = value
def test_to_azure_ai_agent_tools_dict_passthrough() -> None:
"""Test dict tools pass through unchanged."""
tool_dict = {"type": "custom", "config": "value"}
result = to_azure_ai_agent_tools([tool_dict])
assert result[0] == tool_dict
def test_to_azure_ai_agent_tools_unsupported_type() -> None:
"""Test unsupported tool type passes through unchanged."""
class UnsupportedTool:
pass
unsupported = UnsupportedTool()
result = to_azure_ai_agent_tools([unsupported]) # type: ignore
assert len(result) == 1
assert result[0] is unsupported # Passed through unchanged
def test_from_azure_ai_agent_tools_empty() -> None:
"""Test converting empty/None tools list."""
assert from_azure_ai_agent_tools(None) == []
assert from_azure_ai_agent_tools([]) == []
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 result[0] == {"type": "code_interpreter"}
def test_convert_sdk_tool_code_interpreter() -> None:
"""Test _convert_sdk_tool with code_interpreter type."""
tool = MagicMock()
tool.type = "code_interpreter"
result = _convert_sdk_tool(tool)
assert result == {"type": "code_interpreter"}
def test_convert_sdk_tool_function_returns_none() -> None:
"""Test _convert_sdk_tool with function type returns None."""
tool = MagicMock()
tool.type = "function"
result = _convert_sdk_tool(tool)
assert result is None
def test_convert_sdk_tool_mcp_returns_none() -> None:
"""Test _convert_sdk_tool with mcp type returns None."""
tool = MagicMock()
tool.type = "mcp"
result = _convert_sdk_tool(tool)
assert result is None
def test_convert_sdk_tool_file_search() -> None:
"""Test _convert_sdk_tool with file_search type."""
tool = MagicMock()
tool.type = "file_search"
tool.file_search = MagicMock()
tool.file_search.vector_store_ids = ["vs-1", "vs-2"]
result = _convert_sdk_tool(tool)
assert result["type"] == "file_search"
assert result["vector_store_ids"] == ["vs-1", "vs-2"]
def test_convert_sdk_tool_bing_grounding() -> None:
"""Test _convert_sdk_tool with bing_grounding type."""
tool = MagicMock()
tool.type = "bing_grounding"
tool.bing_grounding = MagicMock()
tool.bing_grounding.connection_id = "conn-123"
result = _convert_sdk_tool(tool)
assert result["type"] == "bing_grounding"
assert result["connection_id"] == "conn-123"
def test_convert_sdk_tool_bing_custom_search() -> None:
"""Test _convert_sdk_tool with bing_custom_search type."""
tool = MagicMock()
tool.type = "bing_custom_search"
tool.bing_custom_search = MagicMock()
tool.bing_custom_search.connection_id = "conn-123"
tool.bing_custom_search.instance_name = "my-instance"
result = _convert_sdk_tool(tool)
assert result["type"] == "bing_custom_search"
assert result["connection_id"] == "conn-123"
assert result["instance_name"] == "my-instance"
def test_to_azure_ai_tools_empty() -> None:
"""Test converting empty/None tools list."""
assert to_azure_ai_tools(None) == []
assert to_azure_ai_tools([]) == []
def test_to_azure_ai_tools_code_interpreter_with_file_ids() -> None:
"""Test converting code_interpreter dict tool with file inputs."""
tool = {
"type": "code_interpreter",
"file_ids": ["file-123"],
}
result = to_azure_ai_tools([tool])
assert len(result) == 1
assert result[0]["type"] == "code_interpreter"
def test_to_azure_ai_tools_function_tool() -> None:
"""Test converting FunctionTool."""
def my_func(arg: str) -> str:
"""My function."""
return arg
func_tool = FunctionTool(func=my_func, name="my_func", description="My function.") # type: ignore
result = to_azure_ai_tools([func_tool]) # type: ignore
assert len(result) == 1
assert result[0]["type"] == "function"
assert result[0]["name"] == "my_func"
def test_to_azure_ai_tools_file_search() -> None:
"""Test converting file_search dict tool."""
tool = {
"type": "file_search",
"vector_store_ids": ["vs-123"],
"max_num_results": 10,
}
result = to_azure_ai_tools([tool])
assert len(result) == 1
assert result[0]["type"] == "file_search"
assert result[0]["vector_store_ids"] == ["vs-123"]
assert result[0]["max_num_results"] == 10
def test_to_azure_ai_tools_web_search_with_location() -> None:
"""Test converting web_search dict tool with user location."""
tool = {
"type": "web_search_preview",
"user_location": {
"city": "Seattle",
"country": "US",
"region": "WA",
"timezone": "PST",
},
}
result = to_azure_ai_tools([tool])
assert len(result) == 1
assert result[0]["type"] == "web_search_preview"
def test_to_azure_ai_tools_image_generation() -> None:
"""Test converting image_generation dict tool."""
tool = {
"type": "image_generation",
"model": "gpt-image-1",
"size": "1024x1024",
"quality": "high",
}
result = to_azure_ai_tools([tool])
assert len(result) == 1
assert result[0]["type"] == "image_generation"
assert result[0]["model"] == "gpt-image-1"
def test_prepare_mcp_tool_basic() -> None:
"""Test basic MCP tool conversion."""
tool = {"type": "mcp", "server_label": "my_tool", "server_url": "http://localhost:8080"}
result = _prepare_mcp_tool_for_azure_ai(tool)
assert result["server_label"] == "my_tool"
assert "http://localhost:8080" in result["server_url"]
def test_prepare_mcp_tool_with_description() -> None:
"""Test MCP tool with description."""
tool = {
"type": "mcp",
"server_label": "my_tool",
"server_url": "http://localhost:8080",
"server_description": "My MCP server",
}
result = _prepare_mcp_tool_for_azure_ai(tool)
assert result["server_description"] == "My MCP server"
def test_prepare_mcp_tool_with_headers() -> None:
"""Test MCP tool with headers (no project_connection_id)."""
tool = {
"type": "mcp",
"server_label": "my_tool",
"server_url": "http://localhost:8080",
"headers": {"X-Api-Key": "secret"},
}
result = _prepare_mcp_tool_for_azure_ai(tool)
assert result["headers"] == {"X-Api-Key": "secret"}
def test_prepare_mcp_tool_project_connection_takes_precedence() -> None:
"""Test project_connection_id takes precedence over headers."""
tool = {
"type": "mcp",
"server_label": "my_tool",
"server_url": "http://localhost:8080",
"headers": {"X-Api-Key": "secret"},
"project_connection_id": "my-conn",
}
result = _prepare_mcp_tool_for_azure_ai(tool)
assert result["project_connection_id"] == "my-conn"
assert "headers" not in result
def test_prepare_mcp_tool_approval_mode_always() -> None:
"""Test MCP tool with always_require approval mode."""
tool = {
"type": "mcp",
"server_label": "my_tool",
"server_url": "http://localhost:8080",
"require_approval": "always",
}
result = _prepare_mcp_tool_for_azure_ai(tool)
assert result["require_approval"] == "always"
def test_prepare_mcp_tool_approval_mode_never() -> None:
"""Test MCP tool with never_require approval mode."""
tool = {
"type": "mcp",
"server_label": "my_tool",
"server_url": "http://localhost:8080",
"require_approval": "never",
}
result = _prepare_mcp_tool_for_azure_ai(tool)
assert result["require_approval"] == "never"
def test_prepare_mcp_tool_approval_mode_dict() -> None:
"""Test MCP tool with dict approval mode."""
tool = {
"type": "mcp",
"server_label": "my_tool",
"server_url": "http://localhost:8080",
"require_approval": {"always": {"tool_names": ["sensitive_tool", "dangerous_tool"]}},
}
result = _prepare_mcp_tool_for_azure_ai(tool)
# The approval mode is passed through
assert "require_approval" in result
def test_create_text_format_config_pydantic_model() -> None:
"""Test creating text format config from Pydantic model."""
class MySchema(BaseModel):
name: str
value: int
result = create_text_format_config(MySchema)
assert result["type"] == "json_schema"
assert result["name"] == "MySchema"
assert result["strict"] is True
def test_create_text_format_config_json_schema_mapping() -> None:
"""Test creating text format config from json_schema mapping."""
config = {
"type": "json_schema",
"json_schema": {
"name": "MyResponse",
"schema": {"type": "object", "properties": {"name": {"type": "string"}}},
},
}
result = create_text_format_config(config)
assert result["type"] == "json_schema"
assert result["name"] == "MyResponse"
def test_create_text_format_config_json_object() -> None:
"""Test creating text format config for json_object type."""
result = create_text_format_config({"type": "json_object"})
assert result["type"] == "json_object"
def test_create_text_format_config_text() -> None:
"""Test creating text format config for text type."""
result = create_text_format_config({"type": "text"})
assert result["type"] == "text"
def test_create_text_format_config_invalid_raises() -> None:
"""Test invalid response_format raises error."""
with pytest.raises(IntegrationInvalidRequestException):
create_text_format_config({"type": "invalid"})
def test_convert_response_format_with_format_key() -> None:
"""Test _convert_response_format with nested format key."""
config = {"format": {"type": "json_object"}}
result = _convert_response_format(config)
assert result["type"] == "json_object"
def test_convert_response_format_json_schema_missing_schema_raises() -> None:
"""Test json_schema without schema raises error."""
with pytest.raises(IntegrationInvalidRequestException, match="requires a schema"):
_convert_response_format({"type": "json_schema", "json_schema": {}})
def test_convert_response_format_raw_json_schema_with_properties() -> None:
"""Test raw JSON schema with properties is wrapped in json_schema envelope."""
result = _convert_response_format({"type": "object", "properties": {"x": {"type": "string"}}, "title": "MyOutput"})
assert result["type"] == "json_schema"
assert result["name"] == "MyOutput"
assert result["strict"] is True
assert result["schema"]["additionalProperties"] is False
assert "title" not in result["schema"]
def test_convert_response_format_raw_json_schema_no_title() -> None:
"""Test raw JSON schema without title defaults name to 'response'."""
result = _convert_response_format({"type": "object", "properties": {"x": {"type": "string"}}})
assert result["name"] == "response"
def test_convert_response_format_raw_json_schema_with_anyof() -> None:
"""Test raw JSON schema with anyOf keyword is detected."""
result = _convert_response_format({"anyOf": [{"type": "string"}, {"type": "number"}]})
assert result["type"] == "json_schema"
assert result["strict"] is True
def test_from_azure_ai_tools_mcp_approval_mode_always() -> None:
"""Test from_azure_ai_tools converts MCP require_approval='always' to dict."""
tools = [
{
"type": "mcp",
"server_label": "my_mcp",
"server_url": "http://localhost:8080",
"require_approval": "always",
}
]
result = from_azure_ai_tools(tools)
assert len(result) == 1
assert result[0]["type"] == "mcp"
assert result[0]["require_approval"] == "always"
def test_from_azure_ai_tools_mcp_approval_mode_never() -> None:
"""Test from_azure_ai_tools converts MCP require_approval='never' to dict."""
tools = [
{
"type": "mcp",
"server_label": "my_mcp",
"server_url": "http://localhost:8080",
"require_approval": "never",
}
]
result = from_azure_ai_tools(tools)
assert len(result) == 1
assert result[0]["type"] == "mcp"
assert result[0]["require_approval"] == "never"
def test_from_azure_ai_tools_mcp_approval_mode_dict_always() -> None:
"""Test from_azure_ai_tools converts MCP dict require_approval with 'always' key."""
tools = [
{
"type": "mcp",
"server_label": "my_mcp",
"server_url": "http://localhost:8080",
"require_approval": {"always": {"tool_names": ["sensitive_tool", "dangerous_tool"]}},
}
]
result = from_azure_ai_tools(tools)
assert len(result) == 1
assert result[0]["type"] == "mcp"
assert result[0]["require_approval"] == {"always": {"tool_names": ["sensitive_tool", "dangerous_tool"]}}
def test_from_azure_ai_tools_mcp_approval_mode_dict_never() -> None:
"""Test from_azure_ai_tools converts MCP dict require_approval with 'never' key."""
tools = [
{
"type": "mcp",
"server_label": "my_mcp",
"server_url": "http://localhost:8080",
"require_approval": {"never": {"tool_names": ["safe_tool"]}},
}
]
result = from_azure_ai_tools(tools)
assert len(result) == 1
assert result[0]["type"] == "mcp"
assert result[0]["require_approval"] == {"never": {"tool_names": ["safe_tool"]}}