AgentCore Memory の短期記憶にカスタムメタデータを付けてフィルタする

# プロジェクト初期化
uv init --python 3.13

# 依存パッケージの追加（後半の Strands カスタムツール例を試すなら strands-agents と bedrock-agentcore も）
# uv add すると仮想環境とロックファイルは自動で整います
uv add boto3 strands-agents bedrock-agentcore

import boto3

control = boto3.client("bedrock-agentcore-control", region_name="us-east-1")

res = control.create_memory(
    name="MetadataFilterBlogShort",
    eventExpiryDuration=30,
    memoryStrategies=[
        {"userPreferenceMemoryStrategy": {
            "name": "BlogUserPreferenceShort",
            "namespaces": ["/users/{actorId}/preferences"],
        }},
    ],
)
print(res["memory"]["id"])

uv run create_memory.py

MetadataFilterBlogShort1778050726-xxx

import boto3
from datetime import datetime, timezone

# create_memory.py の実行結果（Memory ID）に差し替えてください
MEMORY_ID = "MetadataFilterBlogShort1778050726-xxx"
# アクター（ユーザー識別子）と会話セッション識別子は任意の文字列で OK
ACTOR_ID = "blog-actor-001"
SESSION_ID = "blog-session-001"

client = boto3.client("bedrock-agentcore", region_name="us-east-1")

events = [
    ("旅行の宿は和室が好きです。京都の和菓子も巡りたい。",
     {"category": "travel", "destination": "kyoto", "priority": "high"}),
    ("出張で東京に行くときは新幹線のグリーン車を選びがちです。",
     {"category": "travel", "destination": "tokyo", "priority": "medium"}),
    ("コーヒーは深煎りのブラックが好みです。",
     {"category": "food", "topic": "coffee", "priority": "low"}),
    ("ホテルは静かなクラブフロアが好きで、朝食は和食派です。",
     {"category": "travel", "destination": "kyoto", "priority": "high"}),
]

for text, md in events:
    ev = client.create_event(
        memoryId=MEMORY_ID, actorId=ACTOR_ID, sessionId=SESSION_ID,
        eventTimestamp=datetime.now(timezone.utc),
        payload=[{"conversational": {"content": {"text": text}, "role": "USER"}}],
        metadata={k: {"stringValue": v} for k, v in md.items()},
    )
    print("created", ev["event"]["eventId"])

uv run create_events.py

created 0000001778050888398#6d9e2dd3
created 0000001778050889421#54c4a9dc
created 0000001778050889787#1c03c316
created 0000001778050890164#ef034a23

import boto3

# create_events.py で使ったものと同じ ID を指定する
MEMORY_ID = "MetadataFilterBlogShort1778050726-xxx"
ACTOR_ID = "blog-actor-001"
SESSION_ID = "blog-session-001"

client = boto3.client("bedrock-agentcore", region_name="us-east-1")

res = client.list_events(
    memoryId=MEMORY_ID, actorId=ACTOR_ID, sessionId=SESSION_ID,
    includePayloads=True,
    filter={
        "eventMetadata": [
            {"left": {"metadataKey": "destination"},
             "operator": "EQUALS_TO",
             "right": {"metadataValue": {"stringValue": "kyoto"}}},
        ]
    },
)

for ev in res.get("events", []):
    text = ev["payload"][0]["conversational"]["content"]["text"]
    print(ev["eventId"], "/", text)

uv run filter_events.py

0000001778050890164#ef034a23 / ホテルは静かなクラブフロアが好きで、朝食は和食派です。
0000001778050888398#6d9e2dd3 / 旅行の宿は和室が好きです。京都の和菓子も巡りたい。

from bedrock_agentcore.memory import MemoryClient

# create_memory.py の実行結果と任意の識別子
MEMORY_ID = "MetadataFilterBlogShort1778050726-xxx"
ACTOR_ID = "blog-actor-001"
SESSION_ID = "blog-session-001"

client = MemoryClient(region_name="us-east-1")

# 書き込み: metadata を Tagged Union 形式で渡す（最大 15 件）
res = client.create_event(
    memory_id=MEMORY_ID,
    actor_id=ACTOR_ID,
    session_id=SESSION_ID,
    messages=[("旅行の宿は和室が好きです。", "USER")],
    metadata={"destination": {"stringValue": "kyoto"}},
)
print("created", res["eventId"])

# 検索: event_metadata に boto3 と同じ形式のフィルタを渡す
events = client.list_events(
    memory_id=MEMORY_ID,
    actor_id=ACTOR_ID,
    session_id=SESSION_ID,
    event_metadata=[
        {"left": {"metadataKey": "destination"},
         "operator": "EQUALS_TO",
         "right": {"metadataValue": {"stringValue": "kyoto"}}},
    ],
)
print("matched events:", len(events))

uv run agentcore_sdk_example.py

created 0000001778051209068#a0891855
matched events: 1

from enum import Enum
from typing import Optional

from bedrock_agentcore.memory import MemorySessionManager
from bedrock_agentcore.memory.constants import ConversationalMessage, MessageRole
from bedrock_agentcore.memory.models.filters import (
    EventMetadataFilter,
    LeftExpression,
    OperatorType,
    RightExpression,
    StringValue,
)
from pydantic import BaseModel, Field
from strands import tool

# ---- 値域を Enum で固定 ----

class Destination(str, Enum):
    """旅行・出張の目的地。kyoto / tokyo のみ許可。"""

    KYOTO = "kyoto"
    TOKYO = "tokyo"

class Category(str, Enum):
    """発話のカテゴリ。"""

    TRAVEL = "travel"
    FOOD = "food"
    BUSINESS_TRIP = "business_trip"

class Priority(str, Enum):
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"

class EventMetadata(BaseModel):
    """イベントに付けるメタデータ集合。フィールドは固定、値は Enum で値域を制約する。
    該当しないフィールドは None / 省略する。
    """

    destination: Optional[Destination] = Field(
        default=None,
        description="旅行・出張の目的地。kyoto か tokyo のみ。該当しない場合は省略する。",
    )
    category: Optional[Category] = Field(
        default=None,
        description="発話のカテゴリ。travel / food / business_trip のいずれか。",
    )
    priority: Optional[Priority] = Field(
        default=None,
        description="優先度。high / medium / low のいずれか。",
    )

def make_memory_tools(
    manager: MemorySessionManager, actor_id: str, session_id: str
):
    """memory_id / actor_id / session_id を設定して、Agent 用のツール 2 種を返す。"""

    @tool
    def remember_user_event(text: str, metadata: EventMetadata) -> str:
        """ユーザー発話をメタデータ付きで AgentCore Memory に記録する。

        Args:
            text: 発話本文
            metadata: イベントに紐付けるメタデータ。EventMetadata のスキーマに従う。

        Returns:
            作成された event の ID
        """
        # None フィールドは AgentCore に送らない
        md = {
            k: StringValue.build(v)
            for k, v in metadata.model_dump().items()
            if v is not None
        }
        event = manager.add_turns(
            actor_id=actor_id,
            session_id=session_id,
            messages=[ConversationalMessage(text=text, role=MessageRole.USER)],
            metadata=md,
        )
        return event.get("eventId") or str(event)

    @tool
    def search_user_events(
        metadata_key: str, metadata_value: Optional[str] = None
    ) -> list[dict]:
        """メタデータで過去のイベントを絞り込んで取得する。metadata_value 省略時は EXISTS で絞る。

        Args:
            metadata_key: 絞り込みに使うメタデータのキー
            metadata_value: 値で絞る場合の値（省略時はキーの有無だけで絞る）

        Returns:
            マッチしたイベントのリスト（{eventId, text, metadata}）
        """
        if metadata_value is None:
            f = EventMetadataFilter.build_expression(
                left_operand=LeftExpression.build(metadata_key),
                operator=OperatorType.EXISTS,
            )
        else:
            f = EventMetadataFilter.build_expression(
                left_operand=LeftExpression.build(metadata_key),
                operator=OperatorType.EQUALS_TO,
                right_operand=RightExpression.build(metadata_value),
            )
        events = manager.list_events(
            actor_id=actor_id,
            session_id=session_id,
            eventMetadata=[f],
            include_payload=True,
        )
        out = []
        for ev in events:
            text = None
            for p in (ev.get("payload") or []):
                content = (p.get("conversational") or {}).get("content") or {}
                text = content.get("text") or text
            meta = ev.get("metadata") or {}
            flat = {k: v.get("stringValue") if isinstance(v, dict) else v
                    for k, v in meta.items()}
            out.append({"eventId": ev.get("eventId"), "text": text, "metadata": flat})
        return out

    return [remember_user_event, search_user_events]

from bedrock_agentcore.memory import MemorySessionManager
from strands import Agent
from strands.models import BedrockModel

from strands_custom_tools import make_memory_tools

# create_memory.py の実行結果と任意の識別子
MEMORY_ID = "MetadataFilterBlogShort1778050726-xxx"
ACTOR_ID = "blog-actor-001"
SESSION_ID = "blog-session-001"

manager = MemorySessionManager(memory_id=MEMORY_ID, region_name="us-east-1")
tools = make_memory_tools(manager, actor_id=ACTOR_ID, session_id=SESSION_ID)

agent = Agent(
    model=BedrockModel(
        model_id="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
        region_name="us-east-1",
    ),
    tools=tools,
    system_prompt=(
        "ユーザー発話を AgentCore Memory に記録するアシスタントです。"
        "発話に応じた適切なメタデータ（category, destination, priority 等）を判断して付与してください。"
        "過去の発話が必要なときは search_user_events で絞り込んでから参照してください。"
    ),
)

from agent_setup import agent

agent("次の発話を記録してください: 「京都の和菓子を巡るのが好きです。」")
agent("次の発話を記録してください: 「東京出張は新幹線のグリーン車が定番です。」")
agent("次の発話を記録してください: 「コーヒーは深煎りのブラックが好みです。」")
agent("次の発話を記録してください: 「京都では和室の宿が一番落ち着きます。」")
agent("destination=kyoto のメタデータが付いている発話だけを取り出してください。")

uv run run.py

Tool #5: search_user_events
destination=kyoto のメタデータが付いている発話を取得しました。2件見つかりました：

1. 「京都では和室の宿が一番落ち着きます。」
   - カテゴリ: travel
   - 目的地: kyoto

2. 「京都の和菓子を巡るのが好きです。」
   - カテゴリ: food
   - 目的地: kyoto