Agent Framework Azure 托管代理

使用 Microsoft Agent Framework Python SDK 在 Azure AI Foundry 上构建持久化代理。

架构

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

安装

bash

完整框架（推荐）

pip install agent-framework --pre

仅安装 Azure 特定包

pip install agent-framework-azure-ai --pre

环境变量

bash
export AZUREAIPROJECT_ENDPOINT=https://.services.ai.azure.com/api/projects/
export AZUREAIMODELDEPLOYMENTNAME=gpt-4o-mini
export BINGCONNECTIONID=your-bing-connection-id # 用于网页搜索

身份验证

python
from azure.identity.aio import AzureCliCredential, DefaultAzureCredential

开发环境

credential = AzureCliCredential()

生产环境

credential = DefaultAzureCredential()

核心工作流

基础代理

python
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())

带函数工具的代理

python
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 getcurrenttime() -> 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=[getweather, getcurrent_time], # 直接传递函数
)

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

带托管工具的代理

python
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)

流式响应

python
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()

对话线程

python
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.getnewthread()

# 第一轮
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})

结构化输出

python
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.modelvalidatejson(result.text)
print(f{weather.location}：{weather.temperature}°{weather.unit})

提供者方法

方法	描述
createagent()	在 Azure AI 服务上创建新代理
getagent(agent_id)

按ID检索现有代理 | | asagent(sdkagent) | 包装SDK代理对象（无HTTP调用） |

托管工具快速参考

工具	导入	用途
HostedCodeInterpreterTool	from agentframework import HostedCodeInterpreterTool	执行Python代码
HostedFileSearchTool

from agentframework import HostedFileSearchTool | 搜索向量存储 | | HostedWebSearchTool | from agent_framework import HostedWebSearchTool | Bing网页搜索 | | HostedMCPTool | from agent_framework import HostedMCPTool | 服务管理的MCP | | MCPStreamableHTTPTool | from agent_framework import MCPStreamableHTTPTool | 客户端管理的MCP |

完整示例

python
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=文档 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.getnewthread()

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

# 流式
print(\n流式：, end=)
async for