🧣
n番煎じだけど、WindowsだとクセがあるMCPサーバ実装
昨日(4/25)、ぬこぬこさん (https://x.com/schroneko) が開催されたMCPサーバの勉強会の投稿を目撃しました!
いきたかったーーー!と思いつつ後の祭り。
で、参加者のみなさんが作ったMCPサーバを投稿していたんですよね。
このMCPサーバのノベル面白かった。
で、目についたのが、このMCPサーバ。
おぉ、すごい!加えて私が住む北海道の情報が取れるといいなと思いました。
ぬこぬこさんのイベントに参加できなかった後悔と、MCPサーバを実際に作る経験ないことへの焦り、JR北海道の遅延情報も取得したい想いが、重なり、自分でも作ってみました!
Githubにあげたので、興味ありましたら、ぜひご覧ください。
Pythonファイルをここにも載せておきますね。
ソースコード(Python)
jrhokkaido_train_info.py
"""
JR 北海道運行情報 MCP サーバー
"""
from mcp.server.fastmcp import FastMCP
import logging
import sys
from typing import Dict, List, Optional
import aiohttp
import asyncio
from bs4 import BeautifulSoup
# ────────────────── ロガー設定(stderr 出力) ──────────────────
logging.basicConfig(
stream=sys.stderr,
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
)
log = logging.getLogger(__name__)
# ────────────────── MCP 初期化 ──────────────────
mcp = FastMCP("JR北海道列車運行情報", log_level="INFO")
log.info("MCP サーバー初期化完了")
# ────────────────── 共通定数 ──────────────────
# JR北海道の運行情報ページURLマップ
AREA_URLS = {
# ローマ字コード
"sapporo": "https://www3.jrhokkaido.co.jp/webunkou/area_spo.html",
"doo": "https://www3.jrhokkaido.co.jp/webunkou/area_doo.html",
"donan": "https://www3.jrhokkaido.co.jp/webunkou/area_donan.html",
"dohoku": "https://www3.jrhokkaido.co.jp/webunkou/area_dohoku.html",
"doto": "https://www3.jrhokkaido.co.jp/webunkou/area_doto.html",
"shinkansen": "https://www3.jrhokkaido.co.jp/webunkou/senku.html?id=24",
# 漢字表記
"札幌": "https://www3.jrhokkaido.co.jp/webunkou/area_spo.html",
"札幌近郊": "https://www3.jrhokkaido.co.jp/webunkou/area_spo.html",
"道央": "https://www3.jrhokkaido.co.jp/webunkou/area_doo.html",
"道南": "https://www3.jrhokkaido.co.jp/webunkou/area_donan.html",
"道北": "https://www3.jrhokkaido.co.jp/webunkou/area_dohoku.html",
"道東": "https://www3.jrhokkaido.co.jp/webunkou/area_doto.html",
"北海道新幹線": "https://www3.jrhokkaido.co.jp/webunkou/senku.html?id=24",
"新幹線": "https://www3.jrhokkaido.co.jp/webunkou/senku.html?id=24",
}
REQUEST_TIMEOUT = 10 # 秒
# ────────────────── 共通関数 ──────────────────
async def fetch_train_info(url: str) -> str:
"""JR北海道運行情報ページのHTMLを取得"""
async with aiohttp.ClientSession() as session:
async with session.get(url, timeout=REQUEST_TIMEOUT) as response:
response.raise_for_status()
return await response.text()
async def scrape_area(url: str) -> List[Dict]:
"""エリアページから遅延情報をスクレイプ"""
html = await fetch_train_info(url)
soup = BeautifulSoup(html, "html.parser")
results = []
# todayGaikyo div内のgaikyo-listを探索
gaikyo_div = soup.find("div", id="todayGaikyo")
if not gaikyo_div:
log.warning(f"todayGaikyo div not found in {url}")
return results
gaikyo_list = gaikyo_div.find("ul", class_="gaikyo-list")
if not gaikyo_list:
log.warning(f"gaikyo-list not found in todayGaikyo div in {url}")
return results
# リスト内の各項目を取得
list_items = gaikyo_list.find_all("li")
for item in list_items:
item_text = item.text.strip()
# 「遅れに関する情報はありません」などの場合はスキップ
if "遅れに関する情報はありません" in item_text or "情報はありません" in item_text:
continue
# 遅延カテゴリの判定
category = "other"
if any(word in item_text for word in ["遅延", "Delay"]):
category = "delay"
elif any(word in item_text for word in ["運休", "Suspension", "Suspend"]):
category = "suspension"
elif any(word in item_text for word in ["お知らせ", "Notice"]):
category = "notice"
# URL中のエリア名から路線名を推定
line_name = "不明"
if "area_spo" in url:
line_name = "札幌近郊"
elif "area_doo" in url:
line_name = "道央"
elif "area_donan" in url:
line_name = "道南"
elif "area_dohoku" in url:
line_name = "道北"
elif "area_doto" in url:
line_name = "道東"
elif "senku.html" in url and "id=24" in url:
line_name = "北海道新幹線"
results.append({
"line": line_name,
"status": item_text,
"category": category
})
return results
# ────────────────── MCP ツール ──────────────────
@mcp.tool()
async def get_delays(area: Optional[str] = None) -> Dict:
"""JR北海道の遅延情報を取得
Args:
area: 取得するエリア ("札幌"/"sapporo", "道央"/"doo", "道南"/"donan", "道北"/"dohoku", "道東"/"doto", "北海道新幹線"/"shinkansen")。省略時は全エリア取得
"""
areas = [area] if area else list(AREA_URLS.keys())
all_results = []
for a in areas:
if a not in AREA_URLS:
all_results.append({
"area": a,
"line": "",
"status": "",
"category": "other",
"error": f"Unknown area: {a}"
})
continue
try:
url = AREA_URLS[a]
results = await scrape_area(url)
for r in results:
all_results.append({"area": a, **r})
except Exception as exc:
log.exception(f"{a}エリアの遅延情報取得エラー")
all_results.append({
"area": a,
"line": "",
"status": "",
"category": "other",
"error": f"{exc}"
})
text = "遅延情報がありません。" if not all_results else "\n".join(
f"[{r['area']}] ERROR: {r['error']}" if 'error' in r else f"[{r['area']}] {r['line']}: {r['status']}"
for r in all_results
)
return {"content": [{"type": "text", "text": text}]}
# ────────────────── エントリーポイント ──────────────────
def start_server():
"""MCPサーバーを開始する関数"""
try:
log.info("利用可能ツール: %s", [get_delays.__name__])
log.info("クライアントからの接続を待機中…")
mcp.run(transport="stdio")
except Exception:
log.exception("致命的エラーによりサーバーを終了します")
raise
if __name__ == "__main__":
start_server()
else:
# モジュールとしてインポートされた場合でもサーバーを起動
start_server()
毎回悩まされるのが、claude_desktop_config.jsonですね。macOSの方ではまっている人あまり見かけないのですが、Windowsだと結構クセがあり苦戦しますね。
藤川さん (https://x.com/hfujikawa77) の記事 (https://zenn.dev/fujihide/articles/88c50159a510f0) も参考にしつつ、どうにか動くようになった。
claude_desktop_config.json
claude_desktop_config.json
{
"mcpServers": {
"jrhokkaido-train-info": {
"command": "uv",
"args": [
"--directory",
"C:\\<your folder>\\jrhokkaido_train_info",
"run",
"jrhokkaido_train_info.py"
],
"workingDirectory": "C:\\<your folder>\\jrhokkaido_train_info"
}
}
}
ぬこぬこさんがオススメしていたinspector (https://github.com/modelcontextprotocol/inspector) はぜひ使いたかったのだが、うまく動かず。(Windowsのせい?笑)
そんなこんなで、Claude Codeや上記有識者さんの知見を拝借しながら、半日でなんとかMCPサーバ作れました!
Discussion