agent-framework-azure-ai-py | Apiara Skills$npx clawhub@latest install agent-framework-azure-ai-py Overview
# 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
```bash
# Full framework (recommended)
pip install agent-framework --pre
# Or Azure-specific package only
pip install agent-framework-azure-ai --pre
```
Environment Variables
```bash
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
```python
from azure.identity.aio import AzureCliCredential, DefaultAzureCredential
# Development
credential = AzureCliCredential()
# Production
credential = DefaultAzureCredential()
Core Workflow
Basic Agent
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
agent = await provider.create_agent(
instructions="You are a helpful assistant.",
result = await agent.run("Hello!")
Agent with Function Tools
from typing import Annotated
from pydantic import Field
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
location: Annotated[str, Field(description="City name to get weather for")],
"""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")
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
agent = await provider.create_agent(
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?")
Agent with Hosted Tools
from agent_framework import (
HostedCodeInterpreterTool,
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
agent = await provider.create_agent(
instructions="You can execute code, search files, and search the web.",
HostedCodeInterpreterTool(),
HostedWebSearchTool(name="Bing"),
result = await agent.run("Calculate the factorial of 20 in Python")
Streaming Responses
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
agent = await provider.create_agent(
instructions="You are a helpful assistant.",
print("Agent: ", end="", flush=True)
async for chunk in agent.run_stream("Tell me a short story"):
print(chunk.text, end="", flush=True)
Conversation Threads
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
agent = await provider.create_agent(
instructions="You are a helpful assistant.",
# Create thread for conversation persistence
thread = agent.get_new_thread()
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")
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
agent = await provider.create_agent(
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
| `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
from typing import Annotated
from pydantic import BaseModel, Field
from agent_framework import (
HostedCodeInterpreterTool,
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
location: Annotated[str, Field(description="City name")],
"""Get weather for a location."""
return f"Weather in {location}: 72°F, sunny"
class AnalysisResult(BaseModel):
AzureCliCredential() as credential,
url="https://learn.microsoft.com/api/mcp",
AzureAIAgentsProvider(credential=credential) as provider,
agent = await provider.create_agent(
name="ResearchAssistant",
instructions="You are a research assistant with multiple capabilities.",
HostedCodeInterpreterTool(),
HostedWebSearchTool(name="Bing"),
thread = agent.get_new_thread()
result = await agent.run(
"Search for Python best practices and summarize",
print(f"Response: {result.text}")
print("\nStreaming: ", end="")
async for chunk in agent.run_stream("Continue with examples", thread=thread):
print(chunk.text, end="", flush=True)
result = await agent.run(
response_format=AnalysisResult,
analysis = AnalysisResult.model_validate_json(result.text)
print(f"\nConfidence: {analysis.confidence}")
if __name__ == "__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.