日曜日, 7月 13, 2025
日曜日, 7月 13, 2025
- Advertisment -
ホームニューステックニュース作って学ぶAIエージェント＆MCP ついでにStrands Agents #AWS - Qiita
テックニュース

作って学ぶAIエージェント＆MCP ついでにStrands Agents #AWS – Qiita

インモビ運営局
By インモビ運営局
0
0



作って学ぶAIエージェント＆MCP ついでにStrands Agents #AWS - Qiita

こんにちは！
株式会社セゾンテクノロジーの島です。

本記事は、
2025 Japan AWS Jr. Champions Qiitaリレー夏
6日目の記事となります！

目次

  1. 環境情報

  2. AIエージェントの実装

  3. MCPの実装

  4. Bedrockとの統合(Strands Agent)

  5. 参考資料

はじめに

本記事は、「AIエージェントとかMCPとかいくら説明聞いてもわからない、、、」という方に向けて、ローカル環境にてそれぞれ簡単な実装を行い、実際の動きを見ながら仕組みを理解することを目的としています。

読むだけでも理解が深まると思いますので、ぜひ最後まで見てください。

1. 環境情報

1-1. 開発環境

  • OS: Windows 11
  • Python: 3.12.6
  • AIサービス:Amazon Bedrock(Claude3.7 sonnet)

1-2. 必要なライブラリ

requirements.txt

boto3==1.34.51
python-dotenv==1.0.1
mcp==1.9.4
strands-agents==0.2.1

1-3. 環境変数(.env)

.env

AWS_PROFILE=your_aws_profile
AWS_REGION=us-west-2
BEDROCK_MODEL_ID=us.anthropic.claude-3-7-sonnet-20250219-v1:0
BEDROCK_MAX_TOKENS=3000
BEDROCK_TEMPERATURE=0.7

# クレデンシャル認証用（プロファイルがない場合）
AWS_ACCESS_KEY_ID=your_access_key_id
AWS_SECRET_ACCESS_KEY=your_secret_access_key

2. AIエージェントの実装

tooluseについて理解している人は読み飛ばして下さい。

2-1. シンプルなAIチャットボット

まずは、単にローカル環境からAmazon BedrockのAPIを叩くだけのシンプルなチャットボットを作成します。

simple_bedrock.py

import os
import boto3
from dotenv import load_dotenv

# 環境変数の読み込み
load_dotenv()

def initialize_bedrock():
    """環境変数をチェックしてプロファイル or クレデンシャル認証"""
    region = os.getenv('AWS_REGION')
    profile = os.getenv('AWS_PROFILE')
    
    print("AWS Bedrockに接続中...")
    
    try:
        if profile:
            # プロファイル認証
            print(f"プロファイル認証: {profile}")
            session = boto3.Session(profile_name=profile, region_name=region)
        else:
            # クレデンシャル認証
            print("クレデンシャル認証")
            access_key = os.getenv('AWS_ACCESS_KEY_ID')
            secret_key = os.getenv('AWS_SECRET_ACCESS_KEY')
            session = boto3.Session(
                aws_access_key_id=access_key,
                aws_secret_access_key=secret_key,
                region_name=region
            )
        
        bedrock = session.client('bedrock-runtime')
        print("接続成功")
        return bedrock
        
    except Exception as e:
        print(f"接続失敗: {str(e)}")
        return None

def chat_with_bedrock(bedrock, user_input):
    """Bedrockとのチャット"""
    try:
        messages = [
            {
                "role": "user",
                "content": [{"text": user_input}]
            }
        ]
        
        response = bedrock.converse(
            modelId=os.getenv('BEDROCK_MODEL_ID'),
            messages=messages,
            inferenceConfig={
                "maxTokens": int(os.getenv('BEDROCK_MAX_TOKENS')),
                "temperature": float(os.getenv('BEDROCK_TEMPERATURE'))
            }
        )
        
        return response['output']['message']['content'][0]['text']
        
    except Exception as e:
        return f"エラー: 応答の生成に失敗しました - {str(e)}"

def main():
    """メイン処理"""
    print("=== シンプルBedrock チャット ===")
    
    # Bedrock初期化
    bedrock = initialize_bedrock()
    if not bedrock:
        return
    
    # チャットループ
    while True:
        user_input = input("\n質問を入力してください (終了: quit): ")
        
        if user_input.lower() in ['quit', 'exit', '終了']:
            print("終了します。")
            break
        
        if not user_input.strip():
            print("質問を入力してください。")
            continue
        
        print("考え中...")
        response = chat_with_bedrock(bedrock, user_input)
        print(f"\nアシスタント: {response}")

if __name__ == "__main__":
    main()

動作確認してみましょう。

terminal

python simple_bedrock.py

=== シンプルBedrock チャット ===
AWS Bedrockに接続中...
プロファイル認証: my-educross-profile
接続成功

質問を入力してください (終了: quit): 現在時刻を教えてください。
考え中...

アシスタント: 申し訳ありませんが、私はリアルタイムの時刻情報にアクセスすることができません。現在の正確な時刻をお知りになりたい場合は、お使いのデバイス（スマートフォン、パソコン、時計など）でご確認いただくか、インターネット検索をご利用ください。

質問を入力してください (終了: quit):

無事会話できました。

以下の部分がAmazon Bedrockにリクエストを送っている部分です。

chat_with_bedrock()

def chat_with_bedrock(bedrock, user_input):
    """Bedrockとのチャット"""
    try:
        messages = [
            {
                "role": "user",
                "content": [{"text": user_input}]
            }
        ]
        
        response = bedrock.converse(
            modelId=os.getenv('BEDROCK_MODEL_ID'),
            messages=messages,
            inferenceConfig={
                "maxTokens": int(os.getenv('BEDROCK_MAX_TOKENS')),
                "temperature": float(os.getenv('BEDROCK_TEMPERATURE'))
            }
        )
        
        # responseの表示
        #print(response)

        return response['output']['message']['content'][0]['text']
        
    except Exception as e:
        return f"エラー: 応答の生成に失敗しました - {str(e)}"

試しにprint(response)で中身を確認してみましょう。
※見やすいようにjson形式に変換しています。

response.json

{
  "ResponseMetadata": {
    "RequestId": "51ecf66e-65bb-4d94-b6d6-cb3df1bea37b",
    "HTTPStatusCode": 200,
    "HTTPHeaders": {
      "date": "Sat, 05 Jul 2025 12:46:45 GMT",
      "content-type": "application/json",
      "content-length": "364",
      "connection": "keep-alive",
      "x-amzn-requestid": "51ecf66e-65bb-4d94-b6d6-cb3df1bea37b"
    },
    "RetryAttempts": 0
  },
  "output": {
    "message": {
      "role": "assistant",
      "content": [
        {
          "text": "申し訳ありませんが、私はリアルタイムの時刻情報にアクセスすることができません。現在の正確な時刻をお知りになりたい場合は、お使いのデバイス（スマートフォン、パソコン、時計など）でご確認いただくか、インターネット検索をご利用ください。"
        }
      ]
    }
  },
  "stopReason": "end_turn",
  "usage": {
    "inputTokens": 19,
    "outputTokens": 106,
    "totalTokens": 125
  },
  "metrics": {
    "latencyMs": 2687
  }
}

AIモデルをAPIで呼び出すだけでは、現在時刻を答えられないことが分かります。

2-2. ツール機能付きエージェント

次に、現在時刻を答えることができるAIエージェントを作成します。

agent.py

import os
import boto3
import json
import datetime
from dotenv import load_dotenv

# 環境変数の読み込み
load_dotenv()

def initialize_bedrock():
    """環境変数をチェックしてプロファイル or クレデンシャル認証"""
    region = os.getenv('AWS_REGION')
    profile = os.getenv('AWS_PROFILE')
    
    print("AWS Bedrockに接続中...")
    
    try:
        if profile:
            # プロファイル認証
            print(f"プロファイル認証: {profile}")
            session = boto3.Session(profile_name=profile, region_name=region)
        else:
            # クレデンシャル認証
            print("クレデンシャル認証")
            access_key = os.getenv('AWS_ACCESS_KEY_ID')
            secret_key = os.getenv('AWS_SECRET_ACCESS_KEY')
            session = boto3.Session(
                aws_access_key_id=access_key,
                aws_secret_access_key=secret_key,
                region_name=region
            )
        
        bedrock = session.client('bedrock-runtime')
        print("接続成功")
        return bedrock
        
    except Exception as e:
        print(f"接続失敗: {str(e)}")
        return None

def get_current_time():
    """現在の日本時間を取得"""
    now = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9)))
    return {
        "time": now.strftime("%Y年%m月%d日 %H時%M分%S秒"),
        "timezone": "Asia/Tokyo"
    }

def chat_with_bedrock_tools(bedrock, user_input):
    """ツール機能付きBedrockとのチャット"""
    # ツール定義
    tools = [
        {
            "toolSpec": {
                "name": "get_current_time",
                "description": "現在の日本時間を取得します",
                "inputSchema": {
                    "json": {
                        "type": "object",
                        "properties": {},
                        "required": []
                    }
                }
            }
        }
    ]
    
    try:
        # 最初のAPI呼び出し
        messages = [
            {
                "role": "user",
                "content": [{"text": user_input}]
            }
        ]
        
        response = bedrock.converse(
            modelId=os.getenv('BEDROCK_MODEL_ID'),
            messages=messages,
            toolConfig={"tools": tools},
            inferenceConfig={
                "maxTokens": int(os.getenv('BEDROCK_MAX_TOKENS')),
                "temperature": float(os.getenv('BEDROCK_TEMPERATURE'))
            }
        )

        # Bedrockからのレスポンスの確認
        #print("【Bedrockレスポンス】")
        #print(response)

        # ツール使用のチェック
        if response['stopReason'] == 'tool_use':
            # ツール使用がある場合
            content = response['output']['message']['content']
            
            # toolUseブロックを探す
            tool_use_block = None
            for item in content:
                if 'toolUse' in item:
                    tool_use_block = item['toolUse']
                    break
            
            if tool_use_block and tool_use_block['name'] == 'get_current_time':
                # 時刻取得ツールを実行
                tool_result = get_current_time()
                
                # ツール結果をメッセージに追加
                messages.append(response['output']['message'])
                messages.append({
                    "role": "user",
                    "content": [
                        {
                            "toolResult": {
                                "toolUseId": tool_use_block['toolUseId'],
                                "content": [{"text": json.dumps(tool_result, ensure_ascii=False)}]
                            }
                        }
                    ]
                })
                
                # メッセージの中身確認
                #print("【送信メッセージ】")
                #print(messages)

                # 最終応答を生成
                final_response = bedrock.converse(
                    modelId=os.getenv('BEDROCK_MODEL_ID'),
                    messages=messages,
                    toolConfig={"tools": tools},
                    inferenceConfig={
                        "maxTokens": int(os.getenv('BEDROCK_MAX_TOKENS')),
                        "temperature": float(os.getenv('BEDROCK_TEMPERATURE'))
                    }
                )
                
                return final_response['output']['message']['content'][0]['text']
        
        # ツール使用がない場合
        return response['output']['message']['content'][0]['text']
        
    except Exception as e:
        return f"エラー: 応答の生成に失敗しました - {str(e)}"

def main():
    """メイン処理"""
    print("=== ツール機能付きBedrock チャット ===")
    
    # Bedrock初期化
    bedrock = initialize_bedrock()
    if not bedrock:
        return
    
    # チャットループ
    while True:
        user_input = input("\n質問を入力してください (終了: quit): ")
        
        if user_input.lower() in ['quit', 'exit', '終了']:
            print("終了します。")
            break
        
        if not user_input.strip():
            print("質問を入力してください。")
            continue
        
        print("考え中...")
        response = chat_with_bedrock_tools(bedrock, user_input)
        print(f"\nアシスタント: {response}")

if __name__ == "__main__":
    main()

先ほどと変わった点を見ていきましょう

まずは、現在時刻を取得する関数が追加されました。

get_current_time()

def get_current_time():
    """現在の日本時間を取得"""
    now = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9)))
    return {
        "time": now.strftime("%Y年%m月%d日 %H時%M分%S秒"),
        "timezone": "Asia/Tokyo"
    }

次に、Amazon Bedrockにリクエストを送っている関数です。
いろいろ付け加えられています。

chat_with_bedrock_tools()

def chat_with_bedrock_tools(bedrock, user_input):
    """ツール機能付きBedrockとのチャット"""
    # ツール定義
    tools = [
        {
            "toolSpec": {
                "name": "get_current_time",
                "description": "現在の日本時間を取得します",
                "inputSchema": {
                    "json": {
                        "type": "object",
                        "properties": {},
                        "required": []
                    }
                }
            }
        }
    ]
    
    try:
        # 最初のAPI呼び出し
        messages = [
            {
                "role": "user",
                "content": [{"text": user_input}]
            }
        ]
        
        response = bedrock.converse(
            modelId=os.getenv('BEDROCK_MODEL_ID'),
            messages=messages,
            toolConfig={"tools": tools},
            inferenceConfig={
                "maxTokens": int(os.getenv('BEDROCK_MAX_TOKENS')),
                "temperature": float(os.getenv('BEDROCK_TEMPERATURE'))
            }
        )

        # Bedrockからのレスポンスの確認
        #print("【Bedrockレスポンス】")
        #print(response)

        # ツール使用のチェック
        if response['stopReason'] == 'tool_use':
            # ツール使用がある場合
            content = response['output']['message']['content']
            
            # toolUseブロックを探す
            tool_use_block = None
            for item in content:
                if 'toolUse' in item:
                    tool_use_block = item['toolUse']
                    break
            
            if tool_use_block and tool_use_block['name'] == 'get_current_time':
                # 時刻取得ツールを実行
                tool_result = get_current_time()
                
                # ツール結果をメッセージに追加
                messages.append(response['output']['message'])
                messages.append({
                    "role": "user",
                    "content": [
                        {
                            "toolResult": {
                                "toolUseId": tool_use_block['toolUseId'],
                                "content": [{"text": json.dumps(tool_result, ensure_ascii=False)}]
                            }
                        }
                    ]
                })
                
                # メッセージの中身確認
                #print("【送信メッセージ】")
                #print(messages)

                # 最終応答を生成
                final_response = bedrock.converse(
                    modelId=os.getenv('BEDROCK_MODEL_ID'),
                    messages=messages,
                    toolConfig={"tools": tools},
                    inferenceConfig={
                        "maxTokens": int(os.getenv('BEDROCK_MAX_TOKENS')),
                        "temperature": float(os.getenv('BEDROCK_TEMPERATURE'))
                    }
                )
                
                return final_response['output']['message']['content'][0]['text']
        
        # ツール使用がない場合
        return response['output']['message']['content'][0]['text']
        
    except Exception as e:
        return f"エラー: 応答の生成に失敗しました - {str(e)}"

まず、以下の部分でtools、つまり生成AIに使わせたい関数の名前とその説明を定義します。
今回の場合は、get_current_timeという関数になります。

tools

 # ツール定義
    tools = [
        {
            "toolSpec": {
                "name": "get_current_time",
                "description": "現在の日本時間を取得します",
                "inputSchema": {
                    "json": {
                        "type": "object",
                        "properties": {},
                        "required": []
                    }
                }
            }
        }
    ]

次に、AmazonBedrockにリクエストを送る部分です。
2-1での実装と比べて、toolConfig={"tools": tools}というパラメータが増えています。
これにより、先ほど定義したツールの一覧(関数とその説明)を生成AIにリクエストとともに渡しているわけです。

リクエスト

 messages = [
            {
                "role": "user",
                "content": [{"text": user_input}]
            }
        ]
        
        response = bedrock.converse(
            modelId=os.getenv('BEDROCK_MODEL_ID'),
            messages=messages,
            toolConfig={"tools": tools},
            inferenceConfig={
                "maxTokens": int(os.getenv('BEDROCK_MAX_TOKENS')),
                "temperature": float(os.getenv('BEDROCK_TEMPERATURE'))
            }
        )

次の処理では、Bedrockから帰ってきたレスポンスでツール使用があるかをチェックしています。

ツール利用確認

 # ツール使用のチェック
        if response['stopReason'] == 'tool_use':

これは、Bedrockから帰ってきたレスポンスの中身を確認するとわかります。

response

{
  "ResponseMetadata": {
    "RequestId": "5217145c-bc01-4131-916a-df18a0371cf7",
    "HTTPStatusCode": 200,
    "HTTPHeaders": {
      "date": "Sat, 05 Jul 2025 14:17:46 GMT",
      "content-type": "application/json",
      "content-length": "435",
      "connection": "keep-alive",
      "x-amzn-requestid": "5217145c-bc01-4131-916a-df18a0371cf7"
    },
    "RetryAttempts": 0
  },
  "output": {
    "message": {
      "role": "assistant",
      "content": [
        {
          "text": "現在の日本時間をお調べします。"
        },
        {
          "toolUse": {
            "toolUseId": "tooluse_lZbLjP5RRq2KNuH0kRHHWg",
            "name": "get_current_time",
            "input": {}
          }
        }
      ]
    }
  },
  "stopReason": "tool_use",
  "usage": {
    "inputTokens": 426,
    "outputTokens": 53,
    "totalTokens": 479
  },
  "metrics": {
    "latencyMs": 1043
  }
}

"stopReason": "tool_use"の部分に注目してください。
2-1で確認した通常時のレスポンスだと、"stopReason": "end_turn"となっていました。
つまり、生成AIが渡された質問とツール定義からツールを使うべきかどうか判断し、ツールを使いたいときは"stopReason": "tool_use"というレスポンスを返す仕組みです。

ではツールを使う場合、すなわちif response['stopReason'] == 'tool_use'の場合の処理を見ていきましょう。

ツールを使う場合は、どのツールを使うかの指示がBedrockからのレスポンス内にあります。

response

{
          "toolUse": {
            "toolUseId": "tooluse_lZbLjP5RRq2KNuH0kRHHWg",
            "name": "get_current_time",
            "input": {}
          }

この情報をもとに、先ほど作成していた関数を呼び出します。
今回はget_current_time()しかありませんが、複数の関数があれば、その分だけ増えていきます。

関数呼び出し

if tool_use_block and tool_use_block['name'] == 'get_current_time':
                # 時刻取得ツールを実行
                tool_result = get_current_time()

関数を実行した結果を得ることが出来たら、その結果を元のメッセージに加えて、もう一度生成AIに送り返します。
それをもとに生成AIが最終的な回答を作成し、無事現在時刻を取得することができます。

terminal

=== ツール機能付きBedrock チャット ===
AWS Bedrockに接続中...
プロファイル認証: my-educross-profile
接続成功

質問を入力してください (終了: quit): 現在時刻を教えてください
考え中...

アシスタント: 現在の日本時間は2025年07月06日 00時03分27秒です。

3. MCPの実装

次に、ローカルのMCPを作っていきましょう。
MCPの通信方式にはstdio(標準入出力)とStreamable HTTPがあります。
とりあえずstdio方式でMCPサーバーを作ってみましょう。

HTTP＋SSE (Server-Sent Events)方式もありますが、legacyとされているためここでは紹介しません。

3-1. MCPサーバーの作成(stdio)

stdio_server.py

from mcp.server.fastmcp import FastMCP
import datetime

# MCPサーバーを作成
mcp = FastMCP("time_server")

@mcp.tool()
async def get_current_time() -> dict:
    # 日本時間で現在時刻を取得
    now = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9)))
    return {
        "time": now.strftime("%Y年%m月%d日 %H時%M分%S秒"),
        "timezone": "Asia/Tokyo"
    }

if __name__ == "__main__":
    # stdioでクライアントと通信
    mcp.run(transport="stdio")

現在時刻を出す関数の部分が、MCPサーバーに切り出されていることが分かります。

3-2. 手動通信でプロトコルを理解

次に、作成したサーバーを起動します。

terminal

python mcp/stdio_server.py

特に何も起こりませんが、stdioは標準入出力でやり取りする方式なので、ターミナルで正しいリクエストを入力すればきちんとしたレスポンスを返してくれるはずです。

以下のjsonをターミナルに一行として貼り付けてください。
これは"method": "initialize"からわかるように、サーバーに対して初期化のメッセージを送信しています。

初期化メッセージ

{"jsonrpc": "2.0", "method": "initialize", "params": {"protocolVersion": "1.0.0", "capabilities": {}, "clientInfo": {"name": "test-client", "version": "1.0.0"}}, "id": 1}

すると、以下のようなレスポンスが返ってきます。
これは、初期化が完了したというサーバーからのメッセージです。

初期化完了

{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-03-26","capabilities":{"experimental":{},"prompts":{"listChanged":false},"resources":{"subscribe":false,"listChanged":false},"tools":{"listChanged":false}},"serverInfo":{"name":"time_server","version":"1.9.4"}}}

次に、以下のjsonをターミナルに一行として貼り付けてください。
これは"method": "notifications/initialized"からわかるように、サーバーに対してクライアント側の準備が完了した通知を送っています。

準備完了通知

{"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}}

ここでは、クライアント側の準備が完了した通知を送るだけなので、サーバー側から何か返答があるわけではありません。

次に、使用可能なツールを問い合わせてみましょう。
以下のjsonをターミナルに一行として貼り付けてください。

ツール問い合わせ

{"jsonrpc": "2.0", "method": "tools/list", "params": {}, "id": 2}

すると、サーバーから以下のレスポンスが返ってきます。

ターミナル

[07/07/25 12:55:52] INFO     Processing request of type ListToolsRequest                                                                                                                                     
{"jsonrpc":"2.0","id":2,"result":{"tools":[{"name":"get_current_time","description":"現在の日本時間を取得します","inputSchema":{"properties":{},"title":"get_current_timeArguments","type":"object"}}]}}

MCPサーバーから使用可能なツールの一覧が取得できました。

では実際にMCPサーバーにツールを実行してもらいましょう。

ツール実行

{"jsonrpc": "2.0", "method": "tools/call", "params": {"name": "get_current_time", "arguments": {}}, "id": 3}

ターミナル

[07/07/25 13:00:27] INFO     Processing request of type CallToolRequest                                                                                                                                      
{"jsonrpc":"2.0","id":3,"result":{"content":[{"type":"text","text":"{\n  \"time\": \"2025年07月07日 13時00分27秒\",\n  \"timezone\": \"Asia/Tokyo\"\n}"}],"isError":false}}

無事現在時刻を取得することができました。

3-3. MCPクライアントの作成

ここまでで、MCPサーバーに対して手動でツールを実行させることができました。
クライアント側では、MCPサーバーの起動から上記の一連の流れを実行します。

stdio_client.py

import asyncio
from mcp.client.stdio import stdio_client, StdioServerParameters
from mcp import ClientSession

async def main():
    # サーバー起動パラメータ
    server_params = StdioServerParameters(
        command="python", 
        args=["mcp/stdio_server.py"]
    )
    
    # MCPサーバーとの通信
    async with stdio_client(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            # 初期化
            await session.initialize()
            
            # ツール呼び出し
            result = await session.call_tool("get_current_time", {})
            print("Tool result:", result.content[0].text)

if __name__ == "__main__":
    asyncio.run(main())

ターミナル

Tool result: {
  "time": "2025年07月08日 09時02分57秒",
  "timezone": "Asia/Tokyo"
}

勝手にMCPサーバーを起動し、現在時刻を取ってきてくれました。

4. Bedrockとの統合(Strands Agent)

最後にMCPのクライアントをBedrockと統合し、MCPを使ったAIエージェントを作成します。
stdioとStreamable HTTPどちらでも可能ですが、せっかくなのでStreamable HTTPで実装してみましょう。
また実装にあたっては、AWSが発表したAIエージェント構築SDK「Strands Agents」を使ってみます。

4-1. MCPサーバー(Streamable HTTP)

http_server.py

import datetime
from mcp.server.fastmcp import FastMCP

# MCPサーバーを作成
mcp = FastMCP("time_server")

@mcp.tool()
async def get_current_time() -> dict:
    """現在の日本時間を取得します"""
    # 日本時間で現在時刻を取得
    now = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9)))
    return {
        "time": now.strftime("%Y年%m月%d日 %H時%M分%S秒"),
        "timezone": "Asia/Tokyo"
    }

if __name__ == "__main__":
    # httpでクライアントと通信
    mcp.run(transport="streamable-http")

基本的にはstdioで作ったMCPサーバーと変わりません。
最後のtransportで"streamable-http"を指定するだけです。

4-2. 統合クライアント

strands_http_client.py

import os
import boto3
from strands import Agent
from strands.models import BedrockModel
from strands.tools.mcp import MCPClient
from mcp.client.streamable_http import streamablehttp_client
from dotenv import load_dotenv

# 環境変数の読み込み
load_dotenv()

# boto3セッションの作成
region = os.getenv('AWS_REGION')
profile = os.getenv('AWS_PROFILE')

if profile:
    # プロファイル認証
    session = boto3.Session(profile_name=profile, region_name=region)
else:
    # クレデンシャル認証
    session = boto3.Session(
        aws_access_key_id=os.getenv('AWS_ACCESS_KEY_ID'),
        aws_secret_access_key=os.getenv('AWS_SECRET_ACCESS_KEY'),
        region_name=region
    )

# BedrockModelの設定
bedrock_model = BedrockModel(
    model_id=os.getenv('BEDROCK_MODEL_ID'),
    boto_session=session
)

# HTTPサーバー用のMCPクライアント
http_mcp_client = MCPClient(lambda: streamablehttp_client(
    "http://localhost:8000/mcp"
))

# 利用可能なツールを読み込んでエージェントに設定
with http_mcp_client:
    agent = Agent(
        model=bedrock_model,
        tools=http_mcp_client.list_tools_sync()
    )
    
    # 継続的な対話ループ
    while True:
        user_input = input("\n質問を入力してください (終了: quit): ")
        
        if user_input.lower() in ['quit', 'exit', '終了']:
            print("終了します。")
            break
        
        if not user_input.strip():
            continue
        
        try:
            response = agent(user_input)
            
        except Exception as e:
            print(f"エラー: {e}\n")

stdioではMCPサーバーをローカルにおいて起動する必要がありましたが、Streamable HTTPを使えばリモートにMCPサーバーを置くこともできるということですね。(今回はローカルですが)

# HTTPサーバー用のMCPクライアント
http_mcp_client = MCPClient(lambda: streamablehttp_client(
    "http://localhost:8000/mcp"
))

また、2-2で作成したagent.pyと比べて、Strands Agentsを使うとかなり簡単にMCP対応のエージェントを構築できることが分かります。

# 3行でツール統合完了
agent = Agent(
    model=bedrock_model,
    tools=http_mcp_client.list_tools_sync()
)

実際に実行した結果は以下の通りです。

ターミナル

質問を入力してください (終了: quit): 現在時刻を教えて
現在の日本時間をお調べします。
Tool #1: get_current_time
現在の日本時間は2025年07月08日 15時29分01秒です。
質問を入力してください (終了: quit): quit
終了します。

無事現在時刻を取得できていますね。

5. 参考資料

以下、参考にさせていただいた記事になります。

今回実際に作ってみたことで、MCPに関する理解がとても深まりました！
実装に関して初歩から一つにまとまっている記事はなかったと思うので、この記事がどなたかの役に立てれば幸いです！





Source link

Views: 0

前の記事
Grok 4 の発表まとめ＆試してみた
次の記事
「JBL、子ども向けノイキャンヘッドホン発売」
RELATED ARTICLES

返事を書く

あなたのコメントを入力してください。
ここにあなたの名前を入力してください

- Advertisment -

ABOUT US

inmobilexion（インモビ）は、動画・ニュース・サイト紹介を中心に、信頼性の高い情報をわかりやすくお届けする情報プラットフォームです。 「in-mobile」×造語の"〜xion"から生まれたインモビは、厳選・接続・導線など多様な意味を内包しながら、次世代の情報体験を提案します。

Contact us: [email protected]

FOLLOW US

© インモビ by inmobilexion.com