生成AIチャットボットでMCP (Model Context Protocl)を使ってみよう #langchain #mcp #ai #llm #slack

AIChatGPT＆AId-higuchiLLMPython

はじめに

生成AIチャットClaudeや生成AIコーディングツールClaude Codeで著名なAnthropicが、Model Context Protocol (MCP)をオープンソース化してから、2025年11月で1年が経とうとしています。新技術の流行り廃りの勢いが物凄い昨今、MCPは一定の影響力を保ったまま推移していると見てよいでしょう。

本稿では、以前に公開した「LangChainでMCP (Model Context Protocol)を使ってみよう」からにさらに一歩進め、生成AI付SlackチャットボットからMCPを使ってみることにします。

前提条件

次の環境で実施しています。

  • Python 3.13.9
  • LangChain 0.3.27
  • LangChain MCP Adapters 0.1.12
  • LangGraph 1.0.1
  • Slack Bolt 1.26.0
  • Slack SDK 3.37.0

生成AIは Azure OpenAIを使用します。

おさらい: 生成AI付Slackチャットボット

元になるプレーンな生成AI付Slackチャットボットのソースコードは次の通りです。

過去記事「SlackボットをAzure App Serviceで動かそう」からは非同期動作用とテスト用にソースコードを少し変更しています。実際に動作させるための設定などは、引き続き過去記事を参照してください。

#!python3

from slack_bolt.async_app import AsyncApp
from slack_bolt.adapter.fastapi.async_handler import AsyncSlackRequestHandler
from fastapi import FastAPI, Request, Response
import uvicorn

import re

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, AIMessage, HumanMessage
from langchain_openai import AzureChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver

import os
from dotenv import load_dotenv
load_dotenv()

# os.getenv("AZURE_OPENAI_API_KEY")
# os.getenv("AZURE_OPENAI_ENDPOINT")
llm = AzureChatOpenAI(
         azure_deployment = os.getenv("AZURE_OPENAI_MODEL_NAME"),
         api_version = os.getenv("AZURE_OPENAI_API_VERSION"),
         temperature = 0.95,
         max_tokens = 1024,
      )

class State(TypedDict):
    messages: Annotated[list, add_messages]

async def chatbot(state: State):
    res = await llm.ainvoke(state["messages"])
    return {"messages": res}

graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.set_entry_point("chatbot")
graph_builder.set_finish_point("chatbot")

app = AsyncApp(
  signing_secret=os.environ.get("SLACK_SIGNING_SECRET"),
  token=os.getenv("SLACK_BOT_TOKEN")
)
slack_bot_id = os.getenv("SLACK_BOT_ID")

async def add_reaction(event, client, emoji):
    await client.reactions_add(
        channel=event["channel"],
        name=emoji,
        timestamp=event["ts"],
    )

@app.event("app_mention")
async def mention_reply(event, say, client):
    await add_reaction(event, client, "eyes")
    memory = InMemorySaver()
    graph = graph_builder.compile(checkpointer=memory)

    user = event['user']
    text = event['text']
    if 'thread_ts' in event:
        thread_ts = event['thread_ts']
    else:
        thread_ts = event['ts']

    msg = re.sub(f'<@{slack_bot_id}>', '', text)
    events = graph.astream(
        {"messages": [("user", msg)]},
        {"configurable": {"thread_id": thread_ts}},
        stream_mode="values"
    )
    async for ev in events:
        message = ev["messages"][-1]
        if type(message) == AIMessage:
            await say(
                text=message.content,
                thread_ts=thread_ts
            )

fastapi_app = FastAPI()
handler = AsyncSlackRequestHandler(app)

@fastapi_app.post("/slack/events")
async def slack_events(request: Request) -> Response:
    return await handler.handle(request)

if __name__ == "__main__":
    subroutine_name = os.path.splitext(os.path.basename(__file__))[0]
    uvicorn.run(f"{subroutine_name}:fastapi_app", host="0.0.0.0", port=8080)

これでSlackでメンションすると、スレッドで回答してくれるボットとなっています。

MCPサーバの準備

以前の記事「LangChainでMCPを使ってみよう」と同様に、MCPサーバを準備していきましょう。

今回も実用的な作業をしないモックですが、一応指示に即した応答「現在の日時を聞くと、それを応答する」機能を持たせます。

#!/usr/bin/env python3
"""MCP 日時サーバー.

現在の日時取得機能を提供します。
"""

from datetime import datetime

from mcp.server import Server
from mcp.types import Tool, TextContent
import mcp.server.stdio

app = Server("time-server")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="time_get_current",
            description="現在の日時を返します",
            inputSchema={
                "type": "object",
                "properties": {},
            },
        ),
    ]

@app.call_tool()
async def call_tool(name: str, _arguments: dict) -> list[TextContent]:
    if name == "time_get_current":
        now = datetime.now()
        return [
            TextContent(
                type="text",
                text=f"現在の日時: {now.strftime('%Y-%m-%d %H:%M:%S')}",
            ),
        ]
    msg = f"Unknown tool: {name}"
    raise ValueError(msg)

async def main() -> None:
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options(),
        )

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

しかしこれは低レベルレイヤーでの作成であり、処理が詳細に分かるのですが、ちょっと冗長です。そこでここではFastMCPを使って書き直しましょう。前回公式クイックスタートでFastMCPのコードを特に言及なく使っていましたが、この仕組みを使うとMCPサーバを簡単に記述することができます。

#!/usr/bin/env python3
"""FastMCP 日時サーバー.

現在の日時取得機能を提供します。
"""

from datetime import datetime
from mcp.server.fastmcp import FastMCP

# FastMCPサーバーを作成
mcp = FastMCP("time-server")

@mcp.tool()
def time_get_current() -> str:
    """現在の日時を返します"""
    now = datetime.now()
    return f"現在の日時: {now.strftime('%Y-%m-%d %H:%M:%S')}"

if __name__ == "__main__":
    mcp.run(transport="stdio")

作業内容は同じで、コードはこんなに短くなりました。本稿ではFastMCP仕様のサーバを使っていきます。

MCPクライアントの準備

では、SlackボットにMCPクライアントを組み込んでいきましょう。全体のソースコードは次のようになります。

#!python3

from slack_bolt.async_app import AsyncApp
from slack_bolt.adapter.fastapi.async_handler import AsyncSlackRequestHandler
from fastapi import FastAPI, Request, Response
import uvicorn

import re
import asyncio

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, AIMessage, HumanMessage, ToolMessage
from langchain_openai import AzureChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langchain_mcp_adapters.client import MultiServerMCPClient

import os
from dotenv import load_dotenv
load_dotenv()

# os.getenv("AZURE_OPENAI_API_KEY")
# os.getenv("AZURE_OPENAI_ENDPOINT")
llm = AzureChatOpenAI(
         azure_deployment = os.getenv("AZURE_OPENAI_MODEL_NAME"),
         api_version = os.getenv("AZURE_OPENAI_API_VERSION"),
         temperature = 0.95,
         max_tokens = 1024,
      )

MCP_SERVERS = {
    "time-server": {
        "command": "python3",
        "args": ["100_2_fastmcp_time_server.py"],
        "transport": "stdio",
        "tools": ["time_get_current"],
        "reaction": "clock3",
    },
}

mcp_client_instance = None

class State(TypedDict):
    messages: Annotated[list, add_messages]

async def chatbot(state: State):
    tools = await mcp_client_instance.get_tools()
    llm_with_tools = llm.bind_tools(tools)
    res = await llm_with_tools.ainvoke(state["messages"])
    return {"messages": res}

async def tool_executor(state: State):
    mcp_tools = await mcp_client_instance.get_tools()
    tool_node = ToolNode(tools=mcp_tools)
    return await tool_node.ainvoke(state)

graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", tool_executor)
graph_builder.set_entry_point("chatbot")
graph_builder.add_conditional_edges("chatbot", tools_condition)
graph_builder.add_edge("tools", "chatbot")

app = AsyncApp(
  signing_secret=os.environ.get("SLACK_SIGNING_SECRET"),
  token=os.getenv("SLACK_BOT_TOKEN")
)
slack_bot_id = os.getenv("SLACK_BOT_ID")

async def add_reaction(event, client, emoji):
    await client.reactions_add(
        channel=event["channel"],
        name=emoji,
        timestamp=event["ts"],
    )

@app.event("app_mention")
async def mention_reply(event, say, client):
    await add_reaction(event, client, "eyes")

    global mcp_client_instance

    mcp_client_config = {}
    for server_name, config in MCP_SERVERS.items():
        mcp_client_config[server_name] = {
            k: v for k, v in config.items()
            if k not in ['tools', 'reaction']
        }

    mcp_client = MultiServerMCPClient(mcp_client_config)
    mcp_client_instance = mcp_client

    memory = InMemorySaver()
    graph = graph_builder.compile(checkpointer=memory)

    user = event['user']
    text = event['text']
    if 'thread_ts' in event:
        thread_ts = event['thread_ts']
    else:
        thread_ts = event['ts']

    msg = re.sub(f'<@{slack_bot_id}>', '', text)
    events = graph.astream(
        {"messages": [("user", msg)]},
        {"configurable": {"thread_id": thread_ts}},
        stream_mode="values"
    )
    async for ev in events:
        messages = ev.get("messages", [])
        if not messages:
            continue
        message = messages[-1]

        if isinstance(message, AIMessage):
            if message.content == "":
                continue

            content_str = str(message.content)
            if content_str.strip():
                await say(
                    text=content_str,
                    thread_ts=thread_ts
                )
        elif isinstance(message, ToolMessage):
            tool_name = message.name or "unknown"
            tool_content = message.content
            content_str = str(tool_content) if isinstance(tool_content, list) else str(tool_content)

            for server_name, server_config in MCP_SERVERS.items():
                if "tools" in server_config and tool_name in server_config["tools"]:
                    if "reaction" in server_config:
                        await add_reaction(event, client, server_config["reaction"])
                    break

            if message.status == "error":
                error_message = f"⚠️ ツール実行エラー: `{tool_name}`\n```\n{content_str}\n```"
                await say(
                    text=error_message,
                    thread_ts=thread_ts
                )

fastapi_app = FastAPI()
handler = AsyncSlackRequestHandler(app)

@fastapi_app.post("/slack/events")
async def slack_events(request: Request) -> Response:
    return await handler.handle(request)

if __name__ == "__main__":
    subroutine_name = os.path.splitext(os.path.basename(__file__))[0]
    uvicorn.run(f"{subroutine_name}:fastapi_app", host="0.0.0.0", port=8080)

変更箇所をかいつまんで見ていきましょう。

+from langchain_mcp_adapters.client import MultiServerMCPClient

前回公式クイックスタートでも見た通り、langchain-mcp-adaptersライブラリのマルチサーバ対応MCPクライアントを利用します。

+MCP_SERVERS = {
+    "time-server": {
+        "command": "python3",
+        "args": ["100_2_fastmcp_time_server.py"],
+        "transport": "stdio",
+        "tools": ["time_get_current"],
+        "reaction": "clock3",
+    },
+}

利用するMCPサーバを定義しています。この 100_2_fastmcp_time_server.py は先に作成したFastMCP仕様のMCPサーバです。なお、 tools と reaction は、本Slackチャットボットで利用するための独自要素です。

 async def chatbot(state: State):
-    res = await llm.ainvoke(state["messages"])
+    tools = await mcp_client_instance.get_tools()
+    llm_with_tools = llm.bind_tools(tools)
+    res = await llm_with_tools.ainvoke(state["messages"])
     return {"messages": res}

+async def tool_executor(state: State):
+    mcp_tools = await mcp_client_instance.get_tools()
+    tool_node = ToolNode(tools=mcp_tools)
+    return await tool_node.ainvoke(state)
+
 graph_builder = StateGraph(State)
 graph_builder.add_node("chatbot", chatbot)
+graph_builder.add_node("tools", tool_executor)
 graph_builder.set_entry_point("chatbot")
-graph_builder.set_finish_point("chatbot")
+graph_builder.add_conditional_edges("chatbot", tools_condition)
+graph_builder.add_edge("tools", "chatbot")

チャットボットのLangGraphノードにMCPクライアントを組み込んでいます。

+    mcp_client_config = {}
+    for server_name, config in MCP_SERVERS.items():
+        mcp_client_config[server_name] = {
+            k: v for k, v in config.items()
+            if k not in ['tools', 'reaction']
+        }
+
+    mcp_client = MultiServerMCPClient(mcp_client_config)
+    mcp_client_instance = mcp_client

先の MCP_SERVERS の定義からマルチサーバ対応MCPクライアントのインスタンスを作成しています。前述の通り tools と reaction は独自要素なので削除しています。保持したままだと初期化エラーになります。

     async for ev in events:
-        message = ev["messages"][-1]
-        if type(message) == AIMessage:
-            await say(
-                text=message.content,
-                thread_ts=thread_ts
-            )
+        messages = ev.get("messages", [])
+        if not messages:
+            continue
+        message = messages[-1]
+
+        if isinstance(message, AIMessage):
+            if message.content == "":
+                continue
+
+            content_str = str(message.content)
+            if content_str.strip():
+                await say(
+                    text=content_str,
+                    thread_ts=thread_ts
+                )

通常のメッセージの処理部分です。

MCPサーバを含むツールからの応答は message.content がカラの場合があるため、スキップして次の応答に移ります。

+        elif isinstance(message, ToolMessage):
+            tool_name = message.name or "unknown"
+            tool_content = message.content
+            content_str = str(tool_content) if isinstance(tool_content, list) else str(tool_content)
+
+            for server_name, server_config in MCP_SERVERS.items():
+                if "tools" in server_config and tool_name in server_config["tools"]:
+                    if "reaction" in server_config:
+                        await add_reaction(event, client, server_config["reaction"])
+                    break
+
+            if message.status == "error":
+                error_message = f"⚠️ ツール実行エラー: `{tool_name}`\n```\n{content_str}\n```"
+                await say(
+                    text=error_message,
+                    thread_ts=thread_ts
+                )

MCPサーバを含むツールからのメッセージの処理部分です。

独自要素の tools と reaction を用いて、メッセージスレッドの先頭メッセージに、指定の絵文字リアクションをつけます。これにより、LLM直通ではなく、ツールを経由して処理したメッセージであることがわかりやすくなります。

また、ツールがエラーを返した場合は専用のエラーメッセージを送信するようにしています。

実行結果

では、実行してみましょう。

このように、MCPサーバを起動して、そちらで日時の取得を行って返答しています。

既存のMCPサーバの追加

MultiServerMCPClient は複数のMCPサーバを取り扱うことができるので、独自の日時取得MCPサーバに加えて、既存のMCPサーバを追加してみましょう。

ここではウェブ検索ソリューションであるTavilyのMCPサーバのTavily MCP Serverを追加します。

--- 100_slack_simple_bot_mcp.py    2025-11-10 18:41:19.421626450 +0900
+++ 100_1_slack_simple_bot_mcp.py    2025-11-10 18:41:08.893055758 +0900
@@ -39,6 +39,16 @@
         "tools": ["time_get_current"],
         "reaction": "clock3",
     },
+    "tavily-mcp": {
+      "command": "npx",
+      "args": ["-y", "tavily-mcp@latest"],
+      "env": {
+         "TAVILY_API_KEY": os.getenv("TAVILY_API_KEY"),
+      },
+      "transport": "stdio",
+      "tools": ["tavily-search"],
+      "reaction": "mag",
+    },
 }

 mcp_client_instance = None

これはNode.jsのパッケージ管理システムnpxでTavilyのMCPサーバを起動（および初回のみインストール）を行っています。繰り返しとなりますが tools と reaction は本稿での独自要素です。

ではこちらも実行してみましょう。

GPT-4.1はカットオフ2024年6月、つまりその時点までの知識で作られています。そのため、2025年11月現在の日本の総理大臣についての知識がありません。そこでTavilyで検索することで、正しい結果を得ることができています。

まとめ

本稿では以前の記事で作成した「生成AI付Slackチャットボット」にMCPクライアントを組み込み、MCPサーバを利用して会話内容を処理できるようにしてみました。

MCPというフレームワークを用いることにより、チャットボットにさまざまな機能を簡単に追加することができます。これまでは独自にシステムを構成しなければいけなかったところが、とても簡単になったと感じます。

一方で、MCPに関しては多くの安全上の懸念点が提起されています。例えば、今回の例のように複数のMCPサーバを組み込んだ場合、予期せぬMCPサーバにメッセージが転送されてしまう可能性があります。ここでは「現在時刻」と「ウェブ検索」の組み合わせだったので、現在時刻の問い合わせのつもりがウェブ検索に流れてしまっても問題ありませんが、「顧客情報」と「ウェブ検索」の組み合わせだったら？ 顧客情報がウェブ検索に流れていってしまうかもしれません。組み込みは簡単になりましたが、簡単なぶん、思わぬ事態を引き起こしやすくなったと言えるかもしれません。

MCPはまだまだ改善の余地がありますが、デファクトスタンダードになると見越されているため、早めにテストし、どのような利点や改善点があるのか知っておくとよいと考えます。

引き続き、生成AIとMCPの動向を追っていきたいと思います。

Author

Daisuke Higuchi

    Chef・Docker・Mirantis製品などの技術要素に加えて、会議の進め方・文章の書き方などの業務改善にも取り組んでいます。「Chef活用ガイド」共著のほか、Debian Official Developerもやっています。

    Daisuke Higuchiの記事一覧

    新規CTA