📖
Bedrock Agent でAWSドキュメントを検索する

AgaHiro
2025/07/15に公開
AWS
tech
 はじめに阿河です。

AWSの公式ドキュメントは膨大で、必要な情報を素早く見つけるのは意外と大変ですよね。特に新しいサービスや機能についての最新情報を探すとき、「このドキュメントのどこに書いてあるんだろう？」と迷った経験は誰にでもあると思います。
そこで今回は、Amazon Bedrock Agentを活用して、自然言語でAWSドキュメントを検索できるAIエージェントを構築してみました。

 目次前提
Bedrock Agentの構築
検証

 1. 前提事前にAmazon Bedrockでモデルアクセスを有効化しておくこと
本記事では、Claude 3.7 Sonnetを利用します。


 2. Bedrock Agentの構築
 Bedrock Agentの設定名前: aws-docs-search
モデル: Claude 3.7 Sonnet
エージェント指示文例
あなたはAWSドキュメント専門のAIアシスタントです。

AWSサービスについて質問されたら、以下の手順で回答してください：
1. まず search_aws_docs 機能を使って関連ドキュメントを検索
2. 取得した情報を基に、わかりやすく日本語での回答を生成

回答時は以下を心がけてください：
- 最新の公式情報を重視する
- 具体的な手順や設定方法を含める
- 参考ドキュメントURLを記載する

重要: 同じ検索を繰り返さないでください。1回の検索で十分な情報を提供してください。

 Action Groupの設定Action Group type: Define with function details
Quick create a new Lambda function
Action group function 1
Name: aws_docs_search
Description: AWS公式ドキュメントの検索と読み込み
Parameters
action
Description: 実行するアクション (search/read)
Type: string
Required: True

query
Description: 検索キーワード (actionがsearchの場合)
Type: string
Required: False

url
Description: 読み込むAWSドキュメントのURL (actionがreadの場合)
Type: string
Required: False



結果的にLambda関数が作成されます。

 Lambdaの実装続いてLambda関数を作成します。

この関数が、AWS公式ドキュメントの検索と取得を担当します。
import logging
import json
import urllib.request
import urllib.parse
import re
from typing import Dict, Any, List, Optional
from html import unescape

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    try:
        logger.info(f"Received event: {json.dumps(event)}")
        
        action_group = event['actionGroup']
        function = event['function']
        message_version = event.get('messageVersion', 1)
        parameters = event.get('parameters', [])

        param_dict = {}
        for param in parameters:
            param_dict[param.get('name')] = param.get('value')

        logger.info(f"Parsed parameters: {param_dict}")

        if function == 'aws_docs_search':
            result = execute_aws_docs_search(param_dict)
        else:
            result = f'Unknown function: {function}'

        response_body = {'TEXT': {'body': result}}
        action_response = {
            'actionGroup': action_group,
            'function': function,
            'functionResponse': {'responseBody': response_body}
        }

        return {'response': action_response, 'messageVersion': message_version}

    except Exception as e:
        logger.error(f'Error in lambda_handler: {str(e)}')
        return create_error_response(
            event.get('actionGroup', ''), 
            event.get('function', ''), 
            event.get('messageVersion', 1), 
            str(e)
        )

def execute_aws_docs_search(params: Dict[str, Any]) -> str:
    try:
        action = params.get('action', '').lower()
        logger.info(f"Executing MCP action: {action}")
        
        if action == 'search':
            query = params.get('query', '')
            logger.info(f"MCP search for: {query}")
            return mcp_search_aws_docs(query)
        elif action == 'read':
            url = params.get('url', '')
            logger.info(f"MCP read document: {url}")
            return mcp_read_aws_doc(url)
        else:
            return f"サポートされていないアクション: {action}. 'search' または 'read' を指定してください。"
            
    except Exception as e:
        logger.error(f'Error in execute_aws_docs_search: {str(e)}')
        return f"MCP処理中にエラーが発生しました: {str(e)}"

def mcp_search_aws_docs(query: str) -> str:

    if not query.strip():
        return "検索クエリが指定されていません。"
    
    try:
        logger.info(f"Starting MCP search for: {query}")
        
        search_results = call_aws_search_api(query)
        
        if not search_results:
 
            return create_fallback_search_result(query)
        
        result = f"【{query}の検索結果】\n\n"
        
        for i, doc in enumerate(search_results, 1):
            result += f"{i}. {doc['title']}\n"
            result += f"概要: {doc['snippet']}\n"
            result += f"URL: {doc['url']}\n"
            result += f"更新日: {doc.get('lastModified', '不明')}\n\n"
        
        result += f"検索完了: {len(search_results)}件のドキュメントが見つかりました。\n"
        result += "詳細が必要な場合は、action='read' でURLを指定してください。"
        
        logger.info(f"MCP search completed. Found {len(search_results)} results.")
        return result
        
    except Exception as e:
        logger.error(f'Error in mcp_search_aws_docs: {str(e)}')
        # エラー時のフォールバック
        return create_fallback_search_result(query)

def call_aws_search_api(query: str) -> List[Dict[str, Any]]:

    try:
        base_url = "https://docs.aws.amazon.com/search/doc-search.html"
        
        params = {
            'searchPath': 'documentation',
            'searchQuery': query,
            'facet_doc_product': 'documentation',
            'this_doc_product': 'documentation',
            'doc-locale': 'en_us'
        }
        
        query_string = urllib.parse.urlencode(params)
        search_url = f"{base_url}?{query_string}"
        
        logger.info(f"Calling AWS search API: {search_url}")
        
        req = urllib.request.Request(
            search_url,
            headers={
                'User-Agent': 'Mozilla/5.0 (compatible; AWS-Docs-Agent/1.0)',
                'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
                'Accept-Language': 'en-US,en;q=0.5',
                'Accept-Encoding': 'gzip, deflate',
                'Connection': 'keep-alive',
                'Upgrade-Insecure-Requests': '1'
            }
        )
        
        with urllib.request.urlopen(req, timeout=15) as response:
            if response.status == 200:
                html_content = response.read()
                
                # gzip解凍が必要な場合
                if response.info().get('Content-Encoding') == 'gzip':
                    import gzip
                    html_content = gzip.decompress(html_content)
                
                content_str = html_content.decode('utf-8')
                logger.info("AWS search API response received")
                
                return parse_aws_search_html(content_str)
            else:
                logger.warning(f"AWS search API returned status: {response.status}")
                return []
                
    except Exception as e:
        logger.error(f"Error calling AWS search API: {str(e)}")
        return []

def parse_aws_search_html(html_content: str) -> List[Dict[str, Any]]:

    try:
        results = []
        
        result_pattern = r'<div[^>]*class="[^"]*search-result[^"]*"[^>]*>(.*?)</div>'
        title_pattern = r'<h[1-6][^>]*>\s*<a[^>]*href="([^"]*)"[^>]*>(.*?)</a>\s*</h[1-6]>'
        snippet_pattern = r'<p[^>]*class="[^"]*search-result-summary[^"]*"[^>]*>(.*?)</p>'
        
        result_blocks = re.findall(result_pattern, html_content, re.DOTALL | re.IGNORECASE)
        
        for block in result_blocks[:5]:  
            
            title_match = re.search(title_pattern, block, re.DOTALL | re.IGNORECASE)
            if title_match:
                url = title_match.group(1)
                title = clean_html_text(title_match.group(2))
                
                if url.startswith('/'):
                    url = f"https://docs.aws.amazon.com{url}"
                
                snippet_match = re.search(snippet_pattern, block, re.DOTALL | re.IGNORECASE)
                snippet = clean_html_text(snippet_match.group(1)) if snippet_match else ""
                
                results.append({
                    'title': title,
                    'url': url,
                    'snippet': snippet[:200] + "..." if len(snippet) > 200 else snippet,
                    'lastModified': extract_last_modified(block)
                })
        
        logger.info(f"Parsed {len(results)} search results from HTML")
        return results
        
    except Exception as e:
        logger.error(f"Error parsing AWS search HTML: {str(e)}")
        return []

def clean_html_text(html_text: str) -> str:

    text = re.sub(r'<[^>]+>', '', html_text)
    text = unescape(text)

    text = re.sub(r'\s+', ' ', text).strip()
    return text

def extract_last_modified(html_block: str) -> str:

    try:
        date_pattern = r'(\d{4}-\d{2}-\d{2}|\d{2}/\d{2}/\d{4})'
        date_match = re.search(date_pattern, html_block)
        return date_match.group(1) if date_match else "不明"
    except:
        return "不明"

def create_fallback_search_result(query: str) -> str:

    encoded_query = urllib.parse.quote(query)
    search_url = f"https://docs.aws.amazon.com/search/doc-search.html?searchPath=documentation&searchQuery={encoded_query}"
    
    
    common_services = {
        's3': ('Amazon S3', 'https://docs.aws.amazon.com/s3/'),
        'lambda': ('AWS Lambda', 'https://docs.aws.amazon.com/lambda/'),
        'ec2': ('Amazon EC2', 'https://docs.aws.amazon.com/ec2/'),
        'rds': ('Amazon RDS', 'https://docs.aws.amazon.com/rds/'),
        'iam': ('AWS IAM', 'https://docs.aws.amazon.com/iam/'),
        'bedrock': ('Amazon Bedrock', 'https://docs.aws.amazon.com/bedrock/'),
        'dynamodb': ('Amazon DynamoDB', 'https://docs.aws.amazon.com/dynamodb/'),
        'cloudformation': ('AWS CloudFormation', 'https://docs.aws.amazon.com/cloudformation/')
    }
    
    query_lower = query.lower()
    matched_services = []
    
    for key, (service_name, base_url) in common_services.items():
        if key in query_lower or service_name.lower() in query_lower:
            matched_services.append({
                'title': f'{service_name} ユーザーガイド',
                'url': f'{base_url}latest/userguide/',
                'snippet': f'{service_name}の包括的なユーザーガイドです。'
            })
    
    if matched_services:
        result = f"【{query}の検索結果】(フォールバック結果)\n\n"
        for i, doc in enumerate(matched_services[:3], 1):
            result += f"{i}. {doc['title']}\n"
            result += f"概要: {doc['snippet']}\n"
            result += f"URL: {doc['url']}\n\n"
        
        result += f"より詳細な検索は: {search_url}"
        return result
    else:
        return f"""【{query}の検索結果】

AWS公式ドキュメントで'{query}'を検索しています。

直接検索: {search_url}

一般的なAWSドキュメント:
1. AWS全般ドキュメント
   URL: https://docs.aws.amazon.com/
   
2. AWSサービス一覧
   URL: https://aws.amazon.com/products/

より具体的なAWSサービス名での検索をお試しください。"""

def mcp_read_aws_doc(url: str) -> str:

    if not url.strip():
        return "ドキュメントURLが指定されていません。"
    
    try:
        if not (url.startswith('https://docs.aws.amazon.com') or 
                url.startswith('https://aws.amazon.com')):
            return "セキュリティ上の理由により、AWS公式ドメインのURLのみサポートしています。"
        
        logger.info(f"Reading AWS document: {url}")
        
        req = urllib.request.Request(
            url,
            headers={
                'User-Agent': 'Mozilla/5.0 (compatible; AWS-Docs-Agent/1.0)',
                'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
                'Accept-Language': 'en-US,en;q=0.5'
            }
        )
        
        with urllib.request.urlopen(req, timeout=20) as response:
            if response.status == 200:
                html_content = response.read()
                
                if response.info().get('Content-Encoding') == 'gzip':
                    import gzip
                    html_content = gzip.decompress(html_content)
                
                content_str = html_content.decode('utf-8')
                logger.info("AWS document content retrieved")
                
                return extract_document_content(content_str, url)
            else:
                return f"ドキュメント取得エラー: HTTP {response.status}"
                
    except Exception as e:
        logger.error(f'Error in mcp_read_aws_doc: {str(e)}')
        return f"ドキュメント読み込みエラー: {str(e)}"

def extract_document_content(html_content: str, url: str) -> str:

    try:
        content_patterns = [
            r'<div[^>]*class="[^"]*awsui-util-container[^"]*"[^>]*>(.*?)</div>',
            r'<main[^>]*>(.*?)</main>',
            r'<article[^>]*>(.*?)</article>',
            r'<div[^>]*id="[^"]*main[^"]*"[^>]*>(.*?)</div>'
        ]
        
        main_content = ""
        
        for pattern in content_patterns:
            matches = re.findall(pattern, html_content, re.DOTALL | re.IGNORECASE)
            if matches:
                main_content = matches[0]
                break
        
        if not main_content:
            title_match = re.search(r'<title[^>]*>(.*?)</title>', html_content, re.IGNORECASE)
            desc_match = re.search(r'<meta[^>]*name="description"[^>]*content="([^"]*)"', html_content, re.IGNORECASE)
            
            title = clean_html_text(title_match.group(1)) if title_match else "AWS ドキュメント"
            description = clean_html_text(desc_match.group(1)) if desc_match else "詳細情報については直接URLを参照してください。"
            
            return f"""【{title}】

URL: {url}

概要:
{description}

このドキュメントの詳細内容を表示するには、直接URLにアクセスしてください。"""
        
        text_content = clean_html_text(main_content)
        
        lines = text_content.split('\n')
        cleaned_lines = [line.strip() for line in lines if line.strip()]
        
        content_lines = []
        char_count = 0
        
        for line in cleaned_lines[:50]:
            if char_count + len(line) > 3000:
                break
            content_lines.append(line)
            char_count += len(line)
        
        final_content = '\n'.join(content_lines)
        
        service_name = extract_service_name_from_url(url)
        
        result = f"【{service_name} ドキュメント詳細】\n\n"
        result += f"URL: {url}\n\n"
        result += f"内容:\n{final_content}\n\n"
        
        if char_count >= 3000:
            result += "※ 内容が長いため一部のみ表示しています。完全な内容は上記URLを参照してください。"
        
        logger.info(f"Extracted {len(final_content)} characters from document")
        return result
        
    except Exception as e:
        logger.error(f"Error extracting document content: {str(e)}")
        return f"ドキュメント内容の抽出に失敗しました: {str(e)}"

def extract_service_name_from_url(url: str) -> str:

    try:
        parts = url.split('/')
        if 'docs.aws.amazon.com' in url and len(parts) >= 4:
            service_key = parts[3]
            
            service_mapping = {
                's3': 'Amazon S3',
                'lambda': 'AWS Lambda', 
                'ec2': 'Amazon EC2',
                'rds': 'Amazon RDS',
                'iam': 'AWS IAM',
                'bedrock': 'Amazon Bedrock',
                'dynamodb': 'Amazon DynamoDB',
                'cloudformation': 'AWS CloudFormation',
                'vpc': 'Amazon VPC',
                'cloudwatch': 'Amazon CloudWatch',
                'apigateway': 'Amazon API Gateway'
            }
            
            return service_mapping.get(service_key, service_key.upper() + ' Service')
        
        return 'AWS Service'
    except:
        return 'AWS Service'

def create_error_response(action_group: str, function: str, message_version: int, error_message: str) -> Dict[str, Any]:

    return {
        'response': {
            'actionGroup': action_group,
            'function': function,
            'functionResponse': {
                'responseBody': {
                    'TEXT': {
                        'body': f'MCP Serverエラー: {error_message}'
                    }
                }
            }
        },
        'messageVersion': message_version
    }

 3. 検証今回はシンプルに、Bedrockコンソールのテスト画面から動作テストを行います。
ドキュメントURL付きで回答が帰ってきました。

精度面でまだまだ課題はありますが、必要な情報が要約されて返ってきています。

 おわりにAIとクラウドサービスの組み合わせにより、従来は困難だった情報検索が手軽に実現できる時代になりました。

今回の実装が、皆さんの業務効率化やシステム開発の参考になれば幸いです。
MEGAZONE株式会社 Tech BlogPublication
MEGAZONEはAWS プレミアティアサービスパートナーとして多数のCompetencyを取得しており、大企業、ゲーム会社、スタートアップ、公共機関などさまざまな分野の5,000以上のお客様にAWSソリューションとサービスを提供しています。
はじめに

目次

1. 前提

2. Bedrock Agentの構築

Bedrock Agentの設定

Action Groupの設定

Lambdaの実装

3. 検証

おわりに

Discussion
