agent-framework-azure-ai-py

User Query → AzureAIAgentsProvider → Azure AI Agent Service (Persistent)
 ↓
 Agent.run() / Agent.run_stream()
 ↓
 Tools: Functions | Hosted (Code/Search/Web) | MCP
 ↓
 AgentThread (conversation persistence)

export AZURE_AI_PROJECT_ENDPOINT="https:// .services.ai.azure.com/api/projects/ " # Required for all auth methods
export AZURE_AI_MODEL_DEPLOYMENT_NAME="gpt-4o-mini" # Required for all auth methods
export BING_CONNECTION_ID="your-bing-connection-id" # For web search
export AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production

from azure.identity.aio import AzureCliCredential, DefaultAzureCredential, ManagedIdentityCredential

# Development
credential = AzureCliCredential()

# Production
# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS= 
credential = DefaultAzureCredential(require_envvar=True)
# Or use a specific credential directly in production:
# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()

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="You are a helpful assistant.",
 )
 
 result = await agent.run("Hello!")
 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="City name to get weather for")],
) -> str:
 """Get the current weather for a location."""
 return f"Weather in {location}: 72°F, sunny"

def get_current_time() -> str:
 """Get the current UTC time."""
 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="You help with weather and time queries.",
 tools=[get_weather, get_current_time], # Pass functions directly
 )
 
 result = await agent.run("What's the weather in Seattle?")
 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="You can execute code, search files, and search the web.",
 tools=[
 HostedCodeInterpreterTool(),
 HostedWebSearchTool(name="Bing"),
 ],
 )
 
 result = await agent.run("Calculate the factorial of 20 in Python")
 print(result.text)

async def main():
 async with (
 AzureCliCredential() as credential,
 AzureAIAgentsProvider(credential=credential) as provider,
 ):
 agent = await provider.create_agent(
 name="StreamingAgent",
 instructions="You are a helpful assistant.",
 )
 
 print("Agent: ", end="", flush=True)
 async for chunk in agent.run_stream("Tell me a short story"):
 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="You are a helpful assistant.",
 tools=[get_weather],
 )
 
 # Create thread for conversation persistence
 thread = agent.get_new_thread()
 
 # First turn
 result1 = await agent.run("What's the weather in Seattle?", thread=thread)
 print(f"Agent: {result1.text}")
 
 # Second turn - context is maintained
 result2 = await agent.run("What about Portland?", thread=thread)
 print(f"Agent: {result2.text}")
 
 # Save thread ID for later resumption
 print(f"Conversation 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="Provide weather information in structured format.",
 response_format=WeatherResponse,
 )
 
 result = await agent.run("Weather in Seattle?")
 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="City name")],
) -> str:
 """Get weather for a location."""
 return f"Weather in {location}: 72°F, sunny"

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="You are a research assistant with multiple capabilities.",
 tools=[
 get_weather,
 HostedCodeInterpreterTool(),
 HostedWebSearchTool(name="Bing"),
 mcp_tool,
 ],
 )
 
 thread = agent.get_new_thread()
 
 # Non-streaming
 result = await agent.run(
 "Search for Python best practices and summarize",
 thread=thread,
 )
 print(f"Response: {result.text}")
 
 # Streaming
 print("\nStreaming: ", end="")
 async for chunk in agent.run_stream("Continue with examples", thread=thread):
 if chunk.text:
 print(chunk.text, end="", flush=True)
 print()
 
 # Structured output
 result = await agent.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())