🍣

MCP（Model Context Protocol）の最小動作サンプル　他

2025/04/14に公開

FastAPI

Model Context Protocol

デモ

動作確認

tech

MCP（Model Context Protocol）の最小動作サンプルを用意しました。

✅ 最小動作サンプル

[mini_server.py]– MCPサーバ（標準入出力で通信）
[mini_client.py]– MCPクライアント（サーバにHelloを送る）

✅ mini_server.py – MCPサーバー

from mcp.server.fastmcp import FastMCP

# "hello" という名前のサーバを作成
mcp = FastMCP("hello")

# MCPツールとしてhello_world関数を登録
@mcp.tool()
async def hello_world(name: str) -> str:
    return f"Hello, {name}!"

if __name__ == "__main__":
    # 標準入出力で通信（stdio使用）
    mcp.run(transport="stdio")

✅ mini_client.py – MCPクライアント

import asyncio
from mcp.client.stdio import stdio_client, StdioServerParameters
from mcp import ClientSession

async def main():
    # mini_server.py をPythonで実行し、stdin/stdoutで接続
    server_params = StdioServerParameters(command="python", args=["mini_server.py"])

    # サーバ接続のためのクライアントストリームを確立
    async with stdio_client(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            # 初期化処理（initialize）
            await session.initialize()

            # "hello_world" ツールを呼び出し
            result = await session.call_tool("hello_world", {"name": "MCP"})
            print("Tool result:", result.content)

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

💡動作手順（ローカル環境）

MCPのライブラリをインストール（必要に応じて仮想環境を使ってください）:
```
pip install fastmcp
```
mini_server.py を保存しておく
mini_client.py を実行
```
python mini_client.py
```
出力例:
```
Tool result: Hello, MCP!
```

🔧 解説

FastMCP("hello") でMCPサーバを構築し、@mcp.tool() でAPI（ツール）を登録
クライアントは call_tool("hello_world", {"name": "MCP"}) でツールを実行

ちょっと物足りないので、
ここでは、現場で本当にありがちなユースケースをベースに、MCPを使った実装サンプルを用意してやってみます。

🎯 失敗したバッチのログを取得し、MCPに自動通知する

📘 シナリオ概要

売上集計バッチが失敗
MCPでログ取得ツールを呼び出し
ログ内容をAIが受け取り、Slack通知 or 要因分析へ　←ここはできてない

✅ MCPサーバ（`batch_ops_server.py`）

# batch_ops_server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("batch_ops")

@mcp.tool()
def get_failed_log(job_name: str) -> str:
    try:
        with open(f"./logs/{job_name}_error.log", "r") as f:
            return f.read()
    except FileNotFoundError:
        return "ログファイルが見つかりませんでした"

@mcp.tool()
def retry_batch(job_name: str) -> str:
    # 実際にはsubprocessなどで実行
    return f"{job_name} を再実行しました"

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

✅ MCPクライアント（`ai_agent_client.py`）

# ai_agent_client.py
import asyncio
from mcp.client.stdio import stdio_client, StdioServerParameters
from mcp import ClientSession

async def main():
    server = StdioServerParameters(command="python", args=["batch_ops_server.py"])

    async with stdio_client(server) as (r, w):
        async with ClientSession(r, w) as session:
            await session.initialize()

            # エラーのログを取得
            log = await session.call_tool("get_failed_log", {"job_name": "sales_batch"})
            print("ログ取得結果：\n", log.content)

            # 条件分岐：タイムアウトだったら再実行
            if "Timeout" in log.content:
                res = await session.call_tool("retry_batch", {"job_name": "sales_batch"})
                print("再実行結果：", res.content)

asyncio.run(main())

✅ テスト用のログファイルを準備

./logs/sales_batch_error.log

内容例:
2025-04-10 02:12:01 - ERROR - Timeout while connecting to DB

✅ 実行方法

# 両方同じディレクトリに配置し、実行
python ai_agent_client.py

結果が出ました。（まだ良さはそこまでわかっていないｗ）
色々とためしていきたい。

🌟 MCPの良さが光るポイント

ポイント	なぜ良いか
🎯 処理が「ツール」として分離されている	誰でも再利用できる、疎結合化
🧠 AIがログを読み、条件で再実行	属人判断からの脱却、再現性ある判断
⚙️ stdioでローカル完結	最小構成ですぐ試せる、クラウド不要
🔁 バッチ・監視の処理が自動連携	現場の「手間がかかる部分」を自動化可能

📦 MCP + FastAPI（Flaskライク）構成で動作確認

MCP + FastAPI (Flask風構成) で Web 経由呼び出しできる MCPサーバ／クライアント のサンプルコードです。

✅ MCP サーバ側（`mcp_flask_style_server.py`）

from fastapi import FastAPI
from mcp.server.fastmcp import FastMCP

app = FastAPI()
mcp = FastMCP("batch-server")

# MCPのツール登録：バッチステータス確認ツール
@mcp.tool()
async def get_batch_status(batch_id: str) -> str:
    # サンプルとして固定メッセージを返す（ここに監視・DBアクセスなど入れる）
    return f"バッチID {batch_id} は SUCCESS 状態です。"

# FastAPIと統合
@app.get("/")
def root():
    return {"message": "MCP + FastAPI サーバ動作中"}

# MCP用エンドポイント追加（Streamable HTTP）
mcp.include_router(app)

✅ MCP クライアント側（`mcp_flask_style_client.py`）

import asyncio
from mcp.client.stream_http import connect_streamable_http
from mcp.client.session import ClientSession
from mcp.client.server_parameters import HTTPServerParameters

async def main():
    # サーバ接続先指定（FastAPI + MCPのサーバ）
    server_url = "http://localhost:8000/mcp"
    server_params = HTTPServerParameters(server_url=server_url)

    # Streamable HTTP接続
    async with connect_streamable_http(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            # MCPツール呼び出し：get_batch_status
            result = await session.call_tool("get_batch_status", {"batch_id": "BATCH-20240412"})
            print("サーバの応答:", result.content)

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

🔧 起動手順（ローカルで）

必要なライブラリをインストール：
```
pip install fastapi uvicorn mcp
```

サーバを起動：

uvicorn mcp_flask_style_server:app --reload

別ターミナルでクライアントを実行：
```
python mcp_flask_style_client.py
```

🌟 この構成のポイント

MCPをWeb（HTTP + SSE）経由で呼び出せる → Flask/REST風に扱える
任意の処理（監視、ログ、DB問い合わせなど）を @mcp.tool() で拡張可能
「バッチ監視サーバ」や「再実行トリガーサーバ」として構築できる

✅ MCP＋FastAPIでバッチジョブを使ったデモプロジェクト

💡 目的

AIやWebアプリから「バッチジョブの実行」「監視」「ログ確認」などを、
MCP（Model Context Protocol）経由で操作できる構成を試す。

💡 構成イメージ

📦 batch_server.py（MCPツール提供） ← MCPで動作（stdio or streamable）
📦 fastapi_dashboard.py（Web）       ← FastAPIで起動
↔️ 両者は HTTP 経由で通信（例：POST /call_tool）

🧱 構成図（簡易）

[ ユーザー or Webフロント（curlやブラウザ） ]
               ↓ POST /run-job
        ┌────────────────────┐
        │ dashboard.py       │ ← FastAPI アプリ
        │（コントローラー） │
        └────────────────────┘
               ↓ MCP経由でツール実行
        ┌────────────────────┐
        │ batch_server.py     │ ← MCPツール群（@mcp.tool()）
        │ run_batch()関数など │
        └────────────────────┘

✅ まず最小の MCP サーバを動かす

batch_server.py（MCP stdio サーバ）

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("batch")

@mcp.tool()
async def run_batch(job_name: str) -> str:
    return f"{job_name} を実行しました！"

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

✅ FastAPI から MCP を呼び出す

dashboard.py（FastAPI）

from fastapi import FastAPI, Request
from pydantic import BaseModel
import asyncio
from mcp.client.stdio import stdio_client, StdioServerParameters
from mcp.client.session import ClientSession

app = FastAPI()

class JobRequest(BaseModel):
    job_name: str

@app.post("/run-job")
async def run_job(req: JobRequest):
    server = StdioServerParameters(command="python", args=["batch_server.py"])
    async with stdio_client(server) as (r, w):
        async with ClientSession(r, w) as session:
            await session.initialize()
            result = await session.call_tool("run_batch", {"job_name": req.job_name})
            return {"message": result.content}

📦 フォルダ構成

mcp_fastapi_demo/
├── dashboard.py       # FastAPIアプリ：Web経由でMCP呼び出し
└──  batch_server.py    # MCPのツール定義（バッチ処理の中身）

🔁 通信フローの解説

ユーザーがWebからバッチ名を指定（例：月次集計）
dashboard.py が Streamable HTTP で batch_server.py に接続
batch_server.py の中の run_batch(job_name) が呼ばれる
「バッチ実行しました！」などのレスポンスが返る
それをWeb/API側で表示・整形

🧪 実行方法

# 依存ライブラリのインストール
pip install mcp fastapi uvicorn

# FastAPI側（ダッシュボード）
uvicorn dashboard:app --reload

# 別ターミナルで curl
curl -X POST http://127.0.0.1:8000/run-job \
     -H "Content-Type: application/json" \
     -d '{"job_name":"月次集計"}'

curlで確認できました。

✅ できたこと

MCPの@mcp.tool()を使って、バッチ処理をAPI化できた
MCPクライアントからMCPサーバへStreamable HTTPで指示
FastAPI経由で、Web UIや他サービスから呼び出せる

MCP + FastAPI + HTTPトリガーでバッチを実行する構成

📁 ファイル構成

mcp_http_batch_demo_v2/
├── batch_tools.py         # MCPツール定義（run_batch / retry_batch）
├── mcp_http_server.py     # FastAPIサーバー（MCP連携）
└── start.sh               # 起動スクリプト

📄 `batch_tools.py`

（MCPサーバとして動作するバッチツール定義）

# batch_tools.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("batch-tools")

@mcp.tool()
async def run_batch(job_name: str) -> str:
    print(f"[RUN] バッチ実行: {job_name}")
    return f"{job_name} を実行しました！"

@mcp.tool()
async def retry_batch(job_name: str) -> str:
    print(f"[RETRY] バッチ再実行: {job_name}")
    return f"{job_name} を再実行しました！"

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

📄 `mcp_http_server.py`

（HTTP経由でMCPツールをトリガーするFastAPIサーバ）

# mcp_http_server.py
from fastapi import FastAPI
from pydantic import BaseModel
from mcp.client.stdio import stdio_client, StdioServerParameters
from mcp import ClientSession

app = FastAPI()

class JobRequest(BaseModel):
    job_name: str

async def call_tool(tool_name: str, job_name: str):
    server_params = StdioServerParameters(command="python", args=["batch_tools.py"])
    async with stdio_client(server_params) as (reader, writer):
        async with ClientSession(reader, writer) as session:
            await session.initialize()
            result = await session.call_tool(tool_name, {"job_name": job_name})
            return result.content  # <- 注意: contentではなく、result自体が値（str）になる場合もある

@app.post("/run-job")
async def run_job(request: JobRequest):
    message = await call_tool("run_batch", request.job_name)
    return {"message": message}

@app.post("/retry-job")
async def retry_job(request: JobRequest):
    message = await call_tool("retry_batch", request.job_name)
    return {"message": message}

📄 `start.sh`

（FastAPIサーバ起動スクリプト）

#!/bin/bash
uvicorn mcp_http_server:app --reload --port 9000

実行権限を与えるのを忘れずに👇

chmod +x start.sh

🚀 起動方法

unzip mcp_http_batch_demo_v2.zip
cd mcp_http_batch_demo_v2
pip install fastapi uvicorn mcp  # まだの方のみ
python batch_tools.py &          # MCPツール登録用のプロセス
bash start.sh                    # FastAPIサーバ起動

✅ 動作確認コマンド

curl -X POST http://127.0.0.1:9000/run-job \
  -H "Content-Type: application/json" \
  -d '{"job_name": "月次集計"}'

curl -X POST http://127.0.0.1:9000/retry-job \
  -H "Content-Type: application/json" \
  -d '{"job_name": "月次集計"}'

📌 期待されるレスポンス

{"message": "月次集計 を実行しました！"}
{"message": "月次集計 を再実行しました！"}

🧠 補足

MCPのFastMCPをFastAPIと組み合わせてHTTPでトリガー可能にした構成です。
run_batchやretry_batchなどの処理は@mcp.tool()で登録されているため、MCPとして独立しても使用可能です。
今後、ログ取得やステータス確認なども拡張可能です。