Amazon Quick × Bedrock AgentでOracle DBのデータを活用する

Amazon Quickには「オートメーション」（Automate）と呼ばれるGUIでAIを活用したワークフローを作成、実行できる機能があります。オートメーションからはAmazon Bedrock Agentを呼び出すこともできます。

また、2025年12月に東京リージョンでも提供が開始されたOracle Database@AWSを利用すると、AWSデータセンター内でOracle Cloud Infrastructure(OCI)が管理するOracle Exadataにアクセスできます。Oracle Database@AWSではOracle Autonomous Database(ADB)を実行することができるため、Oracle Autonomous AI Database MCP Server(ADB-S MCP Server)も利用できます。

これらを組み合わせることで、AWSのサービスからシームレスにOracle DBのデータを取得して活用することができます。
例えば以下のように、Amazon Quickを起点としてADBのデータをAWS上で活用できます。
①Amazon QuickのオートメーションからAmazon Bedrock Agentを呼び出す
②Bedrock Agentが指示内容(入力プロンプト)をもとに、ADB-S MCP Serverを利用して必要なデータをADBから取得
③Bedrock AgentがADBから取得したデータを活用して、指示内容(入力プロンプト)を処理
④処理結果(出力レポート)をAmazon S3にアップロード
⑤ユーザーはレポートを確認して次のアクションに繋げる

本当は上述の構成を実装したかったのですがOracle Database@AWSは個人で気軽に起動できるサービスではないため、今回はOracle Database@AWSの代わりにOCI上にADBを構築する形で実現しています。そのため、通信はインターネットを経由する形になっている点はご容赦ください。

まずはOCI上でADBの構築、MCP Serverの設定をします。実装方法については以下の日本オラクルさんのブログに詳しく記載されていますので本ブログでは割愛します。

Oracle Autonomous AI Database × MCP がもたらすAIエージェント データ基盤の可能性
https://qiita.com/jun110/items/a9d90a7ade9f33f222b4​

ADB上に「sales_transactions」テーブルというサンプルテーブルを作成しました。中身は架空の店舗の売上データです。

Bedrock Agentが利用するツール(アクショングループ)をLambdaを用いて作成しておきます。
今回は以下のようなPythonコードで「oracle-mcp-bedrock-agent」という名前のLambda関数を作成しました。
※実際にはこのコードが動作するためのライブラリ等も一緒にLambdaにアップロードしています。

import json
import os
import asyncio
import traceback
import httpx
from mcp.client.streamable_http import streamable_http_client
from mcp import ClientSession
MCP_SERVER_URL = os.environ['MCP_SERVER_URL']
AUTHORIZATION_TOKEN = os.environ['AUTHORIZATION_TOKEN']
REFRESH_TOKEN = os.environ['REFRESH_TOKEN']
# グローバルクライアントを再利用
_http_client = None
async def get_or_create_session():
    global _http_client
    if _http_client is None:
        headers = {
            "Authorization": AUTHORIZATION_TOKEN,
            "refresh_token": REFRESH_TOKEN
        }
        _http_client = httpx.AsyncClient(headers=headers, timeout=30.0)
    return _http_client
async def query_oracle_mcp(query: str):
    try:
        http_client = await get_or_create_session()
        streams = streamable_http_client(MCP_SERVER_URL, http_client=http_client)
        async with streams as stream_tuple:
            read, write = stream_tuple[0], stream_tuple[1]
            async with ClientSession(read, write) as session:
                await session.initialize()
                # SQLクエリを実行
                result = await session.call_tool(
                    "EXECUTE_SQL",
                    arguments={
                        "SCHEMA_NAME": "ADMIN",
                        "QUERY": query,
                        "OFFSET": 0,
                        "LIMIT": 100
                    }
                )
                if result.content:
                    return {"result": result.content[0].text}
                else:
                    return {"result": str(result)}
    except Exception as e:
        print(traceback.format_exc())
        return {"error": str(e)}
def lambda_handler(event, context):
    print(f"Received event: {json.dumps(event)}")
    action_group = event.get('actionGroup', '')
    function = event.get('function', '')
    parameters = event.get('parameters', [])
    query = None
    for param in parameters:
        if param.get('name') == 'query':
            query = param.get('value')
            break
    if not query:
        response_body = {"error": "No query provided"}
    else:
        try:
            result = asyncio.run(query_oracle_mcp(query))
            response_body = result
        except Exception as e:
            print(traceback.format_exc())
            response_body = {"error": str(e)}
    print(f"Response body: {json.dumps(response_body)}")
    return {
        'messageVersion': '1.0',
        'response': {
            'actionGroup': action_group,
            'function': function,
            'functionResponse': {
                'responseBody': {
                    'TEXT': {
                        'body': json.dumps(response_body)
                    }
                }
            }
        }
    }

続いてBedrock Agentを作成します。

エージェント全体の設定は以下の通りです。

Agentが利用するモデルは「Bedrock Agents向けに最適化されているモデル」の中からClaude 3.5 Haikuを選択しました。また、「エージェント向けの指示」は以下のようにしました。

「アクショングループ」には「oracle_query_action」という名前でグループを追加しています。

アクショングループの定義内容は以下の通りです。

ひとまずBedrock Agentから動作確認をしてみます。Bedrock Agentの画面右側でテストできるので「SALES_TRANSACTIONSテーブルから東京本店の2026年1月の売上合計と、その内訳を教えて」と指示をしてみます。

数秒すると実行結果が返ってきます。しっかりとADBからデータを取得してAIによる分析までできています。「トレースを表示」をクリックしてエージェントがどのように動作したのか詳細を確認していきます。

「ステップ1」を確認すると、最初に入力した指示を受け取っていることが確認できます。

「ステップ1」をスクロールすると以下のように、どのようなSQLを作れば良いかを考えています。

ユーザーの質問から、以下の情報を抽出しました：
- テーブル: SALES_TRANSACTIONS
- 条件:
1. 東京本店の売上
2. 2026年1月の期間
- 必要な情報: 売上合計と内訳
SQLクエリを作成して、この情報を取得します。

考えた内容から以下のSQLを生成しています。

しかし実行結果はエラーとなっています。どうやら「PRODUCT_CATEGORY」という列名が間違っているようです。

「ステップ2」に移ります。エラーを解消するためにテーブルの列名を確認するSQLを実行しています。

「ステップ3」を確認すると、列名が間違っていたことに気づいたようです。エラーが返されたPRODUCT_CATEGORY列だけでなく、TRANSACTION_DATEも間違っていることにエージェント自身で気づけていますね。下の方で修正版の新しいSQLを実行しているのがわかります。

「ステップ4」を確認すると、SQLが無事に実行され、その内容を分析した結果を生成していることが確認できます。

このようにトレースを確認すると、エージェントが目的の動作を完了するために試行錯誤している様子を見ることができます。
以上で動作確認は完了です。無事にエージェントが動作していることが確認できました。

Amazon Quickのオートメーションを作成していきます。今回は「Oracle MCP Search」という名前で作成しました。

オートメーションの詳細を解説していきます。まずStep1では作成したBedrock Agentを呼び出しています。「Input text」にエージェントへの指示を設定します。今回は「SALES_TRANSACTIONSテーブルから東京本店の2026年1月の売上合計と、その内訳を教えて」と設定しています。

Step2ではBedrock Agentで得た回答をExcelに出力して、S3に保存しています。
※保存用のS3バケットは事前に作成しておく必要があります。

主要なアクションの設定内容は以下の通りです。Action 1.1でBedrock Agentからの回答を「bedrock_output」変数に格納しており、その結果をAction2.2で受け取っています。

作成した内容の動作確認をしていきます。動作確認は画面右上の[Debug]から行えます。

実行結果を確認するとステータスが「Completed」になっていることが確認できます。ログの内容からBedrock Agentの正常に実行できていることが確認できます。

該当のS3バケットを確認するとファイルがアップロードされていることが確認できます。

Excelファイルをダウンロードして開いてみると、Bedrock Agentによる分析結果が出力されていることが確認できます。

動作に問題がなければ画面右上の[Commit]で編集内容を確定できます。

作成したオートメーションはトリガー機能を使って定期実行させることができます。トリガーを設定する場合はオートメーションの[Deployments]-[Create trigger]をクリックします。

任意の設定をして[Create]をクリックします。今回は週次実行の設定をしています。

トリガーを設定できました。

Amazon Quickのオートメーションを利用すればAWSのサービスやExcelなどを組み合わせたワークフローをGUIで作成することができました。また、Bedrock AgentやMCPを活用することでOracle Cloud上のデータベースとも連携することができました。このような構成にOracle Database@AWSを組み合わせることで、インターネットに出ることなくセキュアに実現できることが期待できます。

執筆者・シリーズ情報

• 佐伯 竜輔

2008年新卒入社。eラーニング製品、MySQL、Verticaなどのエンジニアを経て、 2019年から2年間限定で採用活動を担当。2021年に技術職に戻り、AWSに関する支援や研修の提案に従事している。...show more