AI代理策略管理服务器

PolicyGuard是一个基于MCP协议的服务器,为AI代理提供策略访问控制、事件跟踪和合规性监控功能。

  • by PrateekKumar1709
  • Python
  • 本地部署
  • 1 星标
  • 更新于
在 GitHub 查看 PrateekKumar1709/policyguard

使用说明

PolicyGuard

用于AI代理的安全与治理MCP服务器

PolicyGuard是一个MCP(模型上下文协议)服务器,为AI代理提供基于策略的访问控制、事件跟踪和合规性监控。

注: 该项目展示了与 凯根特 (AI代理平台),可以扩展 kgateway (API网关),以获得额外的网络级安全性。


目录


概述

随着人工智能代理变得更加自主,组织需要控制来管理他们的行为。PolicyGuard是我的第一个MCP服务器项目,旨在探索如何在MCP层实现安全和治理。

问题

人工智能代理可以调用他们可以访问的任何工具。没有治理:

  • 代理可能会执行破坏性操作
  • 没有代理行为的审计跟踪
  • 无法执行安全策略
  • 无法了解合规性

解决方案

PolicyGuard添加了一个代理调用的安全层 之前 采取行动:

User Request → AI Agent → PolicyGuard (validate_action) → Allowed/Denied

特性

特性 描述
策略强制 根据安全规则验证操作
信任级别 低、中、高管理层次结构
模式匹配 通配符模式,如 delete_*, *_production
自动注册 未知代理的信任度极低
事件跟踪 自动违规记录
审计跟踪 完整的行动历史记录
合规仪表板 安全指标一览

建筑

┌─────────────────────────────────────────────────────────────────┐
│                         AI Agent / LLM                          │
│                                                                 │
│  "Before any action, call validate_action to check permission" │
└─────────────────────────────────────────────────────────────────┘
                              │
                              │ MCP Protocol
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                     PolicyGuard MCP Server                      │
│                                                                 │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐             │
│  │  validate   │  │   create    │  │   report    │             │
│  │   action    │  │   policy    │  │  incident   │             │
│  └─────────────┘  └─────────────┘  └─────────────┘             │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐             │
│  │  register   │  │  get_audit  │  │    get      │             │
│  │   agent     │  │    log      │  │ compliance  │             │
│  └─────────────┘  └─────────────┘  └─────────────┘             │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
                    ┌───────────────────┐
                    │   JSON Storage    │
                    │   (policies,      │
                    │    agents,        │
                    │    audit_log,     │
                    │    incidents)     │
                    └───────────────────┘

MCP工具

PolicyGuard通过MCP公开了6个工具:

1. validate_action ⭐ 主要工具

在执行操作之前,检查是否允许该操作。

{
  "action_type": "tool_call",
  "target": "delete_records",
  "agent_id": "my-agent"
}

答复:

{
  "action_id": "act_a1b2c3d4e5f6",
  "allowed": false,
  "reason": "Delete operations require admin trust level"
}

2. register_agent

注册具有信任级别的代理。

{
  "agent_id": "data-processor",
  "name": "Data Agent",
  "trust_level": "medium"
}

3. create_policy

创建安全规则。

{
  "policy_id": "block-deletes",
  "name": "Block Deletes",
  "rules": "[{\"condition\": {\"tool_pattern\": \"delete_*\"}, \"action\": \"deny\"}]"
}

4. get_audit_log

查询操作历史记录。

5. get_compliance_status

获取安全仪表板指标。

6. report_incident

手动报告安全事件。


快速开始

先决条件

  • Python 3.10+
  • 点

本地安装

# Clone
git clone https://github.com/PrateekKumar1709/policyguard.git
cd policyguard

# Setup
python3 -m venv .venv
source .venv/bin/activate
pip install -e .

# Run
python src/main.py

测试工具

import json
from src.tools.validate_action import validate_action
from src.tools.register_agent import register_agent

# Register an agent
result = json.loads(register_agent.fn(
    agent_id="test-agent",
    name="Test Agent",
    trust_level="medium"
))
print(f"Registered: {result['agent_id']}")

# Validate an action
result = json.loads(validate_action.fn(
    action_type="tool_call",
    target="read_data",
    agent_id="test-agent"
))
print(f"Allowed: {result['allowed']}")

HTTP模式

python src/main.py --transport http --port 8000

Kubernetes部署

PolicyGuard包括一个用于Kubernetes部署的Helm图表。

使用种类

# Create cluster
kind create cluster --name policyguard

# Build and load image
docker build -t policyguard:latest .
kind load docker-image policyguard:latest --name policyguard

# Deploy
kubectl create namespace policyguard
helm install policyguard ./helm/policyguard -n policyguard

# Verify
kubectl get pods -n policyguard

端口转发

kubectl port-forward -n policyguard svc/policyguard 8000:8000

测试

单元测试

pytest tests/ -v

测试结果

tests/test_tools.py::TestValidateAction::test_allow_read_for_low_trust PASSED
tests/test_tools.py::TestValidateAction::test_deny_delete_for_low_trust PASSED
tests/test_tools.py::TestValidateAction::test_allow_delete_for_admin PASSED
tests/test_tools.py::TestValidateAction::test_deny_suspended_agent PASSED
tests/test_tools.py::TestRegisterAgent::test_register_new_agent PASSED
tests/test_tools.py::TestCreatePolicy::test_create_valid_policy PASSED
... (16 tests total)
============================== 16 passed ==============================

E2E测试输出

[1/6] register_agent      ✅ SUCCESS
[2/6] create_policy       ✅ SUCCESS  
[3/6] validate_action     ✅ ALLOWED (read_data)
[4/6] validate_action     ✅ DENIED (delete_records)
[5/6] report_incident     ✅ SUCCESS
[6/6] get_compliance_status ✅ SUCCESS

ALL 6 MCP TOOLS WORKING!

kagent集成

PolicyGuard与 凯根特 用于Kubernetes原生AI代理管理。

截图

代理列表 -PolicyGuard代理已注册并准备就绪:

代理工具 -6个可用的PolicyGuard MCP工具:

合规仪表板 -询问合规状态:

政策验证 -根据策略拒绝操作:

快速设置

# 1. Install kagent CRDs
helm install kagent-crds ./kagent-reference/helm/kagent-crds -n kagent --create-namespace

# 2. Install kagent with Ollama (free local LLM)
helm upgrade --install kagent ./kagent-reference/helm/kagent -n kagent \
  --set providers.default=ollama \
  --set providers.ollama.model=qwen2.5:1.5b \
  --set tag=0.7.13

# 3. Deploy PolicyGuard
helm install policyguard ./helm/policyguard -n policyguard --create-namespace

# 4. Create RemoteMCPServer
kubectl apply -f - <<EOF
apiVersion: kagent.dev/v1alpha2
kind: RemoteMCPServer
metadata:
  name: policyguard
  namespace: kagent
spec:
  protocol: STREAMABLE_HTTP
  url: http://policyguard.policyguard:8000/mcp
  timeout: 30s
EOF

# 5. Create PolicyGuard Agent
kubectl apply -f - <<EOF
apiVersion: kagent.dev/v1alpha2
kind: Agent
metadata:
  name: policyguard-agent
  namespace: kagent
spec:
  description: "Security agent using PolicyGuard"
  type: Declarative
  declarative:
    modelConfig: "default-model-config"
    systemMessage: |
      You are a PolicyGuard Security Agent. Use the PolicyGuard tools to:
      - validate_action: Check if actions are allowed
      - register_agent: Register new agents
      - create_policy: Define security rules
      - get_compliance_status: View compliance dashboard
    tools:
    - type: McpServer
      mcpServer:
        name: policyguard
        kind: RemoteMCPServer
        apiGroup: kagent.dev
        toolNames:
        - validate_action
        - register_agent
        - create_policy
        - get_audit_log
        - get_compliance_status
        - report_incident
EOF

已验证工作

# Check status
$ kubectl get agents,remotemcpservers -n kagent
NAME                TYPE          READY   ACCEPTED
policyguard-agent   Declarative   True    True

NAME          PROTOCOL          URL                                       ACCEPTED
policyguard   STREAMABLE_HTTP   http://policyguard.policyguard:8000/mcp   True

# Invoke agent via CLI
$ kagent invoke -t "Get compliance status" --agent policyguard-agent

Response:
- Security Posture: Healthy
- Total Policies: 1, Enabled: 1  
- Total Incidents: 0
- Agents Registered: 0

# Test validation
$ kagent invoke -t "Can test-agent delete the database?" --agent policyguard-agent

Response:
The action delete_database was not allowed for test-agent.
Reason: Delete operations require admin trust level.

访问kagent用户界面

kubectl port-forward -n kagent svc/kagent-ui 8080:8080
# Open http://localhost:8080

安全模型

信任级别

级别 分数 用例
low 1 未知代理,只读
medium 2 已验证的代理
high 3 受信任的代理
admin 4 完全访问

政策规则

{
  "condition": {
    "tool_pattern": "delete_*",
    "trust_level_below": "admin"
  },
  "action": "deny",
  "message": "Delete requires admin"
}

赋值顺序

  1. 特工停职? → DENY
  2. 工具是否在拒绝列表中? → DENY
  3. 工具不在允许列表中? → DENY
  4. 政策匹配? → 应用规则
  5. 默认→ ALLOW

项目结构

policyguard/
├── helm/
│   └── policyguard/          # Helm chart
│       ├── Chart.yaml
│       ├── values.yaml
│       └── templates/
├── src/
│   ├── main.py               # Entry point
│   ├── core/
│   │   ├── server.py         # MCP server
│   │   └── utils.py          # Utilities
│   └── tools/
│       ├── validate_action.py
│       ├── register_agent.py
│       ├── create_policy.py
│       ├── get_audit_log.py
│       ├── get_compliance_status.py
│       └── report_incident.py
├── tests/
│   └── test_tools.py         # 16 unit tests
├── Dockerfile
├── pyproject.toml
└── README.md

使用的技术

  • FastMCP -Python MCP服务器SDK
  • 派丹蒂克 -数据验证
  • 舵 -Kubernetes打包
  • 测试 -测试

未来改进

  • []数据库后端(PostgreSQL)
  • []Web仪表板用户界面
  • []普罗米修斯指标
  • []RBAC集成
  • []策略版本控制

许可证

MIT许可证


黑客松

MCP_HACK//26 -“MCP和AI代理入门课程”

这是我的第一个MCP服务器项目!我创建PolicyGuard是为了学习:

  • MCP服务器的工作原理
  • 如何将工具暴露给AI代理
  • 如何在Kubernetes上部署MCP服务器
  • 如何在MCP层实现安全/治理

我建造了什么

  • ✅ 用于安全治理的6个MCP工具
  • ✅ 具有模式匹配的策略引擎
  • ✅ 信任级别系统
  • ✅ 审计日志和事件跟踪
  • ✅ Kubernetes的Helm图
  • ✅ 16个单元测试
  • ✅ kagent集成示例