name: agent-framework-azure-ai-py
description: 使用 Microsoft Agent Framework Python SDK (agent-framework-azure-ai) 构建 Azure AI Foundry 智能体。适用于通过 AzureAIAgentsProvider 创建持久化智能体、使用托管工具（代码解释器、文件搜索、网络搜索）、集成 MCP 服务器、管理对话线程或实现流式响应。涵盖函数工具、结构化输出和多工具智能体。
package: agent-framework-azure-ai

Agent Framework Azure 托管智能体

使用 Microsoft Agent Framework Python SDK 在 Azure AI Foundry 上构建持久化智能体。

架构

用户查询 → AzureAIAgentsProvider → Azure AI 智能体服务（持久化）
                    ↓
              Agent.run() / Agent.run_stream()
                    ↓
              工具：函数 | 托管（代码/搜索/网络）| MCP
                    ↓
              AgentThread（对话持久化）

安装

# 完整框架（推荐）
pip install agent-framework --pre

# 或仅安装 Azure 专用包
pip install agent-framework-azure-ai --pre

环境变量

export AZURE_AI_PROJECT_ENDPOINT="https://<project>.services.ai.azure.com/api/projects/<project-id>"
export AZURE_AI_MODEL_DEPLOYMENT_NAME="gpt-4o-mini"
export BING_CONNECTION_ID="your-bing-connection-id"  # 用于网络搜索

身份验证

from azure.identity.aio import AzureCliCredential, DefaultAzureCredential

# 开发环境
credential = AzureCliCredential()

# 生产环境
credential = DefaultAzureCredential()

核心工作流

基础智能体

import asyncio
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential

async def main():
    async with (
        AzureCliCredential() as credential,
        AzureAIAgentsProvider(credential=credential) as provider,
    ):
        agent = await provider.create_agent(
            name="MyAgent",
            instructions="你是一个乐于助人的助手。",
        )

        result = await agent.run("你好！")
        print(result.text)

asyncio.run(main())

带函数工具的智能体

from typing import Annotated
from pydantic import Field
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential

def get_weather(
    location: Annotated[str, Field(description="要查询天气的城市名称")],
) -> str:
    """获取指定地点的当前天气。"""
    return f"{location}的天气：72°F，晴朗"

def get_current_time() -> str:
    """获取当前 UTC 时间。"""
    from datetime import datetime, timezone
    return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")

async def main():
    async with (
        AzureCliCredential() as credential,
        AzureAIAgentsProvider(credential=credential) as provider,
    ):
        agent = await provider.create_agent(
            name="WeatherAgent",
            instructions="你帮助处理天气和时间查询。",
            tools=[get_weather, get_current_time],  # 直接传递函数
        )

        result = await agent.run("西雅图的天气怎么样？")
        print(result.text)

带托管工具的智能体

from agent_framework import (
    HostedCodeInterpreterTool,
    HostedFileSearchTool,
    HostedWebSearchTool,
)
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential

async def main():
    async with (
        AzureCliCredential() as credential,
        AzureAIAgentsProvider(credential=credential) as provider,
    ):
        agent = await provider.create_agent(
            name="MultiToolAgent",
            instructions="你可以执行代码、搜索文件和搜索网络。",
            tools=[
                HostedCodeInterpreterTool(),
                HostedWebSearchTool(name="Bing"),
            ],
        )

        result = await agent.run("用 Python 计算 20 的阶乘")
        print(result.text)

流式响应

async def main():
    async with (
        AzureCliCredential() as credential,
        AzureAIAgentsProvider(credential=credential) as provider,
    ):
        agent = await provider.create_agent(
            name="StreamingAgent",
            instructions="你是一个乐于助人的助手。",
        )

        print("智能体：", end="", flush=True)
        async for chunk in agent.run_stream("给我讲一个短故事"):
            if chunk.text:
                print(chunk.text, end="", flush=True)
        print()

对话线程

from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential

async def main():
    async with (
        AzureCliCredential() as credential,
        AzureAIAgentsProvider(credential=credential) as provider,
    ):
        agent = await provider.create_agent(
            name="ChatAgent",
            instructions="你是一个乐于助人的助手。",
            tools=[get_weather],
        )

        # 创建线程以实现对话持久化
        thread = agent.get_new_thread()

        # 第一轮对话
        result1 = await agent.run("西雅图的天气怎么样？", thread=thread)
        print(f"智能体：{result1.text}")

        # 第二轮对话 - 上下文被保留
        result2 = await agent.run("波特兰呢？", thread=thread)
        print(f"智能体：{result2.text}")

        # 保存线程 ID 以便后续恢复对话
        print(f"对话 ID：{thread.conversation_id}")

结构化输出

from pydantic import BaseModel, ConfigDict
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential

class WeatherResponse(BaseModel):
    model_config = ConfigDict(extra="forbid")

    location: str
    temperature: float
    unit: str
    conditions: str

async def main():
    async with (
        AzureCliCredential() as credential,
        AzureAIAgentsProvider(credential=credential) as provider,
    ):
        agent = await provider.create_agent(
            name="StructuredAgent",
            instructions="以结构化格式提供天气信息。",
            response_format=WeatherResponse,
        )

        result = await agent.run("西雅图的天气？")
        weather = WeatherResponse.model_validate_json(result.text)
        print(f"{weather.location}：{weather.temperature}°{weather.unit}")

提供者方法

托管工具速查表

完整示例

import asyncio
from typing import Annotated
from pydantic import BaseModel, Field
from agent_framework import (
    HostedCodeInterpreterTool,
    HostedWebSearchTool,
    MCPStreamableHTTPTool,
)
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential


def get_weather(
    location: Annotated[str, Field(description="城市名称")],
) -> str:
    """获取指定地点的天气。"""
    return f"{location}的天气：72°F，晴朗"


class AnalysisResult(BaseModel):
    summary: str
    key_findings: list[str]
    confidence: float


async def main():
    async with (
        AzureCliCredential() as credential,
        MCPStreamableHTTPTool(
            name="Docs MCP",
            url="https://learn.microsoft.com/api/mcp",
        ) as mcp_tool,
        AzureAIAgentsProvider(credential=credential) as provider,
    ):
        agent = await provider.create_agent(
            name="ResearchAssistant",
            instructions="你是一个具备多种能力的研究助手。",
            tools=[
                get_weather,
                HostedCodeInterpreterTool(),
                HostedWebSearchTool(name="Bing"),
                mcp_tool,
            ],
        )

        thread = agent.get_new_thread()

        # 非流式响应
        result = await agent.run(
            "搜索 Python 最佳实践并总结",
            thread=thread,
        )
        print(f"响应：{result.text}")

        # 流式响应
        print("\n流式响应：", end="")
        async for chunk in agent.run_stream("继续举例说明", thread=thread):
            if chunk.text:
                print(chunk.text, end="", flush=True)
        print()

        # 结构化输出
        result = await agent.run(
            "分析发现结果",
            thread=thread,
            response_format=AnalysisResult,
        )
        analysis = AnalysisResult.model_validate_json(result.text)
        print(f"\n置信度：{analysis.confidence}")


if __name__ == "__main__":
    asyncio.run(main())

约定

• 始终使用异步上下文管理器：async with provider:
• 直接将函数传递给 tools= 参数（自动转换为 AIFunction）
• 使用 Annotated[type, Field(description=...)] 定义函数参数
• 使用 get_new_thread() 进行多轮对话
• 服务托管 MCP 优先使用 HostedMCPTool，客户端托管使用 MCPStreamableHTTPTool

参考文件

• references/tools.md：详细的托管工具模式
• references/mcp.md：MCP 集成（托管 + 本地）
• references/threads.md：线程和对话管理
• references/advanced.md：OpenAPI、引用、结构化输出