生成AIに「機能追加」できる？Claude + AWSで実現する拡張可能な会話アーキテクチャ

こんにちは、クラウドエンジニアの長谷部・三﨑です。今回は生成AIの話題に触れながら、「Bedrock + Function Calling + Lambda」で構成する拡張可能なAIアーキテクチャについて紹介します。

生成AIと会話するだけでなく、「外部の機能を追加して連携させる」ことで、その可能性は大きく広がります。

最近では、AWS Lambda MCP Server (Multi-Capability Prompting Server) のような、こうした連携をより高度に、そして効率的に行うための仕組みも登場し始めています。MCPの普及・発展によって、生成AIに機能を追加するための便利なライブラリやフレームワークは今後ますます拡がっていくと考えられます。多くの実務シーンでは、将来的にはそうした高レベルなツールを活用することが主流になるでしょう。

とはいえ、それらの根底にある「AIがどのように外部機能と連携するのか」という基本的な仕組みを理解することは、応用力を高める上で非常に重要です。そこでこの記事では、これを実現するためのしくみについて理解を深めるために、原型とも言えるFunction Calling を用い、AWS上でどのように実装できるかを具体的に示すために、「Bedrock + Function Calling + Lambda」を組み合わせたアーキテクチャを詳しく解説します。

Function Callingは、まさにAPIを通じてAIに「できること」を増やすための強力な手段なのです！

アーキテクチャの紹介

今回のシステムは、AWS 上でサーバーレスアーキテクチャを採用しています。以下の図はその構成を示したものです。

この構成では、主に以下のコンポーネントが連携しています。

1. ユーザーリクエストの受信（API Gateway）

ユーザーがクライアント（WebアプリやチャットUIなど）からリクエストを送ると、API Gateway がリクエストを受け取り、AWS Lambda へとルーティングします。

2. Lambda（バックエンド）による処理

API Gatewayから渡されたリクエストは、Lambda（バックエンド）関数で処理されます。この関数では主に以下の役割を担っています。

リクエストの内容に応じた処理分岐
ClaudeのFunction Callingで使う「機能」の実行
DynamoDBへのトークンの記録・取得

3. DynamoDBでトークンを管理

会話で生成されたトークンは Amazon DynamoDB に保存されます。

4. Bedrockを通じた生成AIとの対話

会話生成の処理は、Amazon Bedrock を経由して Claude モデルに問い合わせる形で実現しています。Lambda（生成AI実行用）関数がこの処理を行い、生成された返答を返します。

また、BedrockのFunction Calling機能を使うことで、生成AIは外部のAPI（Lambdaで実装されたもの）を呼び出して拡張機能を実行できるようになっています。

BedrockのFunction Calling機能を活用することで、Lambda関数として機能を追加して、会話の中からその機能を呼び出すことが可能になります。

機能追加できる生成AI：Function Callingの仕組み

Claudeにただ文章を生成させるだけでなく、「特定の操作」をさせたいこともありますよね？

たとえば：

ユーザーの現在の状態を参照したい
他サービスと連携して処理を実行したい

こういったニーズを満たすのがFunction Callingです。 Bedrock経由でClaudeにプロンプトを送信し、Lambdaで定義された関数（たとえば「現在使用中のトークンを取得する関数」）を呼び出す設計です。

class ToolsList:
    @staticmethod
    def get_token(model_id, start_date=None, end_date=None):
        """
        生成AIのインプットトークンとアウトプットトークン、日本円のコストを取得する
        """
        url = "バックエンドAPI URL"
        params = {"model_id": model_id, "start_date": start_date, "end_date": end_date}
        headers = {"Content-Type": "application/json"}
        response = requests.post(url, json=params, headers=headers)
        jsonData = response.json()
        result = f"モデル{model_id}, のコストは{jsonData['total_cost_jpy']}円です．"
        return result


def get_tool_definition():
    """ツール定義を返す"""
    return {
        "toolSpec": {
            "name": "get_token",
            "description": "生成AIのインプットトークンとアウトプットトークン、日本円のコストを取得する",
            "inputSchema": {
                "json": {
                    "type": "object",
                    "properties": {
                        "model_id": {"type": "string", "description": "生成AIのモデル"},
                        "start_date": {
                            "type": "string",
                            "description": "コスト計算の開始期間",
                        },
                        "end_date": {
                            "type": "string",
                            "description": "コスト計算の終了期間",
                        },
                    },
                    "required": ["model_id"],
                }
            },
        }
    }


def get_bedrock_client():
    """Bedrock クライアントを初期化して返す"""
    return boto3.client("bedrock-runtime", region_name="us-west-2")


def get_dynamodb_client():
    """DynamoDB クライアントを初期化して返す"""
    return boto3.client("dynamodb", region_name="us-west-2")


def create_messages(prompt):
    """メッセージを作成して返す"""
    return [
        {
            "role": "user",
            "content": [{"text": prompt}],
        }
    ]


def extract_tool_use(content):
    """コンテンツからツール使用情報を抽出"""
    for item in content:
        if "toolUse" in item:
            return item["toolUse"]
    return None


def get_model_response(client, messages, tool_definition):
    """モデルからレスポンスを取得"""
    return client.converse(
        modelId="anthropic.claude-3-5-haiku-20241022-v1:0",
        messages=messages,
        toolConfig={
            "tools": [tool_definition],
        },
    )

def lambda_handler(event, context):
    """Lambda ハンドラー関数"""
    # 初期化
    client = get_bedrock_client()
    dynamodb_client = get_dynamodb_client()
    tool_definition = get_tool_definition()
    prompt = f" {event['prompt']}"

    # メッセージを作成
    messages = create_messages(prompt)

    # モデルからレスポンスを取得
    response = get_model_response(client, messages, tool_definition)
    messages.append(response["output"]["message"])

    # ツール使用情報を抽出
    response_content = response["output"]["message"]["content"]
    tool_use = extract_tool_use(response_content)

    # ツール関数の使用が不要なケース（正常）
    if tool_use is None:
        return {
            "statusCode": 200,
            "body": json.dumps(
                {"出力結果": response_content[0].get("text", "")}, ensure_ascii=False
            ),
        }

    # ツール関数を実行
    tool_func = getattr(ToolsList, tool_use["name"])
    tool_result = tool_func(**tool_use["input"])

DynamoDBにモデルとトークン情報を保存

生成AIと会話する中で、「その会話で使用したトークン」を取得したい場合もあります。本構成では、Claudeとのすべてのやりとりを DynamoDB に保存しています。 DynamoDBには、modelID、タイムスタンプ、入出力トークンといった情報を格納しています。そのため、後から次のような問い合わせが可能になります：

「今使っているトークンは何？」

Function Callingの使用を使用したレスポンスは以下のようになり、Backend API を利用し、DBからトークン情報を取得した結果が利用できるようになっています。

→ Lambda関数がDynamoDBを検索し、最新のmodelIDやトークン情報を取得 → Claudeの返答としてユーザーに伝えるこのように、会話の中で動的に情報を取得し、AIに返答させることが可能です。

同様に普通の質問を実行してみると以下のようなレスポンスが返却されます。

まとめ：Claude + AWSで作る「機能追加できる」AIシステム

生成AIは「ただのチャットボット」から、「実行できるアシスタント」へと進化しています。今回紹介した構成では、

Claudeで自然な対話を実現しながら
Lambdaを通じて実行処理を追加し
DynamoDBで会話情報を管理する（※DynamoDBには、modelID、タイムスタンプ、入出力トークンを格納）

という形で、会話の中からアクションを実行できる仕組みを実現しています。生成AIをプロダクトに活用する際、Function Calling はとても強力な武器になります。「ただのチャット」にとどまらない、次世代アシスタントの第一歩として、ぜひ試してみてください。

作成したコードの全体の紹介

class ToolsList:
    @staticmethod
    def get_token(model_id, start_date=None, end_date=None):
        """
        生成AIのインプットトークンとアウトプットトークン、日本円のコストを取得する
        """
        url = "バックエンドAPI URL"
        params = {"model_id": model_id, "start_date": start_date, "end_date": end_date}
        headers = {"Content-Type": "application/json"}
        response = requests.post(url, json=params, headers=headers)
        jsonData = response.json()
        result = f"モデル{model_id}, のコストは{jsonData['total_cost_jpy']}円です．"
        return result


def get_tool_definition():
    """ツール定義を返す"""
    return {
        "toolSpec": {
            "name": "get_token",
            "description": "生成AIのインプットトークンとアウトプットトークン、日本円のコストを取得する",
            "inputSchema": {
                "json": {
                    "type": "object",
                    "properties": {
                        "model_id": {"type": "string", "description": "生成AIのモデル"},
                        "start_date": {
                            "type": "string",
                            "description": "コスト計算の開始期間",
                        },
                        "end_date": {
                            "type": "string",
                            "description": "コスト計算の終了期間",
                        },
                    },
                    "required": ["model_id"],
                }
            },
        }
    }


def get_bedrock_client():
    """Bedrock クライアントを初期化して返す"""
    return boto3.client("bedrock-runtime", region_name="us-west-2")


def get_dynamodb_client():
    """DynamoDB クライアントを初期化して返す"""
    return boto3.client("dynamodb", region_name="us-west-2")


def create_messages(prompt):
    """メッセージを作成して返す"""
    return [
        {
            "role": "user",
            "content": [{"text": prompt}],
        }
    ]


def extract_tool_use(content):
    """コンテンツからツール使用情報を抽出"""
    for item in content:
        if "toolUse" in item:
            return item["toolUse"]
    return None


def get_model_response(client, messages, tool_definition):
    """モデルからレスポンスを取得"""
    return client.converse(
        modelId="anthropic.claude-3-5-haiku-20241022-v1:0",
        messages=messages,
        toolConfig={
            "tools": [tool_definition],
        },
    )


def put_item_to_dynamodb(dynamodb_client, model_id, input_tokens, output_tokens):
    """DynamoDBに消費トークン情報を書き込む"""
    table_name = CHAT_HISTORY_TABLE  # DynamoDBのテーブル名
    item = {
        "modelid": {"S": model_id},
        "timestamp": {"S": datetime.datetime.now(datetime.timezone.utc).isoformat()},
        "input_tokens": {"N": str(input_tokens)},
        "output_tokens": {"N": str(output_tokens)},
    }

    dynamodb_client.put_item(TableName=table_name, Item=item)


def lambda_handler(event, context):
    logging.basicConfig(format="%(asctime)s %(message)s", level=logging.DEBUG)
    # 初期化
    client = get_bedrock_client()
    dynamodb_client = get_dynamodb_client()
    tool_definition = get_tool_definition()
    prompt = f" {event['prompt']}"

    # メッセージを作成
    messages = create_messages(prompt)

    # モデルからレスポンスを取得
    response = get_model_response(client, messages, tool_definition)
    messages.append(response["output"]["message"])

    # ツール使用情報を抽出
    response_content = response["output"]["message"]["content"]
    tool_use = extract_tool_use(response_content)

    # DynamoDBに消費トークン情報を追加
    model_id = "claude"
    # usage情報を取得
    usage = response.get("usage")
    if usage:
        input_tokens = usage.get("inputTokens")
        output_tokens = usage.get("outputTokens")
        put_item_to_dynamodb(dynamodb_client, model_id, input_tokens, output_tokens)

    # ツール関数の使用が不要なケース（正常）
    if tool_use is None:
        return {
            "statusCode": 200,
            "body": json.dumps(
                {"出力結果": response_content[0].get("text", "")}, ensure_ascii=False
            ),
        }

    # ツール関数を実行
    tool_func = getattr(ToolsList, tool_use["name"])
    tool_result = tool_func(**tool_use["input"])

    return {
        "statusCode": 200,
        "body": json.dumps({"出力結果": tool_result}, ensure_ascii=False),
    }