Build persistent agents on Azure AI Foundry using the Microsoft Agent Framework Python SDK.
\nUser Query → AzureAIAgentsProvider → Azure AI Agent Service (Persistent)\n ↓\n Agent.run() / Agent.run_stream()\n ↓\n Tools: Functions | Hosted (Code/Search/Web) | MCP\n ↓\n AgentThread (conversation persistence)\n# Full framework (recommended)\npip install agent-framework --pre\n\n# Or Azure-specific package only\npip install agent-framework-azure-ai --pre\nexport AZURE_AI_PROJECT_ENDPOINT=\"https://<project>.services.ai.azure.com/api/projects/<project-id>\" # Required for all auth methods\nexport AZURE_AI_MODEL_DEPLOYMENT_NAME=\"gpt-4o-mini\" # Required for all auth methods\nexport BING_CONNECTION_ID=\"your-bing-connection-id\" # For web search\nexport AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production\n\n\n\uD83D\uDD11 Two rules apply to every code sample below:
\n\n
\n- Prefer
\nDefaultAzureCredential. It works locally (Azure CLI / VS Code / Developer CLI) and in Azure (managed identity, workload identity) with no code change. Avoid connection strings, account/API keys — they bypass Entra audit and rotation.\n\n
\n- Local dev:
\nDefaultAzureCredentialworks as-is.- Production: set
\nAZURE_TOKEN_CREDENTIALS=prod(orAZURE_TOKEN_CREDENTIALS=<specific_credential>) to constrain the credential chain to production-safe credentials.- Wrap every client in a context manager so HTTP transports, sockets, and token caches are released deterministically:\n
\n\n
\n- Sync:
\nwith <Client>(...) as client:- Async:
\nasync with <Client>(...) as client:andasync with DefaultAzureCredential() as credential:(fromazure.identity.aio)Snippets may abbreviate this setup, but production code should always follow both rules.
\n
from azure.identity.aio import AzureCliCredential, DefaultAzureCredential, ManagedIdentityCredential\n\n# Development\ncredential = AzureCliCredential()\n\n# Production\n# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>\ncredential = DefaultAzureCredential(require_envvar=True)\n# Or use a specific credential directly in production:\n# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes\n# credential = ManagedIdentityCredential()\nimport asyncio\nfrom agent_framework.azure import AzureAIAgentsProvider\nfrom azure.identity.aio import AzureCliCredential\n\nasync def main():\n async with (\n AzureCliCredential() as credential,\n AzureAIAgentsProvider(credential=credential) as provider,\n ):\n agent = await provider.create_agent(\n name=\"MyAgent\",\n instructions=\"You are a helpful assistant.\",\n )\n \n result = await agent.run(\"Hello!\")\n print(result.text)\n\nasyncio.run(main())\nfrom typing import Annotated\nfrom pydantic import Field\nfrom agent_framework.azure import AzureAIAgentsProvider\nfrom azure.identity.aio import AzureCliCredential\n\ndef get_weather(\n location: Annotated[str, Field(description=\"City name to get weather for\")],\n) -> str:\n \"\"\"Get the current weather for a location.\"\"\"\n return f\"Weather in {location}: 72°F, sunny\"\n\ndef get_current_time() -> str:\n \"\"\"Get the current UTC time.\"\"\"\n from datetime import datetime, timezone\n return datetime.now(timezone.utc).strftime(\"%Y-%m-%d %H:%M:%S UTC\")\n\nasync def main():\n async with (\n AzureCliCredential() as credential,\n AzureAIAgentsProvider(credential=credential) as provider,\n ):\n agent = await provider.create_agent(\n name=\"WeatherAgent\",\n instructions=\"You help with weather and time queries.\",\n tools=[get_weather, get_current_time], # Pass functions directly\n )\n \n result = await agent.run(\"What's the weather in Seattle?\")\n print(result.text)\nfrom agent_framework import (\n HostedCodeInterpreterTool,\n HostedFileSearchTool,\n HostedWebSearchTool,\n)\nfrom agent_framework.azure import AzureAIAgentsProvider\nfrom azure.identity.aio import AzureCliCredential\n\nasync def main():\n async with (\n AzureCliCredential() as credential,\n AzureAIAgentsProvider(credential=credential) as provider,\n ):\n agent = await provider.create_agent(\n name=\"MultiToolAgent\",\n instructions=\"You can execute code, search files, and search the web.\",\n tools=[\n HostedCodeInterpreterTool(),\n HostedWebSearchTool(name=\"Bing\"),\n ],\n )\n \n result = await agent.run(\"Calculate the factorial of 20 in Python\")\n print(result.text)\nasync def main():\n async with (\n AzureCliCredential() as credential,\n AzureAIAgentsProvider(credential=credential) as provider,\n ):\n agent = await provider.create_agent(\n name=\"StreamingAgent\",\n instructions=\"You are a helpful assistant.\",\n )\n \n print(\"Agent: \", end=\"\", flush=True)\n async for chunk in agent.run_stream(\"Tell me a short story\"):\n if chunk.text:\n print(chunk.text, end=\"\", flush=True)\n print()\nfrom agent_framework.azure import AzureAIAgentsProvider\nfrom azure.identity.aio import AzureCliCredential\n\nasync def main():\n async with (\n AzureCliCredential() as credential,\n AzureAIAgentsProvider(credential=credential) as provider,\n ):\n agent = await provider.create_agent(\n name=\"ChatAgent\",\n instructions=\"You are a helpful assistant.\",\n tools=[get_weather],\n )\n \n # Create thread for conversation persistence\n thread = agent.get_new_thread()\n \n # First turn\n result1 = await agent.run(\"What's the weather in Seattle?\", thread=thread)\n print(f\"Agent: {result1.text}\")\n \n # Second turn - context is maintained\n result2 = await agent.run(\"What about Portland?\", thread=thread)\n print(f\"Agent: {result2.text}\")\n \n # Save thread ID for later resumption\n print(f\"Conversation ID: {thread.conversation_id}\")\nfrom pydantic import BaseModel, ConfigDict\nfrom agent_framework.azure import AzureAIAgentsProvider\nfrom azure.identity.aio import AzureCliCredential\n\nclass WeatherResponse(BaseModel):\n model_config = ConfigDict(extra=\"forbid\")\n \n location: str\n temperature: float\n unit: str\n conditions: str\n\nasync def main():\n async with (\n AzureCliCredential() as credential,\n AzureAIAgentsProvider(credential=credential) as provider,\n ):\n agent = await provider.create_agent(\n name=\"StructuredAgent\",\n instructions=\"Provide weather information in structured format.\",\n response_format=WeatherResponse,\n )\n \n result = await agent.run(\"Weather in Seattle?\")\n weather = WeatherResponse.model_validate_json(result.text)\n print(f\"{weather.location}: {weather.temperature}°{weather.unit}\")\n| Method | \nDescription | \n
|---|---|
create_agent() | \nCreate new agent on Azure AI service | \n
get_agent(agent_id) | \nRetrieve existing agent by ID | \n
as_agent(sdk_agent) | \nWrap SDK Agent object (no HTTP call) | \n
| Tool | \nImport | \nPurpose | \n
|---|---|---|
HostedCodeInterpreterTool | \nfrom agent_framework import HostedCodeInterpreterTool | \nExecute Python code | \n
HostedFileSearchTool | \nfrom agent_framework import HostedFileSearchTool | \nSearch vector stores | \n
HostedWebSearchTool | \nfrom agent_framework import HostedWebSearchTool | \nBing web search | \n
HostedMCPTool | \nfrom agent_framework import HostedMCPTool | \nService-managed MCP | \n
MCPStreamableHTTPTool | \nfrom agent_framework import MCPStreamableHTTPTool | \nClient-managed MCP | \n
import asyncio\nfrom typing import Annotated\nfrom pydantic import BaseModel, Field\nfrom agent_framework import (\n HostedCodeInterpreterTool,\n HostedWebSearchTool,\n MCPStreamableHTTPTool,\n)\nfrom agent_framework.azure import AzureAIAgentsProvider\nfrom azure.identity.aio import AzureCliCredential\n\n\ndef get_weather(\n location: Annotated[str, Field(description=\"City name\")],\n) -> str:\n \"\"\"Get weather for a location.\"\"\"\n return f\"Weather in {location}: 72°F, sunny\"\n\n\nclass AnalysisResult(BaseModel):\n summary: str\n key_findings: list[str]\n confidence: float\n\n\nasync def main():\n async with (\n AzureCliCredential() as credential,\n MCPStreamableHTTPTool(\n name=\"Docs MCP\",\n url=\"https://learn.microsoft.com/api/mcp\",\n ) as mcp_tool,\n AzureAIAgentsProvider(credential=credential) as provider,\n ):\n agent = await provider.create_agent(\n name=\"ResearchAssistant\",\n instructions=\"You are a research assistant with multiple capabilities.\",\n tools=[\n get_weather,\n HostedCodeInterpreterTool(),\n HostedWebSearchTool(name=\"Bing\"),\n mcp_tool,\n ],\n )\n \n thread = agent.get_new_thread()\n \n # Non-streaming\n result = await agent.run(\n \"Search for Python best practices and summarize\",\n thread=thread,\n )\n print(f\"Response: {result.text}\")\n \n # Streaming\n print(\"\\nStreaming: \", end=\"\")\n async for chunk in agent.run_stream(\"Continue with examples\", thread=thread):\n if chunk.text:\n print(chunk.text, end=\"\", flush=True)\n print()\n \n # Structured output\n result = await agent.run(\n \"Analyze findings\",\n thread=thread,\n response_format=AnalysisResult,\n )\n analysis = AnalysisResult.model_validate_json(result.text)\n print(f\"\\nConfidence: {analysis.confidence}\")\n\n\nif __name__ == \"__main__\":\n asyncio.run(main())\nasync with provider:tools= parameter (auto-converted to AIFunction)Annotated[type, Field(description=...)] for function parametersget_new_thread() for multi-turn conversationsHostedMCPTool for service-managed MCP, MCPStreamableHTTPTool for client-managedasync def handlers and async with throughout.with Client(...) as client: (sync) or async with Client(...) as client: (async). For async DefaultAzureCredential from azure.identity.aio, also use async with credential: so tokens and transports are cleaned up.