2024年11月26日にAnthropicが発表したModel Context Protocolがどうやらすごいらしいということで実際に触ってみました。
この記事では「具体的にどんなことができそうか？」という観点で紹介します。
※基本的なセットアップ手順などは割愛します。
1.MCPとは？
MCPの概要について簡単にスライドにまとめましたので参考にしてください。
詳細は下記リンクを参照ください。
公式Doc Introduction - Model Context Protoco
公式Github：Model Context Protocol
2.利用するMCP Serverの機能
今回は、MCP serversが標準で提供しているFilesystemを利用します。
また、Pythonを使ってカスタムなMCS Serverを作ることもできるようなのでこちらも最後に試してみました。
3.事前準備
公式HPの手順だとWindows環境ではFilesystemが正常に動作しないようなので、こちらの記事（Windows環境でのClaude Desktop MCP接続）を参考にして、Node.jsのインストールと、claude_desktop_config.jsonを設定します。
✅claude desktopのインストール
✅Node.jsのインストール
✅claude_desktop_config.jsonの設定
今回は、FilesystemでC:\\test\\claudeと、C:\\Users\\＜ユーザ名＞\\Desktopにアクセスできるように設定しています。
＜ユーザ名＞の部分は自身のユーザ名に置き換えてください。
claude_desktop_config.json
{
"mcpServers": {
"filesystem": {
"command": "node",
"args": [
"C:\\Users\\＜ユーザ名＞\\AppData\\Roaming\\npm\\node_modules\\@modelcontextprotocol\\server-filesystem\\dist\\index.js",
"C:\\test\\claude",
"C:\\Users\\＜ユーザ名＞\\Desktop"
]
}
}
}
以上で、Filesystemが正常に利用できる状態になるはずです。
Claude Desktopのトンカチマークをクリックして、下図のようにFrom Server:filesystemと表示されていれば利用できる状態になっています。
表示されない場合は、claude_desktop_config.jsonの設定が間違っているか、ClaudeDesktopの再起動が正常に行われていない可能性があります。
ちなみに、私のPC環境ではClaude Deskop画面を「×」で閉じても裏プロセスが残っていて正常に再起動されないので、OS再起動するか明示的にプロセスKillしないとダメでした。（バグ？）
4.ユースケース（5つ）
4-1.PDFから解説ドキュメントをローカルPC上に自動生成させる
以下の英語の論文のPDFをダウンロードしておきます。
ClaudeDesktopにPDFファイルを添付した上で以下の指示出しをしてみます。
添付したPDFをすべて確認したうえで、以下の条件で、内容をわかりやすく解説した纏め資料をhtmlドキュメントとして作成し、
「C:\test\claude\論文」フォルダ直下に保存してください。
#条件
・詳細なフロー図を組み込んで内容を理解しやすくする。
・全体的なデザインはアクセンチュア風テイストを基本とし、モダンでおしゃれにする。
・UI/UXを意識して見やすいデザインにする。
・視認性を高めるため、UCDAにそってください。
・htmlコードはできるだけ効率的に少ないコード（４０００字以内を目安）にする。
「PDFファイルの読み込み→htmlコード生成→指定したパスにhtmlファイルの保存」まで処理してくれました。
指定したパスにhtmlファイルが生成される
ローカルPC上で生成されたhtmlファイルを開くと、以下のような素敵な解説ドキュメントが出来上がっていました！（この時点でちょっと感動）
論文等のPDFを添付して、要約文や解説資料をhtmlドキュメントとしてローカルPC上に生成させられるのはかなり便利ですね。
4-2.フォルダの一括生成タスク
ローカルPC上のフォルダを制御できるということなので、複数のフォルダ操作を一括処理するようなタスクをやらせてみます。
事前に、以下のような大量のフォルダ構成定義情報を記載したCSVファイル(PROJECT構成.csv)を作成してMCPがアクセスできるC:\test\claude\PROJECT構成.csvに置いておきます。
この状態で、Claude Desktopアプリから以下のような指示出しをしてみます。
以下のタスクを実行して。
C:\test\claude\PROJECT構成.csvにフォルダ構成が定義されていますので、このフォルダ構成通りにC:\test\claude配下にフォルダを作成して
以下の様にPROJECT構成.csvの内容を確認したうえでフォルダ作成タスクを順に実行してくれました。
最後までタスクが終わった後にtreeコマンドでフォルダ構成を確認すると、PROJECT構成.csvどおりにちゃんとフォルダを自動作成してくれました。
「事前に定義しておいた情報に従って何かのタスクを自動実行させる」というやり方はいろいろ活用できそうですね。
C:\test\claude\XXリプレースプロジェクト>tree /f
フォルダー パスの一覧: ボリューム Windows
ボリューム シリアル番号は 1AE1-5CFD です
C:.
├─00_プロジェクト管理
│ ├─01_プロジェクト計画
│ ├─02_進捗管理
│ ├─03_課題管理
│ ├─04_リスク管理
│ └─05_品質管理
├─10_要件定義
│ ├─01_現行システム調査
│ ├─02_業務要件
│ ├─03_システム要件
│ ├─04_非機能要件
│ └─05_移行要件
├─20_基本設計
│ ├─01_システム方式設計
│ ├─02_機能設計
│ ├─03_画面設計
│ ├─04_帳票設計
│ ├─05_バッチ設計
│ ├─06_DB設計
│ └─07_インターフェース設計
├─30_詳細設計
│ ├─01_システム詳細設計
│ ├─02_プログラム仕様
│ ├─03_テーブル定義
│ └─04_外部インターフェース
└─40_テスト
└─01_テスト計画
4-3.Excel設計書の自動生成
ローカルPC上のファイルを参照した上で、ゴニョゴニョした結果をローカルPC上に保存できるので、「pythonコードを直接参照させてExcelのプログラム設計書を自動で作れるんでは？」と思い試してみました。
後ほど作成する以下のserver.pyというPythonコードのExcel設計書を作成させてみました。
import unittest
import asyncio
import os
from pathlib import Path
import json
import csv
import logging
from datetime import datetime
from server import (
get_system_metrics,
get_process_list,
calculate_metrics_statistics,
calculate_process_statistics,
OUTPUT_DIR
)
class TestWindowsMonitor(unittest.TestCase):
def setUp(self):
self.test_output_dir = OUTPUT_DIR
self.test_output_dir.mkdir(exist_ok=True)
logging.basicConfig(level=logging.INFO)
self.logger = logging.getLogger(__name__)
def test_get_system_metrics(self):
"""システムメトリクス取得のテスト"""
metrics = get_system_metrics()
required_keys = [
"timestamp", "cpu_percent", "memory_percent",
"memory_used_gb", "memory_total_gb"
]
for key in required_keys:
self.assertIn(key, metrics)
self.assertIsInstance(metrics["cpu_percent"], float)
self.assertGreaterEqual(metrics["cpu_percent"], 0)
self.assertLessEqual(metrics["cpu_percent"], 100)
def test_get_process_list(self):
"""プロセスリスト取得のテスト"""
processes = get_process_list()
self.assertGreater(len(processes), 0)
required_keys = ["pid", "name", "cpu_percent", "memory_percent"]
for process in processes:
for key in required_keys:
self.assertIn(key, process)
def test_metrics_collection_and_export(self):
"""メトリクス収集とCSVエクスポートのテスト"""
async def run_collection_test():
self.logger.info(f"出力ディレクトリ: {self.test_output_dir}")
self.logger.info("メトリクス収集開始...")
# メトリクスとプロセスデータの収集
metrics_data = []
process_snapshots = []
# 5秒間、1秒間隔でデータを収集
for _ in range(5):
metrics = get_system_metrics()
processes = get_process_list()
metrics_data.append(metrics)
process_snapshots.append({
"timestamp": metrics["timestamp"],
"processes": processes
})
await asyncio.sleep(1)
# タイムスタンプを含むファイル名の生成
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
metrics_file = self.test_output_dir / f"system_metrics_{timestamp}.csv"
process_file = self.test_output_dir / f"processes_{timestamp}.csv"
stats_file = self.test_output_dir / f"statistics_{timestamp}.json"
# メトリクスデータの出力
with open(metrics_file, 'w', newline='') as f:
writer = csv.DictWriter(f, fieldnames=metrics_data[0].keys())
writer.writeheader()
writer.writerows(metrics_data)
# プロセスデータの出力
with open(process_file, 'w', newline='') as f:
writer = csv.writer(f)
writer.writerow(['timestamp', 'pid', 'name', 'cpu_percent', 'memory_percent'])
for snapshot in process_snapshots:
timestamp = snapshot["timestamp"]
for process in snapshot["processes"]:
writer.writerow([
timestamp,
process["pid"],
process["name"],
process["cpu_percent"],
process["memory_percent"]
])
# 統計情報の計算と出力
statistics = {
"system_metrics": calculate_metrics_statistics(metrics_data),
"processes": calculate_process_statistics(process_snapshots)
}
with open(stats_file, 'w') as f:
json.dump(statistics, f, indent=2)
# ファイルの存在確認
self.assertTrue(metrics_file.exists())
self.assertTrue(process_file.exists())
self.assertTrue(stats_file.exists())
# ファイル内容の確認
self.logger.info(f"生成されたファイル:")
self.logger.info(f"- メトリクス: {metrics_file}")
self.logger.info(f"- プロセス: {process_file}")
self.logger.info(f"- 統計情報: {stats_file}")
# CSVファイルの内容確認
with open(metrics_file, 'r') as f:
reader = csv.reader(f)
header = next(reader)
self.assertGreater(len(header), 0)
data = list(reader)
self.assertGreater(len(data), 0)
self.logger.info(f"メトリクスデータ行数: {len(data)}")
return metrics_file, process_file, stats_file
# テストの実行
files = asyncio.run(run_collection_test())
self.assertTrue(all(f.exists() for f in files))
if __name__ == '__main__':
unittest.main(verbosity=2)
注意点
2024/12/1時点では、直接Excelファイル形式（.xlsx)でファイル生成ができないので、一旦csv形式のファイルを作成してもらうようにしています。
プロンプトは以下の通りです。
C:\test\claude\source-code\server.pyに定義されているPythonコードのプログラム設計書をcsv形式のファイルとして作成してください。
ファイル保存先：C:\test\claude\source-code\プログラム設計書.csv
#注意点
csvファイルを後で.xlsx形式に変更してserver.pyのプログラム設計書としてドキュメント管理する予定なので、その前提で#出力フォーマットのようなきれいな構成のExcel設計書になるように作成してください。
#出力フォーマット
===
Pythonプログラム基本設計書
1. 基本情報
項目 内容
プロジェクト名
作成日
最終更新日
作成者
最終更新者
Pythonバージョン Python 3.x
主要ライブラリ [ライブラリ名]: [用途]
[ライブラリ名]: [用途]
2. プログラム概要
項目 説明
目的
主な機能 1.
2.
3.
4.
5.
入力データ 1.
2.
出力データ 1.
2.
3.
起動方法
3. モジュール構成
ファイル名 役割 主要な機能/クラス 補足
[ファイル名].py
[ファイル名].py
4. 主要クラス・関数
名前 種別 引数 戻り値 説明
[名前] クラス/関数 [引数型] [戻り値型] [説明]
[名前] クラス/関数 [引数型] [戻り値型] [説明]
[名前] クラス/関数 [引数型] [戻り値型] [説明]
5. 重要な変数・定数
名前 データ型 用途 デフォルト値
[変数名] [型] [用途] [値]
[変数名] [型] [用途] [値]
6. 処理の流れ
1 [処理内容]
2 [処理内容]
3 [処理内容]
4 [処理内容]
5 [処理内容]
7. 注意事項・補足
・ [注意事項1]
・ [注意事項2]
・ [注意事項3]
8. 変更履歴
日付 更新者 内容
初版作成
===
上記のプロンプトを実行すると、server.pyと同じフォルダ内にプログラム設計書.csvが自動生成されました。
ただし、文字化けしているのでメモ帳で開いてエンコードをANSIに変更して保存しておきます。
その後、プログラム設計書.csvを開いて.xlsx形式で保存するとExcelファイルに変換できます。
あとは、フォントや体裁を少し整えると以下のようなプログラム設計書がサクッと作れてしまいました。
現状は直接Excelファイルの生成はできませんが、ニーズがめちゃくちゃあると思うのでそのうち対応してくれそうですよね。
直接Excelファイル生成ができるようになり、さらにスタイルの指定もできるようになれば、
「xxxにおいてあるExcelテンプレートを使って●●コードファイルの設計書作って」みたいなことができるようになりそうですよね。
これができるようになったらめちゃくちゃ便利・・。
4-4.プロジェクトファイル一式を自動生成
ローカルPC上のフォルダやファイルを直接操作できるってことは、生成AIに提案してもらったプログラムコード一式をフォルダ階層含めた状態で実際にローカルPC上に生成させられるってことですよね？
Djangoで試してみました。
以下の要件定義を満たすDjangoコードを作成してください。
完了したコードファイル一式を「C:\test\claude\django」に保存してください。
#要件定義
===
# YouTube動画文字起こしアプリ 要件定義書
## 1. プロジェクト概要
### 1.1 目的
- YouTube動画から音声を抽出し、テキストに変換するWebアプリケーションの開発
- ユーザーが簡単に動画の内容をテキストとして取得・保存できる環境の提供
### 1.2 主要機能
1. YouTube URL入力による動画の取得
2. 音声の文字起こし処理
3. 文字起こし結果の表示・保存
## 2. 機能要件
### 2.1 動画URL入力・取得機能
- YouTubeのURLを入力するフォームの提供
- URLの有効性チェック
- 動画の基本情報（タイトル、長さ）の表示
### 2.2 文字起こし機能
- YouTube動画から音声データの抽出
- 音声認識による文字起こし処理
- 進捗状況の表示
### 2.3 結果表示・管理機能
- 文字起こし結果のテキスト表示
- テキストのダウンロード機能（.txtファイル）
- 過去の文字起こし履歴の保存・閲覧
## 3. 技術要件
### 3.1 開発環境
- フレームワーク：Django
- データベース：SQLite（開発）/ PostgreSQL（本番）
- 音声認識：Google Speech-to-Text API
### 3.2 外部サービス・API
- YouTube Data API
- pytube（YouTube動画ダウンロード）
- Google Cloud Speech-to-Text
## 4. 非機能要件
### 4.1 パフォーマンス
- 動画の長さに応じた適切な処理時間の表示
- 同時処理数の制限（サーバー負荷対策）
### 4.2 セキュリティ
- ユーザー認証機能
- APIキーの安全な管理
- 入力値のバリデーション
### 4.3 ユーザビリティ
- レスポンシブデザイン
- 直感的なUI/UX
- エラーメッセージの適切な表示
## 5. 制約事項
- YouTube の利用規約遵守
- 著作権法の遵守
- 無料利用の場合の1日当たりの処理制限
===
以下の様にプログラムの生成が始まりました。
しばらくすると、ローカルPC上に実際にフォルダとプログラムコードが自動生成され始めました。
これは・・・・
最後まで生成終わった後にディレクトリをチェックすると、以下の様に必要なモジュールファイル一式が自動生成されていました。
これはちょっと革新的ですね。
今までだと、開発する際はチャット画面上に表示されたコードをいちいちコピペしてマニュアルでファイルを作成する必要がありましたが、直接コードファイルを作成してもらえるのはすごい。。。
C:\test\claude\django>tree /f
フォルダー パスの一覧: ボリューム Windows
ボリューム シリアル番号は 1AE1-5CFD です
C:.
│ .env.example
│ db.sqlite3
│ manage.py
│ README.md
│ requirements.txt
│
├─static
│ ├─css
│ │ style.css
│ │
│ └─js
├─templates
│ │ base.html
│ │
│ └─transcriber
│ home.html
│ transcription_detail.html
│
├─transcriber
│ admin.py
│ apps.py
│ forms.py
│ models.py
│ urls.py
│ views.py
│
└─youtube_transcriber
asgi.py
settings.py
urls.py
wsgi.py
ここで、ふと気づきました。
「もしかして、今作ってもらったファイルに対して直接修正指示だしできるのでは？」
やってみました。
結果は、以下の通りばっちり修正してくれました。
これは開発の仕方が変わりそうですね。。。。
4-5.Pythonを使ってカスタムなMCS Serverを作る
最後に、Pythonを使ってカスタムのMCS Serverを作ってみました。
今回は、「ローカルPCが重くなったりした際にPCの状態を自動解析してくれる機能」をMCS Serverとして作成してみました。
このカスタムMCS Server-PCの自動解析とFileSystem機能を併用することで、
私：「パソコンがなんか重いので調査してレポートにまとめて」
→ 以下のような調査結果のレポートがローカルPC上に自動生成される
みたいなことができるようになります。
今回作成したカスタムMCS Serverの概要
実際に動作させた際のデモ動画です。
設定手順
pythonのカスタムMCS Serverについては以下の公式マニュアルを参考にしました。
まず、MCS Serverを作成するフォルダに移動した後に、カスタムのMCS Serverの器を作ります。
cd C:\test\claude
uvx create-mcp-server --path windows_monitor
cd windows_monitor
uv add psutil mcp
ファイル構造:
windows_monitor/
├── src/
│ └── windows_monitor/
│ ├── __init__.py
│ └── server.py
├── tests/
└── pyproject.toml
create-mcp-serverコマンドを実行すと以下のようなメッセージが表示されるので、Project name (required)だけ指定してEnterキーで進めます。
Claude.app detected. Would you like to install the server into Claude.app now? [Y/n]:のところで「y」を選ぶと、claude_desktop_config.jsonにエントリーを自動で追加してくれます。
Creating a new MCP server project using uv.
This will set up a Python project with MCP dependency.
Let's begin!
Project name (required): windows_monitor
Project description [A MCP server project]:
Project version [0.1.0]:
Using CPython 3.11.5 interpreter at: C:\Users\<ユーザ名>\AppData\Local\Programs\Python\Python311\python.exe
Creating virtual environment at: .venv
Resolved 19 packages in 133ms
Built windows-monitor @ file:///C:/test/claude/windows_monitor
Prepared 1 package in 1.09s
Installed 19 packages in 123ms
+ annotated-types==0.7.0
+ anyio==4.6.2.post1
+ certifi==2024.8.30
+ click==8.1.7
+ colorama==0.4.6
+ h11==0.14.0
+ httpcore==1.0.7
+ httpx==0.28.0
+ httpx-sse==0.4.0
+ idna==3.10
+ mcp==1.0.0
+ pydantic==2.10.2
+ pydantic-core==2.27.1
+ sniffio==1.3.1
+ sse-starlette==2.1.3
+ starlette==0.41.3
+ typing-extensions==4.12.2
+ uvicorn==0.32.1
+ windows-monitor==0.1.0 (from file:///C:/test/claude/windows_monitor)
Claude.app detected. Would you like to install the server into Claude.app now? [Y/n]: y
✅ Added windows_monitor to Claude.app configuration
Settings file location: C:\Users\<ユーザ名>\AppData\Roaming\Claude\claude_desktop_config.json
✅ Created project windows_monitor in windows_monitor
ℹ️ To install dependencies run:
cd windows_monitor
uv sync --dev --all-extras
今回は、分析結果のcsvファイルを出力する場所を明示的に指定したかったので、envでMONITOR_OUTPUT_DIR変数を定義しています。
winmonの部分が今回作成するカスタムのMSC Serverの定義です。
また、winmonツールで生成した分析結果のファイルを保存するフォルダ"C:\\test\\claude\\windows_monitor\\log"をfilesystemのargsに追加しています。
{
"mcpServers": {
"filesystem": {
"command": "node",
"args": [
"C:\\Users\\<ユーザ名>\\AppData\\Roaming\\npm\\node_modules\\@modelcontextprotocol\\server-filesystem\\dist\\index.js",
"C:\\test\\claude",
"C:\\Users\\<ユーザ名>\\Desktop",
"C:\\test\\claude\\windows_monitor\\log"
]
},
"winmon": {
"command": "uv",
"args": [
"--directory",
"C:\\test\\claude\\windows_monitor",
"run",
"windows-monitor"
],
"env": {
"MONITOR_OUTPUT_DIR": "C:\\test\\claude\\windows_monitor\\log"
}
}
}
}
つづいて、__init__.pyに以下を追加(多分デフォルトで設定されている）します。
from . import server
import asyncio
def main():
asyncio.run(server.main())
__all__ = ['main', 'server']
server.pyを以下のコードに差し替えます。
import os
import json
import logging
from datetime import datetime
import asyncio
from collections.abc import Sequence
from pathlib import Path
import csv
from typing import Any, Dict, List
import psutil
from mcp.server import Server
from mcp.types import (
Resource,
Tool,
TextContent,
EmbeddedResource,
LoggingLevel
)
from pydantic import AnyUrl
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("windows-monitor")
# Output directory configuration - read from environment variable with fallback
OUTPUT_DIR = Path(os.getenv("MONITOR_OUTPUT_DIR", "monitoring_data"))
try:
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
logger.info(f"Using output directory: {OUTPUT_DIR}")
except Exception as e:
logger.error(f"Failed to create output directory {OUTPUT_DIR}: {e}")
logger.info("Falling back to current directory")
OUTPUT_DIR = Path("monitoring_data")
OUTPUT_DIR.mkdir(exist_ok=True)
def get_system_metrics() -> Dict[str, Any]:
"""Get current system resource usage"""
cpu_percent = psutil.cpu_percent(interval=1)
memory = psutil.virtual_memory()
disk = psutil.disk_usage('/')
return {
"timestamp": datetime.now().isoformat(),
"cpu_percent": cpu_percent,
"memory_percent": memory.percent,
"memory_used_gb": round(memory.used / (1024**3), 2),
"memory_total_gb": round(memory.total / (1024**3), 2),
"disk_percent": disk.percent,
"disk_used_gb": round(disk.used / (1024**3), 2),
"disk_total_gb": round(disk.total / (1024**3), 2)
}
def get_process_list() -> List[Dict[str, Any]]:
"""Get list of running processes"""
processes = []
for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']):
try:
processes.append({
'pid': proc.info['pid'],
'name': proc.info['name'],
'cpu_percent': proc.info['cpu_percent'],
'memory_percent': proc.info['memory_percent']
})
except (psutil.NoSuchProcess, psutil.AccessDenied):
pass
return processes
def calculate_metrics_statistics(metrics_data: List[Dict[str, Any]]) -> Dict[str, Any]:
"""Calculate statistics from collected metrics data"""
if not metrics_data:
return {}
return {
"start_time": metrics_data[0]["timestamp"],
"end_time": metrics_data[-1]["timestamp"],
"samples_count": len(metrics_data),
"averages": {
"cpu_percent": sum(m["cpu_percent"] for m in metrics_data) / len(metrics_data),
"memory_percent": sum(m["memory_percent"] for m in metrics_data) / len(metrics_data),
"memory_used_gb": sum(m["memory_used_gb"] for m in metrics_data) / len(metrics_data)
},
"peaks": {
"cpu_percent": max(m["cpu_percent"] for m in metrics_data),
"memory_percent": max(m["memory_percent"] for m in metrics_data),
"memory_used_gb": max(m["memory_used_gb"] for m in metrics_data)
}
}
def calculate_process_statistics(process_snapshots: List[Dict[str, Any]]) -> Dict[str, Any]:
"""Calculate statistics for processes across all snapshots"""
if not process_snapshots:
return {}
process_stats = {}
for snapshot in process_snapshots:
for process in snapshot["processes"]:
name = process["name"]
if name not in process_stats:
process_stats[name] = {
"samples": [],
"cpu_samples": [],
"memory_samples": []
}
process_stats[name]["samples"].append(process)
process_stats[name]["cpu_samples"].append(process["cpu_percent"])
process_stats[name]["memory_samples"].append(process["memory_percent"])
# Calculate statistics for each process
results = {}
for name, stats in process_stats.items():
samples_count = len(stats["samples"])
results[name] = {
"average_cpu": sum(stats["cpu_samples"]) / samples_count,
"average_memory": sum(stats["memory_samples"]) / samples_count,
"peak_cpu": max(stats["cpu_samples"]),
"peak_memory": max(stats["memory_samples"]),
"samples_count": samples_count
}
return results
app = Server("windows-monitor")
@app.list_resources()
async def list_resources() -> list[Resource]:
"""List available monitoring resources."""
return [
Resource(
uri=AnyUrl("winmon://system/current"),
name="Current system metrics",
mimeType="application/json",
description="Current CPU, memory and disk usage"
),
Resource(
uri=AnyUrl("winmon://processes/current"),
name="Current process list",
mimeType="application/json",
description="List of running processes"
)
]
@app.read_resource()
async def read_resource(uri: AnyUrl) -> str:
"""Read system metrics or process list."""
uri_str = str(uri)
if uri_str == "winmon://system/current":
return json.dumps(get_system_metrics(), indent=2)
elif uri_str == "winmon://processes/current":
return json.dumps(get_process_list(), indent=2)
else:
raise ValueError(f"Unknown resource: {uri}")
@app.list_tools()
async def list_tools() -> list[Tool]:
"""List available monitoring tools."""
return [
Tool(
name="export_metrics",
description="Export system metrics and process list to CSV",
inputSchema={
"type": "object",
"properties": {
"duration_seconds": {
"type": "number",
"description": "Duration to monitor (in seconds)",
"minimum": 1,
"maximum": 10
},
"interval_seconds": {
"type": "number",
"description": "Sampling interval (in seconds)",
"minimum": 1,
"maximum": 10
}
},
"required": ["duration_seconds", "interval_seconds"]
}
)
]
@app.call_tool()
async def handle_tool_call(tool_name: str, tool_args: Dict[str, Any]) -> Sequence[TextContent | EmbeddedResource]:
"""Handle tool calls for exporting metrics."""
if tool_name != "export_metrics":
raise ValueError(f"Unknown tool: {tool_name}")
if not isinstance(tool_args, dict):
raise ValueError("Invalid arguments")
duration = int(tool_args.get("duration_seconds", 10))
interval = int(tool_args.get("interval_seconds", 1))
# Create timestamp for filenames
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
metrics_file = OUTPUT_DIR / f"system_metrics_{timestamp}.csv"
process_file = OUTPUT_DIR / f"processes_{timestamp}.csv"
stats_file = OUTPUT_DIR / f"statistics_{timestamp}.json"
# Collect data
metrics_data = []
process_snapshots = []
logger.info(f"Starting metrics collection for {duration} seconds with {interval} second intervals")
for _ in range(0, duration, interval):
metrics = get_system_metrics()
processes = get_process_list()
metrics_data.append(metrics)
process_snapshots.append({
"timestamp": metrics["timestamp"],
"processes": processes
})
await asyncio.sleep(interval)
# Write system metrics
logger.info(f"Writing metrics to {metrics_file}")
with open(metrics_file, 'w', newline='') as f:
writer = csv.DictWriter(f, fieldnames=metrics_data[0].keys())
writer.writeheader()
writer.writerows(metrics_data)
# Write process data
logger.info(f"Writing process data to {process_file}")
with open(process_file, 'w', newline='') as f:
writer = csv.writer(f)
writer.writerow(['timestamp', 'pid', 'name', 'cpu_percent', 'memory_percent'])
for snapshot in process_snapshots:
timestamp = snapshot["timestamp"]
for process in snapshot["processes"]:
writer.writerow([
timestamp,
process["pid"],
process["name"],
process["cpu_percent"],
process["memory_percent"]
])
# Calculate and write statistics
statistics = {
"system_metrics": calculate_metrics_statistics(metrics_data),
"processes": calculate_process_statistics(process_snapshots)
}
logger.info(f"Writing statistics to {stats_file}")
with open(stats_file, 'w') as f:
json.dump(statistics, f, indent=2)
return [
TextContent(
type="text",
text=f"""
Monitoring completed:
- Metrics file: {metrics_file}
- Process data: {process_file}
- Statistics: {stats_file}
Summary:
- Collection period: {statistics['system_metrics']['start_time']} to {statistics['system_metrics']['end_time']}
- Samples collected: {statistics['system_metrics']['samples_count']}
- Average CPU usage: {statistics['system_metrics']['averages']['cpu_percent']:.1f}%
- Peak CPU usage: {statistics['system_metrics']['peaks']['cpu_percent']:.1f}%
- Average memory usage: {statistics['system_metrics']['averages']['memory_percent']:.1f}%
- Peak memory usage: {statistics['system_metrics']['peaks']['memory_percent']:.1f}%
"""
)
]
async def main():
from mcp.server.stdio import stdio_server
logger.info("Starting Windows Monitor MCP Server")
async with stdio_server() as (read_stream, write_stream):
await app.run(
read_stream,
write_stream,
app.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())
このコードの詳細は割愛しますが、以下のような機能を搭載したPythonコードになっています。
| 機能カテゴリ | 詳細 |
|---|---|
| システムリソースの監視 | • CPU使用率を計測 • メモリ使用量を監視 • ディスク使用状況をチェック • 実行中のプロセスの一覧を取得 |
| データの収集と保存 | • 定期的(1秒おきに10秒間）にシステムの状態を測定 • 収集したデータをCSVファイルに保存 • プロセスごとの統計情報をJSONファイルとして出力 |
| 統計情報の計算 | • CPU・メモリ使用率の平均値を計算 • ピーク時の使用率を記録 • プロセスごとの平均使用率を算出 |
一度、Claude Desktopを再起動しておきます。
以上の設定で、このMCSツール（Winmon)がClaude Deskotpから利用できるようになります。
ちなみに、正常に認識されるとClaudeDesktopのDeveloperメニュー上で以下の通り表示されます。
以下のようなパソコンに関する調査依頼を投げると、内部でwinmonツール（カスタムMCSServer）の利用が必要だと判断してくれて、調査処理を開始してくれます。
パソコンがなんか重いので調査して、以下の条件で調査結果のスライドを作成してください。
#条件
・htmlドキュメントとして作成する。
・リソースを多く消費しているプロセスの上位１０個の情報を調査結果に加える。
・収集したデータを簡単に確認できるグラフを加える。
・懸念点、改善点があれば明記する。
・全体的なデザインはアクセンチュア風テイストを基本とし、モダンでおしゃれにする。
・UI/UXを意識して見やすいデザインにする。
・視認性を高めるため、UCDAにそってください。
・htmlコードはできるだけ効率的に少ないコード（４０００字以内を目安）にする。
・完成したhtmlコードをすべて含むhtmlファイルを「C:\test\claude\windows_monitor\log」配下に保存する。
途中、アウトプットの文字数が上限を超えると、途中で処理が止まったりするので、
「htmlファイルの保存処理を再実行して」のように再実行してもらいたい部分を再度依頼すればOKです。
実際に何度か実行してみましたが、まだ動作が不安定のため今後の改善に期待したいところです。
✅処理時間が長いツールだとリクエストがタイムアウトしてエラーになってしまう。
✅処理時間が短い処理でも２回目以降タイムアウトになるケースが多い。ClaudeDesktopを再起動すると初回実行だけは上手くいく。
✅ClaudeDeskotp画面を「×」で閉じてもプロセスが残るので、タスクマネージャで明示的にKillしないとちゃんと再起動にならない。
ほんの一部の機能だけでもこれだけ活用シーンが考えられるので、夢が広がりますね。
アクセンチュア株式会社に所属する社員有志による運営です。アクセンチュアの社員による様々な発信をまとめています。なお、投稿内容は社員個人の見解であり、所属する組織を代表するものではありません。
Discussion