この記事はNTTドコモソリューションズ Advent Calendar 2025 22 日目の記事です。
はじめに
こんにちは、NTTドコモソリューションズの藤井隆裕です。
業務ではパブリッククラウドに関連する開発プロジェクト支援や技術検証を行っています。
何か作ってみたいと思い、MCP サーバを自作してみたので、共有します。
AWS What's New ページから直近 1 週間のアップデート情報を取得する MCP サーバを作成しています。
MCPサーバとは
MCP（Model Context Protocol） サーバは、生成 AI サービスに対して標準化された仕様で外部のツール実行やリソース参照の機能を提供します。
MCP は生成 AI が外部システムにアクセスする時のルール（通信プロトコル）を定義したものです。
生成 AI が MCP サーバを利用することで、学習済みモデルだけでは対応できない最新情報や専門機能を利用可能となり、安全かつ効率よく回答精度の向上が期待できます。
MCPサーバの特徴
- 標準化されたプロトコル: 異なる生成AIサービス間で共通のインターフェース
- 拡張性: 独自の機能を簡単に追加可能
- セキュリティ: 制御されたアクセス権限でデータを取得
- 再利用性: 一度作成すれば複数の生成AIサービスで利用可能
本記事では、MCP サーバをローカル環境に構築して、Amazon Q Developer と MCP サーバの連携を試しています。
前提条件
- Python 3.12 以上
- Visual Studio Code がインストール済みであること
- Amazon Q Developer（VSCode プラグイン）が環境構築済みであること
1. 環境構築
今回は MCP サーバを Python で実装します。
MCP サーバの実装に必要なライブラリを事前にインストールします。
pip install mcp feedparser requests python-dateutil
2. MCP サーバの実装
MCP インターフェース
MCP サーバを構築するために Python で以下の関数を定義しました。
@server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
この関数は MCP の通信プロトコルに対応するインターフェースで、各パラメータは次の役割を持っています。
-
name: str- 実行するツール名（例："get_aws_updates"） -
arguments: Dict[str, Any]- ツールに渡すパラメータ -
List[TextContent]- クライアントに返すレスポンス
処理フロー
- Amazon Q Developer がユーザの質問を解析
- MCP でツール呼び出し
- call_tool() が適切な処理を実行
- 結果データをテキスト形式で返却
- Amazon Q Developer が生成 AIと連携して最終回答を生成
MCP サーバのソースコード作成
任意のプロジェクトディレクトリに server.py を作成します。MCP サーバのインターフェースと AWS 最新情報を取得する処理を実装します。
本記事で実装したコードでは MCP サーバはサーバ名を「aws-whatsnew」に設定し、「get_aws_updates」 「filter_by_service」という 2 つのツールを提供しています。
get_aws_updates では AWS What's New（RSS）から直近 7 日間のアップデート情報を取得します。
filter_by_service では AWS What's New（RSS）から直近 7 日間の指定されたAWSサービスのアップデート情報を取得します。
import asyncio
import json
from datetime import datetime, timedelta
from typing import List, Dict, Any
import feedparser
import requests
from dateutil import parser
from mcp.server import Server
from mcp.types import Tool, TextContent
# サーバインスタンスの作成
server = Server("aws-whatsnew")
class AWSWhatsNewRSSReader:
def __init__(self):
self.rss_url = "https://aws.amazon.com/new/feed/"
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
})
def get_recent_updates(self, days: int = 7) -> List[Dict[str, Any]]:
"""直近N日間のAWSアップデート情報をRSSから取得"""
try:
# RSSフィードを取得
feed = feedparser.parse(self.rss_url)
if feed.bozo:
return [{'error': f'RSSフィード解析エラー: {feed.bozo_exception}'}]
updates = []
cutoff_date = datetime.now() - timedelta(days=days)
for entry in feed.entries:
try:
# 日付解析
if hasattr(entry, 'published_parsed'):
entry_date = datetime(*entry.published_parsed[:6])
elif hasattr(entry, 'updated_parsed'):
entry_date = datetime(*entry.updated_parsed[:6])
else:
continue
# 日付フィルタ
if entry_date < cutoff_date:
continue
# データ抽出
title = entry.title if hasattr(entry, 'title') else 'タイトルなし'
link = entry.link if hasattr(entry, 'link') else ''
description = entry.summary if hasattr(entry, 'summary') else ''
# HTMLタグを除去
import re
description = re.sub(r'<[^>]+>', '', description).strip()
updates.append({
'title': title,
'date': entry_date.strftime('%Y-%m-%d'),
'description': description,
'link': link
})
except Exception as e:
continue
return sorted(updates, key=lambda x: x['date'], reverse=True)
except Exception as e:
return [{'error': f'RSSデータ取得エラー: {str(e)}'}]
def filter_by_service(self, updates: List[Dict], service: str) -> List[Dict]:
"""特定サービスでフィルタリング"""
filtered = []
service_lower = service.lower()
for update in updates:
title_lower = update.get('title', '').lower()
desc_lower = update.get('description', '').lower()
if service_lower in title_lower or service_lower in desc_lower:
filtered.append(update)
return filtered
# RSSリーダーインスタンス
rss_reader = AWSWhatsNewRSSReader()
@server.list_tools()
async def list_tools() -> List[Tool]:
"""利用可能なツール一覧を返す"""
return [
Tool(
name="get_aws_updates",
description="直近1週間のAWSアップデート情報を取得する",
inputSchema={
"type": "object",
"properties": {
"days": {
"type": "integer",
"description": "取得する日数（デフォルト: 7日）",
"default": 7
}
}
}
),
Tool(
name="filter_by_service",
description="特定のAWSサービスでアップデート情報をフィルタリングする",
inputSchema={
"type": "object",
"properties": {
"service": {
"type": "string",
"description": "フィルタリングするサービス名（例: bedrock, lambda, s3）"
},
"days": {
"type": "integer",
"description": "取得する日数（デフォルト: 7日）",
"default": 7
}
},
"required": ["service"]
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
"""ツール実行"""
if name == "get_aws_updates":
days = arguments.get("days", 7)
updates = rss_reader.get_recent_updates(days)
if not updates:
return [TextContent(
type="text",
text="直近のアップデート情報が見つかりませんでした。"
)]
result = f"## 直近{days}日間のAWSアップデート情報\n\n"
for update in updates:
if 'error' in update:
result += f"エラー: {update['error']}\n"
else:
result += f"### {update['title']}\n"
result += f"**日付**: {update['date']}\n"
result += f"**概要**: {update['description']}\n"
if update['link']:
result += f"**詳細**: {update['link']}\n"
result += "\n---\n\n"
return [TextContent(type="text", text=result)]
elif name == "filter_by_service":
service = arguments.get("service", "")
days = arguments.get("days", 7)
if not service:
return [TextContent(
type="text",
text="サービス名を指定してください。"
)]
all_updates = rss_reader.get_recent_updates(days)
filtered_updates = rss_reader.filter_by_service(all_updates, service)
if not filtered_updates:
return [TextContent(
type="text",
text=f"{service}に関連するアップデート情報が見つかりませんでした。"
)]
result = f"## {service}関連のアップデート情報（直近{days}日間）\n\n"
for update in filtered_updates:
result += f"### {update['title']}\n"
result += f"**日付**: {update['date']}\n"
result += f"**概要**: {update['description']}\n"
if update['link']:
result += f"**詳細**: {update['link']}\n"
result += "\n---\n\n"
return [TextContent(type="text", text=result)]
else:
return [TextContent(
type="text",
text=f"不明なツール: {name}"
)]
async def main():
"""サーバ起動"""
from mcp.server.stdio import stdio_server
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())
3. Amazon Q Developer との連携設定
MCP 設定ファイルの作成
Amazon Q Developer の MCP 設定ファイルに以下を追加します。
設定ファイルの場所
-
Windows:
C:\Users\ユーザ名\.aws\amazonq\mcp.json -
Linux:
~/.aws/amazonq/mcp.json
{
"mcpServers": {
"aws-whatsnew": {
"command": "python",
"args": ["server.py の絶対パス"]
}
}
}
MCP サーバの動作確認
VSCode で Amazon Q Developer の MCP 設定画面を開き、MCPサーバが正常に起動していることを確認します。
- VSCode のAmazon Q Developer 設定画面を開く
- 設定画面で「aws-whatsnew」サーバが緑色（正常起動）になっていることを確認
※ 緑色にならない場合は更新ボタンをクリックしてみてください。MCP サーバが再起動します。
利用方法
Amazon Q Developer で以下のように質問します。
- 「直近1週間のAWSアップデート情報を教えて」→
get_aws_updatesツールが実行されます - 「Bedrockに関連するアップデート情報を教えて」→
filter_by_serviceツールが実行されます
4. 動作確認
- VSCode のAmazon Q Developer から質問を入力する
例：「直近1週間のAWSアップデート情報を教えて」 - MCP が正常起動している場合、MCP が提供しているツールを実行するか確認があるので、実行ボタンをクリックする
実行ボタンをクリックした後、アップデート情報が出力されることを確認できました
まとめ
AWS アップデート 情報を自動取得する MCP サーバをローカル環境に構築し、 Amazon Q Developer と連携させて情報取得が成功することを確認できました。
日々アップデートされる実装コードに関連する技術情報の取得処理を MCP サーバにまとめることができるので、Web ブラウザで検索せずに VSCode で完結できるように整備してもよさそうだと思いました。
参考資料
- AWS 公式サイト「AWS What's New RSS フィード」
https://aws.amazon.com/new/feed/ （参照日：2025 年 12 月 18 日）
記載されている会社名、製品名、サービス名は、各社の商標または登録商標です。