Blueprint: The Governed Swarm

Inter-Agent Permission Management in Multi-Agent Workflows

In a multi-agent swarm (e.g., using LangGraph or CrewAI), agents often delegate tasks to each other. Governance must ensure that a "Writer" agent cannot accidentally or maliciously call a "Publisher" tool, or that a supervisor can only delegate high-privilege tasks to verified worker agents.

This blueprint demonstrates Recursive Tool-Call Gating in a multi-agent topology.

Architecture

1. Master Policy Definition

Define separate principal identities for each agent in the swarm.

{
  "name": "swarm-topology-policy",
  "priority": 6000,
  "rules": [
    {
      "id": "writer-read-only",
      "effect": "allow",
      "principals": ["agent:writer"],
      "actions": ["cms:read", "search:*"],
      "resources": ["*"]
    },
    {
      "id": "publisher-full-access",
      "effect": "allow",
      "principals": ["agent:publisher"],
      "actions": ["cms:*"],
      "resources": ["*"]
    },
    {
      "id": "deny-writer-publish",
      "effect": "deny",
      "principals": ["agent:writer"],
      "actions": ["cms:publish", "cms:delete"],
      "resources": ["*"]
    }
  ]
}

2. Implementation

LangGraph (Stateful Swarm)

from typing import TypedDict, Annotated, Sequence
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    messages: Sequence[str]
    current_agent: str

def get_client(agent_name: str):
    return ChatOpenAI(
        model="gpt-4o",
        openai_api_base="http://cz-gateway:8001/v1",
        openai_api_key="cz_live_swarm_key",
        default_headers={
            "X-ControlZero-Agent-ID": agent_name,
            "X-ControlZero-User-Group": "swarm-v1"
        }
    )

def writer_node(state):
    client = get_client("writer")
    # If the writer tries to use a 'publish' tool, CZ will block it
    # even if the supervisor 'told' it to.
    response = client.invoke("Draft and publish a blog post about AI safety.")
    return {"messages": [response.content], "current_agent": "writer"}

def publisher_node(state):
    client = get_client("publisher")
    response = client.invoke("Publish the draft.")
    return {"messages": [response.content], "current_agent": "publisher"}

# Define Graph
workflow = StateGraph(AgentState)
workflow.add_node("writer", writer_node)
workflow.add_node("publisher", publisher_node)
workflow.set_entry_point("writer")
workflow.add_edge("writer", "publisher")
workflow.add_edge("publisher", END)

graph = workflow.compile()

# Execution
# Scenario: The 'writer' node is internally protected.
# Even if its prompt is hijacked, it cannot call 'cms:publish'.
graph.invoke({"messages": [], "current_agent": "init"})

3. Validation Checklist

• Recursive Gating: Verify that every hop in the LangGraph chain is individually audited in ClickHouse with the correct agent_id.
• Cross-Agent Isolation: Ensure that a failure in the writer node (due to policy block) correctly triggers the error handling in the graph state.
• Dynamic Re-routing: Implement a conditional edge that routes tasks to different agents based on the policy_decision received from the Gateway.