はじめに

MCPの概要を理解し、RooCodeでMCPを使ってみて、なんとなく理解した段階でした。
もう少し手を動かして感覚的にもMCPを理解したいということがモチベーションです。

以下の記事を大いに参考にさせて頂きました。
よくまとまっており、この記事を読めばMCPについて一通り理解できる大変優良な記事です。

MCPを自作により得られたもの

FastMCPの使い方を知れた
MCP ClientとMCP Serverの間で共有される情報や、対話の様子を確認できた
LangGraphの理解も少し深まった

使用した主な技術

FastMCP

概要
- MCPサーバーとクライアントの作成を劇的に簡素化するために設計された高レベルのPythonicフレームワーク
- Prefect社が主導のオープンソースプロジェクト(https://gofastmcp.com/)
特徴
- 🚀 簡易で迅速な開発
  - 高レベルのインターフェースにより、コード量を削減
  - ツールやリソースを定義するのは、標準的なPython関数にデコレーターを付けるだけで済む
- 🐍 Pythonic
  - Pythonのベストプラクティスを考慮して設計されており、Pythonに慣れた開発者にとって自然で直感的

LangGraph

概要
- LangChainの上に構築された新しいエージェントオーケストレーションフレームワーク
- エージェントのランタイムにおいて「ループ（循環）」をもつ処理フローを簡単に構築できる
特徴
- エージェントの構築
  - LangChainが「LLMを便利に使うための総合的なフレームワーク」なのに対して、LangGraphは「その中で、処理の流れや状態をグラフで構造化して制御するためのツール」
- 豊富なドキュメントとサービス
  - OSSで開発が活発。利用者も多くweb上の記事が多い
  - LangSmithを使えば、LangGraphで作ったものの動作をロギングしたり、クラウドに上げたりできる

MCPのクラサバ構成

MCP Server

yahooファイナンスから株価情報を取得するMCP Serverを作成

# 以下のコードをyahoofinance_server.pyとして保存

from mcp.server.fastmcp import FastMCP
from typing import Annotated
import yfinance as yf
import pandas as pd

# MCPサーバーインスタンスの作成
mcp = FastMCP("StockData")

# 株価を取得する関数をMCPの「ツール」として登録。
@mcp.tool()
def get_stock_data(
    ticker: Annotated[str, "企業のティッカーシンボル（例：AAPL、7203.T）"],
    start_date: Annotated[str, "データ取得の開始日 (YYYY-MM-DD)"],
    end_date: Annotated[str, "データ取得の終了日 (YYYY-MM-DD)"]
) -> str:
    """
    Yahooファイナンスから株価情報（日付, 始値, 高値, 安値, 終値, 出来高, 銘柄）を取得します。
    """
    try:
        # ティッカーシンボルの株価を検索
        stock = yf.Ticker(ticker)
        df = stock.history(start=start_date, end=end_date)

        if df.empty:
            return f"{ticker} のデータが見つかりませんでした。"

        # 取得データの整形
        df = df.reset_index()
        df["Ticker"] = ticker
        df = df.rename(columns={
            "Date": "日付", "Open": "始値", "High": "高値",
            "Low": "安値", "Close": "終値", "Volume": "出来高",
            "Ticker": "銘柄"
        })

        result = df[["日付", "銘柄", "始値", "高値", "安値", "終値", "出来高"]].to_string(index=False)
        return f"{ticker} の株価データ:\n```\n{result}\n```"

    except Exception as e:
        return f"データ取得中にエラーが発生しました: {repr(e)}"

# サーバー起動（stdio）
if __name__ == "__main__":
    mcp.run(transport="stdio")

PythonREPLと呼ばれるLLMから受け取ったPythonコードを実行するクラスでMCP Serverを作成

# repl_server.py

from langchain_experimental.utilities import PythonREPL
from typing import Annotated
from mcp.server.fastmcp import FastMCP
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
import base64
from io import BytesIO
import contextlib
import sys
import traceback
import os
# 日本語フォント設定
plt.rcParams["font.family"] = "Noto Sans CJK JP"
plt.rcParams["font.sans-serif"] = ["Noto Sans CJK JP", "Yu Gothic", "Meiryo"]

# MCP Server のインスタンス作成
mcp = FastMCP("PythonREPL")

# Python REPL ユーティリティの初期化
repl = PythonREPL()

@mcp.tool()
def python_repl(code: Annotated[str, "チャートを生成するために実行する Python コード"]) -> str:
    """
    Python コードを実行し、matplotlib のグラフが含まれる場合は base64 PNG 画像として返す。
    """
    try:
        # stdout の捕捉
        stdout = BytesIO()
        with contextlib.redirect_stdout(stdout):
            exec(code, globals())

        if not plt.get_fignums():
            return f"実行成功 (グラフは生成されませんでした)\n\n```\n{stdout.getvalue().decode()}\n```"

        os.makedirs("/mnt/data", exist_ok=True)

        # matplotlibの画像を base64 に変換
        buf = BytesIO()
        plt.savefig(buf, format="png")
        buf.seek(0)
        img_base64 = base64.b64encode(buf.read()).decode("utf-8")
        buf.close()
        plt.close()

        return f"Successfully executed:\n```python\n{code}\n```\n\n![chart](data:image/png;base64,{img_base64})"
    except BaseException:
        return "Failed to execute:\n```\n" + traceback.format_exc() + "```"

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

MCP Client

MCP ServerはLangGraphのtoolとして登録可能
MCP Serverを複数登録しておくことも可能
ReActエージェントを用いることで、よしなにMCP Serverを使用してくれる

import os
from langchain_openai import ChatOpenAI
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from IPython.display import Image, display
import re
import base64

from dotenv import load_dotenv
load_dotenv('.env')

model = ChatOpenAI(
    model_name="gpt-4o-mini",
    temperature=0.7,
    max_tokens=500
)

# MCPクライアント初期化
client = MultiServerMCPClient({
    "stock": {
        "command": "python",
        "args": ["/root/src/MCPサーバ自作/yahoofinance_server.py"],
        "transport": "stdio",
    },
    "chart": {
        "command": "python",
        "args": ["/root/src/MCPサーバ自作/repl_server.py"],
        "transport": "stdio",
    }
})

# ツールをロード
tools = await client.get_tools()

# エージェント生成
agent = create_react_agent(model, tools)

user_query = "2025年6月1から2025年6月10日までのMicrosoftの株の終値を取得し、その値を使ってmatplotlibのチャートを作成してください。"
response = await agent.ainvoke({"messages": user_query, "verbose": True})
print(response)

# グラフの画像の表示
match = re.search(r'data:image/png;base64,([A-Za-z0-9+/=]+)', str(response))
if match:
    image_data = match.group(1)
    display(Image(data=base64.b64decode(image_data)))
else:
    print(response)

実行結果

考察

ReActエージェント(MCP Client)がMCP Serverと適切に連携することができた
何度か試行していると、グラフを出力するPythonコードの実行でエラーが発生するノードがあったが、ReActを繰り返すうちにエラーを解消し、適切な結果導いていたのでLangGraphの性能に驚いた
今回gpt-4o-miniを使ったが、思考過程を細かく見ることができなかった
これは、gptモデルの仕様らしく、ClaudeやOracleのモデルを使うと思考過程が出力されるらしい