Agent Framework Azure Hosted Agents Build persistent agents on Azure AI Foundry using the Microsoft Agent Framework Python SDK. Architecture User Query → AzureAIAgentsProvider → Azure AI Agent Service (Persistent) ↓ Agent.run() / Agent.run_stream() ↓ Tools: Functions | Hosted (Code/Search/Web) | MCP ↓ AgentThread (conversation persistence) Installation
Full framework (recommended)
pip install agent-framework --pre
Or Azure-specific package only
pip
install
agent-framework-azure-ai
--pre
Environment Variables
export
AZURE_AI_PROJECT_ENDPOINT
=
"https://
For web search
Authentication from azure . identity . aio import AzureCliCredential , DefaultAzureCredential
Development
credential
AzureCliCredential ( )
Production
credential
- DefaultAzureCredential
- (
- )
- Core Workflow
- Basic Agent
- 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!"
- )
- (
- result
- .
- text
- )
- asyncio
- .
- run
- (
- main
- (
- )
- )
- Agent with Function Tools
- 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 ) Agent with Hosted Tools 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 ) Streaming Responses 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 ( ) Conversation Threads 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
- (
- f"Conversation ID:
- {
- thread
- .
- conversation_id
- }
- "
- )
- Structured Outputs
- 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
- )
- (
- f"
- {
- weather
- .
- location
- }
- :
- {
- weather
- .
- temperature
- }
- °
- {
- weather
- .
- unit
- }
- "
- )
- Provider Methods
- Method
- Description
- create_agent()
- Create new agent on Azure AI service
- get_agent(agent_id)
- Retrieve existing agent by ID
- as_agent(sdk_agent)
- Wrap SDK Agent object (no HTTP call)
- Hosted Tools Quick Reference
- Tool
- Import
- Purpose
- HostedCodeInterpreterTool
- from agent_framework import HostedCodeInterpreterTool
- Execute Python code
- HostedFileSearchTool
- from agent_framework import HostedFileSearchTool
- Search vector stores
- HostedWebSearchTool
- from agent_framework import HostedWebSearchTool
- Bing web search
- HostedMCPTool
- from agent_framework import HostedMCPTool
- Service-managed MCP
- MCPStreamableHTTPTool
- from agent_framework import MCPStreamableHTTPTool
- Client-managed MCP
- Complete Example
- 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 ( ) ) Conventions Always use async context managers: async with provider: Pass functions directly to tools= parameter (auto-converted to AIFunction) Use Annotated[type, Field(description=...)] for function parameters Use get_new_thread() for multi-turn conversations Prefer HostedMCPTool for service-managed MCP, MCPStreamableHTTPTool for client-managed Reference Files references/tools.md: Detailed hosted tool patterns references/mcp.md: MCP integration (hosted + local) references/threads.md: Thread and conversation management references/advanced.md: OpenAPI, citations, structured outputs When to Use This skill is applicable to execute the workflow or actions described in the overview.