AWS Strands Agents + Neo4j 集成

简介

此示例部署了一个作为 AgentCore 运行时的 AWS Strands Agent，它通过 AgentCore 网关连接到 Neo4j。该代理使用 Cognito OAuth 进行 MCP 身份验证，并使用 AgentCore Memory 进行用户偏好跟踪。

前提条件：此堆栈要求首先部署 示例 2：带有外部 Neo4j MCP 的网关。网关堆栈提供了此代理连接所需的 Cognito OAuth 设置和 MCP 网关端点。

核心功能

  • Strands Agent 框架：支持流式响应的与模型无关的代理

  • AgentCore Gateway MCP：通过 AgentCore 网关（来自示例 2）连接到 Neo4j MCP

  • Cognito OAuth (M2M)：用于安全 MCP 访问的客户端凭证流

  • AgentCore Memory：具有按用户命名空间的内置用户偏好记忆功能

  • CDK 基础设施：完整的即代码基础设施部署

架构

User Request
    ↓
AgentCore Runtime (Strands Agent)
    ├─ Bedrock LLM (Claude / cross-region inference)
    ├─ AgentCore Memory (user preferences)
    └─ MCP Client → [OAuth token from Cognito]
                        ↓
                  AgentCore Gateway (from Sample 2)
                        ↓
                  Neo4j MCP Server
                        ↓
                  Neo4j Database

组件

  1. AgentCore Runtime — 运行从 S3 部署的 Strands 代理代码

  2. Strands Agent — 编排 LLM 调用、MCP 工具使用和内存

  3. MCP Client — 获取 Cognito OAuth 令牌并调用 AgentCore 网关

  4. AgentCore Memory — 在会话间存储和检索用户偏好

  5. Secrets Manager — 安全存储 Cognito 客户端凭证

CDK 上下文参数

上下文键 源 (Source) 描述

cognito_client_id

示例 2 堆栈输出

Cognito 应用客户端 ID

cognito_client_secret

示例 2 堆栈输出

Cognito 应用客户端密码

cognito_scope

示例 2 堆栈输出

OAuth 作用域（例如 neo4j-mcp-gateway/invoke）

cognito_token_endpoint

示例 2 堆栈输出

Cognito 令牌端点 URL

gateway_url

示例 2 堆栈输出

AgentCore 网关 MCP URL

model_id

命令行参数或默认值

用于推理的模型 ID（有关可用 ID，请参阅 https://docs.aws.amazon.com/bedrock/latest/userguide/inference-profiles-support.html），默认值为 global.anthropic.claude-sonnet-4-6

深度分析

身份验证流程

MCP 客户端 (strands_agent/mcp_client/client.py) 实现了带有令牌缓存的 OAuth 2.0 client_credentials 流

Agent invocation
    ↓
MCP Client checks token cache
    ├─ Valid & >30 s until expiry → reuse
    └─ Missing / expiring:
         ├─ Load credentials from Secrets Manager (cached after first call)
         ├─ POST client_credentials grant to Cognito
         └─ Cache access_token
    ↓
httpx.AsyncClient with "Authorization: Bearer <token>"
    ↓
AgentCore Gateway (JWT validation → Interceptor swaps for Neo4j Basic Auth)
    ↓
Neo4j MCP Server

  • 凭证来自 AgentCore 上的 Secrets Manager (SECRET_ARN)，本地开发时可回退使用环境变量

  • 使用 MCP SDK 中的 streamable_http_client，并封装在 Strands MCPClient 中

代码部署

代理作为打包到 S3 的 Python 源代码进行部署（无需容器镜像）

  1. CDK 使用 uv pip install 打包 strands_agent/（目标为 manylinux_2_17_aarch64 / Python 3.13）

  2. 包作为 CDK S3 资产上传

  3. CfnRuntime 引用 S3 位置，并设置 entry_point=main.py 和 PYTHON_3_13 运行时

  4. AgentCore 在调用时下载并运行代码

代理入口点

strands_agent/main.py 中的处理程序使用 BedrockAgentCoreApp

  • `@app.entrypoint` — 接收 payload 和 context 的异步处理程序

  • 会话 ID — 取自 context.session_id，若无则回退使用 uuid

  • 执行者 ID (Actor ID) — 有效载荷中必需；用于对每个用户的内存记录进行命名空间隔离

  • 流式传输 — agent.stream_async() 向调用者生成文本块

内存

带有 UserPreferenceMemoryStrategy 的 AgentCore 内存

  • 命名空间： /users/{actorId}/preferences/

  • 检索： 前 5 个结果，0.5 相关性阈值

  • 自定义工具： clear_preferences 允许用户重置存储的偏好设置

AgentCoreMemorySessionManager 在每次对话轮次前保存对话并检索偏好设置。

模型

  • 模型： global.anthropic.claude-sonnet-4-6（跨区域推理配置文件）

  • 身份验证： 通过运行时执行角色使用 IAM

CDK 堆栈资源

  • S3 资产 — 通过 uv 打包的代理代码 + 依赖项

  • Secrets Manager — Cognito client_id / client_secret

  • IAM 角色 — S3, Secrets Manager, Bedrock, AgentCore Memory, CloudWatch, X-Ray, 工作负载身份

  • AgentCore Memory — 带有用户偏好策略的 CfnMemory

  • AgentCore Runtime — CfnRuntime（基于代码，HTTP 协议，公网）

环境变量

变量 描述

SECRET_ARN

Cognito 凭证的 Secrets Manager ARN

BEDROCK_AGENTCORE_MEMORY_ID

AgentCore Memory 资源 ID

GATEWAY_URL

AgentCore 网关 MCP 端点

COGNITO_SCOPE

网关的 OAuth 作用域

COGNITO_TOKEN_ENDPOINT

Cognito 令牌端点 URL

AWS_DEFAULT_REGION

AWS 区域（为 AgentCore 显式设置）

MODEL_ID

用于推理的模型 ID（有关可用 ID，请参阅 https://docs.aws.amazon.com/bedrock/latest/userguide/inference-profiles-support.html），默认值为 global.anthropic.claude-sonnet-4-6

可用的 MCP 工具

请参阅 官方 Neo4j MCP 服务器文档。

如何使用此示例

先决条件

第 1 步：克隆仓库

git clone https://github.com/neo4j-labs/neo4j-agent-integrations.git
cd neo4j-agent-integrations/aws-strands-agents

第 2 步：安装依赖项

pip install -r requirements.txt

第 3 步：获取示例 2 的输出

从已部署的网关堆栈中检索 Cognito 和网关值

STACK_NAME=Neo4jAgentCoreGatewayStack

GATEWAY_URL=$(aws cloudformation describe-stacks \
  --stack-name "$STACK_NAME" \
  --query "Stacks[0].Outputs[?OutputKey=='GatewayUrl'].OutputValue" \
  --output text)

COGNITO_CLIENT_ID=$(aws cloudformation describe-stacks \
  --stack-name "$STACK_NAME" \
  --query "Stacks[0].Outputs[?OutputKey=='CognitoAppClientId'].OutputValue" \
  --output text)

COGNITO_TOKEN_ENDPOINT=$(aws cloudformation describe-stacks \
  --stack-name "$STACK_NAME" \
  --query "Stacks[0].Outputs[?OutputKey=='CognitoTokenEndpoint'].OutputValue" \
  --output text)

COGNITO_SCOPE=$(aws cloudformation describe-stacks \
  --stack-name "$STACK_NAME" \
  --query "Stacks[0].Outputs[?OutputKey=='CognitoScope'].OutputValue" \
  --output text)

COGNITO_USER_POOL_ID=$(aws cloudformation describe-stacks \
  --stack-name "$STACK_NAME" \
  --query "Stacks[0].Outputs[?OutputKey=='CognitoUserPoolId'].OutputValue" \
  --output text)

COGNITO_CLIENT_SECRET=$(aws cognito-idp describe-user-pool-client \
  --user-pool-id "$COGNITO_USER_POOL_ID" \
  --client-id "$COGNITO_CLIENT_ID" \
  --query "UserPoolClient.ClientSecret" \
  --output text)

第 4 步：部署基础设施

# Bootstrap CDK (first time only)
cdk bootstrap

# Deploy the stack
cdk deploy Neo4jStrandsAgentStack \
  -c cognito_client_id="$COGNITO_CLIENT_ID" \
  -c cognito_client_secret="$COGNITO_CLIENT_SECRET" \
  -c cognito_scope="$COGNITO_SCOPE" \
  -c cognito_token_endpoint="$COGNITO_TOKEN_ENDPOINT" \
  -c gateway_url="$GATEWAY_URL" \
  -c model_id="global.anthropic.claude-sonnet-4-6"

堆栈输出

输出 描述

AgentRuntimeArn

已部署的 AgentCore 运行时的 ARN

AgentRuntimeRoleArn

运行时的 IAM 角色 ARN

CognitoSecretArn

Cognito 凭证密钥的 ARN

AgentCoreMemoryId

AgentCore Memory 资源 ID

第 5 步：测试代理

打开 demo.ipynb，将 arn 变量设置为 CDK 输出中的 AgentRuntimeArn，然后运行笔记本。

第 6 步：清理

cdk destroy Neo4jStrandsAgentStack

参考资料

AWS 文档

Neo4j 资源

© . This site is unofficial and not affiliated with Neo4j, Inc.