はじめに 前回の記事 では、Webアプリで映像を扱う際の産業用途におけるリアルタイムコミュニケーション実現のための検討ポイントと、intdashとintdash-RTC SDKという選択肢を紹介しました。 要点をおさらいすると: SFU選定時はベンダー固有の実装に依存する点も考慮が必要 センサーデータとの統合や録画・解析は追加の設計が必要になる可能性 intdashは、産業用途に適したもう一つの選択肢 前編では概要しか触れなかったため、実際の使用感について興味をお持ちいただいた方もいらっしゃると思います。 本記事では、intdash-RTC SDKの具体的な実装方法をサンプルコード付きで解説します。環境が準備できていれば、約30分で動作するビデオチャットが完成しますので、ぜひお試しください。 ライブラリの概要 intdashは映像・音声・センサーデータを統合して伝送できるプラットフォームです。intdash-RTC SDKは、intdashで扱うデータのうち映像・音声をWebアプリから簡単に扱えるようにするライブラリです。 本ライブラリは3つのインターフェースで構成されています。 ┌─────────────────────────────────────────────┐ │ MediaConnection │ 統合管理 ├─────────────────────┬───────────────────────┤ │ MediaSender │ MediaReceiver │ 送受信 └─────────────────────┴───────────────────────┘ | ^ v | ┌─────────────────────────────────────────────┐ │ intdash (iSCP) │ 伝送 └─────────────────────────────────────────────┘ MediaConnection: 送受信を統合管理。 start() / stop() で接続制御 MediaSender: ローカルメディアをintdashに送信 MediaReceiver: intdashからメディアを受信し、video要素に直接接続 本ライブラリは、TypeScript 向けのintdash通信ライブラリ iscp-ts のラッパーとしても利用できますし、今回のようなアプリケーションでは接続設定を直接指定するシンプルな使い方も可能です。 実装ステップ 以下では、基本的なビデオチャットを実装するサンプルコードを5つのステップに分けて紹介します。全量は記事末尾の「Appendix: サンプルコード全文」に再掲しています。 1. ライブラリのimportとMediaConnectionの作成 // 1. ライブラリのimport import { createMediaConnection } from "@aptpod/intdash-rtc" ; // 2. MediaConnectionの作成（intdashへの接続も内部で行います） const { mediaConnection } = await createMediaConnection( { address : "your-intdash.example.jp" , // intdash接続アドレス projectUuid : "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" , // intdashプロジェクトのUUID nodeId : "yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyy" , // 送信元（自分側）のノードUUID enableTLS : true , apiToken : "your_token_here" , // intdash Web Consoleから作成したAPIトークン（OAuth2サインインを使う場合はapiTokenを省略） } , { sender : { // 音声に関する設定 audio : { codec : "PCM" , // PCM, OPUS, AACをサポート } , // 映像に関する設定 video : { codec : "H264" , // H264, VP9をサポート } , } , receiver : { sourceNodeIds : [ "zzzzzzzz-zzzz-zzzz-zzzz-zzzzzzzzzzzz" ] , // 通信相手のノードUUID } , } , ); @aptpod/intdash-rtc が今回のintdash-RTC SDKです。これをimportするだけで、intdashを介したビデオチャットが実装できます（ @aptpod/iscp-ts は peer dependency として併せてインストールが必要です）。 createMediaConnection 関数で、intdashへの接続とMediaConnectionの作成をまとめて行います。第1引数で接続情報（アドレス・プロジェクトUUID・ノードUUID・APIトークン）を、第2引数で送受信オプションを指定します。これらの値は intdash Web Console から取得できます。APIトークンは長期間有効な固定トークンですが、より安全なワンタイムトークンやOAuth2による認証も利用可能です。送受信オプションは階層化された構造になっており、 sender.audio で音声設定、 sender.video で映像設定、 receiver で受信対象ノードを指定します。コーデックは型安全な文字列リテラル型で指定でき、設定ミスを防ぎます。 2. ローカルメディアの取得と表示 // 3. ローカルメディアストリームの取得と設定 const localStream = await navigator .mediaDevices. getUserMedia ( { video : true , audio : true , } ); await mediaConnection.sender.setLocalMediaStream(localStream); // ローカル映像をvideo要素に表示 const localVideoEl = document . getElementById ( "local-video" ) as HTMLVideoElement ; localVideoEl.srcObject = new MediaStream (localStream.getVideoTracks()); localVideoEl.muted = true ; await localVideoEl.play(); getUserMedia でカメラとマイクからストリームを取得し、 setLocalMediaStream で設定します。取得したストリームはそのままローカルのvideo要素にも表示できます。 3. 受信メディアのvideo要素への接続 // 4. 受信メディアストリームの処理 const remoteVideoEl = document . getElementById ( "remote-video" , ) as HTMLVideoElement ; mediaConnection.receiver.on( "statechange" , ( state ) => { if (state === "running" ) { // 受信が開始されたら、video要素に直接接続 mediaConnection.receiver.attach(remoteVideoEl); } else if (state === "stopped" || state === "failed" ) { // 受信が停止したら、video要素から切断 mediaConnection.receiver.detach(remoteVideoEl); } } ); 受信側の状態変化は statechange イベントで監視します。 running になったら attach() でvideo要素に直接接続できます。内部の変換処理（デコード等）はintdash-RTC SDKが処理するため、意識する必要がありません。 4. 接続開始 // 5. 接続開始（running状態まで待機） await mediaConnection.start( { waitUntil : "running" } ); console . log ( "接続が確立しました" ); start({ waitUntil: 'running' }) で接続を開始し、 running 状態になるまで待機します。これにより、接続が完全に確立してから次の処理に進めます。 5. UI操作と停止処理 // 6. UI操作（映像・音声の有効/無効切り替え） document . getElementById ( "toggle-video" )?. addEventListener ( "click" , async () => { const enabled = await mediaConnection.sender.setVideoEnabled( !mediaConnection.sender.videoEnabled, ); console .log( `映像: ${ enabled ? "有効" : "無効" } ` ); } ); document . getElementById ( "toggle-audio" )?. addEventListener ( "click" , async () => { const enabled = await mediaConnection.sender.setAudioEnabled( !mediaConnection.sender.audioEnabled, ); console .log( `音声: ${ enabled ? "有効" : "無効" } ` ); } ); // 7. 停止処理（クリーンアップ） document . getElementById ( "stop" )?. addEventListener ( "click" , async () => { await mediaConnection. stop (); await mediaConnection.iscpConn. close (); localStream.getTracks(). forEach (( track ) => track. stop ()); console .log( "接続を終了しました" ); } ); setVideoEnabled / setAudioEnabled で映像・音声を個別に制御できます。終了時は stop() でメディア処理を停止し、 iscpConn.close() でintdash接続もクローズします。内部のワークレットやバッファも自動的にクリーンアップされます。 実際の動作の確認 このように、WebRTCと遜色ない簡潔さで、intdashを使った映像・音声通信を実装できます。 実際にサンプルアプリを動作させた様子が以下の動画です。 youtu.be 画面の左側がサンプルアプリ「Easy Video Chat」、右側がintdashに付属する簡易可視化ツール「Edge Finder」です。 サンプルアプリ「Easy Video Chat」には「あなた」（左）と「相手」（右）の2つの映像表示があります。今回のデモでは送信元と受信元のノードを同一に設定しているため、左右に同じ映像が表示されています。左の「あなた」は getUserMedia で取得したローカル映像をそのまま表示したもの、右の「相手」はそのローカルメディアをintdashへ送信したあと再度受信してデコード・表示したものです。 実際に二者間で通話する場合は、相手側でも同様にintdashへの接続と受信元ノードの指定を行うことで、双方向での映像・音声通話が成立します。 同時にEdge Finder側では、送信されたH.264映像フレームとPCM音声データが、それぞれ時系列のデータポイントとしてintdashに記録されていく様子が確認できます。 さらなる拡張性 データの可視化とダッシュボードのカスタマイズ 動画で紹介した「Edge Finder」はintdashに標準で付属しており、送受信されているデータをすぐに確認できる簡易ツールです。 より高度な可視化やダッシュボードのカスタマイズが必要な場合は、弊社の可視化アプリケーション「VM2M Data Visualizer」をご利用いただけます。ノーコードで時系列データのダッシュボードを構築でき、映像とセンサーデータを時刻同期させた再生や、多彩なウィジェットを組み合わせた解析画面の作成に対応しています。 VM2M Data Visualizer: 複数の映像と時系列データを時刻同期して再生するタイムラインビュー VM2M Data Visualizer: ゲージ・グラフ・レーダーチャートなど多彩なウィジェットを組み合わせたカスタムダッシュボード Appendix: サンプルコード全文 ここまでステップごとに分割して紹介してきたサンプルコードの全量を、以下に再掲します。 // 1. ライブラリのimport import { createMediaConnection } from "@aptpod/intdash-rtc" ; // 2. MediaConnectionの作成（intdashへの接続も内部で行います） const { mediaConnection } = await createMediaConnection( { address : "your-intdash.example.jp" , // intdash接続アドレス projectUuid : "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" , // intdashプロジェクトのUUID nodeId : "yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyy" , // 送信元（自分側）のノードUUID enableTLS : true , apiToken : "your_token_here" , // intdash Web Consoleから作成したAPIトークン（OAuth2サインインを使う場合はapiTokenを省略） } , { sender : { // 音声に関する設定 audio : { codec : "PCM" , // PCM, OPUS, AACをサポート } , // 映像に関する設定 video : { codec : "H264" , // H264, VP9をサポート } , } , receiver : { sourceNodeIds : [ "zzzzzzzz-zzzz-zzzz-zzzz-zzzzzzzzzzzz" ] , // 通信相手のノードUUID } , } , ); // 3. ローカルメディアストリームの取得と設定 const localStream = await navigator .mediaDevices. getUserMedia ( { video : true , audio : true , } ); await mediaConnection.sender.setLocalMediaStream(localStream); // ローカル映像をvideo要素に表示 const localVideoEl = document . getElementById ( "local-video" ) as HTMLVideoElement ; localVideoEl.srcObject = new MediaStream (localStream.getVideoTracks()); localVideoEl.muted = true ; await localVideoEl.play(); // 4. 受信メディアストリームの処理 const remoteVideoEl = document . getElementById ( "remote-video" , ) as HTMLVideoElement ; mediaConnection.receiver.on( "statechange" , ( state ) => { if (state === "running" ) { // 受信が開始されたら、video要素に直接接続 mediaConnection.receiver.attach(remoteVideoEl); } else if (state === "stopped" || state === "failed" ) { // 受信が停止したら、video要素から切断 mediaConnection.receiver.detach(remoteVideoEl); } } ); // 5. 接続開始（running状態まで待機） await mediaConnection.start( { waitUntil : "running" } ); console . log ( "接続が確立しました" ); // 6. UI操作（映像・音声の有効/無効切り替え） document . getElementById ( "toggle-video" )?. addEventListener ( "click" , async () => { const enabled = await mediaConnection.sender.setVideoEnabled( !mediaConnection.sender.videoEnabled, ); console .log( `映像: ${ enabled ? "有効" : "無効" } ` ); } ); document . getElementById ( "toggle-audio" )?. addEventListener ( "click" , async () => { const enabled = await mediaConnection.sender.setAudioEnabled( !mediaConnection.sender.audioEnabled, ); console .log( `音声: ${ enabled ? "有効" : "無効" } ` ); } ); // 7. 停止処理（クリーンアップ） document . getElementById ( "stop" )?. addEventListener ( "click" , async () => { await mediaConnection. stop (); await mediaConnection.iscpConn. close (); localStream.getTracks(). forEach (( track ) => track. stop ()); console .log( "接続を終了しました" ); } ); おわりに 本記事では、intdash-RTC SDKを使ったビデオチャットの実装方法を見てきました。 ポイントのまとめ: WebRTCと同等の簡潔さで、intdash経由のビデオチャットを実装可能 最初はビデオチャットから始めて、将来はセンサーデータを含むマルチモーダルシステムへ拡張できるのがintdashの強みです。同じプラットフォーム、同じSDKで対応できるため、技術選定で後悔するリスクを減らせます。 「試してみたい」「もう少し詳しく知りたい」という方は、ぜひお気軽に お問い合わせ ください。本ライブラリは現在お問い合わせベースでの個別提供となっています。

はじめに こんにちは。開発本部 開発1部 デリッシュリサーチチームの江﨑です。 デリッシュリサーチは、デリッシュキッチンに蓄積された検索ログやレシピへの反応をもとに食トレンドを分析できるサービスです。 本記事では、社内用にデリッシュリサーチのデータを Claude から自然言語で問い合わせられるようにする MCP サーバーを自作した話を紹介します。FastMCP と Databricks Apps で実装した構成、運用上のノウハウ、そしてリリース後に社内で広がった活用事例をまとめます。 はじめに 背景：なぜ自前の MCP サーバーを作ったか システム全体像 使用技術 MCP サーバーの実装 ツール一覧 ツールの基本パターン シノニムの代表語正規化 Databricks Apps での運用ノウハウ app.yaml で起動コマンドを定義 Resource 宣言の上限 20 個 ツール呼び出し履歴を Unity Catalog に蓄積 実際の呼び出しの流れ 試してみた事例 ダッシュボードでは出せない切り口に対応 MCP × Web 検索を組み合わせたトレンド分析 商材タイアップ提案の素材集め まとめ 参考リンク 背景：なぜ自前の MCP サーバーを作ったか デリッシュリサーチには各種分析機能があり、検索トレンド・レビュー・お気に入りレシピなど、マーケティングに使えるデータが一通り揃っています。ただ、実際に業務で使おうとすると、いくつかの壁がありました。 分析の手間 ：デリッシュリサーチを開いて、目当ての機能を探して、フィルタを設定して、結果から示唆を得る、という手順が必要です。気になったことをちょっと確認する用途には、やや手数が多くなります。 分析の仕方に一定のスキルが必要 ：どの切り口で見るか、どのフィルタを組むか、数字をどう読み解くかは訓練が要る作業で、誰でもすぐに使いこなせるとはいきません。 画面を横断した分析がしづらい ：それぞれの機能はタブごとに分かれており、「検索データ × 気温 × 都道府県」のような掛け合わせの問いには答えづらい構造でした。 実際のデリッシュリサーチの画面は次のようになっています。 デリッシュリサーチの画面 デリッシュリサーチは社外のクライアント企業に提供しているサービスですが、社内のメンバーも利用しており、同じ壁にぶつかっていました。一方で、ちょうど社内には Claude が広く配布され始めており、 「Claude で何ができるか」の具体例を作って社内全体での活用を推進したい という思いもありました。デリッシュリサーチのデータを Claude のチャットから自然言語で聞ける MCP を作れば、上の 3 つの壁を越えつつ、社内向けの Claude 活用事例にもなります。 加えて、Databricks には Managed MCP サーバーという仕組みがあります。これは Unity Catalog の汎用ツール（Vector Search、Genie Space、Databricks SQL、Unity Catalog Functions）を MCP として AI クライアントに提供するものです。ただし、シノニム正規化や複数テーブル結合を伴うドメイン特有の集計を Managed MCP に組み込むのは難しく、求めている精度のデータを提供するには物足りませんでした。 そこで、リサーチデータの集計操作そのものをツールとして公開する 自前の MCP サーバー を作ることにしました。 システム全体像 構成は以下のとおりです。 システム全体像 チームの admin が Custom Connector に MCP サーバーを一度登録すれば、利用者は Claude.ai や Claude Desktop から Databricks の SSO でログインするだけで接続できます。MCP サーバー本体は Databricks Apps 上で稼働しており、Databricks SDK を経由して Unity Catalog のテーブルにアクセスします。 Databricks Apps は、Databricks ワークスペース上で Web アプリケーションをホストできる機能です。Unity Catalog と同じワークスペース上で動かせるため、アプリに自動で割り当てられるサービスプリンシパルからそのまま Unity Catalog のテーブルにアクセスできます。 認証は Databricks ワークスペースの OAuth で行います。利用者が Claude から MCP に接続すると、最初に Databricks のログイン画面に飛ばされてログインします。以降の MCP リクエストには発行されたアクセストークンが付与されます。Databricks Apps の手前にあるリバースプロキシがそれを検証したうえで、認証済みのエンドユーザー情報を x-forwarded-email などのヘッダーに載せてアプリに転送してくれます。MCP サーバーのコード側ではこのヘッダーを読むだけで「誰が呼び出しているか」が分かります。 使用技術 本 MCP サーバーで使用している主な技術は以下のとおりです。 項目 技術 言語 Python 3.11 以上 MCP フレームワーク FastMCP 2.x Web フレームワーク FastAPI + uvicorn ホスティング Databricks Apps データ層 Unity Catalog（Delta テーブル） 観測 OpenTelemetry パッケージ管理 uv FastMCP は Python で MCP サーバーを書くためのフレームワークです。MCP の HTTP ベースの通信方式（Streamable HTTP）に対応しており、Databricks Apps 上で動かせます。 MCP サーバーの実装 ツール一覧 現在、MCP サーバーには次のようなツールが登録されています。 キーワードの検索数を期間・粒度を指定して取得 検索ワードランキング 組み合わせ検索ランキング 主ワードに対する副ワードの傾向分析 都道府県別の検索ワードランキング 都道府県別の組み合わせ検索ランキング お気に入りに追加されたレシピのランキング レシピ単位のレビューと平均評価 気温と検索数の相関 食材の物価データ 食材の物価の前年同月比ランキング ツールの基本パターン FastMCP では、Python の関数に @mcp_server.tool デコレータを付けるだけでツールとして登録できます。本 MCP では、関数の docstring・引数のバリデーション・SQL 実行・整形して返却、という流れを全ツールで揃えています。 def register (mcp_server): @ mcp_server.tool # FastMCP にツールとして登録 def get_search_trends ( search_word: str = "" , start_date: str = "" , end_date: str = "" , granularity: str = "monthly" , ) -> dict : """指定ワードの検索数推移を期間・粒度を指定して取得する。 Args: search_word: 検索ワード（例: "キャベツ"）。シノニムは自動で代表語に正規化される。 start_date: 開始日（YYYY-MM-DD、省略時は前年同月の月初） end_date: 終了日（YYYY-MM-DD、省略時は前日） granularity: 粒度（daily / weekly / monthly / quarterly / yearly） """ # 引数のバリデーション search_word = search_word.strip() if not search_word: return { "error" : "search_word is required" } # 表記ゆれを代表語に正規化（後述のシノニム正規化） main_word = resolve_to_main(search_word) # SQL を実行し、結果を整形して返却 try : rows = _fetch_basic_trends(main_word, start_date, end_date, granularity) return { "search_word" : main_word, "data_points" : len (rows), "data" : rows} except Exception : # エラーは Unity Catalog の観測テーブルに記録される logger.exception( "get_search_trends failed" , extra={ "tool" : "get_search_trends" }) return { "error" : "Internal error" } ツールの docstring は AI クライアントがそのままツールの説明として参照します。AI 側がどのツールをどんな引数で呼ぶかは、この docstring の質に強く依存します。 Args: の各引数に「何を渡してよいか」「省略時の挙動」を必ず明記する運用にしています。 シノニムの代表語正規化 集計テーブルの search_word は、ETL の段階で表記ゆれや言い換えが代表語に寄せられた状態で保存されています。たとえば「キャベツ」「きゃべつ」のような表記の違いは代表語「キャベツ」に集約されています。MCP のツール側でも、入力されたワードを resolve_to_main() で代表語に変換してから集計テーブルを参照するようにしました。これにより、ユーザーがどの表記で問い合わせても同じ結果を返せます。 Databricks Apps での運用ノウハウ app.yaml で起動コマンドを定義 Databricks Apps では、リポジトリのルートに app.yaml を置いたうえで、Workspace 上の管理画面から GitHub リポジトリを連携してデプロイしています。ソースを更新した後は Workspace 側で再デプロイを実行するだけで反映されます。 command : [ "uv" , "run" , "opentelemetry-instrument" , "custom-mcp-server" ] env : - name : WAREHOUSE_ID value : "<warehouse-id>" opentelemetry-instrument を経由して起動することで、アプリのログ・トレース・メトリクスが自動で Unity Catalog 上の観測テーブルに流れます。 Resource 宣言の上限 20 個 Databricks Apps では、参照したいテーブルや SQL Warehouse を UI上で宣言できます。宣言したリソースはデプロイ時にアプリのサービスプリンシパルへ自動で GRANT され、削除時には自動で REVOKE されます。 運用していて詰まったのが、 1 アプリあたりに宣言できる resource の数は 20 個まで という制約です。これは 2026 年 5 月時点で公式ドキュメントには明記されておらず、実際に 21 個目を追加しようとしてエラーで気付きました。今回の MCP では参照するテーブルが 20 個を超えており、すぐに上限に到達してしまいました。 そのため、UIからではなく Unity Catalog 側で 手動 GRANT を運用する方式 に切り替えました。 GRANT USE CATALOG ON CATALOG external_marketing_research TO `<sp-uuid>`; GRANT USE SCHEMA ON SCHEMA external_marketing_research.search TO `<sp-uuid>`; GRANT SELECT ON TABLE external_marketing_research.search.search_count TO `<sp-uuid>`; ツール呼び出し履歴を Unity Catalog に蓄積 社内向けに公開している以上、誰がどのツールをどの引数で呼んだかを追えるようにしておきたい要件がありました。 FastMCP には Middleware クラスがあり、ツール呼び出しの前後にフックを差し込めます。これを使い、呼び出し履歴を Unity Catalog のテーブルに INSERT する仕組みを入れました。MCP が参照している他の集計テーブルと同じカタログに置くことで一元化できます。 class HistoryMiddleware (Middleware): async def on_call_tool (self, context, call_next): tool_name = context.message.name user = get_request_user() # x-forwarded-email から取得 start = time.perf_counter() is_error = False try : return await call_next(context) except Exception : is_error = True raise # 例外は握りつぶさず外側に伝播させる finally : # 成功/失敗どちらの場合も履歴を書く duration_ms = (time.perf_counter() - start) * 1000 self._schedule_write( tool_name=tool_name, user_email=user[ "email" ] if user else None , arguments_json=_serialize_arguments(context.message.arguments), duration_ms=duration_ms, is_error=is_error, ) Databricks Apps では、リクエストヘッダーの x-forwarded-email にエンドユーザーのメールアドレスが入ります。これを contextvars で受け取り、INSERT 時に user_email として記録しています。 このテーブルを使うと、誰が何を呼んだかを追える監査記録としてだけでなく、次のような切り口で利用状況を集計できます。 ツール別の呼び出し頻度 ：どのツールが実際によく使われているかを把握し、機能改善の優先度を判断する ツール別のエラー率と所要時間 ：例外発生率と平均レスポンス時間からツールの健全性を確認する ユーザー別の利用状況 ：誰がどれくらい使っているかを見て、活用事例のヒアリング相手を決める 実際の呼び出しの流れ たとえば「リサーチ MCP を使用して、キャベツのトレンドを分析してください」と Claude に投げかけると、必要なツールを Claude が自分で判断して順番に呼び出し、結果を分析・可視化して回答してくれます。 MCPの呼び出し例 この例では、次の流れで動いています。 まず利用可能なツールを把握する 過去 2 年分の検索トレンドと組み合わせワードのツールを呼び出す 続けて物価データのツールも呼び出して相関を確認する 取得したデータをもとに、要約（前年同期比、価格との相関係数、ピーク月など）とグラフを生成する 利用者は自然言語で問いを投げるだけで、その背後で複数のツールが呼ばれ、検索データ・物価データ・組み合わせワードを横断したアウトプットが返ってきます。 試してみた事例 社内リリース後、エンジニアからビジネス職まで、さまざまなメンバーに触ってもらいました。まだ実業務での運用に組み込んでいるわけではなく、各自で試しながら可能性を探っている段階ですが、いくつか興味深い使い方が出てきたので紹介します。 ダッシュボードでは出せない切り口に対応 「平日と比較して休日に検索頻度が増加する料理を教えてください」のように、デリッシュリサーチの管理画面上のプリセットでは答えられない切り口の問いにも対応できます。AI 側で日次データを取得するツールを呼び出し、平日／休日の比率を計算する手順を自分で組み立てて回答します。 MCP × Web 検索を組み合わせたトレンド分析 「リサーチ MCP の検索データ」と「Web 検索」を併用して、特定の食材・調理法のブームを背景情報込みで分析するような使い方も試されています。たとえば、せいろブームの背景に Web 上のどんな話題があり、検索数とどう連動しているかをまとめる、といった分析です。 商材タイアップ提案の素材集め 「ビール商材向けにおつまみレシピのタイアップ提案を考えて」のような依頼でも、MCP からレビュー数の多いレシピや時期別の検索傾向を取得し、提案スライドまで作るところまで行われました。 数値や因果関係の最終確認は必要ですし、プロンプトや精度にもまだ改善の余地があります。とはいえ、提案のドラフトを素早く試作する用途として手応えを感じています。 まとめ 今回は社内向けのデリッシュリサーチ MCP サーバーを自作した話を紹介しました。 リリース後、エンジニアからビジネス職までいろいろなメンバーに触ってもらっています。デリッシュリサーチ単体では難しい横断的な分析や、分析からスライド作成までを一気通貫でやれる点に可能性を感じています。SNS など外部のデータソースも組み合わせて、活用の幅を伸ばしていきたいです。 最後まで読んでいただきありがとうございました！ 参考リンク Model Context Protocol FastMCP Databricks Apps Databricks マネージド MCP サーバーを使用する Unity Catalog 権限リファレンス

本記事は、シリーズ「AWS における AI エージェント対応のデータ基盤」の第 2 回です。 第 1 回 では、AI エージェントが組織の本番データに対して正しく動くために必要な 3 要素（認可・ビジネスデータカタログ・ドメイン知識）を紹介し、認可が効いている様子をデモで示しました。本記事では、3 要素のうち認可に焦点を当て、AI エージェント経由のデータアクセスに Amazon SageMaker Catalog のアクセス制御を透過的に効かせる実装パターンを解説します。 サンプルリポジトリ: aws-samples/sample-sagemaker-agentic-analyst aws-samples sample-sagemaker-agentic-analyst A demo application that demonstrates how fine-grained access controls configured in SageMaker Unified Studio are transparently enforced on data access through AI agents. AI エージェントにアクセス制御を効かせる 3 つの壁 AI エージェントにデータアクセスを任せるとき、守るべき原則があります。エージェントは独自の認可ロジックを持たず、ユーザーが SageMaker Catalog で付与されている権限を、エージェント経由でもそのまま透過的に利用させることです（SageMaker Catalog は Amazon DataZone の上に構築されており、権限設定は DataZone API で操作します）。エージェント内に独自の認可ロジックを作ると、既存のガバナンスと二重管理になり、整合性を保つのが難しくなるからです。 この原則を実現しようとすると、素直な実装ではたどり着けない 3 つの壁があります。これらは本記事で解説する設計上の工夫によって越えられます。 壁 1: コンピュートリソース自体のロールで権限を取得すると、全ユーザーが同一権限になる。 AI エージェントのツールは AWS Lambda 、 AWS Fargate 、 Amazon EC2 など何らかのコンピュートリソース上で実行されます。そのコンピュートリソース自体のロール（たとえば Lambda の実行ロール）で DataZone の GetEnvironmentCredentials API を呼ぶと、返ってくるのはそのロール自身のメンバーシップに基づくプロジェクトロールです。どのユーザーがリクエストしても同じ認証情報が返るため、ユーザー個別のアクセス制御を効かせるには工夫が必要です。 壁 2: SAML フェデレーション経由のトークンには、IdP 側のグループ情報が乗らない。 AWS IAM Identity Center （以下 IdC）と、たとえば Amazon Cognito を SAML フェデレーションで連携する構成では、IdC 側のグループ情報が Cognito のトークンに自動的には含まれません。「データコンシューマーには athena_query を許可し、ドメイン管理者には cloudtrail_query のみ許可する」といったツール単位の認可を IdC のグループベースで行うには、グループ情報をトークンに載せる工夫が必要です。 壁 3: 設定時と実行時で関与するサービスが異なり、認可の全体像を把握する必要がある。 SageMaker Catalog の裏側では複数のサービスが連携しています。設定時に使うサービスと、クエリ実行時に評価されるサービスが異なるため、全体像を把握しないと正しい実装にたどり着けません。 本記事では、サンプルリポジトリ sample-sagemaker-agentic-analyst がこれらの壁をどう越えているかを解説します。 サービスの役割分担を整理する 本記事で扱うサービスの関係を先に整理します。 Amazon SageMaker Catalog は、データと AI の発見・ガバナンス・コラボレーションを担うサービスで、Amazon DataZone の上に構築されています。データの Publish/Subscribe やアクセス権の設定は SageMaker Catalog（および Amazon SageMaker Unified Studio の UI）で行いますが、実際のクエリ実行時に行・列レベルのアクセス制御を評価するのは AWS Lake Formation です。S3 上のファイルに対するアクセス制御は Amazon S3 Access Grants が担います。 つまり、SageMaker Catalog で「誰がどのデータを見てよいか」を 設定 し、Lake Formation と S3 Access Grants が 実行時 にその設定を評価する、という役割分担です。 重要な原則として、DataZone はクエリ実行パスに入りません。後述する認証情報変換フローでは DataZone API（ RedeemAccessToken / GetEnvironmentCredentials ）を呼びますが、これは AgentCore Gateway に接続された Lambda（以下 Tool Lambda）が プロジェクトロールの認証情報を取得する ための前段であり、Athena クエリや S3 オブジェクト取得そのものには DataZone は介在しません。クエリ実行時の認可評価は Lake Formation と S3 Access Grants が担います。この「認証情報取得の経路」と「データアクセスの経路」の分離を理解しておくと、以降のフローが読みやすくなります。 認証情報変換フロー: 5 つのステップ 壁 1 と壁 3 に対応するため、本サンプルでは 5 つのステップで認証情報を変換します。ブラウザでサインインしたユーザーの Cognito トークンから出発し、最終的にそのユーザーの権限が反映された SageMaker プロジェクトロールの一時認証情報を Tool Lambda の手元に届けます（下図）。 Step 1: ブラウザから AgentCore Runtime へ ブラウザ上の React アプリが、 Amazon Bedrock AgentCore Runtime のエンドポイントに HTTPS リクエストを送ります。このリクエストには 2 つのトークンが乗っています。 Authorization: Bearer <Cognito Access Token> — Runtime の Cognito Authorizer が検証します カスタムヘッダー X-Amzn-Bedrock-AgentCore-Runtime-Custom-Cognito-Id-Token: <Cognito ID Token> — 次のステップで使います Cognito Access Token と Cognito ID Token はどちらも Amazon Cognito が発行する JSON Web Token（JWT）ですが、役割が異なります。Access Token は「このリクエストは正当なユーザーから来たか」を Runtime と Gateway が判定するために使います。ID Token はユーザーのアイデンティティ情報（メールアドレスなど）を含んでおり、次のステップで IdC のユーザーと突き合わせるために使います。 Step 2: chat-agent が Cognito ID Token を IdC Access Token に引き換える Amazon Bedrock AgentCore Runtime はエージェントプロセスをホストするマネージドサービスです。呼び出し主体はその上で動くユーザーコード（本サンプルでは chat-agent ）であり、Runtime サービス自身がトークン変換を自動で行うわけではありません。 chat-agent が IdC の CreateTokenWithIAM API を呼びます。 const tokenRes = await new SSOOIDCClient({ region }).send( new CreateTokenWithIAMCommand({ clientId: idcApplicationArn, grantType: 'urn:ietf:params:oauth:grant-type:jwt-bearer', assertion: cognitoIdToken, }), ); const idcAccessToken = tokenRes.accessToken; jwt-bearer grant で Cognito ID Token を渡すと、IdC はそのクレームから IdC ユーザーを特定し、IdC Access Token を返します。 ここで 2 つの補足があります。 Cognito JWT と IdC Access Token は別物です。 発行者が違い（Cognito vs IdC）、形式も違います（JWT vs 不透明トークン）。Cognito JWT は Cognito 連携アプリでしか通用しませんが、IdC Access Token は IdC の Trusted Identity Propagation（TIP） に対応した AWS サービスで通用します。TIP は、IdC ユーザーのアイデンティティを IAM ロールの STS セッションに identity context として伝播させる仕組みです。DataZone は TIP 対応サービスの 1 つで、 RedeemAccessToken はその入口に位置します。 CreateTokenWithIAM と RedeemAccessToken は、このトークンの世界をまたぐブリッジの役割を果たします。 ただし本サンプルでは、TIP の identity-enhanced session をそのままデータ層まで持ち込んで Lake Formation や S3 Access Grants に評価させる構成は採っていません。SageMaker Catalog の Publish/Subscribe モデルが プロジェクトロール に行・列・オブジェクトレベルの権限を付与する設計になっているため、IdC ユーザーのアイデンティティは RedeemAccessToken → GetEnvironmentCredentials を経由してプロジェクトロールへ引き換えられ、以降のデータアクセスはプロジェクトロールの権限で評価されます。TIP の役割はこの引き換えの前段に限定され、本記事の焦点もそこにあります。 CreateTokenWithIAM には jti 制約があります。 JWT には jti （JWT ID）という一意識別子のクレームがあり、IdC は jwt-bearer grant で受け取った JWT の jti を記録します。同じ jti の JWT が再度送られると拒否されるため、同一の Cognito ID Token で CreateTokenWithIAM を 2 回呼ぶことはできません。このため、chat-agent で 1 リクエストあたり 1 回だけ実行し、得られた IdC Access Token を x-idc-access-token カスタムヘッダーで AgentCore Gateway 経由で全 Tool Lambda に伝播する設計になっています。 なお、 CreateTokenWithIAM を呼ぶための IAM アクション名は sso-oauth:CreateTokenWithIAM です。SDK クライアントは SSOOIDCClient を使いますが、IAM ポリシー側のサービス名は sso-oauth になります。また、IdC の OAuth Customer Managed Application に datazone:domain:access スコープを事前に登録しておく必要があります。 Step 3: Tool Lambda が IdC Access Token を DomainExecutionRole の認証情報に引き換える Tool Lambda が Amazon DataZone の RedeemAccessToken エンドポイントに HTTP POST を投げます。 const redeemRes = await fetch( `https://datazone.${region}.api.aws/sso/redeem-token`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ domainId, accessToken: idcAccessToken }), }, ); const { credentials: domainExecRoleCreds } = await redeemRes.json(); RedeemAccessToken は AWS SDK に含まれていません。 公開ドキュメントでは Athena JDBC ドライバ経由の利用例（ Analyze subscribed data via JDBC ）が示されていますが、サーバーサイドアプリからの直接呼び出しは生の HTTP リクエストで行う必要があります。このため、エンドポイント URL も通常の DataZone SDK が使う datazone.{region}.amazonaws.com ではなく datazone.{region}.api.aws を指定します。 この API には 2 つの特徴があります。SigV4 署名が不要であること（認証は IdC Access Token 自体が行う）、そして jti 制約がないため並列の Tool 呼び出しで複数の Lambda が同じ IdC Access Token を使っても問題ないことです。 返ってくるのは DomainExecutionRole の一時認証情報（ accessKeyId / secretAccessKey / sessionToken / expiration ）です。DomainExecutionRole は SageMaker Unified Studio ドメインに紐付く IAM ロールで、この認証情報には IdC ユーザーのアイデンティティが紐付いています 。内部的には、 RedeemAccessToken が DomainExecutionRole を assume する際に、Step 2 で触れた TIP の仕組みにより STS セッションに IdC ユーザーの identity context が埋め込まれます。これが次のステップで効いてきます。 Step 4: Tool Lambda が DomainExecutionRole の認証情報でプロジェクトロールを取得する Tool Lambda が Amazon DataZone SDK を DomainExecutionRole の認証情報で 初期化し、 GetEnvironmentCredentials を呼びます。 const envCreds = await new DataZoneClient({ region, credentials: domainExecRoleCreds, }).send( new GetEnvironmentCredentialsCommand({ domainIdentifier: domainId, environmentIdentifier: environmentId, }), ); Amazon DataZone は呼び出し元の STS セッションから identity context を取り出し、IdC ユーザーを特定します。そのユーザーがプロジェクトのメンバーであれば、プロジェクトロールの一時認証情報を返します。メンバーでなければ拒否されます。 ここが本サンプルの核心です。DomainExecutionRole の認証情報で呼ぶからこそ、 ユーザー本人のメンバーシップ で認可が評価されます。もし Lambda 実行ロールで直接 GetEnvironmentCredentials を呼んでいたら、Lambda 実行ロール自身のメンバーシップで判定されてしまい、ユーザーごとの権限差が消えます。 Step 5: プロジェクトロールで Athena / S3 を呼び出す Tool Lambda が Athena や S3 のクライアントを プロジェクトロールの認証情報で 初期化し、クエリやオブジェクト取得を実行します。 const athena = new AthenaClient({ region, credentials: { accessKeyId: envCreds.accessKeyId, secretAccessKey: envCreds.secretAccessKey, sessionToken: envCreds.sessionToken, }, }); Athena がクエリを実行すると、Lake Formation がプロジェクトロールの権限に基づいて行・列レベルのフィルタリングを透過的に適用します。ユーザーの権限の範囲内にある行と列だけが結果として返り、範囲外の情報は応答に含まれません。 ステップのまとめ Step 主体 API 入力 出力 1 ブラウザ → Runtime POST /invocations — Cognito Access Token + ID Token（Runtime に到達） 2 chat-agent → IdC CreateTokenWithIAM Cognito ID Token IdC Access Token 3 Tool Lambda → DataZone RedeemAccessToken IdC Access Token DomainExecutionRole 認証情報 4 Tool Lambda → DataZone GetEnvironmentCredentials DomainExecutionRole 認証情報 プロジェクトロール認証情報 5 Tool Lambda → Athena/S3 StartQueryExecution / GetObject 等 プロジェクトロール認証情報 クエリ結果 / オブジェクト Step 5 の S3 アクセスは Publisher と Subscriber で経路が分かれます。後述の「Publisher と Subscriber で異なる S3 アクセス方式」で詳述します。 この設計では、 Lambda 実行ロールはデータアクセスに一切使いません 。権限判定はすべてユーザーに紐付いた認証情報で行われます。 Policy in AgentCore によるツール単位認可 認証情報変換フローはデータアクセスの認可を扱いますが、「どのユーザーがどのツールを呼べるか」という別軸の認可も必要です。AgentCore の認可モデルは、 Inbound（ユーザー → AI エージェントへの入口） と Outbound（AI エージェント → ツールへの出口） の 2 軸で整理されます。Inbound は AgentCore Runtime の JWT Authorizer が Cognito Access Token を検証して「この呼び出し元はエージェントを呼んでよいか」を判定します。Outbound はツール単位の認可で、本サンプルでは AgentCore Gateway の Policy in AgentCore で実現しています（下図）。 グループ情報の埋め込み 壁 2 で述べた通り、IdC と Cognito を SAML フェデレーションで連携する構成では、IdC 側のグループ情報は Cognito トークンに自動的には含まれません。本サンプルでは、 Amazon Cognito の Pre token generation Lambda trigger の V2 イベントを使い、Cognito Access Token に cedar_groups カスタムクレームを埋め込みます。値は |data-producers|security-auditors| のようにパイプ区切りの文字列です。 Cedar ポリシーによる評価 Policy in AgentCore では、Cedar 言語で記述されたポリシーを policy engine に登録し、それを AgentCore Gateway に関連付けます。Gateway にリクエストが到達すると、policy engine が Cognito Access Token の cedar_groups クレームを読み取り、Cedar ポリシーで評価します。Gateway はリクエスト時点で JWT のクレームを AgentCore::OAuthUser エンティティのタグとして Entity Store に格納するため、ポリシー上は principal.getTag("cedar_groups") のようにタグとして参照します。「JWT では cedar_groups クレーム、Cedar ポリシーでは cedar_groups タグ」という名前の対応関係です。 permit( principal is AgentCore::OAuthUser, action, resource == AgentCore::Gateway::"<gateway-arn>" ) when { principal.hasTag("cedar_groups") && principal.getTag("cedar_groups") like "*|security-auditors|*" && (action == AgentCore::Action::"cloudtrail-query___cloudtrail_query") }; このポリシーは「 security-auditors グループに属するユーザーだけが cloudtrail_query ツールを呼べる」ことを宣言しています。アクション名の cloudtrail-query___cloudtrail_query は、AgentCore Gateway が MCP ツール定義から自動生成する命名で、 ターゲット名（ cloudtrail-query ） ___ ツール名（ cloudtrail_query ） の形を取ります。 Cedar の policy engine は default-deny （明示的に許可されない限り拒否）で動作します。上記の 1 本のポリシーだけでは security-auditors 向けの 1 ツールしか許可されていないため、他のユーザー・他のツールはすべて拒否されます。同様のポリシーを複数定義することで、たとえばデータコンシューマーには athena_query と s3_read のみを許可し、データプロデューサーにはカタログ管理ツールも許可する、といった職務分離を実現できます。 Policy in AgentCore によるツール単位認可と、認証情報変換フローによるデータアクセス認可は独立した 2 つの軸です。Gateway で「このユーザーはこのツールを呼んでよいか」を判定し、Tool Lambda で「このユーザーはこのデータを見てよいか」を判定します。 Publisher と Subscriber で異なる S3 アクセス方式 非構造化データ（S3 上のファイル）へのアクセスは、プロジェクトの役割によって経路が異なります。Amazon S3 Access Grants は、S3 の prefix / bucket / object 単位で IAM プリンシパルやディレクトリユーザーに READ/WRITE/READWRITE 権限を付与する仕組みで、 GetDataAccess API で当該対象への一時認証情報を取得してから S3 API を呼ぶ形で利用します。SageMaker Catalog の Publish/Subscribe は、この Grant を Subscriber のプロジェクトロールに対してのみ自動作成する 設計です。Publisher 側のプロジェクトロールには明示的な Grant が作られず、Publisher は別のパス（IAM ポリシーによる直接アクセス）でバケットを読みます。これが Publisher/Subscriber で経路が分かれる理由です。 Publisher（プロジェクトがバケットを所有する場合）: プロジェクトロールの認証情報で直接 S3:GetObject を呼びます。アクセス権は、プロジェクトロールに付与された IAM インラインポリシー（プロジェクト配下のプレフィックスに限定）によって許可されます。Publisher プロジェクトロール自身への明示的な S3 Access Grants の Grant は存在しないため、 GetDataAccess は失敗します。 Subscriber（別プロジェクト経由で購読する場合）: SageMaker Unified Studio の Publish/Subscribe が Subscriber ロールに対して IAM タイプの Grant を自動作成します。Tool Lambda はまず S3Control:GetDataAccess で一時認証情報を取得し、その認証情報で S3:GetObject を呼びます。 判断ロジックは、コネクションの accessRole の有無とプロジェクトの所有関係で決まります。プロジェクトレベルのコネクションで accessRole があれば S3 Access Grants 経由、なければ直接アクセスです。 まとめと次のアクション 本記事では、AI エージェント経由のデータアクセスに SageMaker Catalog のアクセス制御を透過的に効かせる実装パターンを解説しました。 設定時と実行時の役割分担 を整理し、SageMaker Catalog / DataZone が設定を担い、Lake Formation / S3 Access Grants が実行時に認可を評価する構造を明確にする 認証情報変換フロー （ CreateTokenWithIAM → RedeemAccessToken → GetEnvironmentCredentials ）で、ユーザーに紐付いたプロジェクトロールの認証情報を Tool Lambda に届ける Policy in AgentCore で、ツール単位の認可をデータアクセス認可とは独立した軸で制御する Lambda 実行ロールはデータアクセスに一切使わず、権限判定はすべてユーザーに紐付いた認証情報で行われます。これにより、SageMaker Catalog で設定された行・列・オブジェクトレベルのアクセス権は、AI エージェント経由でも透過的に適用されます。 第 1 回 で紹介した拡張性（ゼロショット時系列予測のオンデマンド実行）も、本記事で解説した認可の仕組みに支えられています。アクセス制御されたデータを、追加の認証設計なしで Amazon SageMaker AI の推論エンドポイントに流し込めるのは、プロジェクトロールの認証情報が Tool Lambda の手元まで届いているからです。サンプルリポジトリの apps/gateway-tools/time-series-forecast/ と design/data-access-control.md で実装の詳細を確認できます。 高野 賢司 AWS のシニアソリューションアーキテクトとして、東海以西の製造業のお客様を中心に支援しています。Infrastructure as Code や AI 駆動開発を中心とした開発者ツールのエキスパートでもあり、 Kiro によって広がったソフトウェア開発の世界を楽しんでいます。