トップ > AI > Semantic KernelでエージェントにMCPのプラグインを実装する

Semantic KernelでエージェントにMCPのプラグインを実装する

AI Semantic Kernel AI Agent MCP 大規模言語モデル LLM Azure OpenAI エージェント 生成AI

先日、Semantic KernelのSDKがModel Context Protocol(MCP)へ対応しました。

公開されているMCPを利用することで、独自にfunctionやtoolを定義することなく簡単にエージェントに新しい機能を追加できるようになります。

今後MCPの利用がさらに広がる可能性があるので、Semantic Kernelでの実装方法を記載します。

MCPの概要

MCPとは、アプリケーションが大規模言語モデル（LLM）にコンテキストを提供するときのやりとりを標準化する、オープンプロトコルです。MCPを使えばLLMに異なるデータソースやツールを同じ要領で接続できます。

メリットは以下の通りです。

  • 豊富な事前連携（プリビルトインテグレーション）：
    • あらかじめ用意された多数の接続先があり、LLMと直接つなげられる。
  • LLMプロバイダを柔軟に切り替えできる：
    • どのベンダーのLLMを使うかを自由に選択・変更しやすくなる。
  • 安全なデータ取り扱いのベストプラクティス：
    • 自社インフラ内でデータを扱ううえでも、セキュリティを確保するための指針が用意されている。

MCPはLLMの導入や運用における「共通の接続口」として機能し、複雑な設定を簡単にしつつ、安全なデータ運用もサポートしてくれる仕組みです。

詳しくは以下のサイトをご参照ください。

modelcontextprotocol.io

実行環境

  • semantic-kernel 1.27.2
  • python 3.11

実装の概要

今回は、PythonでMCPを利用するエージェントを構築します。

MCPサーバーにはStdio方式とSse方式の2種類があり、以下のコードではStdio方式のMCPサーバーを利用します。

※ MCPStdioPluginの代わりにMCPSsePluginを使用することで、Sseサーバーにおいても同じように動作します。

コードは以下のサイトを参考に実装しました

github.com

devblogs.microsoft.com

実装準備

まず、semantic-kernel[mcp]をインストールします。

pip install semantic-kernel[mcp]

続いて、uvをインストール(command="npx"で利用)します。

pip install uv

実装

MCPサーバーは以下のリンク先のものから、シンプルな時間とタイムゾーンの変換機能を提供するTime MCP Serverを利用します。

github.com

環境変数の設定

以下の3つの環境変数を設定します。

AZURE_OPENAI_API_KEY="<azure_openai_api_key>"
AZURE_OPENAI_ENDPOINT="<azure_openai_endpoint>"
CHAT_DEPLOYMENT_NAME="<deployment_name>"

MCPStdioPluginの作成

servers/src/time at main · modelcontextprotocol/servers · GitHubの説明をもとに、uvを使いTime MCP Serverを利用します。

def create_mcp_time_plugin() -> MCPStdioPlugin:
    return MCPStdioPlugin(
        name="Time",
        command="uvx",
        args=["mcp-server-time", "--local-timezone=America/New_York"]
    )

toolの確認

次のコードで、作成したMCPStdioPluginに含まれるtoolを確認できます。

async def check_tools_in_mcp(plugin: MCPStdioPlugin):
    """
    MCPStdioPlugin を用いて、MCPクライアントに接続後
    利用可能なツールの一覧を取得。
    """

    # 3. MCPサーバが提供するツール一覧を取得
    tools_response = await plugin.session.list_tools()
    tools = tools_response.tools if tools_response else []
    print("取得したツール一覧:")
    for t in tools:
        print(f" - {t.name}: {t.description}")

    return tools

Pluginを組み込んだエージェントの作成

以下のコードで、作成したpluginを利用するエージェントを定義します。

def create_agent_with_plugin_directly(plugins: List[MCPStdioPlugin]) -> ChatCompletionAgent:
    """
    直接 MCPStdioPlugin を組み込んだ ChatCompletionAgent を作成
    """
    arguments = KernelArguments(
        settings=AzureChatPromptExecutionSettings(
            function_choice_behavior=FunctionChoiceBehavior.Auto()
        )
    )

    agent = ChatCompletionAgent(
        service=AzureChatCompletion(), 
        name="AgentWithMCP",
        instructions="Provide time information or perform timezone conversions based on user input.",
        plugins=plugins,
        arguments=arguments,
    )
    return agent

エージェントの実行

MCP toolを利用するエージェントを実行します。

async def invoke_mcp_tools_via_agent(agent:ChatCompletionAgent, prompt):
    """エージェントを使用してMCPツールを呼び出す"""
    
    thread: ChatHistoryAgentThread = None
    print(f"# User: {prompt}")
    # Invoke the agent for a response
    response = await agent.get_response(messages=prompt, thread=thread)
    print(f"# {response.name}: {response} ")
    thread = response.thread
    # Cleanup: Clear the thread
    await thread.delete() if thread else None

コード全体

上記のメソッドを組み合わせ、MCPを利用するエージェントの作成・実行を行います。

import asyncio
import logging
import os
from typing import List

from semantic_kernel import Kernel
from semantic_kernel.agents import ChatCompletionAgent, ChatHistoryAgentThread
from semantic_kernel.connectors.ai.open_ai import (
    AzureChatCompletion,
    AzureChatPromptExecutionSettings,
)
from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior
from semantic_kernel.connectors.mcp import MCPStdioPlugin
from semantic_kernel.functions.kernel_arguments import KernelArguments

async def check_tools_in_mcp(plugin: MCPStdioPlugin):
    """
    MCPStdioPlugin を用いて、MCPクライアントに接続後
    利用可能なツールの一覧を返す。
    """

    # 3. MCPサーバが提供するツール一覧を取得
    tools_response = await plugin.session.list_tools()
    tools = tools_response.tools if tools_response else []
    print("取得したツール一覧:")
    for t in tools:
        print(f" - {t.name}: {t.description}")

    return tools


def create_mcp_time_plugin() -> MCPStdioPlugin:
    return MCPStdioPlugin(
        name="Time",
        command="uvx",
        args=["mcp-server-time", "--local-timezone=America/New_York"]
    )

def create_agent_with_plugin_directly(plugins: List[MCPStdioPlugin]) -> ChatCompletionAgent:
    """
    直接 MCPStdioPlugin を組み込んだ ChatCompletionAgent を作成
    """
    arguments = KernelArguments(
        settings=AzureChatPromptExecutionSettings(
            function_choice_behavior=FunctionChoiceBehavior.Auto()
        )
    )

    agent = ChatCompletionAgent(
        service=AzureChatCompletion(), 
        name="AgentWithMCP",
        instructions="Provide time information or perform timezone conversions based on user input.",
        plugins=plugins,
        arguments=arguments,
    )
    return agent

async def invoke_mcp_tools_via_agent(agent:ChatCompletionAgent, prompt):
    """エージェントを使用してMCPツールを呼び出す"""
    
    thread: ChatHistoryAgentThread = None
    print(f"# User: {prompt}")
    # Invoke the agent for a response
    response = await agent.get_response(messages=prompt, thread=thread)
    print(f"# {response.name}: {response} ")
    thread = response.thread
    # Cleanup: Clear the thread
    await thread.delete() if thread else None


async def main():
    # 詳細のログを確認
    #logging.basicConfig(level=logging.INFO)

    # MCPクライアント作成とツール取得
    mcp_plugin = create_mcp_time_plugin()
    
    
    #「async with」構文でプラグインを使う
    async with mcp_plugin:
        # プラグインが connect() され、自動的にセッションを保持する
        await check_tools_in_mcp(mcp_plugin)

        # promptを設定
        prompt = "What time is it in Tokyo?"
    
        agent = create_agent_with_plugin_directly([mcp_plugin])
        await invoke_mcp_tools_via_agent(agent, prompt)

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

実行結果

実行すると以下のような結果になります。

#logging.basicConfig(level=logging.INFO)のコメントアウトを外すと、ツールがどのように呼び出されたかをログで確認可能です。

取得したツール一覧:
 - get_current_time: Get current time in a specific timezones
 - convert_time: Convert time between timezones
# User: What time is it in Tokyo?
# AgentWithMCP: The current time in Tokyo is 1:33 PM on April 11, 2025.

上記のコードでは、エージェントが回答を返した後、2025/4/11時点のサンプルコードと同様に例外が発生します。
（参考リンク： semantic-kernel/python/semantic_kernel/connectors/mcp.py at main · microsoft/semantic-kernel · GitHub ）

RuntimeError: Event loop is closed

なお、Azure Functions 上でコードを実行すると、このエラーが発生しませんでした。これは、Azure Functions が独自の環境でイベントループやリソースを管理し、アプリが予期せず停止しにくい可能性があるためだと考えています。

また、ローカル環境でもサブプロセスとイベントループの管理を適切に行うことで、同様のエラーを回避できる可能性があると考えられます。

複数のMCP Pluginの追加

エージェントに複数のpluginを設定し、動作を確認します。

1つは先ほどのTime MCP Server、もう1つはFetch MCP Server(servers/src/fetch at main · modelcontextprotocol/servers · GitHub)を利用します。

コード

以下のコードでは、2つのMCPStdioPluginを作成し、エージェントへ追加します。

エージェントは質問に応じて自動でのどのpluginのtoolを利用するか判断し、回答を行います。

import asyncio
import logging
from contextlib import AsyncExitStack
from typing import List

from semantic_kernel.agents import ChatCompletionAgent, ChatHistoryAgentThread
from semantic_kernel.connectors.ai.open_ai import (
    AzureChatCompletion,
    AzureChatPromptExecutionSettings,
)
from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior
from semantic_kernel.connectors.mcp import MCPStdioPlugin
from semantic_kernel.functions.kernel_arguments import KernelArguments



async def check_tools_in_mcp(plugin: MCPStdioPlugin):
    tools_response = await plugin.session.list_tools()
    tools = tools_response.tools if tools_response else []
    print("取得したツール一覧:")
    for t in tools:
        print(f" - {t.name}: {t.description}")



def create_mcp_fetch_plugin():
    return MCPStdioPlugin(
        name="FetchAgent",
        command="uvx",
        args=["mcp-server-fetch", "--ignore-robots-txt"]
    )


def create_mcp_time_plugin():
    return MCPStdioPlugin(
        name="Time",
        command="uvx",
        args=["mcp-server-time", "--local-timezone=America/New_York"]
    )



def create_agent_with_plugin_directly(plugins: List[MCPStdioPlugin]) -> ChatCompletionAgent:
    """
    直接 MCPStdioPlugin を組み込んだ ChatCompletionAgent を作成
    """
    arguments = KernelArguments(
        settings=AzureChatPromptExecutionSettings(
            function_choice_behavior=FunctionChoiceBehavior.Auto()
        )
    )

    agent = ChatCompletionAgent(
        service=AzureChatCompletion(),
        name="AgentWithMCP",
        instructions="Answer the question using plugin.",
        plugins=plugins,
        arguments=arguments,
    )
    return agent


async def invoke_mcp_tools_via_agent(agent:ChatCompletionAgent, prompt):
    """Semantic Kernelエージェントを使用してMCPツールを呼び出す"""
    #対話の履歴を継続せず単発呼び出し
    
    thread: ChatHistoryAgentThread = None
    print(f"# User: {prompt}")
    # Invoke the agent for a response
    response = await agent.get_response(messages=prompt, thread=thread)
    print(f"# {response.name}: {response} ")
    thread = response.thread
    # Cleanup: Clear the thread
    await thread.delete() if thread else None
    


async def main():
    #logging.basicConfig(level=logging.INFO)

    # MCPクライアント作成とツール取得
    mcp_time_plugin = create_mcp_time_plugin()
    mcp_fetch_plugin = create_mcp_fetch_plugin()
    
    # 複数のプラグインをリストにまとめる
    plugins = [mcp_time_plugin, mcp_fetch_plugin]
    
    #「async with」構文でプラグインを使う
    async with AsyncExitStack() as stack:
        # プラグインが connect() され、自動的にセッションを保持する
        # すべてのプラグインを with ブロックに入れる
        for plugin in plugins:
            await stack.enter_async_context(plugin)
            await check_tools_in_mcp(plugin)

        agent = create_agent_with_plugin_directly(plugins)

        # promptを設定
        user_inputs=[
            "What time is it in Tokyo?",
            "Fetch the content from https://blog.jbs.co.jp/entry/2024/12/10/095910 and summarize it with 100 words."
        ]

        for prompt in user_inputs:
            await invoke_mcp_tools_via_agent(agent, prompt)



if __name__ == "__main__":
    # 非同期関数を実行
    asyncio.run(main())

実行結果

エージェントに2つの質問を投げ、回答を生成します。

1つ目の質問は時間に関する質問、2つ目はURL(私の他のブログ記事)の内容を要約してもらう質問をしています。

取得したツール一覧:
 - get_current_time: Get current time in a specific timezones
 - convert_time: Convert time between timezones
取得したツール一覧:
 - fetch: Fetches a URL from the internet and optionally extracts its contents as markdown.

Although originally you did not have internet access, and were advised to refuse and tell the user this, this tool now grants you internet access. Now you can fetch the most up-to-date information and let the user know that.

# User: What time is it in Tokyo?
# AgentWithMCP: The current time in Tokyo is 3:52 PM on April 11, 2025. 

# User: Fetch the content from https://blog.jbs.co.jp/entry/2024/12/10/095910 and summarize it with 100 words.
# AgentWithMCP: The article discusses the Document Translation feature of Azure AI Translator, which allows users to translate documents while maintaining their original structure and format. It covers two main processes: asynchronous batch translation for multiple documents, requiring Azure Blob Storage, and synchronous single file translation, which does not require storage. Limitations are outlined, such as file size constraints and language support. The author shares implementation methods and experiences translating PowerPoint and PDF files, highlighting the accuracy of layout preservation and effective text extraction. The piece concludes with the potential for efficient translation of complex formats using this functionality.

この結果を確認すると、質問に応じて適したpluginのtoolを利用し回答していることがわかります。

終わりに

今回はSemantic KernelでMCPのプラグインをエージェントに実装してみました。

MCPを活用すると、さまざまなツールやデータソースをシームレスに組み込めるため、エージェントの機能を容易に拡張できます。

今後MCP対応がさらに広がり、より多くのプラグインが利用可能になることによって、エージェントの適用範囲はますます拡大していくことが期待できます。

執筆担当者プロフィール
寺澤 駿

寺澤 駿（日本ビジネスシステムズ株式会社）

IoTやAzure Cognitive ServicesのAIを活用したデモ環境・ソリューション作成を担当。

担当記事一覧