agent-framework-azure-ai-py

バンドルされた上流サポート

サポートファイルはスキルからの相対パスに含まれています。ピン留めされたソース、ライセンス、ハッシュ、インポートの変更については、UPSTREAM-NOTICE.md を参照してください。

Agent Framework Azure Hosted Agents

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

アーキテクチャ

User Query → AzureAIAgentsProvider → Azure AI Agent Service (Persistent)
                    ↓
              Agent.run() / Agent.run_stream()
                    ↓
              Tools: Functions | Hosted (Code/Search/Web) | MCP
                    ↓
              AgentThread (conversation persistence)

SDK の世代と実行スコープ

以下の例は、古い世代の Agent Framework（AzureAIAgentsProvider、AgentThread、Hosted*Tool）をそのまま使用しています。現在のパッケージで動作することは検証されていません。現在の Microsoft ドキュメントではモデルプロバイダーとエージェントサービスが分けられており、Python の README には agent-framework-foundry が記載されています。これらのスニペットを適用する前に、インストールされているバージョンを特定し、それに対応する公式 API／サンプルを使用してください。互換性のあるレガシーバージョンは確認されていません。古いクラスと、バージョンを固定していない最新バージョンのインストールを組み合わせて、動作すると想定しないでください。

正確な Azure テナント／プロジェクト／デプロイ、プリンシパル、最小権限のロールを選択してください。エージェントの作成、ファイルのアップロード、コード／ウェブ／MCP ツールの有効化、プロンプトの実行では、サービスにデータが送信され、記録が保持され、料金が発生する場合があります。まず、認可の範囲、予算、タイムアウト、ツールの許可リスト、リソースのクリーンアップを設定してください。クライアントを閉じても、ホストされたエージェント／ファイル／ベクトルストア／スレッドは削除されません。取得したテキストやツールからのテキストは信頼できないものとして扱い、ツールでテナント／リソース／アクションの認可を強制してください。プロンプトはアクセス制御ではありません。

インストール

# Current package discovery only; pin reviewed versions in your project lockfile.
# These installs do not make the legacy snippets below compatible.
pip install agent-framework
# Selective current Foundry integration:
pip install agent-framework-foundry

環境変数

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

認証

from azure.identity.aio import AzureCliCredential, DefaultAzureCredential

# Development
credential = AzureCliCredential()

# Production
credential = DefaultAzureCredential()

基本的なワークフロー

基本的なエージェント

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())

関数ツールを使うエージェント

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:
    """Return fixture weather (not a live observation)."""
    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)

ホストされたツールを使うエージェント

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)

ストリーミング応答

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()

会話スレッド

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}")

構造化出力

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}")

プロバイダーのメソッド

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

ホストされたツールのクイックリファレンス

ツール	インポート	目的
HostedCodeInterpreterTool	from agent_framework import HostedCodeInterpreterTool	Python コードを実行します
HostedFileSearchTool	from agent_framework import HostedFileSearchTool	ベクトルストアを検索します
HostedWebSearchTool	from agent_framework import HostedWebSearchTool	Bing ウェブ検索
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="City name")],
) -> str:
    """Return fixture weather (not a live observation)."""
    return f"Weather in {location}: 72°F, sunny"


class AnalysisResult(BaseModel):
    summary: str
    key_findings: list[str]
    confidence: float = Field(ge=0.0, le=1.0)


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"\nModel self-rating (not calibrated reliability): {analysis.confidence}")


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

現在の参照資料：Microsoft のプロバイダーモデル、Python のインストール方法とサンプル。

規約

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

参考ファイル

• references/tools.md：ホストされたツールの詳しいパターン
• references/mcp.md：MCP 統合（ホスト型＋ローカル）
• references/threads.md：スレッドと会話の管理
• references/advanced.md：OpenAPI、引用、構造化出力

使用する場面

このスキルは、概要で説明されているワークフローまたはアクションを実行する場合に適用できます。

制限事項

• このスキルは、タスクが上記のスコープに明確に一致する場合にのみ使用してください。
• 出力を、環境固有の検証、テスト、専門家によるレビューの代替として扱わないでください。
• 必要な入力、権限、安全境界、成功基準が欠けている場合は、作業を中断して確認を求めてください。