用 MCP 服务器把 Kiro CLI 的对话记忆接入 Amazon Bedrock AgentCore Memory

预计阅读时间：11 分钟

Kiro CLI 让你在终端里直接和 AI 代理对话，但默认的对话记忆有限——关掉终端，上下文就丢了。Amazon Bedrock AgentCore Memory 是一个全托管服务，能让 AI 代理跨会话保留信息，对话越用越"懂你"。问题在于：两者之间没有现成的桥梁。这篇文章讲的是如何写一个自定义 MCP（Model Context Protocol）服务器，把 AgentCore Memory 的存储、检索、监控能力以工具形式暴露给 Kiro CLI，让每次终端对话都带着历史记忆。

为什么需要自定义 MCP 服务器

Kiro CLI 本身支持 MCP 协议——只要你给它提供一个 MCP 服务器，它就能调用服务器里注册的工具。而 Bedrock AgentCore Memory 提供的是 REST API 和 SDK，不是 MCP 工具。所以核心工作就是：

用 MCP SDK 写一个服务器，把 AgentCore Memory 的关键操作封装成工具。
在 Kiro CLI 的配置里注册这个 MCP 服务器。
对话时 Kiro CLI 自动发现并调用这些工具，完成记忆的存取。

这样 Kiro CLI 不需要改代码，AgentCore Memory 也不需要改接口，中间层就是你的 MCP 服务器。

MCP 服务器要暴露哪些工具

根据 AgentCore Memory 的能力和对话场景的实际需求，至少需要以下几类工具：

工具名	作用	对应 AgentCore 操作
`store_memory`	把当前对话的关键信息写入长期记忆	Create / Put memory entry
`retrieve_memory`	根据关键词或语义检索历史记忆	Query / Search memory
`list_sessions`	查看已有的记忆会话列表	List memory sessions
`delete_memory`	清除指定的记忆条目	Delete memory entry
`get_memory_status`	查看记忆存储用量和健康状态	Get memory usage metrics

下面直接看代码。

实现一个可运行的 MCP 服务器

这个示例用 Python 的 mcp SDK 和 boto3 实现。假设你已经有一个 Bedrock AgentCore Memory 的实例（memory ID），以及 AWS 凭证配置完毕。

先安装依赖：

pip install mcp boto3

然后创建服务器文件 kiro_memory_mcp_server.py：

import json
import boto3
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

# —— 配置区：按你的实际环境修改 ——
MEMORY_ID = "your-memory-id"          # AgentCore Memory 实例 ID
REGION = "us-east-1"                  # AWS 区域
SESSION_NAMESPACE = "kiro-cli"        # 记忆会话命名空间，区分不同项目

bedrock_client = boto3.client(
    "bedrock-agentcore-memory",       # AgentCore Memory 的 boto3 service name
    region_name=REGION
)

server = Server("kiro-memory-mcp")

# 注册工具元数据
@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="store_memory",
            description="将对话中的关键信息存入长期记忆，供后续会话检索",
            inputSchema={
                "type": "object",
                "properties": {
                    "content": {"type": "string", "description": "要存储的内容"},
                    "tags":    {"type": "array",  "items": {"type": "string"},
                                "description": "标签，便于后续过滤"}
                },
                "required": ["content"]
            }
        ),
        Tool(
            name="retrieve_memory",
            description="从长期记忆中检索与查询相关的历史信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "检索关键词或语义描述"},
                    "max_results": {"type": "integer", "description": "最多返回条数", "default": 5}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="get_memory_status",
            description="查看记忆存储的用量和会话概况",
            inputSchema={"type": "object", "properties": {}}
        ),
    ]

# —— store_memory 工具实现 ——
@server.call_tool()
async def call_tool(name, arguments):
    if name == "store_memory":
        content = arguments["content"]
        tags = arguments.get("tags", [])

        # 调用 AgentCore Memory API 写入记忆
        response = bedrock_client.create_memory_entry(
            memoryId=MEMORY_ID,
            sessionNamespace=SESSION_NAMESPACE,
            content=content,
            tags=tags
        )
        entry_id = response.get("entryId", "unknown")

        return [TextContent(type="text", text=f"已存入记忆，entry ID: {entry_id}")]

    elif name == "retrieve_memory":
        query = arguments["query"]
        max_results = arguments.get("max_results", 5)

        # 语义检索历史记忆
        response = bedrock_client.query_memory(
            memoryId=MEMORY_ID,
            query=query,
            maxResults=max_results
        )
        entries = response.get("entries", [])

        if not entries:
            return [TextContent(type="text", text="未找到相关记忆")]

        results = []
        for e in entries:
            results.append(f"[{e.get('entryId')}] {e.get('content')} (tags: {e.get('tags', [])})")

        return [TextContent(type="text", text="\n".join(results))]

    elif name == "get_memory_status":
        response = bedrock_client.get_memory_usage(
            memoryId=MEMORY_ID
        )
        status = json.dumps(response, indent=2, default=str)
        return [TextContent(type="text", text=f"记忆状态:\n{status}")]

    else:
        return [TextContent(type="text", text=f"未知工具: {name}")]

# 启动服务器（stdio 模式，Kiro CLI 通过标准输入/输出与之通信）
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, server.create_initialization_options())

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

运行前需要修改的地方：

MEMORY_ID：替换为你自己在 Bedrock AgentCore Memory 控制台创建的 memory 实例 ID。
REGION：替换为你部署 AgentCore Memory 的 AWS 区域。
boto3 的 service name：AgentCore Memory 是新服务，实际 SDK 中的 service name 请参照最新 AWS 文档确认，可能需要用 bedrock-agentcore 或通过 session.client 指定 endpoint。
AWS 凭证：确保环境变量 AWS_ACCESS_KEY_ID、AWS_SECRET_ACCESS_KEY 已设置，或使用 aws configure 配置了 profile。

在 Kiro CLI 中注册 MCP 服务器

Kiro CLI 的 MCP 配置写在项目根目录的 .kiro/mcp.json 中。把上面的服务器加进去：

{
  "mcpServers": {
    "kiro-memory": {
      "command": "python",
      "args": ["kiro_memory_mcp_server.py"],
      "env": {
        "AWS_REGION": "us-east-1"
      }
    }
  }
}

command 和 args 指定了启动 MCP 服务器的方式——Kiro CLI 会以子进程方式运行它，通过 stdin/stdin 通信。env 可以额外注入环境变量。

配置完成后，在终端启动 Kiro CLI：

kiro

进入对话后，Kiro CLI 会自动发现 store_memory、retrieve_memory、get_memory_status 三个工具。你可以直接在对话中让代理使用它们：

> 把刚才讨论的数据库连接池配置存到记忆里
> 我上次问过什么关于 Redis 部署的问题？检索一下
> 看一下当前记忆存储用了多少

代理会自行决定何时调用哪个工具，你不需要手动触发。

对话记忆的实际工作流

一次完整的跨会话记忆流程大致如下：

首次对话：你告诉代理项目的技术栈和架构偏好，代理调用 store_memory 写入 AgentCore Memory。
关闭终端：Kiro CLI 进程结束，MCP 服务器随之退出，但记忆已经持久化在 Bedrock 端。
再次打开 Kiro CLI：新会话启动，MCP 服务器重新被拉起。代理在对话开始时调用 retrieve_memory，拉回之前存的技术栈偏好。
持续积累：每次有价值的决策、配置、踩坑经验都被存入记忆，检索结果越来越精准。

这就是"对话越用越懂你"的机制——不是靠本地文件，而是靠托管服务的跨会话持久化。

需要注意的边界和取舍

记忆不是越多越好。 AgentCore Memory 有存储配额和检索延迟。存入太多碎片信息会降低检索精度，建议只存储两类内容：

重复出现的偏好（语言、框架、代码风格）
关键决策和踩坑记录（为什么选 A 不选 B）

标签是检索质量的关键。 store_memory 的 tags 参数不要偷懒。好的标签体系比如 ["architecture", "database", "redis"]，能让后续语义检索更快命中。

成本意识。 AgentCore Memory 是托管服务，按存储量和 API 调用计费。get_memory_status 工具就是为了让你随时掌握用量，避免无意识地膨胀。

安全边界。 记忆内容会经过 Bedrock 的存储管线。不要把密钥、密码、敏感客户数据存入记忆——这些应该走 Secrets Manager 或环境变量。

上手清单

如果你准备动手，按这个顺序走：

在 AWS 控制台创建一个 Bedrock AgentCore Memory 实例，拿到 memoryId。
配置好 boto3 凭证和区域。
复制上面的 MCP 服务器代码，修改配置区的三个变量。
在项目 .kiro/mcp.json 中注册服务器。
启动 Kiro CLI，进行一次对话，明确要求代理存储一条信息。
关闭终端，重新启动，要求代理检索刚才存的信息——验证跨会话生效。
用 get_memory_status 查看用量，确认一切正常。

整个过程不需要改 Kiro CLI 的源码，也不需要改 AgentCore Memory 的配置——MCP 服务器就是全部的集成工作。