使用说明
AWS基岩ARC MCP服务器
一种模型上下文协议(MCP)服务器,提供对AWS基岩自动推理检查(ARC)的标准化访问。该服务器使AI代理和应用程序能够以数学精度根据形式逻辑策略验证内容,实现高达99%的验证准确率。
特性
- 符合MCP标准的服务器:将ARC验证作为标准化的MCP工具公开
- AWS基岩集成:与AWS基岩护栏和ARC直接集成
- 智能响应重写:自动重写LLM响应以修复策略冲突
- 轻松部署:只需一个命令即可部署到Amazon Bedrock AgentCore运行时
- OAuth 2.0身份验证:通过Amazon Cognito进行安全身份验证
- 综合工具:验证、重写、列出护栏并检索护栏详细信息
- 无服务器且可扩展:基于AWS无服务器基础架构构建
用例
- 根据业务规则进行LLM输出验证
- 受监管行业的内容验证
- 多步推理验证
- 策略合规性检查
- 基于数学证明的幻觉检测
建筑
MCP Clients (Claude, Cursor, Custom Agents)
↓ HTTP/SSE (OAuth Bearer)
Amazon Bedrock AgentCore Runtime
↓ MCP Protocol
ARC MCP Server (Python + FastMCP)
↓ AWS SDK (boto3)
AWS Bedrock Guardrails + ARC Engine
先决条件
- AWS帐户 具有适当的权限
- AWS-CLI 已配置凭据
- Python 3.11+
- 基岩试剂核心入门工具包:
pip install bedrock-agentcore-starter-toolkit - 至少一个具有ARC策略的AWS基岩护栏 (用于测试)
快速开始
1.克隆存储库
git clone https://github.com/your-org/ARC-MCP.git
cd ARC-MCP
2.部署AWS基础设施
这将部署IAM执行角色和Cognito用户池进行身份验证。
部署前检查表
在运行部署脚本之前,请决定:
-
Cognito用户池:您想创建新的用户池还是使用现有的用户池?
- 创建新项:该脚本将为此MCP服务器创建一个专用的Cognito用户池
- 使用现有:重用现有的用户池(如果您已经配置了用户,则很有用)
- 要列出现有的用户池,请执行以下操作:
aws cognito-idp list-user-pools --max-results 20 --region us-east-1
- 要列出现有的用户池,请执行以下操作:
-
测试用户:是否要在部署期间创建测试用户?
- 是:该脚本将使用您提供的凭据创建测试用户
- 不:您需要手动创建用户或使用用户池中的现有用户
部署基础设施
# Set your AWS region (optional, defaults to us-east-1)
export AWS_REGION=us-east-1
# Run the interactive deployment script
./scripts/deploy-infrastructure.sh
脚本将提示您:
- 是创建新的Cognito用户池还是使用现有的用户池
- 如果使用现有:用户池ID
- 是否创建测试用户
- 如果创建测试用户:用户名和密码
这将创建:
- 具有基岩ARC权限的IAM执行角色
- 用于OAuth身份验证的Cognito用户池(如果创建新的)
- Cognito用户池客户端
- 测试用户帐户(如果需要)
- CloudWatch日志组
部署后,脚本将配置详细信息保存到 deployment-info.txt.
手动用户创建(可选)
如果在部署期间选择不创建测试用户,则可以手动创建一个:
# Get your User Pool ID from deployment-info.txt or CloudFormation outputs
USER_POOL_ID="<your-pool-id>"
# Create user
aws cognito-idp admin-create-user \
--user-pool-id $USER_POOL_ID \
--username testuser \
--user-attributes Name=email,Value=testuser@example.com Name=email_verified,Value=true \
--message-action SUPPRESS \
--region us-east-1
# Set permanent password
aws cognito-idp admin-set-user-password \
--user-pool-id $USER_POOL_ID \
--username testuser \
--password YourPassword123! \
--permanent \
--region us-east-1
3.将MCP服务器部署到AgentCore
./scripts/deploy-agentcore.sh
这将:
- 安装Python依赖项
- 可选本地测试
- 部署到Amazon Bedrock AgentCore运行时
4.获取身份验证令牌
./scripts/get-bearer-token.sh
这将从Cognito检索承载令牌,以便与MCP服务器进行身份验证。令牌保存到 bearer-token.txt 并显示在终端中。
5.测试服务器
# Export the bearer token
export BEARER_TOKEN="<your-token-from-previous-step>"
# Run tests
./scripts/test-mcp-server.sh
MCP工具
服务器公开了四个MCP工具:
1. validate_content
根据AWS Bedrock ARC策略验证文本内容。
参数:
guardrail_id(字符串,必填):护栏标识符(ID或ARN)content(字符串,必填):要验证的文本内容guardrail_version(字符串,可选):版本号或“草稿”(默认值:“草稿”)source(字符串,可选):“输入”或“输出”(默认值:“输出”)
例子:
result = await mcp_client.call_tool(
"validate_content",
{
"guardrail_id": "abc123xyz",
"content": "Employee with 8 months tenure is eligible for benefits",
"guardrail_version": "1",
"source": "OUTPUT"
}
)
答复:
{
"action": "NONE",
"valid": true,
"assessments": {
"automatedReasoningPolicy": {
"findings": [{
"result": "VALID",
"variables": {"tenure_months": 8},
"appliedRules": ["MinimumTenureRule"],
"explanation": "Statement verified..."
}]
}
},
"usage": {
"automatedReasoningPolicyUnits": 3,
"processingTimeMs": 245
}
}
2. list_guardrails
列出可用的AWS基岩护栏和ARC策略。
参数:
max_results(整数,可选):返回的最大数字(默认值:20,最大值:100)
例子:
result = await mcp_client.call_tool("list_guardrails", {"max_results": 10})
3. get_guardrail_info
获取特定护栏的详细信息。
参数:
guardrail_id(字符串,必填):护栏标识符version(字符串,可选):版本或“草稿”(默认值:“草稿”)
例子:
result = await mcp_client.call_tool(
"get_guardrail_info",
{"guardrail_id": "abc123xyz", "version": "1"}
)
4. rewrite_response
验证LLM响应,并根据ARC发现自动重写。此工具结合了验证和智能重写来修复策略违规。
参数:
user_query(string,必填):原始用户问题/提示llm_response(string,必填):LLM对验证和重写的响应guardrail_id(字符串,必填):护栏标识符(ID或ARN)guardrail_version(字符串,可选):版本号或“草稿”(默认值:“草稿”)model_id(字符串,可选):用于重写的模型ID(默认值:“anthropic.claude-3-5-connect-20241022-v2:0”)domain(字符串,可选):域上下文,如“医疗保健”、“金融”(默认值:“常规”)policy_definition(字符串,可选):用于其他上下文的可选策略文本
例子:
result = await mcp_client.call_tool(
"rewrite_response",
{
"user_query": "What are the eligibility requirements for benefits?",
"llm_response": "Employees can apply for benefits immediately after joining.",
"guardrail_id": "abc123xyz",
"guardrail_version": "1",
"domain": "Human Resources"
}
)
答复:
{
"query": "What are the eligibility requirements for benefits?",
"original_response": "Employees can apply for benefits immediately after joining.",
"rewritten_response": "Employees are eligible for benefits after completing 6 months of continuous employment, as required by company policy.",
"rewritten": true,
"findings": [
{
"result": "INVALID",
"explanation": "Response violates minimum tenure requirement",
"violations": ["Incorrect tenure period stated"],
"suggestions": ["Must specify 6-month minimum requirement"]
}
],
"finding_types": ["INVALID"],
"findings_count": 1,
"domain": "Human Resources",
"message": "Successfully rewrote response for INVALID"
}
查找已处理的类型:
- 无效:完全违反政策-重写以遵守规则
- 可满足式:技术上正确但不完整-提高清晰度
- 无数据:没有支持数据的索赔-添加适当的资格
- 翻译歧义:模棱两可的陈述-澄清语言
- TOO_COMPLEX:过于复杂的问题-建议简化
- 有效:未发现问题-返回原始响应不变
集成示例
克劳德桌面
添加到您的Claude Desktop配置(claude_desktop_config.json):
{
"mcpServers": {
"bedrock-arc-validator": {
"type": "streamable-http",
"url": "https://bedrock-agentcore.us-east-1.amazonaws.com/runtimes/YOUR_ENCODED_ARN/invocations?qualifier=DEFAULT",
"headers": {
"authorization": "Bearer YOUR_BEARER_TOKEN",
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream"
}
}
}
}
Python MCP客户端
import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
async def validate_with_arc():
mcp_url = "https://bedrock-agentcore.us-east-1.amazonaws.com/runtimes/YOUR_ENCODED_ARN/invocations?qualifier=DEFAULT"
headers = {
"authorization": f"Bearer {bearer_token}",
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream"
}
async with streamablehttp_client(mcp_url, headers) as (read_stream, write_stream, _):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
result = await session.call_tool(
"validate_content",
{
"guardrail_id": "abc123xyz",
"content": "Your content here",
"guardrail_version": "DRAFT"
}
)
print(f"Valid: {result['valid']}")
asyncio.run(validate_with_arc())
项目结构
ARC-MCP/
├── main.py # MCP server entry point
├── handlers/
│ ├── __init__.py
│ ├── validation.py # ARC validation logic
│ ├── discovery.py # Guardrail discovery
│ ├── rewrite_handler.py # Response rewriting orchestration
│ ├── response_rewriter.py # ResponseRewriter class
│ ├── rewrite_utils.py # Finding processing utilities
│ └── template_manager.py # Prompt template management
├── response_rewriting_prompts/ # Rewriting prompt templates
│ ├── INVALID.txt # Template for INVALID findings
│ ├── SATISFIABLE.txt # Template for SATISFIABLE findings
│ ├── NO_DATA.txt # Template for NO_DATA findings
│ └── TRANSLATION_AMBIGUOUS.txt # Template for ambiguous findings
├── tests/
│ ├── __init__.py
│ ├── test_validation.py # Validation tests
│ ├── test_discovery.py # Discovery tests
│ ├── test_rewrite_utils.py # Rewrite utilities tests
│ └── test_response_rewriter.py # Response rewriter tests
├── scripts/
│ ├── deploy-infrastructure.sh # Deploy CloudFormation
│ ├── deploy-agentcore.sh # Deploy to AgentCore
│ ├── get-bearer-token.sh # Get auth token
│ └── test-mcp-server.sh # Test deployed server
├── .agentcore/
│ └── config.json # AgentCore configuration
├── requirements.txt # Python dependencies
├── cloudformation-template.yaml # Infrastructure as Code
├── pytest.ini # Pytest configuration
└── README.md # This file
发展
再进行
pip install -r requirements.txt
# For development
pip install pytest pytest-asyncio
运行测试
# Unit tests
pytest
# Specific test file
pytest tests/test_validation.py
# With coverage
pytest --cov=handlers
地方发展
# Run server locally
python main.py
# Or with agentcore CLI
agentcore launch --local
配置
环境变量
AWS_REGION:AWS区域(默认:us-east-1)EXECUTION_ROLE_ARN:IAM执行角色ARNCOGNITO_USER_POOL_ID:Cognito用户池IDCOGNITO_CLIENT_ID:Cognito客户端IDBEARER_TOKEN:用于身份验证的OAuth承载令牌
代理核心配置
编辑 .agentcore/config.json 自定义:
- 运行时设置
- 执行角色
- OAuth授权人
- 环境变量
安全
认证流程
- 客户端通过Cognito进行身份验证
- Cognito返回JWT令牌(IdToken)
- 客户端在授权标头中包含令牌
- AgentCore验证JWT
- MCP服务器为AWS API调用使用IAM执行角色
最佳实践
- 从不硬编码AWS凭据
- 使用IAM执行角色
- 定期旋转Cognito代币
- IAM权限遵循最小特权原则
- 监控CloudWatch日志以查找可疑活动
监控
CloudWatch指标
服务器向CloudWatch发送指标:
InvocationCount:MCP工具调用总数InvocationErrors:工具调用失败InvocationDuration:工具执行时间
日志
在CloudWatch中查看日志:
aws logs tail /aws/bedrock-agentcore/arc-mcp-server --follow
成本跟踪
ARC定价:每1000个文本单位0.17美元(1个文本单位=最多1000个字符)
通过CloudWatch指标监控使用情况或检查 usage 验证响应中的字段。
故障排除
常见问题
“ResourceNotFoundException:找不到护栏”
- 验证护栏ID是否正确
- 确保IAM角色可以访问护栏
- 检查护栏是否配置了ARC策略
“AccessDeniedException”
- 验证IAM执行角色是否具有所需权限
- 检查CloudFormation堆栈是否已成功部署
- 审查IAM政策
cloudformation-template.yaml
“令牌已过期”
- Cognito代币在60分钟后过期
- 跑
./scripts/get-bearer-token.sh获取新令牌
“找不到代理”
- 确保AgentCore部署成功完成
- 跑
agentcore status --agent arc-mcp-server检查状态
调试模式
启用调试登录 main.py:
logging.basicConfig(level=logging.DEBUG)
boto3.set_stream_logger('', logging.DEBUG)
贡献
- 分叉存储库
- 创建要素分支:
git checkout -b feature-name - 进行更改
- 运行测试:
pytest - 承诺:
git commit -am 'Add feature' - 推:
git push origin feature-name - 创建拉取请求
许可证
看 许可证 文件以获取详细信息。
资源
支持
对于问题、疑问或贡献:
- 在GitHub上打开一个问题
- 检查 设计文档 详细规格
- 查看AWS基岩文档
致谢
内置:
