こんにちは、石坂です。
Slackからプロダクトのシステム仕様を気軽に聞けるようにしたくて、Devin MCPとGoogle Cloud上のサービスを組み合わせてボットを作りました。
DevinにはDeepWikiという機能があります。GitHubリポジトリを連携するとコードを解析して、自動的にドキュメントを生成してくれます。2025年7月時点では英語のみですが、それでも十分理解しやすい内容です。
ただし、それはあくまで大きな書物という存在で、全てを読み込んで理解するのは現実的ではありません。
そして、Devinは「Ask Devin」という機能を提供しておりという質問を投げると、DeepWikiの内容を基に自然言語で質問に答えてくれます。
Devinはアカウント数による従量課金ではないので、カスタマーサクセスチームなど非エンジニア含めて多くの人にアカウントを付与しやすい利点があります。
しかし、以下の課題がありました：
- 全員にアカウントを発行する手間
- 無料の「Ask Devin」以外の有料機能を誤って使用するリスク
- Slack以外のツールを開く必要がある
そこで、Slackから直接質問できる仕組みを作ることにしました。
技術スタック
- Python 3.13
- MCP Python SDK
- Google Cloud Functions（第2世代）
- Cloud Pub/Sub
- Devin MCP (Deepwiki API)
- Slack Event API
技術選定：なぜVertex AIではなくMCPライブラリを選んだか
当初はVertex AIの利用を検討しましたが、外部のMCPサービス（Devin MCPなど）との連携には課題がありました：
- MCP Toolbox for Databasesは、BigQueryなどデータベース接続用
- Agent Development Kit（ADK）でも実装が必要。今回のシンプルなユースケースには過剰
Devin MCPの特徴
重要な点は、Devin MCPは内部でLLMを使って回答を自然文で生成して返却しています。
つまり：
- リポジトリの解析 → 回答生成まで一貫して処理
- 追加のLLMによる加工は不要
- 非エンジニア向けの分かりやすい説明が直接得られる
そのため、シンプル化のためにPython MCPライブラリを使って直接Devin MCPと連携する方法を選択しました。
MCP接続部分の実装例
from mcp import ClientSession
from mcp.client.sse import sse_client
# Devin MCPに直接接続
async with sse_client(server_url, headers=headers) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# Deepwikiに質問
result = await session.call_tool(
"ask_question",
arguments={
"repoName": "owner/repo",
"question": "質問内容"
}
)
アーキテクチャ
実装のポイント
MCPクライアントの実装
Python MCPライブラリを使ったシンプルな実装：
async def generate_response_with_devin(query_text: str) -> Optional[str]:
server_url = "https://mcp.devin.ai/sse"
headers = {"Authorization": f"Bearer {DEVIN_API_KEY}"}
async with sse_client(server_url, headers=headers) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# インストラクションを含めて質問
full_question = f"""
[インストラクション]
非エンジニアにも分かりやすく、親しみやすい口調で回答してください。
[質問] {query_text}
"""
result = await session.call_tool(
"ask_question",
arguments={
"repoName": REPO_NAME,
"question": full_question
}
)
# 生成済みの回答を取得
response_text = ""
for content in result.content:
if hasattr(content, 'text'):
response_text += content.text
return response_text
実装時の注意点
1. Pub/SubのACKデッドライン設定
問題: Devin MCPとのやりとりには時間がかかる場合があります。Pub/SubのACKデッドラインがデフォルトの10秒だと、処理成功のACKを返せず、処理中にイベントが再送信されてしまいます。
結果、1つ質問すると、複数の回答(しかも同じ内容)が送られてしまいます。
解決策①: ACKデッドラインを延長
# ACKデッドラインを10分に延長
gcloud pubsub subscriptions update [SUBSCRIPTION_NAME] \
--ack-deadline=600
解決策②: 即座にACKを返す実装
@functions_framework.cloud_event
def process_message(cloud_event: CloudEvent) -> None:
# バックグラウンド処理を開始
threading.Thread(
target=process_in_background,
args=(cloud_event.data,),
daemon=True
).start()
# 即座に成功を返す
return "OK", 200
2. Slackの3秒タイムアウト対策
問題: Slackは3秒以内にレスポンスがないとリトライします。Cloud Functionsのコールドスタート時は要注意です。
コスト削減のためインスタンス数は最小0個にしていたため、一定時間質問がないとインスタンスが停止してしまい、コールドスタートになるケースが度々ありました。結果、3秒以内にSlackにレスポンスを返せず、こちらも複数回同じ回答が送られてくるケースがありました。
解決策: event_idによる重複防止
processed_events = {} # グローバル変数
@functions_framework.http
def handle_slack_event(request: Request):
event_id = request.get_json().get("event_id")
# 重複チェック
if event_id in processed_events:
return "Already processed", 200
processed_events[event_id] = True
この実装はメモリ上で管理しているため、Cloud Functionsのインスタンスが変わると効果がありません。しかし、今回は最大1インスタンスでの運用のため、実用上は十分な回避策となっています。もし複数インスタンスでの運用が必要な場合は、RedisやFirestoreなどの外部ストレージを使った実装を検討してください。
まとめ
Vertex AIのような包括的なプラットフォームではなく、Python MCPライブラリで直接連携することで、シンプルかつ効果的なソリューションを実現できました。
ポイントは：
- MCPプロトコルを直接利用 - 不要な抽象化レイヤーを避ける
- Devin MCPが回答を生成 - 追加の文章加工が不要
- 実装時の課題への対処 - タイムアウトや重複処理への適切な対応
これにより、社内メンバーがSlackからAIに聞くだけで、プロダクト仕様を知れる環境が整いました。
最後に、絶賛採用強化中です！
もし興味を持っていただけた方はこちらをご覧ください！
Discussion