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://<project>.services.ai.azure.com/api/projects/<project-id>" export AZURE_AI_MODEL_DEPLOYMENT_NAME="gpt-4o-mini" export BING_CONNECTION_ID="your-bing-connection-id" # 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!")
    print(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
    print(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)
    print(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