Moto 中的 Bedrock AgentCore 模拟：事件 API 实现与实战指南

Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载导读Amazon Bedrock AgentCore 是 AWS 面向智能体Agent应用的内存事件服务负责记录与检索 agent 运行过程中产生的结构化事件流。Moto 在bedrock-agentcore服务模块中提供了该服务核心事件 API 的本地模拟实现目前支持create_event、get_event、list_events、delete_event四个操作。读完本文你将掌握该服务的实现范围与状态、四个事件 API 的完整参数语义与调用方式、底层数据模型与路由原理以及如何基于 Moto 编写可离线运行的单元测试。服务定位与实现范围bedrock-agentcore是 Moto 仓库中针对 AWS Bedrock AgentCore 服务提供的模拟后端。官方实现状态清单位于 docs/docs/services/bedrock-agentcore.rst该文档通过autoclass指令直接引用 moto/bedrockagentcore/models.py 中的BedrockAgentCoreBackend类作为整个模拟后端的入口文档。实现特性清单完整继承原文档列出了 AgentCore 的全部 API 及当前实现状态其中标记[X]表示已实现已实现[X]未实现[ ]create_eventbatch_create_memory_records、batch_delete_memory_records、batch_update_memory_records、complete_resource_token_auth、create_ab_test、create_payment_instrument、create_payment_session、delete_ab_test、delete_batch_evaluation、delete_capacity_provider_session、delete_memory_record、delete_payment_instrument、delete_payment_session、delete_recommendation、evaluate、get_ab_test、get_agent_card、get_batch_evaluation、get_browser_session、get_code_interpreter_session、get_memory_record、get_payment_instrument、get_payment_instrument_balance、get_payment_session、get_recommendation、get_resource_api_key、get_resource_oauth2_token、get_resource_payment_token、get_workload_access_token、get_workload_access_token_for_jwt、get_workload_access_token_for_user_id、ingest_data、invoke_agent_runtime、invoke_agent_runtime_command、invoke_browser、invoke_code_interpreter、invoke_harness、list_ab_tests、list_actors、list_batch_evaluations、list_browser_sessions、list_code_interpreter_sessions、list_memory_extraction_jobs、list_memory_records、list_payment_instruments、list_payment_sessions、list_recommendations、list_sessions、process_payment、retrieve_memory_records、save_browser_session_profile、search_registry_records、start_batch_evaluation、start_browser_session、start_code_interpreter_session、start_memory_extraction_job、start_recommendation、stop_batch_evaluation、stop_browser_session、stop_code_interpreter_session、stop_runtime_session、update_ab_test、update_browser_stream从上表可以清晰地看到当前 Moto 对 AgentCore 的覆盖聚焦于事件Event生命周期管理即create_event、get_event、list_events、delete_event而记忆记录Memory Records、AB 测试、支付会话、浏览器会话、代码解释器会话、运行时调用等大批 API 尚未模拟。因此在使用前请确认你的测试路径只涉及事件类操作。已实现的四个事件 API 详解四个 API 的请求处理逻辑集中在 moto/bedrockagentcore/responses.py 的BedrockAgentCoreResponse类中每个方法对应一个 AWS 操作通过self.backend调用 moto/bedrockagentcore/models.py 中的BedrockAgentCoreBackend。create_event写入一条事件记录create_event用于在指定的memoryId下为某个actorId与sessionId组合创建一条事件。请求参数如下memoryId必填事件所属的内存标识对应测试常量如mtest12345-abcdefghijactorId必填执行动作的 actor 标识sessionId必填事件所属会话标识eventTimestamp可选事件时间戳如2026-01-01T00:00:00Z缺省时使用当前 UTC 时间payload必填非空数组事件负载通常为对话内容数组例如[{conversational: {content: {text: hello}, role: USER}}]branch可选分支信息字典缺省为{name: main}metadata可选附加元数据字典仅当非空时才会出现在响应中。根据 models.py 的实现后端会对actorId、sessionId为空以及payload为空的情况抛出ValidationException对应 AWS 的 400 校验错误错误消息与真实服务一致地描述了 Member must not be null / Member must not be empty 约束。调用示例来自 tests/test_bedrockagentcore/test_bedrockagentcore.py 的等价逻辑import boto3 from moto import mock_aws mock_aws def test_create_event(): client boto3.client(bedrock-agentcore, region_nameus-east-1) resp client.create_event( memoryIdmtest12345-abcdefghij, actorIdactor-1, sessionIdsession-1, eventTimestamp2026-01-01T00:00:00Z, payload[{conversational: {content: {text: hello}, role: USER}}], ) event resp[event] print(event[eventId]) # 形如 epoch-millis#hex print(event[branch]) # {name: main}响应中的event对象包含memoryId、actorId、sessionId、eventId、eventTimestamp、branch以及payload和若提供metadata。get_event按 ID 读取事件get_event需要memoryId、actorId、sessionId、eventId四个参数返回指定事件的完整信息含payload。若事件不存在后端会抛出ResourceNotFoundException错误码ResourceNotFoundException消息形如Event eventId not found。resp client.get_event( memoryIdMEMORY_ID, actorIdactor-1, sessionIdsession-1, eventIdcreated_event_id, ) assert resp[event][payload] expected_payloadlist_events按作用域列出事件list_events按(memoryId, actorId, sessionId)组合作用域列出事件支持两个可选参数includePayloads是否在列表中包含每条事件的payload默认为False。这与真实服务一致——列表默认不返回负载从而避免大体积传输maxResults限制返回条数后端按插入顺序截取前maxResults条events[:max_results]。响应结构为{events: [...], nextToken: None}。需要特别说明的是虽然响应中带有nextToken字段但当前实现始终返回None即尚未实现分页续读当事件总数超过maxResults时多余的事件将被直接截断。列表按插入顺序即时间顺序返回因为后端存储按插入顺序维护字典。resp client.list_events( memoryIdMEMORY_ID, actorIdactor-1, sessionIdsession-1, includePayloadsTrue, maxResults2, )delete_event删除事件delete_event同样以四元组定位事件并删除成功时返回被删除的eventId。若目标事件不存在抛出ResourceNotFoundException。删除后可再次get_event验证此时会得到ResourceNotFoundException参见 tests/test_bedrockagentcore/test_bedrockagentcore.py。resp client.delete_event( memoryIdMEMORY_ID, actorIdactor-1, sessionIdsession-1, eventIdcreated_event_id, ) assert resp[eventId] created_event_id事件数据模型与底层存储Event 模型models.py 中的Event类封装了事件的全部字段并定义了序列化逻辑to_dict(include_payloadTrue)event_id的生成格式与真实 AWS 对齐epoch-millis#hex即当前 Unix 毫秒时间戳加#加 16 位随机十六进制串通过mock_random.get_random_hex(16)生成形如1735689600000#a1b2c3d4e5f60718event_timestamp缺省时取utcnow()branch缺省为{name: main}metadata缺省为空字典且仅在非空时序列化到响应中序列化时payload是否包含由调用方控制get_event/create_event返回完整负载list_events默认剔除。后端存储结构BedrockAgentCoreBackend使用嵌套字典作为内存存储models.pyself.events: dict[tuple[str, str, str], dict[str, Event]] {}外层以(memoryId, actorId, sessionId)三元组为键内层以eventId为键。字典的插入顺序即事件的 chronological 顺序因此list_events天然按时间先后返回且不同sessionId之间的事件彼此隔离测试test_list_events_scoped_to_session专门验证了这一点。这也解释了为什么列表截断是取前 N 条而非取任意 N 条。请求路由与区域支持URL 路由urls.py 定义了该服务的端点匹配规则url_bases [rhttps?://bedrock-agentcore\.(.)\.amazonaws\.com] url_paths {{0}/.*$: BedrockAgentCoreResponse.dispatch}即所有发送到bedrock-agentcore.region.amazonaws.com的请求都会被BedrockAgentCoreResponse.dispatch拦截分发。该服务也已注册进 moto/backend_index.py 的 URL 索引中模式为https?://bedrock-agentcore\.(.)\.amazonaws\.com。区域支持与同源的bedrock-agentcore-control服务采用相同的处理方式models.py由于 Moto 自身的区域清单尚未收录该服务BedrockAgentCoreBackend通过BackendDict的use_boto3_regionsFalse并显式声明additional_regions来支持以下 16 个区域us-east-1、us-east-2、us-west-1、us-west-2、ap-south-1、ap-northeast-1、ap-northeast-2、ap-southeast-1、ap-southeast-2、ca-central-1、eu-central-1、eu-west-1、eu-west-2、eu-west-3、eu-north-1、sa-east-1。在使用boto3.client(bedrock-agentcore, region_name...)时请从上述列表中选择区域否则可能无法命中模拟端点。测试默认使用us-east-1。参数解析机制BedrockAgentCoreResponse构造时设置了service_namebedrock-agentcore与automated_parameter_parsing Trueresponses.py后者意味着请求体中的驼峰命名参数如memoryId、maxResults、includePayloads会被自动解析为 Python 字典键再交由各 handler 通过params.get(...)提取。异常模型exceptions.py 定义了该服务的两类客户端错误均继承自ServiceException异常类Error Code触发场景ValidationExceptionValidationExceptionactorId/sessionId缺失或payload为空ResourceNotFoundExceptionResourceNotFoundException查询/删除不存在的eventId在 boto3 客户端中这两类错误会以ClientError形式抛出测试中通过exc.value.response[Error][Code]断言见 test_bedrockagentcore.py。完整实战在 Moto 中测试 AgentCore 事件流以下是一个覆盖事件全生命周期的完整测试集合了 tests/test_bedrockagentcore/test_bedrockagentcore.py 中的核心场景可直接作为你项目测试的起点import boto3 import pytest from botocore.exceptions import ClientError from moto import mock_aws TEST_REGION us-east-1 MEMORY_ID mtest12345-abcdefghij def _payload(texthello): return [{conversational: {content: {text: text}, role: USER}}] mock_aws def test_agentcore_event_lifecycle(): client boto3.client(bedrock-agentcore, region_nameTEST_REGION) # 1. 创建事件 created client.create_event( memoryIdMEMORY_ID, actorIdactor-1, sessionIdsession-1, eventTimestamp2026-01-01T00:00:00Z, payload_payload(), branch{name: feature-branch}, metadata{source: {stringValue: unit-test}}, )[event] assert created[branch] {name: feature-branch} assert created[metadata] {source: {stringValue: unit-test}} # 2. 按 ID 读取 fetched client.get_event( memoryIdMEMORY_ID, actorIdactor-1, sessionIdsession-1, eventIdcreated[eventId], )[event] assert fetched[payload] _payload() # 3. 列表默认不含 payload支持 includePayloads / maxResults listed client.list_events( memoryIdMEMORY_ID, actorIdactor-1, sessionIdsession-1, includePayloadsTrue, ) assert len(listed[events]) 1 assert listed[events][0][payload] _payload() # 4. 删除后不可再读取 client.delete_event( memoryIdMEMORY_ID, actorIdactor-1, sessionIdsession-1, eventIdcreated[eventId], ) with pytest.raises(ClientError) as exc: client.get_event( memoryIdMEMORY_ID, actorIdactor-1, sessionIdsession-1, eventIdcreated[eventId], ) assert exc.value.response[Error][Code] ResourceNotFoundException边界行为速查缺少actorId或sessionId→ValidationExceptionpayload[]空数组→ValidationException查询不存在的eventId→ResourceNotFoundExceptionlist_events默认不返回payload只有includePayloadsTrue时才包含list_events返回nextTokenNone超量事件直接截断无分页续读事件按(memoryId, actorId, sessionId)隔离不同会话互不可见。与相关服务的区分仓库中还存在同源的bedrock-agentcore-control服务moto/bedrockagentcorecontrol/ 与 moto/backend_index.py其端点域名为bedrock-agentcore-control.region.amazonaws.com两者在 Moto 中是独立的模拟后端。写测试时请确认你调用的 client 名称与端点属于哪一个服务boto3.client(bedrock-agentcore)→BedrockAgentCoreBackend本文所述事件 APIboto3.client(bedrock-agentcore-control)→ AgentCore 控制平面后端。小结Moto 对bedrock-agentcore的模拟虽尚处早期仅 4/69 个 API但事件生命周期这一最核心的读写路径已经可用且有完整测试覆盖。对于需要离线测试 Agent 事件采集、回放与清理逻辑的开发者create_event/get_event/list_events/delete_event已足够支撑绝大多数场景而涉及记忆记录、运行时调用等尚未实现的 API 时建议等待后续版本或对相应调用做本地打桩处理。深入阅读 moto/bedrockagentcore/models.py、moto/bedrockagentcore/responses.py 与 tests/test_bedrockagentcore/test_bedrockagentcore.py 可获得最准确的实现细节。赞分享Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载相关推荐5分钟搞定微信群消息自动转发wechat-forwarding终极指南5分钟搞定微信群消息自动转发wechat forwarding终极指南 还在为手动转发微信群消息而烦恼吗wechat forwarding是一款能够实现微信Mock测试Moto 中 Amazon Managed Prometheusamp服务的模拟实现与实战指南Moto 中 Amazon Managed Prometheusamp服务的模拟实现与实战指南 Amazon Managed PrometheusAMPMock测试moto 中的 AWS FSx 模拟已实现 API 全景、源码结构与测试实战指南moto 中的 AWS FSx 模拟已实现 API 全景、源码结构与测试实战指南 本文基于 moto 仓库中 docs/docs/services/fsx.rMock测试上一篇3步实现App RTL完美适配让阿拉伯语用户体验翻倍的UI改造指南下一篇3分钟搞定RoomGPT环境配置Replicate API密钥获取与存储指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考