🤖
Microsoft Agent Frameworkをためしてみた（python）

yu shimohama
2025/10/07に公開
Python
Agent
AIエージェント
Microsoft Agent Framework
tech
2025/10/1にMicrosoftからエージェント開発の最新フレームワークである『Microsoft Agent Framework』が公開されました！
ザックリと概要を説明して、Pythonでの実装例をかいつまんで載せていきます。

※LLMモデルはAzure Open AIを利用しております。
https://devblogs.microsoft.com/foundry/introducing-microsoft-agent-framework-the-open-source-engine-for-agentic-ai-apps/

 概要
 Microsoft Agent Framework = Semantic Kernel + AutoGenまず背景として、従来のAIエージェントシステムの開発において、プロトタイプから生産まで様々な障害がありました。具体的には以下です。
オープンソースフレームワークの多くは断片化していて、それぞれに独自のAPIを持ち、抽象化がなされていた
ローカル環境（開発者のPCなど）で構築・テストしたシステムやコードが、そのままクラウド環境にスムーズに移行できるとは限らない
可観測性、コンプライアンスフック、セキュリティ、長期実行の耐久性などのエンタープライズの準備が欠けていること
重要なものとしては3つ目の部分で、Microsoftとしては「エンタープライズシステムが構築できるSemanticKernelとマルチエージェントオーケストレーションシステムが構築できるAutoGenを統合することで、両方のいいとこどりをして上記問題を解決しよう！」ということで、Microsoft Agent Frameworkが構築されました。
実際の各フレームワークの特徴の対応関係は以下の表となります。


 4つの主要機能Microsoft Agent Frameworkでは、次のことが可能になります。

 1. オープンスタンダードと相互運用性
以下をサポートし、エージェントの移植性とベンダーニュートラル性を保証

MCP: エージェントは、MCP を介して公開されている外部ツールまたはデータサーバーを動的に検出して呼び出すことができます。Microsoft エージェント フレームワークを使用すると、カスタム グルー コードなしで、成長する MCP 準拠サービスのエコシステムに簡単に接続できます。

A2A: エージェントは、構造化されたプロトコル駆動型メッセージングを使用して、ランタイム間でコラボレーションできます。A2A サポートにより、開発者は、異なるフレームワークや環境で実行されている場合でも、1 つのエージェントがデータを取得し、別のエージェントがデータを分析し、3 番目のエージェントが結果を検証するワークフローを作成できます。

OpenAPI ファーストの設計: OpenAPI 仕様を持つ REST API は、呼び出し可能なツールとして即座にインポートできます。Microsoft Agent Framework は、スキーマの解析、ツール定義、セキュリティで保護された呼び出しを処理するため、開発者はラッパーを手動で構築することなく、何千ものエンタープライズ API を活用できます。

クラウドに依存しないランタイム: エージェントは、コンテナー、オンプレミス、または複数のクラウド間で実行できるため、環境間で移植できます。開発者は、好みの SDK (Azure OpenAI、OpenAI など) を使用して 1 つのエージェントをスピンアップし、既存のメソッドを AIFunctions としてラップしてツールを追加し、すぐに外部 API に接続できます。

 2. 研究から運用までのパイプライン
以下のMicrosoft Researchの最先端のオーケストレーションパターンがエンタープライズで使用可能
ステップバイステップのワークフローのためのシーケンシャルオーケストレーション
エージェントが並行して作業するコンカレントオーケストレーション
エージェントが共同でブレインストーミングを行うグループチャットオーケストレーション
コンテキストの進化に応じてエージェント間で責任が移動するハンドオフオーケストレーション

マネージャーエージェントが動的なタスク台帳を構築および改良し、複雑でオープンエンドの問題に対して専門のエージェント(場合によっては人間)を調整するマジェンティックオーケストレーション

 3. コミュニティ主導の拡張性
Microsoft Agent Framework は 100% オープン ソースであり、コミュニティとともに成長できるように設計されています。モジュール設計により、拡張、カスタマイズが簡単になります。

エンタープライズ システムへのコネクタ: Agent Framework は、組み込みコネクタの幅広いセット (Azure AI Foundry、Microsoft Graph、Microsoft Fabric、SharePoint、Oracle、Amazon Bedrock、MongoDB、Azure Logic Apps を介したさまざまな SaaS システム) を継承するため、エージェントは初日からエンタープライズ データを操作できます。

プラガブルメモリモジュール: 開発者は、Redis、Pinecone、Qdrant、Weaviate、Elasticsearch、Postgres、または会話メモリ用の独自のストアを選択できます。Agent Frameworkは抽象化を提供します。

宣言型エージェント: YAML または JSON 定義を使用すると、開発者はプロンプト、ロール、ツールを宣言的に指定できます。これらのファイルは、バージョン管理、テンプレート化、チーム間で共有できます。

コミュニティのイノベーション: Agent Framework は、コミュニティ主導のオーケストレーション戦略、新しいコネクタ、ベスト プラクティスを吸収するように設計されています。

 4. エンタープライズ対応
組み込みの可観測性、承認、セキュリティ、長期実行の耐久性

可観測性: OpenTelemetry は、すべてのエージェント アクション、ツール呼び出し、オーケストレーション ステップをインストルメント化して視覚化できるため、Azure AI Foundry ダッシュボードを通じて推論フローを簡単に追跡し、パフォーマンスを監視できます。

安全なクラウドホスティング: エージェントは、仮想ネットワーク統合、ロールベースのアクセス、プライベート データ処理、組み込みのコンテンツの安全性などのエンタープライズ制御を使用して、Azure AI Foundry でネイティブに実行されます。

セキュリティとコンプライアンス: Azure AI Content Safety の統合、Entra ID 認証、構造化ログにより、Agent Framework エージェントは規制された業界で実行できます。

長時間の耐久性: エージェントのスレッドとワークフローは、中断からの一時停止、再開、回復を行うことができ、再試行とエラー処理ロジックにより、長時間実行されるプロセスの信頼性を大規模に維持できます。

Human in the loop: ガバナンスが必要なシナリオでは、ツールを人間の承認を必要とするものとしてマークできます。Agent Framework は、UI またはキューにルーティングできる保留中の承認要求を自動的に発行し、それに応じて実行を続行または拒否します。これはローカルツールやリモートサービス呼び出し全体で機能し、機密性の高い操作を確実に制御できます。

CI/CD 統合: このフレームワークは GitHub Actions と Azure DevOps パイプラインに直接統合され、テレメトリはエンタープライズ グレードのデプロイと根本原因分析のためにAzure MonitorとApplication Insightsに流れ込みます。

 将来的な動き今後の展開として、Microsoft Agent Framework は、Microsoft 365 Agents SDK との統合や Azure AI Foundry Agent Service との共有ランタイムなど、Microsoft のエージェント開発スタック全体の統合がさらに進むとのこと


 実装
 インストール
 前提条件以下の条件を満たす必要があります
Pythonは3.10以上
Azure AI Projectにモデルをデプロイしておく
Azure CLIでaz loginしておく

 pipでインストール以下を実行します

pip install agent-framework

 Azure OpenAI ChatCompletion Agentでチャットするまずはシンプルにgpt-4oでチャットしてみます。
以下の環境変数をsetしておく必要があります。

AZURE_OPENAI_ENDPOINT：Azure OpenAIリソースのエンドポイント

AZURE_OPENAI_CHAT_DEPLOYMENT_NAME：Azure OpenAIリソースにデプロイされたモデルのデプロイ名

※venvを使われている方は、.envで定義してload_dotenvしてください。
AzureOpenAIChatClientに対して、引数としてchat_clientにAzureCliCredentialを設定し、create_agentで、Azure OpenAI ChatCompletion Agentを生成します。
create_agentの際に、以下を引数で渡します。

instructions：エージェントの説明

name：エージェントの名前
最後にagent.run()でチャットを実行します。
import asyncio
from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential
from dotenv import load_dotenv

load_dotenv(override=True)

agent = AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
    instructions="あなたはドラえもんです。ユーザーの要望に対して、適切な道具の名前を答えてください。",
    name="Doraemon"
)

async def main():
    result = await agent.run("空を自由に飛びたい")
    print(result.text)

asyncio.run(main())
回答内容
「タケコプター」がぴったりだよ！頭に装着してスイッチを入れると、自由に空を飛べるよ！

 エージェントで画像を使用する先ほどは、agent.runで引数に直接文字列を渡していましたが、今回はChatMessageを渡します。
引数のcontentsには、ユーザーのメッセージであるTextContentを、画像データであるDataContentを設定します。
import asyncio
from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential
from agent_framework import ChatMessage, TextContent, Role, DataContent
from dotenv import load_dotenv

load_dotenv(override=True)

agent = AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
    name="VisionAgent",
    instructions="あなたは画像を分析できる有能なエージェントです。"
)

# Load image from local file
with open("./image.jpg", "rb") as f:
    image_bytes = f.read()

message = ChatMessage(
    role=Role.USER, 
    contents=[
        TextContent(text="この画像で何が見えますか？"),
        DataContent(
            data=image_bytes,
            media_type="image/jpeg"
        )
    ]
)

async def main():
    result = await agent.run(message)
    print(result.text)

asyncio.run(main())
使用画像

回答内容
この画像は一見、屋外のプールの底から空を見上げているように見えますが、実際には「レアンドロ・エルリッヒ（Leandro Erlich）」が制作したインスタレーションアート「スイミング・プール」に似ています。この作品は、透明なアクリル板の上に薄い水の層を載せている構造で、その下は実際には乾いた空間となっています。梯子があり、空と水面の反射がリアルな視覚効果を生んでいます。

 エージェントでツールを利用する指定された場所の天気を取得する関数と、名産品を取得する関数をtoolsとして定義してみました。

回答結果をみると、質問に対して適切な関数が選択されていることがわかります。
from typing import Annotated
from pydantic import Field
from dotenv import load_dotenv
import asyncio
from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential

load_dotenv(override=True)

class WeatherTools:
    def __init__(self):
        self.last_location = None

    def get_weather(
        self,
        location: Annotated[str, Field(description="天気を取得する場所")],
    ) -> str:
        """指定された場所の天気を取得します。"""
        return f"{location}の天気は曇りで最高気温は15°Cです。"

    def get_specialty_products(
        self, 
        location: Annotated[str, Field(description="名産品を取得する場所")],
    ) -> str:
        """指定された場所の名産品を取得します。"""
        return f"{location}の名産品は、ワッフル、チーズ、チョコレートです。"

tools = WeatherTools()
agent = AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
    instructions="あなたは親切なアシスタントです",
    tools=[tools.get_weather, tools.get_specialty_products]
)

async def main():
    result = await agent.run("アムステルダムの天気はどうですか？")
    print(result.text)

asyncio.run(main())
回答結果
アムステルダムの現在の天気は曇りで、最高気温は15°Cです。外出の際は少し肌寒いかもしれませんので、上着を持っていくと良いでしょう。

 エージェントでMCPツールを利用するMCPのツールのタイプは、以下の3種類があります。

MCPStdioTool - ローカルのMCPサーバー

MCPStreamableHTTPTool - HTTP/SSE MCPサーバー

MCPWebsocketTool - WebSocket MCPサーバー
試しにMCPStreamableHTTPToolを使ってみます。
import asyncio
from agent_framework import ChatAgent, MCPStreamableHTTPTool
from agent_framework.azure import AzureAIAgentClient
from azure.identity.aio import AzureCliCredential
from dotenv import load_dotenv

load_dotenv(override=True)

async def http_mcp_example():
    """HTTP ベースの MCP サーバーを使用する例"""
    async with (
        AzureCliCredential() as credential,
        MCPStreamableHTTPTool(
            name="Microsoft Learn MCP",
            url="https://learn.microsoft.com/api/mcp",
            headers={"Authorization": "Bearer your-token"},
        ) as mcp_server,
        ChatAgent(
            chat_client=AzureAIAgentClient(async_credential=credential),
            name="DocsAgent",
            instructions="Microsoft ドキュメントに関する質問をサポートします。",
        ) as agent,
    ):
        result = await agent.run(
            "az cli を使用して Azure ストレージアカウントを作成する方法は？",
            tools=mcp_server
        )
        print(result)

if __name__ == "__main__":
    asyncio.run(http_mcp_example())
やり方は、先ほどのtoolsにMCPサーバーを指定するだけです。
回答結果
--resource-group <resource-group> 

--location <location> 

--sku Standard_RAGRS 

--kind StorageV2 

--min-tls-version TLS1_2 

--allow-blob-public-access false

--kind: ストレージアカウントの種類 (一般的には StorageV2 を選択)。

--min-tls-version: TLS の最小バージョン。

--allow-blob-public-access: パブリックアクセスを許可するかどうか (デフォルトではfalse に設定)。
これらのコマンドを必要なパラメーターとともに実行して、ストレージアカウントをプロビジョニングすることができます。
詳細については、公式ドキュメントをご覧ください:

Azure CLI のストレージアカウント作成に関するドキュメント

 ワークフローオーケストレーション※ワークフローを画像に保存するため、あらかじめ以下をpipでインストールする必要があります。
pip install graphviz
pip install agent-framework[viz]
ワークフローオーケストレーションパターンのうち、コンカレントオーケストレーションの実装を行います。
コンカレント・オーケストレーションでは、複数のエージェントが同じタスクを並行して処理できます。各エージェントは入力を個別に処理し、その結果が収集および集計されます。このアプローチは、ブレインストーミング、アンサンブル推論、投票システムなど、多様な視点やソリューションが価値のあるシナリオに適しています。

今回は、ランダムの10個の整数値リストを入力として、以下のようなワークフローを組んで、平均と合計をまとめて回答させてみます。

フローの処理手順は以下となります。
入力をフロー開始Executorであるdispatcherに渡す
受け取った入力をdispatcherが以下の2つのExecutorに分配する

・平均を算出するaverage

・合計を算出するsummation
averageとsummationの結果をフローの最後のExecutorであるaggregatorが収集し、まとめてリストとして配信する
import asyncio
import random

from agent_framework import Executor, WorkflowBuilder, WorkflowContext, WorkflowOutputEvent, handler, WorkflowViz
from typing_extensions import Never

class Dispatcher(Executor):
    """
    このエグゼキューターの唯一の目的は、ワークフローの入力を
    他のエグゼキューターに分配することです。
    """

    @handler
    async def handle(self, numbers: list[int], ctx: WorkflowContext[list[int]]):
        if not numbers:
            raise RuntimeError("入力は有効な整数のリストである必要があります。")

        await ctx.send_message(numbers)

class Average(Executor):
    """整数のリストの平均を計算します。"""

    @handler
    async def handle(self, numbers: list[int], ctx: WorkflowContext[float]):
        average: float = sum(numbers) / len(numbers)
        await ctx.send_message(average)


class Sum(Executor):
    """整数のリストの合計を計算します。"""

    @handler
    async def handle(self, numbers: list[int], ctx: WorkflowContext[int]):
        total: int = sum(numbers)
        await ctx.send_message(total)

class Aggregator(Executor):
    """異なるタスクからの結果を集約し、最終的な出力を生成します。"""

    @handler
    async def handle(self, results: list[int | float], ctx: WorkflowContext[Never, list[int | float]]):
        """ソースエグゼキューターからの結果を受信します。

        フレームワークは自動的にソースエグゼキューターからのメッセージを収集し、
        リストとして配信します。

        Args:
            results (list[int | float]): 上流のエグゼキューターからの実行結果。
                型注釈は、上流のエグゼキューターが生成するユニオン型のリストである必要があります。
            ctx (WorkflowContext[Never, list[int | float]]): 最終的な出力を生成できるワークフローコンテキスト。
        """
        await ctx.yield_output(results)

async def main() -> None:
    # 1) エグゼキューターを作成
    dispatcher = Dispatcher(id="dispatcher")
    average = Average(id="average")
    summation = Sum(id="summation")
    aggregator = Aggregator(id="aggregator")

    # 2) シンプルなファンアウト・ファンインワークフローを構築
    workflow = (
        WorkflowBuilder()
        .set_start_executor(dispatcher)
        .add_fan_out_edges(dispatcher, [average, summation])
        .add_fan_in_edges([average, summation], aggregator)
        .build()
    )

    viz = WorkflowViz(workflow)
    viz.save_svg("workflow_architecture.svg")

    # 3) ワークフローを実行
    output: list[int | float] | None = None
    input_numbers = [random.randint(1, 100) for _ in range(10)]
    print(f"Input numbers: {input_numbers}")
    async for event in workflow.run_stream(input_numbers):
        if isinstance(event, WorkflowOutputEvent):
            output = event.data

    if output is not None:
        print(output)

if __name__ == "__main__":
    asyncio.run(main())
回答結果
Input numbers: [43, 4, 42, 66, 66, 38, 12, 25, 87, 65]
Output: [44.8, 448]
ヘッドウォータース
株式会社ヘッドウォータースのテックブログです。 AIエージェント、生成AI、LLM、Azureのサービスや資格、IoT、XR系などData&AIとApp modernizeに関して幅広く投稿します！
概要

Microsoft Agent Framework = Semantic Kernel + AutoGen

4つの主要機能

1. オープンスタンダードと相互運用性

2. 研究から運用までのパイプライン

3. コミュニティ主導の拡張性

4. エンタープライズ対応

将来的な動き

実装

インストール

前提条件

pipでインストール

Azure OpenAI ChatCompletion Agentでチャットする

エージェントで画像を使用する

エージェントでツールを利用する

エージェントでMCPツールを利用する

ワークフローオーケストレーション

Discussion
