# SkillPatch skill: agent-framework-azure-ai-py

This skill guides agents in building persistent Azure AI Foundry agents using the Microsoft Agent Framework Python SDK. It covers authentication with DefaultAzureCredential, environment setup, core agent creation workflows, function tools, hosted tools (code interpreter, file search, web search), MCP server integration, conversation thread management, and streaming responses.

You (the agent) were given this URL and asked to use this skill. This is a **public** skill — no SkillPatch account, API key, or prior setup is required. Two ways to use it:

**1. Use it right now** — the skill's complete file tree (SKILL.md + all reference files) is inlined below; read `SKILL.md` first, then follow it, consulting the other files as it directs.

**2. Install the exact package onto disk** (recommended if you can run a shell — this reproduces the skill byte-for-byte, including any binary assets that can't be inlined):

```bash
mkdir -p .claude/skills/agent-framework-azure-ai-py
curl -sSL https://skillpatch.dev/install_skill/agent-framework-azure-ai-py | tar -xz -C .claude/skills/
```

(`.claude/skills/` is Claude Code's convention; use whatever directory your agent loads skills from.)


---

## Skill files (5)

- `SKILL.md`
- `references/advanced.md`
- `references/mcp.md`
- `references/threads.md`
- `references/tools.md`


### `SKILL.md`

````markdown
---
name: agent-framework-azure-ai-py
description: Build Azure AI Foundry agents using the Microsoft Agent Framework Python SDK (agent-framework-azure-ai). Use when creating persistent agents with AzureAIAgentsProvider, using hosted tools (code interpreter, file search, web search), integrating MCP servers, managing conversation threads, or implementing streaming responses. Covers function tools, structured outputs, and multi-tool agents.
license: MIT
metadata:
  author: Microsoft
  version: "1.0.0"
  package: agent-framework-azure-ai
---

# 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>"  # 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
```

## Authentication & Lifecycle

> **🔑 Two rules apply to every code sample below:**
>
> 1. **Prefer `DefaultAzureCredential`.** 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.
>    - Local dev: `DefaultAzureCredential` works as-is.
>    - Production: set `AZURE_TOKEN_CREDENTIALS=prod` (or `AZURE_TOKEN_CREDENTIALS=<specific_credential>`) to constrain the credential chain to production-safe credentials.
> 2. **Wrap every client in a context manager** so HTTP transports, sockets, and token caches are released deterministically:
>    - Sync: `with <Client>(...) as client:`
>    - Async: `async with <Client>(...) as client:` **and** `async with DefaultAzureCredential() as credential:` (from `azure.identity.aio`)
>
> Snippets may abbreviate this setup, but production code should always follow both rules.

```python
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=<specific_credential>
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()
```

## Core Workflow

### Basic Agent

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

## Best Practices

1. **This SDK is async-first** — use `async def` handlers and `async with` throughout.
2. **Always use context managers for clients and async credentials.** Wrap every client in `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.

## Reference Files

- [references/tools.md](references/tools.md): Detailed hosted tool patterns
- [references/mcp.md](references/mcp.md): MCP integration (hosted + local)
- [references/threads.md](references/threads.md): Thread and conversation management
- [references/advanced.md](references/advanced.md): OpenAPI, citations, structured outputs

````


### `references/advanced.md`

````markdown
# Advanced Patterns Reference

Advanced patterns including structured outputs, OpenAPI tools, file handling, and more.

## Structured Outputs with Pydantic

### Basic Response Format

```python
from pydantic import BaseModel, ConfigDict
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential

class MovieRecommendation(BaseModel):
    model_config = ConfigDict(extra="forbid")  # Strict validation
    
    title: str
    year: int
    genre: str
    rating: float
    summary: str

async with (
    AzureCliCredential() as credential,
    AzureAIAgentsProvider(credential=credential) as provider,
):
    agent = await provider.create_agent(
        name="MovieAgent",
        instructions="Recommend movies based on user preferences.",
        response_format=MovieRecommendation,  # Set at creation
    )
    
    result = await agent.run("Recommend a sci-fi movie")
    movie = MovieRecommendation.model_validate_json(result.text)
    print(f"{movie.title} ({movie.year}) - {movie.rating}/10")
```

### Complex Nested Structures

```python
from pydantic import BaseModel, ConfigDict, Field
from typing import Optional

class Address(BaseModel):
    model_config = ConfigDict(extra="forbid")
    street: str
    city: str
    country: str
    postal_code: Optional[str] = None

class Person(BaseModel):
    model_config = ConfigDict(extra="forbid")
    name: str
    age: int
    email: str
    address: Address
    hobbies: list[str] = Field(default_factory=list)

class TeamResponse(BaseModel):
    model_config = ConfigDict(extra="forbid")
    team_name: str
    members: list[Person]
    total_members: int

agent = await provider.create_agent(
    name="TeamGenerator",
    instructions="Generate fictional team data.",
    response_format=TeamResponse,
)

result = await agent.run("Create a team of 3 software developers")
team = TeamResponse.model_validate_json(result.text)
for member in team.members:
    print(f"- {member.name}, {member.age}, {member.address.city}")
```

### Runtime Response Format Override

```python
class QuickAnswer(BaseModel):
    answer: str
    confidence: float

class DetailedAnalysis(BaseModel):
    summary: str
    key_points: list[str]
    recommendations: list[str]
    sources: list[str]

# Agent created without default response format
agent = await provider.create_agent(
    name="FlexibleAgent",
    instructions="Provide information in the requested format.",
)

# Quick answer format
quick_result = await agent.run(
    "What is Python?",
    response_format=QuickAnswer,
)

# Detailed analysis format (same agent)
detailed_result = await agent.run(
    "Analyze the benefits of microservices architecture",
    response_format=DetailedAnalysis,
)
```

---

## OpenAPI Tools

Integrate external APIs using OpenAPI specifications.

### Basic OpenAPI Integration

```python
from agent_framework import OpenAPITool
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential

# OpenAPI spec can be URL or inline dict
openapi_spec = {
    "openapi": "3.0.0",
    "info": {"title": "Weather API", "version": "1.0.0"},
    "paths": {
        "/weather/{city}": {
            "get": {
                "operationId": "getWeather",
                "summary": "Get weather for a city",
                "parameters": [
                    {
                        "name": "city",
                        "in": "path",
                        "required": True,
                        "schema": {"type": "string"}
                    }
                ],
                "responses": {
                    "200": {
                        "description": "Weather data",
                        "content": {
                            "application/json": {
                                "schema": {
                                    "type": "object",
                                    "properties": {
                                        "temperature": {"type": "number"},
                                        "conditions": {"type": "string"}
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
    }
}

async with (
    AzureCliCredential() as credential,
    AzureAIAgentsProvider(credential=credential) as provider,
):
    agent = await provider.create_agent(
        name="WeatherAPIAgent",
        instructions="Use the weather API to answer weather questions.",
        tools=OpenAPITool(
            name="WeatherAPI",
            spec=openapi_spec,
            base_url="https://api.weather.example.com",
        ),
    )
```

### OpenAPI with Authentication

```python
from agent_framework import OpenAPITool

openapi_tool = OpenAPITool(
    name="SecureAPI",
    spec="https://api.example.com/openapi.json",
    base_url="https://api.example.com",
    headers={
        "Authorization": "Bearer your-api-key",
        "X-API-Version": "2024-01",
    },
)
```

---

## File Generation and Handling

### Code Interpreter File Output

```python
from agent_framework import HostedCodeInterpreterTool
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential

async with (
    AzureCliCredential() as credential,
    AzureAIAgentsProvider(credential=credential) as provider,
):
    agent = await provider.create_agent(
        name="DataAnalyst",
        instructions="Analyze data and create visualizations.",
        tools=HostedCodeInterpreterTool(),
    )
    
    result = await agent.run(
        "Create a bar chart of sales data: Q1=100, Q2=150, Q3=120, Q4=200. Save as PNG."
    )
    
    # Check for generated files in the response
    print(result.text)
    
    # Files generated by code interpreter are typically referenced in the response
    # and can be downloaded via the files API
```

### Working with File IDs

```python
from azure.ai.agents.aio import AgentsClient

async with (
    AzureCliCredential() as credential,
    AgentsClient(endpoint=endpoint, credential=credential) as agents_client,
    AzureAIAgentsProvider(agents_client=agents_client) as provider,
):
    # Upload a file
    from pathlib import Path
    
    file = await agents_client.files.upload(
        file_path=Path("data/sales.csv"),
        purpose="agents"
    )
    print(f"Uploaded file ID: {file.id}")
    
    # Use file with code interpreter
    from agent_framework import HostedCodeInterpreterTool, HostedFileContent
    
    agent = await provider.create_agent(
        name="CSVAnalyst",
        instructions="Analyze the provided CSV file.",
        tools=HostedCodeInterpreterTool(
            inputs=[HostedFileContent(file_id=file.id)]
        ),
    )
    
    result = await agent.run("Summarize the data in the uploaded file")
```

---

## Citations and Source Attribution

### Enabling Citations

```python
agent = await provider.create_agent(
    name="ResearchAgent",
    instructions="""Answer questions using the knowledge base.
    
    IMPORTANT: Always cite your sources using this format:
    【message_idx:search_idx†source_name】
    
    Example: "Azure Functions supports Python【1:0†azure-docs】"
    """,
    tools=[
        HostedFileSearchTool(inputs=[...]),
    ],
)
```

### Parsing Citations

```python
import re

def parse_citations(text: str) -> list[dict]:
    """Extract citations from agent response."""
    pattern = r'【(\d+):(\d+)†([^】]+)】'
    citations = []
    
    for match in re.finditer(pattern, text):
        citations.append({
            "message_idx": int(match.group(1)),
            "search_idx": int(match.group(2)),
            "source": match.group(3),
        })
    
    return citations

result = await agent.run("What is Azure Functions?")
citations = parse_citations(result.text)
for cite in citations:
    print(f"Source: {cite['source']}")
```

---

## Provider Configuration Options

### Custom Model and Endpoint

```python
from agent_framework.azure import AzureAIAgentsProvider

provider = AzureAIAgentsProvider(
    credential=credential,
    project_endpoint="https://my-project.services.ai.azure.com/api/projects/my-project-id",
    model_deployment_name="gpt-4o",  # Override default model
)
```

### Using Existing AgentsClient

```python
from azure.ai.agents.aio import AgentsClient
from agent_framework.azure import AzureAIAgentsProvider

# Create and configure client separately
agents_client = AgentsClient(
    endpoint=endpoint,
    credential=credential,
)

# Pass to provider
provider = AzureAIAgentsProvider(agents_client=agents_client)
```

---

## Agent Lifecycle Management

### Retrieving Existing Agents

```python
async with (
    AzureCliCredential() as credential,
    AzureAIAgentsProvider(credential=credential) as provider,
):
    # Create agent and save ID
    agent = await provider.create_agent(
        name="PersistentAgent",
        instructions="You remember everything.",
    )
    agent_id = agent.id  # Save this
    
    # Later: retrieve the same agent
    same_agent = await provider.get_agent(agent_id=agent_id)
```

### Wrapping SDK Agents

```python
from azure.ai.agents.aio import AgentsClient

async with (
    AzureCliCredential() as credential,
    AgentsClient(endpoint=endpoint, credential=credential) as agents_client,
    AzureAIAgentsProvider(agents_client=agents_client) as provider,
):
    # Get agent via SDK
    sdk_agent = await agents_client.get_agent("agent-id")
    
    # Wrap as ChatAgent (no HTTP call)
    agent = provider.as_agent(sdk_agent)
    
    # Now use with agent framework
    result = await agent.run("Hello!")
```

---

## Error Handling Patterns

### Graceful Degradation

```python
from agent_framework import HostedWebSearchTool, HostedCodeInterpreterTool

async def run_with_fallback(agent, query: str, thread=None):
    """Run query with fallback on tool failures."""
    try:
        result = await agent.run(query, thread=thread)
        return result.text
    except Exception as e:
        # Log the error
        print(f"Tool execution error: {e}")
        
        # Create fallback agent without tools
        fallback_agent = await provider.create_agent(
            name="FallbackAgent",
            instructions="Answer based on your knowledge only.",
        )
        result = await fallback_agent.run(query)
        return f"[Fallback response] {result.text}"
```

### Retry Logic

```python
import asyncio
from typing import Optional

async def run_with_retry(
    agent,
    query: str,
    thread=None,
    max_retries: int = 3,
    delay: float = 1.0,
) -> Optional[str]:
    """Run query with exponential backoff retry."""
    for attempt in range(max_retries):
        try:
            result = await agent.run(query, thread=thread)
            return result.text
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            wait_time = delay * (2 ** attempt)
            print(f"Attempt {attempt + 1} failed, retrying in {wait_time}s: {e}")
            await asyncio.sleep(wait_time)
    return None
```

---

## Performance Optimization

### Connection Reuse

```python
# ✅ Good: Reuse provider and client
async with (
    AzureCliCredential() as credential,
    AzureAIAgentsProvider(credential=credential) as provider,
):
    # Create multiple agents with same provider
    agent1 = await provider.create_agent(name="Agent1", instructions="...")
    agent2 = await provider.create_agent(name="Agent2", instructions="...")
    
    # Process multiple requests
    for query in queries:
        await agent1.run(query)

# ❌ Bad: Creating new provider for each request
for query in queries:
    async with AzureAIAgentsProvider(credential=credential) as provider:
        agent = await provider.create_agent(...)
        await agent.run(query)
```

### Concurrent Requests

```python
import asyncio

async def process_queries(provider, queries: list[str]) -> list[str]:
    """Process multiple queries concurrently."""
    agent = await provider.create_agent(
        name="BatchAgent",
        instructions="Answer questions concisely.",
    )
    
    # Each query gets its own thread
    async def process_one(query: str) -> str:
        thread = agent.get_new_thread()
        result = await agent.run(query, thread=thread)
        return result.text
    
    results = await asyncio.gather(*[process_one(q) for q in queries])
    return results
```

---

## Debugging and Logging

### Enable Verbose Logging

```python
import logging

# Enable Azure SDK logging
logging.basicConfig(level=logging.DEBUG)
azure_logger = logging.getLogger("azure")
azure_logger.setLevel(logging.DEBUG)

# Enable agent framework logging
af_logger = logging.getLogger("agent_framework")
af_logger.setLevel(logging.DEBUG)
```

### Inspecting Tool Calls in Streaming

```python
from agent_framework import AgentResponseUpdate

async for chunk in agent.run_stream("Calculate something"):
    if isinstance(chunk, AgentResponseUpdate):
        if chunk.tool_calls:
            for tool_call in chunk.tool_calls:
                print(f"[DEBUG] Tool: {tool_call.name}")
                print(f"[DEBUG] Args: {tool_call.arguments}")
    if chunk.text:
        print(chunk.text, end="", flush=True)
```

````


### `references/mcp.md`

````markdown
# MCP Integration Reference

Model Context Protocol (MCP) integration patterns for Azure AI agents.

## Overview

The Agent Framework supports two MCP tool types:

| Tool | Management | Use Case |
|------|------------|----------|
| `HostedMCPTool` | Service-managed | MCP servers the Azure AI service connects to |
| `MCPStreamableHTTPTool` | Client-managed | MCP servers your code connects to |

---

## HostedMCPTool (Service-Managed)

The Azure AI service manages the MCP connection lifecycle.

### Basic Usage

```python
from agent_framework import HostedMCPTool
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential

async with (
    AzureCliCredential() as credential,
    AzureAIAgentsProvider(credential=credential) as provider,
):
    agent = await provider.create_agent(
        name="DocsAgent",
        instructions="Answer questions using Microsoft documentation.",
        tools=HostedMCPTool(
            name="Microsoft Learn MCP",
            url="https://learn.microsoft.com/api/mcp",
            approval_mode="never_require",  # Don't ask for approval
        ),
    )
    
    result = await agent.run("How do I use Azure Functions?")
    print(result.text)
```

### With Allowed Tools Filter

Restrict which MCP tools the agent can use:

```python
mcp_tool = HostedMCPTool(
    name="Microsoft Learn MCP",
    url="https://learn.microsoft.com/api/mcp",
    approval_mode="never_require",
    allowed_tools=["microsoft_docs_search", "microsoft_docs_read"],  # Only these tools
)
```

### With Authentication Headers

```python
mcp_tool = HostedMCPTool(
    name="Private MCP Server",
    url="https://my-mcp-server.example.com/mcp",
    approval_mode="never_require",
    headers={
        "Authorization": "Bearer your-api-key",
        "X-Custom-Header": "custom-value",
    },
)
```

### Approval Modes

Control when tool execution requires user approval:

```python
# Never require approval (automatic execution)
mcp_tool = HostedMCPTool(
    name="Safe MCP",
    url="https://safe-mcp.example.com/mcp",
    approval_mode="never_require",
)

# Always require approval
mcp_tool = HostedMCPTool(
    name="Sensitive MCP",
    url="https://sensitive-mcp.example.com/mcp",
    approval_mode="always_require",
)

# Per-tool approval configuration
mcp_tool = HostedMCPTool(
    name="Mixed MCP",
    url="https://mcp.example.com/mcp",
    approval_mode={
        "always_require_approval": ["delete_resource", "modify_config"],
        "never_require_approval": ["search", "read"],
    },
)
```

---

## MCPStreamableHTTPTool (Client-Managed)

You manage the MCP connection lifecycle in your code.

### Basic Usage

```python
from agent_framework import MCPStreamableHTTPTool
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential

async with (
    AzureCliCredential() as credential,
    MCPStreamableHTTPTool(
        name="Microsoft Learn MCP",
        url="https://learn.microsoft.com/api/mcp",
    ) as mcp_tool,  # MUST use context manager
    AzureAIAgentsProvider(credential=credential) as provider,
):
    agent = await provider.create_agent(
        name="DocsAgent",
        instructions="Answer questions using the documentation.",
        tools=mcp_tool,
    )
    
    result = await agent.run("What is Azure AI Foundry?")
    print(result.text)
```

### With Custom HTTP Client

For authentication or custom headers:

```python
from httpx import AsyncClient
from agent_framework import MCPStreamableHTTPTool

# Create HTTP client with authentication
http_client = AsyncClient(
    headers={
        "Authorization": f"Bearer {github_pat}",
        "User-Agent": "MyApp/1.0",
    },
    timeout=30.0,
)

async with (
    MCPStreamableHTTPTool(
        name="GitHub MCP",
        url="https://api.github.com/mcp",
        http_client=http_client,
    ) as github_mcp,
    AzureAIAgentsProvider(credential=credential) as provider,
):
    agent = await provider.create_agent(
        name="GitHubAgent",
        instructions="Help with GitHub operations.",
        tools=github_mcp,
    )
```

### Multiple MCP Tools

```python
async with (
    AzureCliCredential() as credential,
    MCPStreamableHTTPTool(
        name="Docs MCP",
        url="https://learn.microsoft.com/api/mcp",
    ) as docs_mcp,
    MCPStreamableHTTPTool(
        name="GitHub MCP",
        url="https://api.github.com/mcp",
        http_client=authenticated_client,
    ) as github_mcp,
    AzureAIAgentsProvider(credential=credential) as provider,
):
    agent = await provider.create_agent(
        name="MultiMCPAgent",
        instructions="You can search docs and interact with GitHub.",
        tools=[docs_mcp, github_mcp],
    )
```

---

## HostedMCPTool vs MCPStreamableHTTPTool

| Aspect | HostedMCPTool | MCPStreamableHTTPTool |
|--------|---------------|----------------------|
| Connection managed by | Azure AI Service | Your code |
| Context manager required | No | Yes |
| Best for | Public MCP servers | Authenticated/private servers |
| Connection lifecycle | Automatic | Manual (via context manager) |
| Headers | Via `headers` param | Via custom `http_client` |

### When to Use Which

**Use HostedMCPTool when:**
- MCP server is publicly accessible
- Azure AI service can reach the MCP endpoint
- You want simpler code (no context manager)
- Approval workflows are needed

**Use MCPStreamableHTTPTool when:**
- MCP server requires authentication
- MCP server is private/internal
- You need custom HTTP client configuration
- You want explicit connection control

---

## Combining MCP with Other Tools

```python
from typing import Annotated
from pydantic import Field
from agent_framework import (
    HostedCodeInterpreterTool,
    MCPStreamableHTTPTool,
)

def get_user_id() -> str:
    """Get the current user's ID."""
    return "user-123"

async with (
    AzureCliCredential() as credential,
    MCPStreamableHTTPTool(
        name="Company API MCP",
        url="https://internal-api.company.com/mcp",
    ) as company_mcp,
    AzureAIAgentsProvider(credential=credential) as provider,
):
    agent = await provider.create_agent(
        name="EnterpriseAgent",
        instructions="""You are an enterprise assistant that can:
        - Execute Python code for analysis
        - Access company internal APIs via MCP
        - Get user information
        
        Always verify user identity before accessing sensitive data.""",
        tools=[
            get_user_id,
            HostedCodeInterpreterTool(),
            company_mcp,
        ],
    )
```

---

## Error Handling for MCP

```python
try:
    async with MCPStreamableHTTPTool(
        name="MCP Server",
        url="https://mcp.example.com",
    ) as mcp_tool:
        # MCP connection established
        agent = await provider.create_agent(
            name="Agent",
            instructions="...",
            tools=mcp_tool,
        )
        result = await agent.run("Query using MCP")
        
except ConnectionError as e:
    print(f"Failed to connect to MCP server: {e}")
except TimeoutError as e:
    print(f"MCP connection timed out: {e}")
```

---

## Knowledge Base MCP Integration

For Azure AI Search knowledge bases exposed via MCP:

```python
# Knowledge base MCP endpoint format
kb_mcp_endpoint = f"{search_endpoint}/knowledgebases/{kb_name}/mcp?api-version=2025-11-01-preview"

mcp_tool = HostedMCPTool(
    name="Knowledge Base",
    url=kb_mcp_endpoint,
    approval_mode="never_require",
    allowed_tools=["knowledge_base_retrieve"],
)

agent = await provider.create_agent(
    name="KBAgent",
    instructions="""Answer questions using the knowledge base.
    Always cite sources using the format: 【source†title】""",
    tools=mcp_tool,
)
```

````


### `references/threads.md`

````markdown
# Thread Management Reference

Patterns for managing conversation state and multi-turn interactions.

## Overview

`AgentThread` links agent execution to server-side conversation state, enabling:
- Multi-turn conversations with context
- Conversation persistence and resumption
- Thread-based message history

---

## Creating and Using Threads

### Basic Multi-Turn Conversation

```python
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential

async with (
    AzureCliCredential() as credential,
    AzureAIAgentsProvider(credential=credential) as provider,
):
    agent = await provider.create_agent(
        name="ChatAgent",
        instructions="You are a helpful assistant.",
    )
    
    # Create a new thread for the conversation
    thread = agent.get_new_thread()
    
    # First turn
    result1 = await agent.run("My name is Alice", thread=thread)
    print(f"Agent: {result1.text}")
    
    # Second turn - agent remembers the name
    result2 = await agent.run("What's my name?", thread=thread)
    print(f"Agent: {result2.text}")
    
    # Third turn - context continues
    result3 = await agent.run("Tell me a joke about my name", thread=thread)
    print(f"Agent: {result3.text}")
```

### Accessing Thread Information

```python
thread = agent.get_new_thread()

# Run a conversation
await agent.run("Hello!", thread=thread)

# Access the conversation ID for persistence
print(f"Conversation ID: {thread.conversation_id}")

# Thread also tracks service-side thread ID (for Azure AI agents)
print(f"Service Thread ID: {thread.service_thread_id}")
```

---

## Conversation Persistence

### Saving Thread ID

```python
import json

async def save_conversation(thread, filepath: str):
    """Save thread ID for later resumption."""
    data = {
        "conversation_id": thread.conversation_id,
        "service_thread_id": thread.service_thread_id,
    }
    with open(filepath, "w") as f:
        json.dump(data, f)

# Usage
thread = agent.get_new_thread()
await agent.run("Start a conversation", thread=thread)
await save_conversation(thread, "conversation.json")
```

### Resuming Conversations

For Azure AI agents with persistent server-side threads, you can resume conversations:

```python
import json
from agent_framework import AgentThread

async def load_and_resume(provider, agent_id: str, filepath: str):
    """Resume a previous conversation."""
    with open(filepath) as f:
        data = json.load(f)
    
    # Get the existing agent
    agent = await provider.get_agent(agent_id=agent_id)
    
    # Create thread with existing service thread ID
    thread = AgentThread(service_thread_id=data["service_thread_id"])
    
    # Continue the conversation
    result = await agent.run("Continue our conversation", thread=thread)
    return result
```

---

## Thread with Streaming

Threads work the same way with streaming responses:

```python
thread = agent.get_new_thread()

# First turn - streaming
print("Agent: ", end="", flush=True)
async for chunk in agent.run_stream("Tell me about Python", thread=thread):
    if chunk.text:
        print(chunk.text, end="", flush=True)
print()

# Second turn - non-streaming (context maintained)
result = await agent.run("What was that language again?", thread=thread)
print(f"Agent: {result.text}")

# Third turn - streaming again
print("Agent: ", end="", flush=True)
async for chunk in agent.run_stream("Give me a code example", thread=thread):
    if chunk.text:
        print(chunk.text, end="", flush=True)
print()
```

---

## Thread with Tools

Tools work seamlessly within threaded conversations:

```python
from typing import Annotated
from pydantic import Field

def search_database(
    query: Annotated[str, Field(description="Search query")]
) -> str:
    """Search the database for information."""
    return f"Results for '{query}': Item A, Item B, Item C"

def get_item_details(
    item_name: Annotated[str, Field(description="Name of the item")]
) -> str:
    """Get details for a specific item."""
    return f"Details for {item_name}: Price $99, In Stock: Yes"

async with (
    AzureCliCredential() as credential,
    AzureAIAgentsProvider(credential=credential) as provider,
):
    agent = await provider.create_agent(
        name="ShoppingAgent",
        instructions="Help users find and learn about products.",
        tools=[search_database, get_item_details],
    )
    
    thread = agent.get_new_thread()
    
    # Turn 1: Search
    result1 = await agent.run("Search for laptops", thread=thread)
    print(f"Agent: {result1.text}")
    
    # Turn 2: Follow-up (context aware)
    result2 = await agent.run("Tell me more about Item A", thread=thread)
    print(f"Agent: {result2.text}")
    
    # Turn 3: Another follow-up
    result3 = await agent.run("Is it available?", thread=thread)
    print(f"Agent: {result3.text}")
```

---

## Multiple Parallel Conversations

Handle multiple users/conversations simultaneously:

```python
async def handle_user_session(provider, user_id: str, messages: list[str]):
    """Handle a single user's conversation."""
    agent = await provider.create_agent(
        name=f"Agent-{user_id}",
        instructions="You are a helpful assistant.",
    )
    
    # Each user gets their own thread
    thread = agent.get_new_thread()
    
    for message in messages:
        result = await agent.run(message, thread=thread)
        print(f"[{user_id}] User: {message}")
        print(f"[{user_id}] Agent: {result.text}")

# Handle multiple users concurrently
import asyncio

async def main():
    async with (
        AzureCliCredential() as credential,
        AzureAIAgentsProvider(credential=credential) as provider,
    ):
        await asyncio.gather(
            handle_user_session(provider, "user1", ["Hello", "What's 2+2?"]),
            handle_user_session(provider, "user2", ["Hi there", "Tell me a joke"]),
            handle_user_session(provider, "user3", ["Good morning", "Weather today?"]),
        )
```

---

## Thread Best Practices

### Do's

```python
# ✅ Create a new thread for each logical conversation
thread = agent.get_new_thread()

# ✅ Pass the same thread to maintain context
await agent.run("Message 1", thread=thread)
await agent.run("Message 2", thread=thread)

# ✅ Save thread IDs for conversations that need resumption
conversation_id = thread.conversation_id
```

### Don'ts

```python
# ❌ Don't create a new thread for each message (loses context)
for msg in messages:
    thread = agent.get_new_thread()  # Wrong!
    await agent.run(msg, thread=thread)

# ❌ Don't share threads between different agents
agent1_thread = agent1.get_new_thread()
await agent2.run("Hello", thread=agent1_thread)  # May cause issues

# ❌ Don't forget to pass the thread (single-turn only)
await agent.run("Message 1")  # No thread - no context saved
await agent.run("Message 2")  # Can't reference previous message
```

---

## Thread Lifecycle

```
1. agent.get_new_thread()
   └── Creates new AgentThread object
   └── Server-side thread created on first run

2. agent.run(..., thread=thread)
   └── Message added to thread
   └── Agent response added to thread
   └── Context accumulated

3. (Optional) Save thread.conversation_id
   └── For later resumption

4. (Optional) Resume with AgentThread(service_thread_id=...)
   └── Continues existing conversation
```

---

## Stateless vs Stateful Patterns

### Stateless (No Thread)

Each call is independent:

```python
# Good for one-shot queries
result = await agent.run("What is 2+2?")
```

### Stateful (With Thread)

Context persists across calls:

```python
# Good for conversations
thread = agent.get_new_thread()
result1 = await agent.run("My favorite color is blue", thread=thread)
result2 = await agent.run("What's my favorite color?", thread=thread)  # Knows it's blue
```

````


### `references/tools.md`

````markdown
# Hosted Tools Reference

Detailed patterns for all hosted tools available in the Agent Framework.

## HostedCodeInterpreterTool

Enables agents to execute Python code on the Azure AI service.

### Basic Usage

```python
from agent_framework import HostedCodeInterpreterTool
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential

async with (
    AzureCliCredential() as credential,
    AzureAIAgentsProvider(credential=credential) as provider,
):
    agent = await provider.create_agent(
        name="CodingAgent",
        instructions="You can write and execute Python code to solve problems.",
        tools=HostedCodeInterpreterTool(),
    )
    
    result = await agent.run("Calculate the factorial of 20 using Python")
    print(result.text)
```

### With File Inputs

```python
from agent_framework import HostedCodeInterpreterTool, HostedFileContent

# Reference a file already uploaded to the service
code_tool = HostedCodeInterpreterTool(
    inputs=[
        HostedFileContent(file_id="file-abc123"),
    ]
)

agent = await provider.create_agent(
    name="DataAnalyst",
    instructions="Analyze the provided data file.",
    tools=code_tool,
)
```

### Common Use Cases

- Data analysis and visualization
- Mathematical calculations
- File processing (CSV, JSON, etc.)
- Code generation and testing

---

## HostedFileSearchTool

Enables agents to search through documents using vector stores.

### Setup with Vector Store

```python
from pathlib import Path
from agent_framework import HostedFileSearchTool, HostedVectorStoreContent
from agent_framework.azure import AzureAIAgentsProvider
from azure.ai.agents.aio import AgentsClient
from azure.identity.aio import AzureCliCredential

async with (
    AzureCliCredential() as credential,
    AgentsClient(endpoint=endpoint, credential=credential) as agents_client,
    AzureAIAgentsProvider(agents_client=agents_client) as provider,
):
    # Upload file to the service
    file = await agents_client.files.upload(
        file_path=Path("data/knowledge_base.txt"),
        purpose="agents"
    )
    
    # Create vector store from file
    vector_store = await agents_client.vector_stores.create_and_poll(
        file_ids=[file.id],
        name="my_knowledge_store"
    )
    
    # Create file search tool with vector store
    file_search_tool = HostedFileSearchTool(
        inputs=[HostedVectorStoreContent(vector_store_id=vector_store.id)],
        max_results=10,  # Optional: limit search results
    )
    
    agent = await provider.create_agent(
        name="ResearchAgent",
        instructions="Search the knowledge base to answer questions accurately.",
        tools=file_search_tool,
    )
    
    result = await agent.run("What are the key findings in the document?")
    print(result.text)
```

### Multiple Vector Stores

```python
file_search_tool = HostedFileSearchTool(
    inputs=[
        HostedVectorStoreContent(vector_store_id="vs-policy-docs"),
        HostedVectorStoreContent(vector_store_id="vs-technical-specs"),
    ],
    max_results=20,
)
```

### Common Use Cases

- Document Q&A
- Knowledge base retrieval
- Policy/procedure lookup
- Technical documentation search

---

## HostedWebSearchTool

Enables agents to search the web using Bing.

### Basic Bing Grounding

```python
import os
from agent_framework import HostedWebSearchTool
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential

# Requires BING_CONNECTION_ID environment variable
os.environ["BING_CONNECTION_ID"] = "your-bing-connection-id"

async with (
    AzureCliCredential() as credential,
    AzureAIAgentsProvider(credential=credential) as provider,
):
    agent = await provider.create_agent(
        name="SearchAgent",
        instructions="Search the web for current information to answer questions.",
        tools=HostedWebSearchTool(
            name="Bing Grounding Search",
            description="Search the web for current information",
        ),
    )
    
    result = await agent.run("What are the latest developments in AI?")
    print(result.text)
```

### Bing Custom Search

For searching a custom index of websites:

```python
import os

# Requires custom search configuration
os.environ["BING_CUSTOM_CONNECTION_ID"] = "your-custom-bing-connection-id"
os.environ["BING_CUSTOM_INSTANCE_NAME"] = "your-custom-instance"

bing_custom_tool = HostedWebSearchTool(
    name="Bing Custom Search",
    description="Search specific websites for relevant information",
)
```

### Common Use Cases

- Current events and news
- Real-time information lookup
- Fact-checking
- Research assistance

---

## HostedImageGenerationTool

Enables agents to generate images (when available on the service).

```python
from agent_framework import HostedImageGenerationTool

agent = await provider.create_agent(
    name="CreativeAgent",
    instructions="You can generate images based on descriptions.",
    tools=HostedImageGenerationTool(),
)
```

---

## Combining Multiple Tools

Agents can use multiple tools simultaneously:

```python
from typing import Annotated
from pydantic import Field
from agent_framework import (
    HostedCodeInterpreterTool,
    HostedFileSearchTool,
    HostedWebSearchTool,
    HostedVectorStoreContent,
)

# Custom function tool
def get_current_date() -> str:
    """Get today's date."""
    from datetime import date
    return date.today().isoformat()

async with (
    AzureCliCredential() as credential,
    AgentsClient(endpoint=endpoint, credential=credential) as agents_client,
    AzureAIAgentsProvider(agents_client=agents_client) as provider,
):
    # Setup vector store first
    vector_store = await agents_client.vector_stores.create_and_poll(
        file_ids=[uploaded_file.id],
        name="docs_store"
    )
    
    agent = await provider.create_agent(
        name="SuperAgent",
        instructions="""You are a versatile assistant with multiple capabilities:
        - Execute Python code for calculations and data analysis
        - Search internal documents for company information
        - Search the web for current external information
        - Provide current date when needed
        
        Choose the appropriate tool based on the user's question.""",
        tools=[
            get_current_date,  # Function tool
            HostedCodeInterpreterTool(),
            HostedFileSearchTool(
                inputs=[HostedVectorStoreContent(vector_store_id=vector_store.id)]
            ),
            HostedWebSearchTool(name="Bing"),
        ],
    )
```

---

## Tool Selection Guidelines

| Need | Tool |
|------|------|
| Code execution, math, data analysis | `HostedCodeInterpreterTool` |
| Search uploaded documents | `HostedFileSearchTool` |
| Current web information | `HostedWebSearchTool` |
| Custom business logic | Function tools |
| External API integration | `HostedMCPTool` or `MCPStreamableHTTPTool` |

---

## Error Handling

```python
from agent_framework import AgentResponseUpdate

async for chunk in agent.run_stream("Analyze this data"):
    if isinstance(chunk, AgentResponseUpdate):
        # Check for tool execution errors
        if chunk.tool_calls:
            for tool_call in chunk.tool_calls:
                if hasattr(tool_call, 'error') and tool_call.error:
                    print(f"Tool error: {tool_call.error}")
    if chunk.text:
        print(chunk.text, end="", flush=True)
```

````