【Strands Agents】MCP サーバーと AI エージェントで AWS サービス情報を効率収集・比較

はじめに

皆さんこんにちは！パーソル＆サーバーワークスの小泉です。
この記事では、Strands AgentsとModel Context Protocol (MCP) サーバーを活用して、AWSサービスの情報を自動的に収集・要約・比較するAI Agentの実装方法を紹介します。
実際に作成したadvanced_workflow.py を基に説明します。

Strands Agents とは

Strands Agentsは、数行のコードでAI エージェントを構築できるオープンソースSDKです。

MCP (Model Context Protocol) サーバーとは

MCPは、アプリケーションが大規模言語モデル（LLM）にコンテキストを提供する方法を標準化するプロトコルです。 Strands AgentsはMCPと統合することで、外部ツールやサービスを通じてエージェントの機能を拡張します。
MCPサーバーは、エージェントに追加のツールやリソースを提供し、AWSドキュメントなどの外部情報源へのアクセスを可能にします。
これにより、エージェントは最新のAWSサービス情報を取得し、分析することができます。

AWS サービス情報収集の課題

AWSのサービスは多岐にわたり、頻繁に更新されるため、以下のような課題があります：

• 膨大な量のドキュメントから必要な情報を見つけ出すのに時間がかかる
• 複数のサービスを比較検討する際に情報の整理が難しい
• 技術的な詳細を簡潔に要約して理解するのが困難

これらの課題に対して、AI エージェントを活用することで、情報収集と要約のプロセスを自動化し、効率化することができます。

ワークフローの概要

今回作成したadvanced_workflow.py は、AWSサービスに関する情報を収集、要約、比較、推奨するための高度なエージェントワークフローを実装しています。
このワークフローは以下の主要コンポーネントで構成されています：

1. BaseAgent: すべてのエージェントの基本クラス
2. 専門エージェント: 特定のタスクを実行する専門エージェント※1
3. AdvancedWorkflowManager: 複数のワークフロータイプを管理するマネージャー

※1

• ResearchAgent: AWSドキュメントMCPサーバーを使用して情報を収集
• SummaryAgent: 収集された情報を要約
• ComparisonAgent: 複数のサービスを比較
• RecommendationAgent: ユースケースに基づいて最適なサービスを推奨

ワークフローの種類

advanced_workflow.py では、以下の3種類のワークフローを実行できます：

1. 単一サービスの調査と要約: 特定のAWSサービスに関する情報を収集し、要約する
2. 複数サービスの比較: 複数のAWSサービスを比較分析する
3. ユースケースに基づく推奨: 特定のユースケースと要件に基づいて最適なAWSサービスを推奨する

ワークフロー図

コードの構造

作成した、advanced_workflow.py について内容を一部抜粋して解説します。

BaseAgent クラス

class BaseAgent:
    """すべてのエージェントの基本クラス"""
    def __init__(self, name: str, model_id: str = "us.anthropic.claude-3-7-sonnet-20250219-v1:0"):
        self.name = name
        self.model_id = model_id
        # セッションマネージャーを作成
        self.session_manager = FileSessionManager(
            session_id=f"{SESSION_ID_PREFIX}_{name.lower().replace(' ', '_')}",
            storage_dir="./sessions"
        )
        print(f"{name} Agent initialized")
    def process(self, input_data: Any) -> Any:
        """入力データを処理して結果を返す（サブクラスでオーバーライド）"""
        raise NotImplementedError("Subclasses must implement process()")

BaseAgent クラスは、すべてのエージェントの基本クラスとして機能し、共通の機能を提供します。

専門エージェント

各専門エージェントは BaseAgent を継承し、特定のタスクを実行します。

class ResearchAgent(BaseAgent):
    """AWS ドキュメントを検索して情報を収集するエージェント"""
    def __init__(self, name: str = "Research"):
        super().__init__(name)
        # AWS Documentation MCP serverに接続
        self.aws_docs_mcp_client = MCPClient(lambda: stdio_client(
            StdioServerParameters(
                command="uvx",
                args=["awslabs.aws-documentation-mcp-server@latest"]
            )
        ))
        # MCP clientのコンテキストを開始
        self.aws_docs_mcp_client.__enter__()
        # MCP serverからツールを取得
        self.tools = self.aws_docs_mcp_client.list_tools_sync()
        # エージェントを作成
        self.agent = Agent(
            self.model_id, 
            tools=self.tools,
            session_manager=self.session_manager
        )

class ComparisonAgent(BaseAgent):
    """複数のAWSサービスを比較するエージェント"""
    def __init__(self, name: str = "Comparison"):
        super().__init__(name)
        # エージェントを作成（ツールなし）
        self.agent = Agent(
            self.model_id,
            session_manager=self.session_manager
        )
    def process(self, data: Dict[str, Any]) -> str:
        """サービスを比較"""
        services = data.get("services", [])
        service_info = data.get("service_info", {})
        comparison_type = data.get("comparison_type", "general")
        # 各サービスの情報を結合して比較プロンプトを作成
        # ...

AdvancedWorkflowManager クラス

class AdvancedWorkflowManager:
    """高度なエージェントワークフローを管理"""
    def __init__(self):
        # エージェントを作成
        self.research_agent = ResearchAgent()
        self.summary_agent = SummaryAgent()
        self.comparison_agent = ComparisonAgent()
        self.recommendation_agent = RecommendationAgent()
        print("Advanced Workflow Manager initialized")
    def run_comparison_workflow(self, services: List[str], comparison_type: str = "general") -> Dict[str, Any]:
        """複数サービスの比較ワークフローを実行"""
        print(f"\n🚀 Starting comparison workflow for services: {', '.join(services)}")
        try:
            # 各サービスの情報を収集
            service_info = {}
            for service in services:
                service_info[service] = self.research_agent.process(service)
            # サービスを比較
            comparison = self.comparison_agent.process({
                "services": services,
                "service_info": service_info,
                "comparison_type": comparison_type
            })
            # 結果を返す
            return {
                "services": services,
                "comparison_type": comparison_type,
                "service_info": service_info,
                "comparison": comparison,
                "success": True
            }
        except Exception as e:
            print(f"Error in comparison workflow: {e}")
            return {
                "services": services,
                "error": str(e),
                "comparison": "エラーが発生したため比較を生成できませんでした。",
                "service_info": {},
                "comparison_type": comparison_type,
                "success": False
            }

セッション管理

各エージェントは独自のセッション管理を持ち、会話履歴をファイルシステムに保存します。これにより、エージェントの状態が維持され、一貫した対話が可能になります。セッションデータは ./sessions ディレクトリに保存され、同じセッション ID プレフィックスを使用すれば後で会話を再開することもできます。

実際の使用例

複数サービスの比較例

今回は、EC2とECSの比較を行いました。

入力:

• サービス: "EC2, ECS"
• 比較タイプ: "コスト比較"

処理:

1. ResearchAgent が各サービス（EC2,ECS）に関する情報を収集
2. ComparisonAgent が収集された情報を基に比較分析を実行

実行画面:

まとめ

今回はローカル環境でのテストになりましたが、最近出たAWS Bedrock AgentCoreにデプロイして、Slackと連携して社内全員がAI Agentを使える仕組みを作っていきたいです。

この記事は私が書きました

小泉 和貴

記事一覧

全国を旅行することを目標に、仕事を頑張っています。