Amazon Bedrockのconverse APIからMCPを使用してみる

概要

以前にAnthropicのMCPについて簡単に調べた。

また、別の記事としてgradioでチャットアプリケーション形式のAmazon Bedrockのインターフェースを作っていた。

今回はこれらを組み合わせて、チャットアプリケーション形式のインターフェースでAmazon Bedrockを呼び出す時にMCPサーバを使用してプロンプトでURLが指定されていた場合に、そんURLをFetchして回答に反映する機能を追加した。

最終的に出来上がったのが以下。

プロンプトでURLを指定するとテキスト情報を取得してプロンプトの回答に反映してくれるようになる。
以下リポジトリを参考にして実装した。

参考

URLをfetchするMCPサーバは以下リポジトリで公開されていたものをそのまま持ってきている。

実装内容

作成したコードは上記のリポジトリに配置する。

MCPサーバとconverseAPIを使用したモデルの呼び出し部分はbedrock.pyを参照。
以下にモデル呼び出しと関連する処理を記載する。

bedrock.py

# 関連する処理だけ抜きだしている
# 完全なコードはGithubのリポジトリ参照

async with stdio_client(
        StdioServerParameters(command="uv", args=["run", "mcp-fetch-website"])
    ) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # MCPサーバの一覧を取得
            tools_result = await session.list_tools()
            tools_list = [{"name": tool.name, "description": tool.description,
                          "inputSchema": tool.inputSchema} for tool in tools_result.tools]

            # 取得したMCPサーバの一覧をプロンプトに渡している。
            system = [
                {
                    "text": "You are a helpful AI assistant. You have access to the following tools: " +
                    json.dumps(tools_list) + "Speak in Japanese"
                }
            ]
            state.append({
                        "role": "user",
                        "content": [{"text": input_text}]
                    })
            

            try:
                client = boto3.client("bedrock-runtime")
            except Exception as e:
                return f"Error creating boto3 client: {e}", history

            try:
                while True:
                    # converse APIのリクエストでtoodConfigとしてMCPサーバの情報を渡している
                    response = client.converse(
                        modelId=BEDROCK_MODEL_ID,
                        messages=messages,
                        system=system,
                        inferenceConfig={
                            "maxTokens": 300,
                            "topP": 0.1,
                            "temperature": 0.3
                        },
                        toolConfig=convert_tool_format(tools_result.tools)

                        stop_reason = response['stopReason']
                    if stop_reason == 'tool_use':
                    #プロンプトの回答がtool_useの場合は指定のMCPサーバにリクエストする
                        tool_requests = response['output']['message']['content']
                        for tool_request in tool_requests:
                            if 'toolUse' in tool_request:
                                tool = tool_request['toolUse']
                                logger.info("Requesting tool %s. Request: %s",
                                            tool['name'], tool['toolUseId'])

                                try:
                                    # MCPセッションでツールを呼び出す
                                    tool_response = await session.call_tool(tool['name'], tool['input'])

                                    # ツールのレスポンスを整形する
                                    tool_result = {
                                        "toolUseId": tool['toolUseId'],
                                        "content": [{"text": str(tool_response)}]
                                    }

                                メッセージ履歴にツールのレスポンスを追加する
                                messages.append({
                                    "role": "user",
                                    "content": [{"toolResult": tool_result}]
                                })
                    else:
                        response_text = ""
                        for content in output_message['content']:
                            if 'text' in content:
                                response_text += content['text'] + "\n"
                        return response_text,messages
            except Exception as e:
                return f"Error invoking Bedrock Converse API: {e}",messages



                        

今後

以下で有用なMCPサーバがまとめられている。
確認して便利そうなのがあれば取り入れていきたい。