动作过滤工具

基于能力和资源状态的MCP服务器动作过滤工具,用于控制AI代理可见和可执行的操作。

  • by anon767
  • Python
  • 本地部署
  • 1 星标
  • 更新于
在 GitHub 查看 anon767/mcp-action-guard

使用说明

mcp-action-guard

Capability and resource-aware action filtering for MCP servers.

AI agents discover tools via MCP's tools/list and invoke them via tools/call. This library controls what an agent can see and do based on:

  • Capabilities — what the agent is allowed to do (e.g. order:confirm)
  • Resource state — what's valid right now (e.g. only confirm pending orders)
  • Resource attributes — instance-level checks (e.g. ownership, amount thresholds)

Deny-by-default. If no rule matches, the tool is hidden and blocked.

How It Works

Agent → MCP Server → GuardedMCPServer → PolicyEngine → filtered tools

You write a JSON policy that maps (resource_type, state, capability) to tool names. The GuardedMCPServer wraps the MCP SDK and uses that policy to filter tools/list and guard tools/call.

For instance-level checks (can this agent modify this specific order?), you provide a resource predicate function — no need to encode every business rule in JSON.

Quick Start

1. Install

pip install -e .

2. Define a policy

{
  "rules": [
    {
      "resource_type": "order",
      "state": "*",
      "capability": "order:read",
      "tools": ["get_order", "list_orders"]
    },
    {
      "resource_type": "order",
      "state": "pending",
      "capability": "order:confirm",
      "tools": ["confirm_order"]
    }
  ]
}

state: "*" matches any state. Each rule lists the tool names the agent can discover/invoke when the rule matches.

3. Build an MCP server

from mcp_action_guard import GuardedMCPServer, PolicyEngine

policy = PolicyEngine.from_file("policy.json")
mcp = GuardedMCPServer(policy=policy, name="order-service")

@mcp.tool(
    name="confirm_order",
    capabilities=["order:confirm"],
    description="Confirm a pending order.",
    resource_type="order",
    input_schema={
        "type": "object",
        "properties": {"order_id": {"type": "string"}},
        "required": ["order_id"],
    },
)
async def confirm_order(order_id: str) -> str:
    return f"Order {order_id} confirmed"

if __name__ == "__main__":
    mcp.run()

4. Filter tools for an agent

When you know the agent's capabilities (from a JWT, session, gateway, etc.):

from mcp_action_guard import ResourceContext

resource = ResourceContext(resource_type="order", state="pending")

# Agent with confirm capability sees confirm_order
tools = mcp.filter_tools(["order:read", "order:confirm"], resource)

# Read-only agent sees only get_order, list_orders
tools = mcp.filter_tools(["order:read"], resource)

5. Guard a tool call

from mcp_action_guard import ResourceContext

resource = ResourceContext(resource_type="order", state="pending")

# Passes — agent has the capability and state is valid
mcp.guard_call("confirm_order", resource, ["order:confirm"])

# Raises StateConflictError — can't confirm a shipped order
resource_shipped = ResourceContext(resource_type="order", state="shipped")
mcp.guard_call("confirm_order", resource_shipped, ["order:confirm"])

# Raises InsufficientCapabilitiesError — agent lacks order:confirm
mcp.guard_call("confirm_order", resource, ["order:read"])

Resource-Level Filtering

For instance-level checks, pass a resource_predicate to the policy engine:

from mcp_action_guard import PolicyEngine, ResourceContext

def ownership_check(resource: ResourceContext, capabilities: list[str]) -> bool:
    """Only allow access if the requesting agent owns this resource."""
    owner = resource.attributes.get("owner")
    agent_id = resource.attributes.get("agent_id")
    if owner is None or agent_id is None:
        return True  # no ownership constraint
    return owner == agent_id

policy = PolicyEngine.from_file("policy.json", resource_predicate=ownership_check)

# Owner accesses their own order — predicate passes, rules evaluated normally
owned = ResourceContext("order", "pending", {"owner": "agent-A", "agent_id": "agent-A"})
policy.allowed_tools(owned, ["order:confirm"])  # ["confirm_order"]

# Different agent tries — predicate returns False, zero tools
not_owned = ResourceContext("order", "pending", {"owner": "agent-A", "agent_id": "agent-B"})
policy.allowed_tools(not_owned, ["order:confirm"])  # []

The predicate receives the full ResourceContext (including arbitrary attributes) and the agent's capabilities. Return False to deny all access to that resource, regardless of rules.

Cross-MCP Coordination

The policy engine is transport-agnostic. For multi-server setups:

order_policy = PolicyEngine.from_file("order_policy.json")
payment_policy = PolicyEngine.from_file("payment_policy.json")

order_server = GuardedMCPServer(policy=order_policy, name="orders")
payment_server = GuardedMCPServer(policy=payment_policy, name="payments")

# Capabilities come from the same source (JWT, gateway, session).
agent_caps = ["order:read", "order:confirm", "payment:charge"]

order_tools = order_server.filter_tools(agent_caps)
payment_tools = payment_server.filter_tools(agent_caps)

# Agent sees a unified, filtered tool list across both servers.
all_tools = order_tools + payment_tools

The coordination layer (gateway, orchestrator, or agent framework) extracts capabilities once and passes them to each server's filter_tools() / guard_call(). This library provides the filtering primitive, not the orchestration.

Running the Example

The repo includes an MCP order server and an OpenAI-powered agent that walks an order through its full lifecycle.

pip install -e ".[examples]"

# Run the agent (starts the MCP server as a subprocess automatically)
python -m examples.agent_client

The agent discovers tools, creates an order, then follows the available actions through pending → confirmed → shipped → delivered. At each state, the policy engine controls which tools are visible.

License

MIT