Agent Framework を使って Microsoft Foundry の MCP ツールを利用するエージェントを開発する

本記事では、Microsoft Foundry と Agent Framework（Python）を利用して、Model Context Protocol (MCP) ツールを扱えるエージェントを開発する方法を紹介します。

Agent Framework からMicrosoft Foundry のエージェントを利用することで、複数エージェントの切り替えや、それらを組み合わせたオーケストレーションが行いやすくなることが期待できます。

実装環境
環境変数
実装
- サンプルコード
- 実行結果
Microsoft Foundry のエージェントを Agent Framework でオーケストレーションする
- サンプルコード
Microsoft Foundry SDK で直接エージェントを利用する方法
最後に

実装環境

Python 3.12
agent-framework 1.0.0b251216

環境変数

今回の記事では、Microsoft Foundry のリソースを利用します。以下の環境変数を設定します。

AZURE_AI_PROJECT_ENDPOINT="https://<your-project>.services.ai.azure.com/api/projects/<project-id>"
AZURE_AI_MODEL_DEPLOYMENT_NAME = "<deployment_name>"

実装

以下のコードでは、Agent Framework から Azure AI Foundry の Agent Service に接続し、MCP Toolを利用するエージェントを作成・実行します。

※ 実装は下記リンクの内容を参考にしています。

MCP および Foundry エージェント | Microsoft Learn

サンプルコード

import os
import asyncio
from agent_framework import HostedMCPTool
from agent_framework_azure_ai import AzureAIAgentClient
from azure.identity.aio import DefaultAzureCredential


# ローカル環境で.envから環境変数を読み取る場合に利用
from dotenv import load_dotenv
load_dotenv(dotenv_path=".env", override=True)


async def basic_foundry_mcp_example():
    """Basic example of Microsoft Foundry agent with hosted MCP tools."""
    print("[start] basic_foundry_mcp_example", flush=True)

    required_env_vars = ["AZURE_AI_PROJECT_ENDPOINT", "AZURE_AI_MODEL_DEPLOYMENT_NAME"]
    for name in required_env_vars:
        print(f"[env] {name} is set: {bool(os.getenv(name))}", flush=True)


    # --- エージェントを実行後に「削除」されないようにする設定 ---
    # 既定では should_cleanup_agent=True のため、agent_id を指定せずに自動作成されたエージェントは
    # `async with` を抜けて close されるタイミングで削除されます。
    # 1) この実行で作ったエージェントを close 時に削除しない
    # async with AzureAIAgentClient(credential=credential, should_cleanup_agent=False) as chat_client:

    # 2) 既存エージェントを使う（agent_id を指定した既存エージェントは削除されません）
    # existing_agent_id = os.getenv("AZURE_AI_AGENT_ID")
    # async with AzureAIAgentClient(
    #     credential=credential,
    #     agent_id=existing_agent_id,
    #     should_cleanup_agent=False,
    # ) as chat_client:
    # agent = chat_client.create_agent()

    async with (
        DefaultAzureCredential() as credential,
        AzureAIAgentClient(credential=credential) as chat_client,
    ):

        # Create agent with hosted MCP tool
        agent = chat_client.create_agent(
            name="MicrosoftLearnAgent", 
            instructions="You answer questions by searching Microsoft Learn content only.",
            tools=HostedMCPTool(
                name="Microsoft Learn MCP",
                url="https://learn.microsoft.com/api/mcp",
                approval_mode="never_require",
            ),
        )

        # Stream output so you can see progress even if it takes time.
        prompt = "Microsoft FoundryとAgent Frameworkの最新のアップデートを要約してください。"
        print("[run] starting agent.run_stream...", flush=True)
        try:
            async def _run() -> None:
                async for update in agent.run_stream(prompt):
                    text = getattr(update, "text", None)
                    if text:
                        print(text, end="", flush=True)
                print("\n[run] completed", flush=True)

            await asyncio.wait_for(_run(), timeout=180)
        except asyncio.TimeoutError:
            print("\n[run] timed out (180s). Check auth/network/project settings.", flush=True)

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

実行結果

コードを実行すると、Microsoft Foundry に MCP Tool を利用するエージェントが作成され、回答生成が実行されます。

エージェントが実行後に削除されない設定にしておくと、Microsoft Foundry で作成されたエージェントを確認することができます。

Agents playground では、エージェントの実行内容を確認することができ、MCP Tool を利用して回答を生成していることが確認可能です。

Microsoft Foundry のエージェントを Agent Framework でオーケストレーションする

以下のコードでは、Agent Framework を使い既存の Microsoft Foundry のエージェントと新しい Microsoft Foundry のエージェントを作成し、並列で実行します。

Agent Framework を利用することで、Microsoft Foundry のエージェントを含めた複数エージェントをオーケストレーションすることが可能です。

サンプルコード

長いため折りたたみます。

エージェントの並列実行(クリックで開く)

import os
import asyncio

from agent_framework import (
    AgentExecutor,
    ChatMessage,
    ConcurrentBuilder,
    FunctionApprovalRequestContent,
    HostedMCPTool,
    RequestInfoEvent,
    WorkflowFailedEvent,
    WorkflowOutputEvent,
)
from agent_framework_azure_ai import AzureAIAgentClient
from azure.identity.aio import DefaultAzureCredential

# ローカル環境で.envから環境変数を読み取る場合に利用
from dotenv import load_dotenv
load_dotenv(dotenv_path=".env", override=True)


async def _run_workflow_with_auto_responses(
    *,
    workflow,
    prompt: str,
    auto_approve: bool,
    request_info_default_text: str = "",
    timeout_seconds: int = 300,
) -> list[ChatMessage] | None:
    """Run a workflow, auto-responding to RequestInfoEvent when needed.

    This is critical for Foundry agents when a run requires tool approval:
    the workflow will emit RequestInfoEvent(s) and become idle until you send responses.
    """

    async def _drain_event_stream(event_stream) -> tuple[list[ChatMessage] | None, dict[str, object]]:
        final_messages: list[ChatMessage] | None = None
        responses: dict[str, object] = {}

        async for event in event_stream:
            if isinstance(event, WorkflowOutputEvent):
                data = event.data
                if isinstance(data, list) and (not data or isinstance(data[0], ChatMessage)):
                    final_messages = data
                continue

            if isinstance(event, RequestInfoEvent):
                req = event.data
                if isinstance(req, FunctionApprovalRequestContent):
                    if auto_approve:
                        responses[event.request_id] = req.create_response(approved=True)
                    else:
                        raise RuntimeError("Tool approval required, but auto_approve=False")
                elif event.response_type is str:
                    responses[event.request_id] = request_info_default_text
                else:
                    raise RuntimeError(
                        f"Unhandled RequestInfoEvent response_type={event.response_type} request_type={type(req)}"
                    )
                continue

            if isinstance(event, WorkflowFailedEvent):
                raise RuntimeError(f"Workflow failed: {event.details.error_type}: {event.details.message}")

        return final_messages, responses

    async def _inner() -> list[ChatMessage] | None:
        final_messages, pending_responses = await _drain_event_stream(workflow.run_stream(prompt))

        while pending_responses:
            final2, pending_responses = await _drain_event_stream(
                workflow.send_responses_streaming(pending_responses)
            )
            if final2 is not None:
                final_messages = final2

        return final_messages

    return await asyncio.wait_for(_inner(), timeout=timeout_seconds)


async def main() -> None:
    print("[start] agent_framework_foundry_agents_concurrent", flush=True)

    print(f"[env] AZURE_AI_PROJECT_ENDPOINT is set: {bool(project_endpoint)}", flush=True)
    print(f"[env] AZURE_AI_MODEL_DEPLOYMENT_NAME is set: {bool(model_deployment_name)}", flush=True)
    print(f"[env] AZURE_AI_AGENT_ID is set: {bool(existing_agent_id)}", flush=True)

    if not project_endpoint:
        raise RuntimeError("AZURE_AI_PROJECT_ENDPOINT is required")
    if not model_deployment_name:
        raise RuntimeError("AZURE_AI_MODEL_DEPLOYMENT_NAME is required")
    if not existing_agent_id:
        raise RuntimeError("AZURE_AI_AGENT_ID is required (existing agent)")

    # 既存エージェントは agent_id 指定で利用するため、このスクリプト側で name を明示しないと
    # フレームワークのメッセージ帰属表示が 'UnnamedAgent' になることがある。
    existing_agent_display_name = os.getenv("AZURE_AI_EXISTING_AGENT_DISPLAY_NAME") or "MicrosoftLearnAgent"

    prompt = (
        "Microsoft FoundryとAgent Frameworkの最新のアップデートを要約し、"
        "開発者が注意すべき点を3つ挙げてください。日本語で簡潔に回答してください。"
    )

    async with DefaultAzureCredential() as credential:
        # 既存エージェント用クライアント: agent_id を渡す (既存エージェントは削除されない)
        async with AzureAIAgentClient(
            credential=credential,
            agent_id=existing_agent_id,
            should_cleanup_agent=False,
        ) as existing_client:
            existing_agent = existing_client.create_agent(name=existing_agent_display_name)

            # 新規エージェント用クライアント: agent_id なし (必要なら close 時に削除される)
            async with AzureAIAgentClient(
                credential=credential,
                # should_cleanup_agent=False にすると、この実行で作ったエージェントを残せます
            ) as new_client:
                new_agent = new_client.create_agent(
                    name="NewFoundryAgent",
                    instructions="一般的な情報を答えるエージェントです。"
                )

                # ConcurrentBuilder で (既存1つ + 新規1つ) を同時実行
                existing_exec = AgentExecutor(agent=existing_agent, id=f"foundry_existing:{existing_agent_id}")
                new_exec = AgentExecutor(agent=new_agent, id="foundry_new")
                workflow = ConcurrentBuilder().participants([existing_exec, new_exec]).build()
                final_messages = await _run_workflow_with_auto_responses(
                    workflow=workflow,
                    prompt=prompt,
                    auto_approve=True,
                )      

    print("\n===== Final Aggregated Conversation (messages) =====", flush=True)
    if not final_messages:
        print("No final ChatMessage list was produced.")
        return

    for i, msg in enumerate(final_messages, start=1):
        name = getattr(msg, "author_name", None) or getattr(msg, "role", None) or "user"
        sep = "-" * 60
        print(f"{sep}\n\n{i:02d} [{name}]:\n{msg.text}")


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