name: agent-framework-azure-ai-py
description: “Build persistent agents on Azure AI Foundry using the Microsoft Agent Framework Python SDK.”
risk: critical
source: community
date_added: “2026-02-27”
Azure 托管代理的 Agent Framework
使用 Microsoft Agent Framework Python SDK 在 Azure AI Foundry 上构建持久化代理。
架构
用户查询 → AzureAIAgentsProvider → Azure AI 代理服务（持久化） ↓ Agent.run() / Agent.run_stream() ↓ 工具：函数 | 托管（代码/搜索/Web）| MCP ↓ AgentThread（对话持久化）安装
# 完整框架（推荐）pipinstallagent-framework--pre# 或仅 Azure 特定包pipinstallagent-framework-azure-ai--pre环境变量
exportAZURE_AI_PROJECT_ENDPOINT="https://<project>.services.ai.azure.com/api/projects/<project-id>"exportAZURE_AI_MODEL_DEPLOYMENT_NAME="gpt-4o-mini"exportBING_CONNECTION_ID="your-bing-connection-id"# 用于网络搜索身份验证
fromazure.identity.aioimportAzureCliCredential,DefaultAzureCredential# 开发环境credential=AzureCliCredential()# 生产环境credential=DefaultAzureCredential()核心工作流
基本代理
importasynciofromagent_framework.azureimportAzureAIAgentsProviderfromazure.identity.aioimportAzureCliCredentialasyncdefmain():asyncwith(AzureCliCredential()ascredential,AzureAIAgentsProvider(credential=credential)asprovider,):agent=awaitprovider.create_agent(name="MyAgent",instructions="You are a helpful assistant.",)result=awaitagent.run("Hello!")print(result.text)asyncio.run(main())带函数工具的代理
fromtypingimportAnnotatedfrompydanticimportFieldfromagent_framework.azureimportAzureAIAgentsProviderfromazure.identity.aioimportAzureCliCredentialdefget_weather(location:Annotated[str,Field(description="City name to get weather for")],)->str:"""Get the current weather for a location."""returnf"Weather in{location}: 72°F, sunny"defget_current_time()->str:"""Get the current UTC time."""fromdatetimeimportdatetime,timezonereturndatetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")asyncdefmain():asyncwith(AzureCliCredential()ascredential,AzureAIAgentsProvider(credential=credential)asprovider,):agent=awaitprovider.create_agent(name="WeatherAgent",instructions="You help with weather and time queries.",tools=[get_weather,get_current_time],# Pass functions directly)result=awaitagent.run("What's the weather in Seattle?")print(result.text)带托管工具的代理
fromagent_frameworkimport(HostedCodeInterpreterTool,HostedFileSearchTool,HostedWebSearchTool,)fromagent_framework.azureimportAzureAIAgentsProviderfromazure.identity.aioimportAzureCliCredentialasyncdefmain():asyncwith(AzureCliCredential()ascredential,AzureAIAgentsProvider(credential=credential)asprovider,):agent=awaitprovider.create_agent(name="MultiToolAgent",instructions="You can execute code, search files, and search the web.",tools=[HostedCodeInterpreterTool(),HostedWebSearchTool(name="Bing"),],)result=awaitagent.run("Calculate the factorial of 20 in Python")print(result.text)流式响应
asyncdefmain():asyncwith(AzureCliCredential()ascredential,AzureAIAgentsProvider(credential=credential)asprovider,):agent=awaitprovider.create_agent(name="StreamingAgent",instructions="You are a helpful assistant.",)print("Agent: ",end="",flush=True)asyncforchunkinagent.run_stream("Tell me a short story"):ifchunk.text:print(chunk.text,end="",flush=True)print()对话线程
fromagent_framework.azureimportAzureAIAgentsProviderfromazure.identity.aioimportAzureCliCredentialasyncdefmain():asyncwith(AzureCliCredential()ascredential,AzureAIAgentsProvider(credential=credential)asprovider,):agent=awaitprovider.create_agent(name="ChatAgent",instructions="You are a helpful assistant.",tools=[get_weather],)# 创建线程以保持对话持久化thread=agent.get_new_thread()# 第一轮result1=awaitagent.run("What's the weather in Seattle?",thread=thread)print(f"Agent:{result1.text}")# 第二轮 - 上下文保持result2=awaitagent.run("What about Portland?",thread=thread)print(f"Agent:{result2.text}")# 保存线程 ID 以便后续恢复print(f"Conversation ID:{thread.conversation_id}")结构化输出
frompydanticimportBaseModel,ConfigDictfromagent_framework.azureimportAzureAIAgentsProviderfromazure.identity.aioimportAzureCliCredentialclassWeatherResponse(BaseModel):model_config=ConfigDict(extra="forbid")location:strtemperature:floatunit:strconditions:strasyncdefmain():asyncwith(AzureCliCredential()ascredential,AzureAIAgentsProvider(credential=credential)asprovider,):agent=awaitprovider.create_agent(name="StructuredAgent",instructions="Provide weather information in structured format.",response_format=WeatherResponse,)result=awaitagent.run("Weather in Seattle?")weather=WeatherResponse.model_validate_json(result.text)print(f"{weather.location}:{weather.temperature}°{weather.unit}")提供者方法
| 方法 | 描述 |
|---|---|
create_agent() | 在 Azure AI 服务上创建新代理 |
get_agent(agent_id) | 按 ID 检索现有代理 |
as_agent(sdk_agent) | 包装 SDK 代理对象（无 HTTP 调用） |
托管工具快速参考
| 工具 | 导入 | 用途 |
|---|---|---|
HostedCodeInterpreterTool | from agent_framework import HostedCodeInterpreterTool | 执行 Python 代码 |
HostedFileSearchTool | from agent_framework import HostedFileSearchTool | 搜索向量存储 |
HostedWebSearchTool | from agent_framework import HostedWebSearchTool | Bing 网络搜索 |
HostedMCPTool | from agent_framework import HostedMCPTool | 服务管理的 MCP |
MCPStreamableHTTPTool | from agent_framework import MCPStreamableHTTPTool | 客户端管理的 MCP |
完整示例
importasynciofromtypingimportAnnotatedfrompydanticimportBaseModel,Fieldfromagent_frameworkimport(HostedCodeInterpreterTool,HostedWebSearchTool,MCPStreamableHTTPTool,)fromagent_framework.azureimportAzureAIAgentsProviderfromazure.identity.aioimportAzureCliCredentialdefget_weather(location:Annotated[str,Field(description="City name")],)->str:"""Get weather for a location."""returnf"Weather in{location}: 72°F, sunny"classAnalysisResult(BaseModel):summary:strkey_findings:list[str]confidence:floatasyncdefmain():asyncwith(AzureCliCredential()ascredential,MCPStreamableHTTPTool(name="Docs MCP",url="https://learn.microsoft.com/api/mcp",)asmcp_tool,AzureAIAgentsProvider(credential=credential)asprovider,):agent=awaitprovider.create_agent(name="ResearchAssistant",instructions="You are a research assistant with multiple capabilities.",tools=[get_weather,HostedCodeInterpreterTool(),HostedWebSearchTool(name="Bing"),mcp_tool,],)thread=agent.get_new_thread()# 非流式result=awaitagent.run("Search for Python best practices and summarize",thread=thread,)print(f"Response:{result.text}")# 流式print("\nStreaming: ",end="")asyncforchunkinagent.run_stream("Continue with examples",thread=thread):ifchunk.text:print(chunk.text,end="",flush=True)print()# 结构化输出result=awaitagent.run("Analyze findings",thread=thread,response_format=AnalysisResult,)analysis=AnalysisResult.model_validate_json(result.text)print(f"\nConfidence:{analysis.confidence}")if__name__=="__main__":asyncio.run(main())约定
- 始终使用异步上下文管理器：
async with provider: - 将函数直接传递给
tools=参数（自动转换为 AIFunction） - 对函数参数使用
Annotated[type, Field(description=...)] - 对多轮对话使用
get_new_thread() - 对服务管理的 MCP 优先使用
HostedMCPTool，对客户端管理的 MCP 使用MCPStreamableHTTPTool
参考文件
- references/tools.md：详细托管工具模式
- references/mcp.md：MCP 集成（托管 + 本地）
- references/threads.md：线程和对话管理
- references/advanced.md：OpenAPI、引用、结构化输出
何时使用
本技能适用于执行概述中描述的工作流或操作。
限制
- 仅当任务明确匹配上述范围时使用本技能。
- 不要将输出视为特定环境验证、测试或专家审查的替代品。
- 如果缺少必需的输入、权限、安全边界或成功标准，请停下来要求澄清。
