SamuelChien/mega-skills-collection

Agent Framework Azure ホステッドエージェント

Microsoft Agent Framework Python SDK を使用して、Azure AI Foundry 上に永続的なエージェントを構築します。

アーキテクチャ

ユーザークエリ → AzureAIAgentsProvider → Azure AI エージェントサービス (永続的)
                    ↓
              Agent.run() / Agent.run_stream()
                    ↓
              ツール: 関数 | ホステッド (コード/検索/Web) | MCP
                    ↓
              AgentThread (会話の永続化)

インストール

# フルフレームワーク (推奨)
pip install agent-framework --pre

# Azure 専用パッケージのみ
pip install agent-framework-azure-ai --pre

環境変数

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"  # Web 検索用
export AZURE_TOKEN_CREDENTIALS=prod # 本番環境で DefaultAzureCredential を使用する場合のみ必須

認証とライフサイクル

🔑 以下のすべてのコードサンプルに適用される2つのルール:

1. DefaultAzureCredential を推奨します。 ローカル (Azure CLI / VS Code / Developer CLI) と Azure (マネージド ID、ワークロード ID) の両方でコード変更なしに動作します。接続文字列やアカウント/API キーは Entra の監査とローテーションをバイパスするため避けてください。
   • ローカル開発: DefaultAzureCredential はそのまま使用できます。
   • 本番環境: AZURE_TOKEN_CREDENTIALS=prod (または AZURE_TOKEN_CREDENTIALS=<特定の資格情報>) を設定して、資格情報チェーンを本番環境で安全な資格情報に制限します。
2. すべてのクライアントをコンテキストマネージャーでラップ して、HTTP トランスポート、ソケット、トークンキャッシュが確定的に解放されるようにします:
   • 同期: with <Client>(...) as client:
   • 非同期: async with <Client>(...) as client: かつ async with DefaultAzureCredential() as credential: (azure.identity.aio より)

スニペットではこの設定を省略する場合がありますが、本番コードでは常に両方のルールに従う必要があります。

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

# 開発環境
credential = AzureCliCredential()

# 本番環境
# ローカル開発: DefaultAzureCredential。本番: AZURE_TOKEN_CREDENTIALS=prod または AZURE_TOKEN_CREDENTIALS=<特定の資格情報> を設定
credential = DefaultAzureCredential(require_envvar=True)
# または本番環境で特定の資格情報を直接使用:
# 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="あなたは役立つアシスタントです。",
        )
        
        result = await agent.run("こんにちは!")
        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="天気を取得する都市名")],
) -> str:
    """指定された場所の現在の天気を取得します。"""
    return f"{location}の天気: 22°C, 晴れ"

def get_current_time() -> str:
    """現在のUTC時刻を取得します。"""
    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="あなたは天気と時刻の質問に答えます。",
            tools=[get_weather, get_current_time],  # 関数を直接渡す
        )
        
        result = await agent.run("東京の天気は?")
        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="あなたはコードを実行し、ファイルを検索し、Webを検索できます。",
            tools=[
                HostedCodeInterpreterTool(),
                HostedWebSearchTool(name="Bing"),
            ],
        )
        
        result = await agent.run("Pythonで20の階乗を計算してください")
        print(result.text)

ストリーミング応答

async def main():
    async with (
        AzureCliCredential() as credential,
        AzureAIAgentsProvider(credential=credential) as provider,
    ):
        agent = await provider.create_agent(
            name="StreamingAgent",
            instructions="あなたは役立つアシスタントです。",
        )
        
        print("エージェント: ", end="", flush=True)
        async for chunk in agent.run_stream("短い物語を聞かせてください"):
            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="あなたは役立つアシスタントです。",
            tools=[get_weather],
        )
        
        # 会話の永続化のためにスレッドを作成
        thread = agent.get_new_thread()
        
        # 最初のターン
        result1 = await agent.run("東京の天気は?", thread=thread)
        print(f"エージェント: {result1.text}")
        
        # 2番目のターン - コンテキストが維持される
        result2 = await agent.run("大阪はどうですか?", thread=thread)
        print(f"エージェント: {result2.text}")
        
        # 後で再開するためにスレッドIDを保存
        print(f"会話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="構造化形式で天気情報を提供してください。",
            response_format=WeatherResponse,
        )
        
        result = await agent.run("東京の天気は?")
        weather = WeatherResponse.model_validate_json(result.text)
        print(f"{weather.location}: {weather.temperature}°{weather.unit}")

プロバイダーメソッド

メソッド	説明
create_agent()	Azure AI サービスに新しいエージェントを作成
get_agent(agent_id)	ID で既存のエージェントを取得
as_agent(sdk_agent)	SDK エージェントオブジェクトをラップ (HTTP 呼び出しなし)

ホステッドツールクイックリファレンス

ツール	インポート	目的
HostedCodeInterpreterTool	from agent_framework import HostedCodeInterpreterTool	Python コードを実行
HostedFileSearchTool	from agent_framework import HostedFileSearchTool	ベクターストアを検索
HostedWebSearchTool	from agent_framework import HostedWebSearchTool	Bing Web 検索
HostedMCPTool	from agent_framework import HostedMCPTool	サービス管理型 MCP
MCPStreamableHTTPTool	from agent_framework import MCPStreamableHTTPTool	クライアント管理型 MCP

完全な例

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="都市名")],
) -> str:
    """場所の天気を取得します。"""
    return f"{location}の天気: 22°C, 晴れ"


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="あなたは複数の機能を持つリサーチアシスタントです。",
            tools=[
                get_weather,
                HostedCodeInterpreterTool(),
                HostedWebSearchTool(name="Bing"),
                mcp_tool,
            ],
        )
        
        thread = agent.get_new_thread()
        
        # 非ストリーミング
        result = await agent.run(
            "Pythonのベストプラクティスを検索して要約してください",
            thread=thread,
        )
        print(f"応答: {result.text}")
        
        # ストリーミング
        print("\nストリーミング: ", end="")
        async for chunk in agent.run_stream("例を続けてください", thread=thread):
            if chunk.text:
                print(chunk.text, end="", flush=True)
        print()
        
        # 構造化出力
        result = await agent.run(
            "調査結果を分析してください",
            thread=thread,
            response_format=AnalysisResult,
        )
        analysis = AnalysisResult.model_validate_json(result.text)
        print(f"\n信頼度: {analysis.confidence}")


if __name__ == "__main__":
    asyncio.run(main())

規約

• 常に非同期コンテキストマネージャーを使用: async with provider:
• 関数を tools= パラメーターに直接渡す (自動的に AIFunction に変換されます)
• 関数パラメーターには Annotated[type, Field(description=...)] を使用する
• マルチターン会話には get_new_thread() を使用する
• サービス管理型 MCP には HostedMCPTool、クライアント管理型 MCP には MCPStreamableHTTPTool を推奨

ベストプラクティス

1. この SDK は非同期ファーストです — ハンドラーには async def を、全体で async with を使用してください。
2. クライアントと非同期資格情報には常にコンテキストマネージャーを使用してください。 すべてのクライアントを with Client(...) as client: (同期) または async with Client(...) as client: (非同期) でラップします。azure.identity.aio からの非同期 DefaultAzureCredential については、トークンとトランスポートがクリーンアップされるように async with credential: も使用してください。

リファレンスファイル

• references/tools.md: 詳細なホステッドツールパターン
• references/mcp.md: MCP 統合 (ホステッド + ローカル)
• references/threads.md: スレッドと会話の管理
• references/advanced.md: OpenAPI、引用、構造化出力