ARC验证服务器

AWS Bedrock ARC MCP服务器是一个提供标准化访问AWS Bedrock自动推理检查(ARC)的服务器,使AI代理和应用程序能够以数学精度验证内容是否符合正式逻辑策略。

  • by jacksodj
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 jacksodj/ARC-MCP

使用说明

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用户池进行身份验证。

部署前检查表

在运行部署脚本之前,请决定:

  1. Cognito用户池:您想创建新的用户池还是使用现有的用户池?

    • 创建新项:该脚本将为此MCP服务器创建一个专用的Cognito用户池
    • 使用现有:重用现有的用户池(如果您已经配置了用户,则很有用)
      • 要列出现有的用户池,请执行以下操作: aws cognito-idp list-user-pools --max-results 20 --region us-east-1
  2. 测试用户:是否要在部署期间创建测试用户?

    • 是:该脚本将使用您提供的凭据创建测试用户
    • 不:您需要手动创建用户或使用用户池中的现有用户

部署基础设施

# 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执行角色ARN
  • COGNITO_USER_POOL_ID:Cognito用户池ID
  • COGNITO_CLIENT_ID:Cognito客户端ID
  • BEARER_TOKEN:用于身份验证的OAuth承载令牌

代理核心配置

编辑 .agentcore/config.json 自定义:

  • 运行时设置
  • 执行角色
  • OAuth授权人
  • 环境变量

安全

认证流程

  1. 客户端通过Cognito进行身份验证
  2. Cognito返回JWT令牌(IdToken)
  3. 客户端在授权标头中包含令牌
  4. AgentCore验证JWT
  5. 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)

贡献

  1. 分叉存储库
  2. 创建要素分支: git checkout -b feature-name
  3. 进行更改
  4. 运行测试: pytest
  5. 承诺: git commit -am 'Add feature'
  6. 推: git push origin feature-name
  7. 创建拉取请求

许可证

看 许可证 文件以获取详细信息。

资源

支持

对于问题、疑问或贡献:

  • 在GitHub上打开一个问题
  • 检查 设计文档 详细规格
  • 查看AWS基岩文档

致谢

内置: