Python: Add support for Foundry Toolboxes (#5346)

* Add support for the Foundry Toolbox in MAF

Introduces a Foundry Toolbox integration: FoundryChatClient gains a
get_toolbox() helper plus select_toolbox_tools(), normalize_tools in
the core package flattens tool-collection wrappers (ToolboxVersionObject
and generic iterables, while leaving Pydantic BaseModel instances
alone), and the new agent_framework.foundry namespace re-exports the
toolbox helpers. Ships with unit tests, a sample, and a design doc.

azure-ai-projects is pinned to the public >=2.0.0,<3.0 range and the
lockfile resolves from public PyPI. The toolbox test module skips when
Toolbox* types are unavailable so CI stays green until the public 2.1.0
SDK lands. OMC tooling directories (.omc/, .omx/) are gitignored.

* Update to latest azure ai projects package

* Improve sample

* Rename ADR to 0025

* Update ADR

* Apply suggestion from @alliscode

Co-authored-by: Ben Thomas <ben.thomas@microsoft.com>

* Improve samples

* Update test

---------

Co-authored-by: Ben Thomas <ben.thomas@microsoft.com>
This commit is contained in:
Evan Mattson
2026-04-21 08:56:01 +09:00
committed by GitHub
Unverified
parent 3e54a689fc
commit 04aaf0c1fe
21 changed files with 1980 additions and 6 deletions
@@ -7,6 +7,7 @@ These samples demonstrate how to use context providers to enrich agent conversat
| File / Folder | Description |
|---------------|-------------|
| [`simple_context_provider.py`](simple_context_provider.py) | Implement a custom context provider by extending `ContextProvider` to extract and inject structured user information across turns. |
| [`foundry_toolbox_context_provider.py`](foundry_toolbox_context_provider.py) | Compose a Microsoft Foundry toolbox with a `ContextProvider` that caches the toolbox once and picks a subset of its tools per-turn via `select_toolbox_tools`, driven by keywords in the latest user message. |
| [`azure_ai_foundry_memory.py`](azure_ai_foundry_memory.py) | Use `FoundryMemoryProvider` to add semantic memory — automatically retrieves, searches, and stores memories via Azure AI Foundry. |
| [`azure_ai_search/`](azure_ai_search/) | Retrieval Augmented Generation (RAG) with Azure AI Search in semantic and agentic modes. See its own [README](azure_ai_search/README.md). |
| [`mem0/`](mem0/) | Memory-powered context using the Mem0 integration (open-source and managed). See its own [README](mem0/README.md). |
@@ -19,6 +20,12 @@ These samples demonstrate how to use context providers to enrich agent conversat
- `FOUNDRY_MODEL`: Model deployment name
- Azure CLI authentication (`az login`)
**For `foundry_toolbox_context_provider.py`:**
- `FOUNDRY_PROJECT_ENDPOINT`: Your Microsoft Foundry project endpoint
- `FOUNDRY_MODEL`: Model deployment name
- A toolbox already configured in that project; set `TOOLBOX_NAME` / `TOOLBOX_VERSION` at the top of the sample
- Azure CLI authentication (`az login`)
**For `azure_ai_foundry_memory.py`:**
- `FOUNDRY_PROJECT_ENDPOINT`: Your Azure AI Foundry project endpoint
- `FOUNDRY_MODEL`: Chat/responses model deployment name
@@ -0,0 +1,207 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from typing import Any
from agent_framework import Agent, AgentSession, ContextProvider, Message, SessionContext
from agent_framework.foundry import (
FoundryChatClient,
get_toolbox_tool_name,
get_toolbox_tool_type,
select_toolbox_tools,
)
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
from pydantic import BaseModel
# Load environment variables from .env file
load_dotenv()
"""
Foundry Toolbox + Context Provider Example
This sample composes a Foundry toolbox with a ContextProvider so the agent's
tool list is chosen dynamically per-turn. It uses the chat client itself as a lightweight "tool router": the
latest user message plus a short menu of toolbox tools is sent to the model
with a Pydantic ``response_format``, and the returned tool names drive
``select_toolbox_tools``. The toolbox is fetched once and cached on the
provider's state dict; subsequent turns reuse the cache.
Prerequisites:
- A Microsoft Foundry project
- A toolbox already configured in that project (set TOOLBOX_NAME below)
- FOUNDRY_PROJECT_ENDPOINT and FOUNDRY_MODEL environment variables set
- Azure CLI authentication (`az login`)
"""
# Replace with your own Foundry toolbox name and version.
TOOLBOX_NAME = "research_toolbox"
# Set to None to resolve the toolbox's current default version at fetch time.
TOOLBOX_VERSION: str | None = None
# Generic queries that exercise the router without assuming any specific tool
# types are configured. The first is introspective, the second forces a
# non-empty pick for whichever tools the toolbox actually contains, and the
# third should route to nothing.
QUERIES: list[str] = [
"Introduce yourself and briefly describe the tools you can use to help me.",
"Pick the tool you think is most useful and demonstrate it with a short example.",
"Say hi in one short sentence - no tools needed.",
]
def create_sample_toolbox(name: str) -> str:
"""Create (or replace) a toolbox version in the Foundry project.
Toolboxes are normally configured in the Foundry portal or a deployment
script, not the application itself. This helper exists so the sample can
be run end-to-end without first setting a toolbox up by hand — delete any
existing toolbox under ``name``, then create a fresh version containing a
single MCP tool. Returns the created version identifier.
"""
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import MCPTool, Tool
from azure.core.exceptions import ResourceNotFoundError
with (
AzureCliCredential() as credential,
AIProjectClient(credential=credential, endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"]) as project_client,
):
try:
project_client.beta.toolboxes.delete(name)
print(f"Toolbox `{name}` deleted")
except ResourceNotFoundError:
pass
tools: list[Tool] = [
MCPTool(
server_label="api_specs",
server_url="https://gitmcp.io/Azure/azure-rest-api-specs",
require_approval="never",
)
]
created = project_client.beta.toolboxes.create_version(
name=name,
description="Toolbox version with MCP require_approval set to 'never'.",
tools=tools,
)
print(f"Created toolbox {created.name}@{created.version} ({len(created.tools)} tool(s))")
return created.version
class ToolSelection(BaseModel):
"""Structured output for the per-turn tool router."""
tool_names: list[str]
ROUTER_INSTRUCTIONS = (
"You are a tool router. Given the user's latest message and a menu of "
"available tools (one per line, formatted as 'NAME - TYPE'), return the "
"NAMES of the tools that would plausibly help answer the message. Return "
"an empty list if no tool is needed."
)
class DynamicToolboxProvider(ContextProvider):
"""Fetches a Foundry toolbox once and lets the model pick tools per-turn."""
DEFAULT_SOURCE_ID = "foundry_toolbox"
def __init__(
self,
source_id: str = DEFAULT_SOURCE_ID,
*,
client: FoundryChatClient,
toolbox_name: str,
toolbox_version: str | None = None,
) -> None:
super().__init__(source_id)
self._client = client
self._toolbox_name = toolbox_name
self._toolbox_version = toolbox_version
async def before_run(
self,
*,
agent: Any,
session: AgentSession | None,
context: SessionContext,
state: dict[str, Any],
) -> None:
"""Cache the toolbox on first call, then let the model pick tools per-turn."""
toolbox = state.get("toolbox")
if toolbox is None:
toolbox = await self._client.get_toolbox(self._toolbox_name, version=self._toolbox_version)
state["toolbox"] = toolbox
print(f"[{self.source_id}] Loaded toolbox {toolbox.name}@{toolbox.version} ({len(toolbox.tools)} tool(s))")
user_messages = [m for m in context.get_messages(include_input=True) if getattr(m, "role", None) == "user"]
if not user_messages:
context.extend_tools(self.source_id, list(toolbox.tools))
return
picks = await self._route_tools(user_messages[-1].text, toolbox.tools)
if picks:
tools = select_toolbox_tools(toolbox, include_names=picks)
print(f"[{self.source_id}] Router picked {sorted(picks)} - surfacing {len(tools)} tool(s)")
else:
tools = list(toolbox.tools)
print(f"[{self.source_id}] Router picked nothing - surfacing all {len(tools)} tool(s)")
context.extend_tools(self.source_id, tools)
async def _route_tools(self, user_text: str, tools: Any) -> list[str]:
"""Ask the model which toolbox tools to surface for this turn."""
menu = "\n".join(f"- {get_toolbox_tool_name(t)} - {get_toolbox_tool_type(t)}" for t in tools)
prompt = (
f"User message:\n{user_text}\n\n"
f"Available tools:\n{menu}\n\n"
"Return the names of tools that should be surfaced for this turn."
)
response = await self._client.get_response(
messages=[Message("user", [prompt])],
options={
"instructions": ROUTER_INSTRUCTIONS,
"response_format": ToolSelection,
},
)
selection: ToolSelection = response.value # type: ignore
return selection.tool_names
async def main() -> None:
client = FoundryChatClient(
project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
model=os.environ["FOUNDRY_MODEL"],
credential=AzureCliCredential(),
)
# Comment out if the toolbox already exists in your Foundry project.
create_sample_toolbox(TOOLBOX_NAME)
toolbox_provider = DynamicToolboxProvider(
client=client,
toolbox_name=TOOLBOX_NAME,
toolbox_version=TOOLBOX_VERSION,
)
async with Agent(
client=client,
instructions=(
"You are a helpful assistant. Use the tools available to you on each "
"turn to answer the user. If no tools are relevant, reply directly."
),
context_providers=[toolbox_provider],
) as agent:
session = agent.create_session()
for query in QUERIES:
print(f"\nUser: {query}")
result = await agent.run(query, session=session)
print(f"Assistant: {result}")
if __name__ == "__main__":
asyncio.run(main())
@@ -26,6 +26,8 @@ This folder contains Azure AI Foundry and Foundry Local samples for Agent Framew
| [`foundry_chat_client_with_hosted_mcp.py`](foundry_chat_client_with_hosted_mcp.py) | Foundry Chat Client with hosted MCP |
| [`foundry_chat_client_with_local_mcp.py`](foundry_chat_client_with_local_mcp.py) | Foundry Chat Client with local MCP |
| [`foundry_chat_client_with_session.py`](foundry_chat_client_with_session.py) | Foundry Chat Client with session management |
| [`foundry_chat_client_with_toolbox.py`](foundry_chat_client_with_toolbox.py) | Foundry Chat Client with Foundry toolbox loading and multi-toolbox composition |
| [`foundry_chat_client_with_toolbox_mcp.py`](foundry_chat_client_with_toolbox_mcp.py) | Foundry Chat Client connected to a toolbox via its MCP endpoint using `MCPStreamableHTTPTool` |
## FoundryLocalClient Samples
@@ -0,0 +1,174 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient, select_toolbox_tools
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
"""
Foundry Chat Client with Toolbox Example
This sample demonstrates loading a named, versioned Foundry toolbox into an
Agent via ``FoundryChatClient.get_toolbox()``. A toolbox is a server-side
bundle of tool configurations (code interpreter, file search, MCP, web search,
etc.) configured in the Foundry portal or via the raw SDK.
Prerequisites:
- A Microsoft Foundry project
- A toolbox already configured in that project (set TOOLBOX_NAME below)
- FOUNDRY_PROJECT_ENDPOINT and FOUNDRY_MODEL environment variables set
"""
# Replace with your own Foundry toolbox name and version.
TOOLBOX_NAME = "research_toolbox"
TOOLBOX_VERSION = "1"
# Used only by combine_toolboxes() — swap in a second toolbox you own.
SECOND_TOOLBOX_NAME = "analysis_toolbox"
SECOND_TOOLBOX_VERSION = "1"
# Replace with any question that exercises the tools configured in your toolbox.
QUERY = "Introduce yourself and briefly describe the tools you can use to help me."
def create_sample_toolbox(name: str) -> str:
"""Create (or replace) a toolbox version in the Foundry project.
Toolboxes are normally configured in the Foundry portal or a deployment
script, not the application itself. This helper exists so the samples can
be run end-to-end without first setting a toolbox up by hand — delete any
existing toolbox under ``name``, then create a fresh version containing a
single MCP tool. Returns the created version identifier.
"""
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import MCPTool, Tool
from azure.core.exceptions import ResourceNotFoundError
with (
AzureCliCredential() as credential,
AIProjectClient(credential=credential, endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"]) as project_client,
):
try:
project_client.beta.toolboxes.delete(name)
print(f"Toolbox `{name}` deleted")
except ResourceNotFoundError:
pass
tools: list[Tool] = [
MCPTool(
server_label="api_specs",
server_url="https://gitmcp.io/Azure/azure-rest-api-specs",
require_approval="never",
)
]
created = project_client.beta.toolboxes.create_version(
name=name,
description="Toolbox version with MCP require_approval set to 'never'.",
tools=tools,
)
print(f"Created toolbox {created.name}@{created.version} ({len(created.tools)} tool(s))")
return created.version
async def main() -> None:
"""Example showing how to use a single Foundry toolbox with FoundryChatClient."""
print("=== Foundry Chat Client with Toolbox Example ===")
# For authentication, run `az login` in your terminal or replace
# AzureCliCredential with your preferred authentication option.
client = FoundryChatClient(
credential=AzureCliCredential(),
project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
model=os.environ["FOUNDRY_MODEL"],
)
# Comment out if the toolbox already exists in your Foundry project.
create_sample_toolbox(TOOLBOX_NAME)
# Omit ``version`` to resolve the toolbox's current default version at runtime.
toolbox = await client.get_toolbox(TOOLBOX_NAME)
print(f"Loaded toolbox {toolbox.name}@{toolbox.version} ({len(toolbox.tools)} tool(s))")
agent = Agent(
client=client,
instructions="You are a research assistant. Use the available tools to answer questions.",
tools=toolbox,
)
print(f"User: {QUERY}")
result = await agent.run(QUERY)
print(f"Result: {result}\n")
async def combine_toolboxes() -> None:
"""Alternative flow: combine the tools from multiple Foundry toolboxes."""
client = FoundryChatClient(
credential=AzureCliCredential(),
project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
model=os.environ["FOUNDRY_MODEL"],
)
# Comment out if the toolboxes already exist in your Foundry project.
create_sample_toolbox(TOOLBOX_NAME)
create_sample_toolbox(SECOND_TOOLBOX_NAME)
toolbox_a = await client.get_toolbox(TOOLBOX_NAME, version=TOOLBOX_VERSION)
toolbox_b = await client.get_toolbox(SECOND_TOOLBOX_NAME, version=SECOND_TOOLBOX_VERSION)
print(
"Loaded toolboxes: "
f"{toolbox_a.name}@{toolbox_a.version} ({len(toolbox_a.tools)} tool(s)), "
f"{toolbox_b.name}@{toolbox_b.version} ({len(toolbox_b.tools)} tool(s))"
)
agent = Agent(
client=client,
instructions="You are a research assistant. Use all available tools to answer questions.",
tools=[toolbox_a, toolbox_b],
)
print(f"User: {QUERY}")
result = await agent.run(QUERY)
print(f"Combined-toolbox result: {result}\n")
async def select_tools_from_toolbox() -> None:
"""Alternative flow: keep only a subset of toolbox tools before agent creation."""
client = FoundryChatClient(
credential=AzureCliCredential(),
project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
model=os.environ["FOUNDRY_MODEL"],
)
# Comment out if the toolbox already exists in your Foundry project.
create_sample_toolbox(TOOLBOX_NAME)
toolbox = await client.get_toolbox(TOOLBOX_NAME, version=TOOLBOX_VERSION)
print(f"Loaded toolbox {toolbox.name}@{toolbox.version} ({len(toolbox.tools)} tool(s))")
selected_tools = select_toolbox_tools(
toolbox,
include_types=["code_interpreter", "mcp"],
)
print(f"Selected {len(selected_tools)} toolbox tools for the agent")
agent = Agent(
client=client,
instructions="You are a research assistant. Use only the selected toolbox tools.",
tools=selected_tools,
)
print(f"User: {QUERY}")
result = await agent.run(QUERY)
print(f"Selected-toolbox result: {result}\n")
if __name__ == "__main__":
asyncio.run(main())
# asyncio.run(combine_toolboxes())
# asyncio.run(select_tools_from_toolbox())
@@ -0,0 +1,118 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from collections.abc import Callable
from typing import Any
from agent_framework import Agent, MCPStreamableHTTPTool
from agent_framework.foundry import FoundryChatClient
from azure.core.credentials import TokenCredential
from azure.identity import AzureCliCredential, DefaultAzureCredential, get_bearer_token_provider
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
"""
Foundry Toolbox via MAF ``MCPStreamableHTTPTool``
Instead of fetching the toolbox and fanning out individual tool specs, point
MAF's ``MCPStreamableHTTPTool`` at the toolbox's MCP endpoint. The agent
discovers and calls the toolbox's tools over MCP at runtime.
Prerequisites:
- A Microsoft Foundry project with a toolbox configured
- FOUNDRY_PROJECT_ENDPOINT and FOUNDRY_MODEL environment variables set
- FOUNDRY_TOOLBOX_ENDPOINT: the toolbox's MCP endpoint URL, e.g.
``https://<account>.services.ai.azure.com/api/projects/<project>/toolsets/<name>/mcp?api-version=v1``
- Azure CLI authentication (``az login``)
"""
# Must match the ``<name>`` segment of FOUNDRY_TOOLBOX_ENDPOINT.
TOOLBOX_NAME = "research_toolbox"
def create_sample_toolbox(name: str) -> str:
"""Create (or replace) a toolbox version in the Foundry project.
Toolboxes are normally configured in the Foundry portal or a deployment
script, not the application itself. This helper exists so the sample can
be run end-to-end without first setting a toolbox up by hand — delete any
existing toolbox under ``name``, then create a fresh version containing a
single MCP tool. Returns the created version identifier.
"""
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import MCPTool, Tool
from azure.core.exceptions import ResourceNotFoundError
with (
AzureCliCredential() as credential,
AIProjectClient(credential=credential, endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"]) as project_client,
):
try:
project_client.beta.toolboxes.delete(name)
print(f"Toolbox `{name}` deleted")
except ResourceNotFoundError:
pass
tools: list[Tool] = [
MCPTool(
server_label="api_specs",
server_url="https://gitmcp.io/Azure/azure-rest-api-specs",
require_approval="never",
)
]
created = project_client.beta.toolboxes.create_version(
name=name,
description="Toolbox version with MCP require_approval set to 'never'.",
tools=tools,
)
print(f"Created toolbox {created.name}@{created.version} ({len(created.tools)} tool(s))")
return created.version
def make_toolbox_header_provider(credential: TokenCredential) -> Callable[[dict[str, Any]], dict[str, str]]:
"""Build a header_provider that injects a fresh Azure AI bearer token on every MCP request."""
get_token = get_bearer_token_provider(credential, "https://ai.azure.com/.default")
def provide(_kwargs: dict[str, Any]) -> dict[str, str]:
return {
"Authorization": f"Bearer {get_token()}",
}
return provide
async def main() -> None:
credential = DefaultAzureCredential()
# Comment out if the toolbox already exists in your Foundry project.
create_sample_toolbox(TOOLBOX_NAME)
toolbox_tool = MCPStreamableHTTPTool(
name="foundry_toolbox",
description="Tools exposed by the configured Foundry toolbox",
url=os.environ["FOUNDRY_TOOLBOX_ENDPOINT"],
header_provider=make_toolbox_header_provider(credential),
load_prompts=False,
)
async with Agent(
client=FoundryChatClient(
project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
model=os.environ["FOUNDRY_MODEL"],
credential=credential,
),
instructions="You are a helpful assistant. Use the available toolbox tools to answer the user.",
tools=toolbox_tool,
) as agent:
query = "What tools do you have access to?"
print(f"User: {query}")
result = await agent.run(query)
print(f"Assistant: {result}")
if __name__ == "__main__":
asyncio.run(main())