こんにちは！
株式会社セゾンテクノロジーの島です。
今回は、Amazon Q Developer CLI をMCPサーバーにするという試みをしたので、そこでの知見を共有します。
目次
1. はじめに
本記事は、Amazon Q Developer CLI をMCPサーバーにするという試みです。
やってみて気づいたことをまとめています。
Amazon Q Developer CLI についてや、MCPについては以下記事をご参照ください。
2. やりたいこと
Amazon Q Developer CLI を使うには「q chat」と打って起動してから使用します。
これをMCPのツールとして定義すれば、Amazon Q Developer CLI 自体をMCPサーバーにできてしまうのではないかと考えて作成してみました。
実際に作成した構成図が以下の通りです。
3. 実装手順
3-1. EC2上にAmazon Q Developer CLI をインストール
以下の記事を参考に、Amazon Q Developer CLIをEC2上にインストールして、使用できる状態にします。
3-2. MCPサーバー化
次に、Amazon Q Developer CLI をMCPサーバーにするためのpythonスクリプトを作成します。
mcp_server.py
import os
import subprocess
from dotenv import load_dotenv
from fastmcp import FastMCP, Context
import re
ANSI_RE = re.compile(r'\x1b\[[0-9;]*[A-Za-z]')
load_dotenv()
mcp = FastMCP(
name="EC2 MCP Server",
version="1.0.0"
)
@mcp.tool
def run_q_with_tools(prompt: str, timeout: int = 60) -> str:
"""
Amazon Q Developer CLI を --trust-all-tools 付きで呼び出し、
提示された処理を自動実行して結果を返す。
"""
# Amazon Q CLIの絶対パスを指定（ホームディレクトリを展開）
cmd = [os.path.expanduser("~/q/bin/q"), "chat", "--no-interactive", "--trust-all-tools",prompt]
print(f"Q Developer に処理を依頼: {prompt}")
try:
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
raw = proc.stdout or proc.stderr
# ANSI コードを消す
clean = ANSI_RE.sub('', raw).strip()
if proc.returncode == 0:
return clean
else:
return f"実行エラー: {clean}"
except subprocess.TimeoutExpired:
return "タイムアウト: 処理が完了しませんでした"
except Exception as e:
print(f"Q CLI 実行失敗: {e}")
return f"失敗: {e}"
if __name__ == "__main__":
if not os.getenv("FASTMCP_AUTH_BEARER_TOKEN"):
print("警告: FASTMCP_AUTH_BEARER_TOKEN が設定されていません。認証なしで起動します。")
mcp.run(
transport="streamable-http",
host="0.0.0.0",
port=8000,
path="/mcp"
)
Amazon Q Developer CLI をインストールしたパスを指定して、そのコマンドを実行するMCPツールを定義します。
コマンドのオプションには--trust-all-toolsを指定することで、途中で実行が止まってしまうことを防ぎます。
3-3. Systemdサービス登録
次に、作成したpythonファイルをSystemdにサービス登録します。
これにより、MCPサーバーが常駐プロセスとして起動します。
EC2
sudo tee /etc/systemd/system/mcp.service >/dev/null <<'EOF'
[Unit]
Description=MCP Server (mcp_server.py)
After=network-online.target
Wants=network-online.target
[Service]
Type=simple
User=ec2-user
Group=ec2-user
WorkingDirectory=<mcp_server.pyを配置したパス>
ExecStart=<pythonのパス(仮想環境を作成している場合はそのパス)> mcp_server.pyのフルパス>
Restart=on-failure
RestartSec=3
Environment=PYTHONUNBUFFERED=1
[Install]
WantedBy=multi-user.target
EOF
デーモンをリロードします。
EC2
sudo systemctl daemon-reload
mcpのサービスを有効化します。
EC2
sudo systemctl enable mcp
無事、MCPサーバーがEC2上で起動していることが分かります。
3-4. Lambda実装
作成したMCPサーバーを使うAIエージェントをLambdaで作成します。
今回はAIエージェント開発フレームワークとしてStrands Agentsを使用しています。
strands_http_client.py
import boto3
import os
import json
from strands import Agent, tool
from strands.models import BedrockModel
from strands.tools.mcp.mcp_client import MCPClient
from mcp.client.streamable_http import streamablehttp_client
from slack_sdk import WebClient
import logging
from botocore.config import Config
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
session = boto3.Session(region_name=os.environ.get("REGION", "us-west-2"))
@tool
def assess_alert_severity(event_data: str) -> str:
"""アラートの重要度を評価"""
bedrock_model = BedrockModel(
model_id=os.environ.get("MODEL_ID", "us.anthropic.claude-3-7-sonnet-20250219-v1:0"),
boto_session=session
)
agent = Agent(
model=bedrock_model,
system_prompt="""アラート通知の重要度を評価してください。
以下の形式で簡潔に出力してください：
- 重要度: CRITICAL/HIGH/MEDIUM/LOW
- 判定理由: 要点のみ
- 調査要否: 必要/不要"""
)
return str(agent(event_data))
@tool
def generate_investigation_prompt(alert_info: str, severity_info: str) -> str:
"""Amazon Q調査用プロンプトを生成"""
bedrock_model = BedrockModel(
model_id=os.environ.get("MODEL_ID", "us.anthropic.claude-3-7-sonnet-20250219-v1:0"),
boto_session=session
)
agent = Agent(
model=bedrock_model,
system_prompt="""Amazon Q Developer CLI用の調査指示を作成してください。
以下の形式で簡潔に出力してください：
- 問題の状況共有: 現状の説明
- 原因調査依頼
- 問題解決依頼"""
)
return str(agent(f"アラート情報:\n{alert_info}\n\n重要度評価:\n{severity_info}"))
@tool
def analyze_results(investigation_results: str, alert_context: str) -> str:
"""調査結果を分析してレポート作成"""
bedrock_model = BedrockModel(
model_id=os.environ.get("MODEL_ID", "us.anthropic.claude-3-7-sonnet-20250219-v1:0"),
boto_session=session
)
agent = Agent(
model=bedrock_model,
system_prompt="""調査結果を分析してレポートを作成してください。
以下の形式で簡潔に出力してください：
- 問題概要: 要点のみ
- 根本原因: 特定された原因
- 対応策: 推奨アクション"""
)
return str(agent(f"調査結果:\n{investigation_results}\n\nアラート情報:\n{alert_context}"))
@tool
def send_slack_notification(message: str) -> str:
"""Slackにメッセージを送信"""
token = os.environ.get('SLACK_BOT_TOKEN')
channel = os.environ.get('SLACK_CHANNEL')
client = WebClient(token=token)
response = client.chat_postMessage(channel=channel, text=message)
return "Slack notification sent successfully"
def lambda_handler(event, context):
"""Lambda関数メインハンドラー"""
try:
bedrock_model = BedrockModel(
model_id=os.environ.get("MODEL_ID", "us.anthropic.claude-3-7-sonnet-20250219-v1:0"),
boto_session=session,
boto_client_config=Config(
read_timeout=900,
connect_timeout=900,
retries={"max_attempts": 3, "mode": "adaptive"},
),
)
# MCP接続
mcp_url = os.environ.get("MCP_URL")
token = os.environ.get('MCP_TOKEN')
headers = {"Authorization": f"Bearer {token}"} if token else {}
mcp_client = MCPClient(lambda: streamablehttp_client(mcp_url, headers=headers))
with mcp_client:
mcp_tools = mcp_client.list_tools_sync()
logger.info(f"Retrieved {len(mcp_tools)} tools from MCP server")
# 監督者エージェント作成（専門ツール + MCPツール + Slack通知ツール）
supervisor = Agent(
model=bedrock_model,
system_prompt="""あなたはAWSの障害通知を分析し、適切な調査・対応対応を指揮する監督者エージェントです。
以下の手順で作業を進めてください：
1. assess_alert_severityツールを使用してアラートの重要度を評価
2. 調査・対応が必要な場合は、generate_investigation_promptツールで調査プロンプトを生成
3. 利用可能なMCPツールを使用して実際の調査・対応を実行
4. analyze_resultsツールで最終レポートを作成
5. send_slack_notificationツールで結果をSlackに通知
各ステップの結果を次のステップに適切に引き継いでください。
""",
tools=[assess_alert_severity, generate_investigation_prompt, analyze_results, send_slack_notification] + mcp_tools
)
# イベントデータを文字列として渡す
event_str = json.dumps(event, ensure_ascii=False, indent=2)
# 監督者エージェント実行
supervisor_prompt = f"""以下の障害通知を分析し、適切な対応を実施してください：
{event_str}
"""
response = supervisor(supervisor_prompt)
# 監督者エージェントの応答をログに出力
logger.info(f"Supervisor Agent Response: {str(response)}")
return {
'statusCode': 200,
'body': json.dumps({
'message': 'Alert investigation completed',
'response': str(response)
}, ensure_ascii=False)
}
except Exception as e:
logger.error(f"Lambda handler error: {e}", exc_info=True)
return {
'statusCode': 500,
'body': json.dumps({
'error': str(e),
'type': type(e).__name__
}, ensure_ascii=False)
}
注意
Strands Agentsに渡すboto3セッションのタイムアウト時間が短いとエラーになります。
3-5. 疑似障害を起こす
実際に監視対象のEC2で疑似障害を起こしてみましょう。
EC2
stress --cpu 2
そうすると、slackに以下のような通知が届きました。
ログを見ると、原因プロセスをAmazon Q Developer CLIがkillしてくれていることが分かります。
4. やってみて気づいたMCPの意義
4-1. 課題
ここまでで、Amazon Q Develper CLI をMCPサーバーにしてみましたが果たしてこれはMCPにする意味があったのでしょうか？
今回MCPとしたことで、以下のような課題があると思います。
- 無駄に複雑性が増してしまう
- 通信コストがかかってしまう
- MCPサーバー分のコストがかかってしまう
上記をまとめるとつまり、今回のユースケースに対して無駄が多いということになります。
もっとシンプルなアーキテクチャで同じことを実装する方法はたくさんあります。
4-2. MCPのメリット
では、MCPの利点は何でしょうか？
自分が考えるMCPの利点は以下の2点です。
- 1つの Tool Use を多くの人が同じように利用できる
- 統合されたクライアントに対して、簡単に組み込むことができる
4-3. 展望
これらを踏まえると、例えば以下のようなユースケースならメリットがありそうです。
- Amazon Q Developer CLI をリモートMCPサーバーとして、それぞれのローカルクライアントが使用
こうすることで、それぞれのクライアントでAmazon Q Developer CLI をインストールする手間が減ります。
また、環境を触ることができるAIエージェントを統一することで、管理が簡単になります。
5. さいごに
今回はAmazon Q Developer CLI をMCPサーバーにするという試みをしてみました。
その中でMCP活用のアンチパターンを作ってしまいましたが、MCPの意義について考え直すいい機会になったと思います。
この記事がどなたかの参考になれば幸いです。