Observed since 20 Oct, 21:42 UTC
No superseded metadata states collected. Unchanged observations do not create new revisions.
Layer 3: The Framework - A governance and management layer for autonomous AI agents
A governance and management layer for autonomous AI agents. The Agent Control Plane treats the LLM as a raw compute component and provides a kernel-like layer for safe, controlled execution.
🎯 Benchmark Results: The Control Plane achieves 0% safety violations vs 26.67% for prompt-based safety, with 98% fewer tokens. See comparative study →
🎥 Demo Video: Watch 2-minute demo → | Full tutorial (12 min) →
We need to stop treating the LLM as a magic box and start treating it as a raw compute component that requires a kernel.
In distributed systems, we don't ask a microservice nicely to respect a rate limit—we enforce it at the gateway. We don't ask a database query nicely not to drop a table—we enforce it via permissions. With AI agents, we need the same deterministic enforcement.
As we move from chatbots to autonomous agents—systems that can execute code, modify data, and trigger workflows—the biggest bottleneck isn't intelligence. It's governance. The Agent Control Plane solves this by providing:
Traditional LLM applications lack proper governance:
The Agent Control Plane sits between the LLM (raw compute) and the execution environment, providing:
# Install the latest stable version
pip install agent-control-plane
# Or install with development dependencies
pip install agent-control-plane[dev]
# Clone the repository
git clone https://github.com/imran-siddique/agent-control-plane.git
cd agent-control-plane
# Install the package in editable mode
pip install -e .
# Or install with development dependencies
pip install -e ".[dev]"
agent-control-plane/
├── src/
│ └── agent_control_plane/ # Main package source code
│ ├── agent_kernel.py # Core kernel functionality
│ ├── control_plane.py # Main control plane interface
│ ├── adapter.py # OpenAI SDK adapter (drop-in middleware)
│ ├── policy_engine.py # Policy enforcement
│ ├── execution_engine.py # Safe execution
│ ├── constraint_graphs.py # Multi-dimensional context
│ ├── shadow_mode.py # Simulation mode
│ ├── mute_agent.py # Capability-based execution
│ ├── supervisor_agents.py # Recursive governance
│ └── flight_recorder.py # Audit logging (SQLite)
├── tests/ # Test suite
├── examples/ # Example scripts
├── benchmark/ # Red team safety benchmarks
├── docs/ # Documentation
└── README.md # This file
from agent_control_plane import AgentControlPlane, create_standard_agent
from agent_control_plane.agent_kernel import ActionType
# Create the control plane
control_plane = AgentControlPlane()
# Create an agent with standard permissions
agent = create_standard_agent(control_plane, "my-agent")
# Execute an action
result = control_plane.execute_action(
agent,
ActionType.FILE_READ,
{"path": "/data/myfile.txt"}
)
if result["success"]:
print(f"Result: {result['result']}")
else:
print(f"Error: {result['error']}")
NEW: Zero-friction integration! Wrap your OpenAI client to automatically govern LLM tool calls:
from openai import OpenAI
from agent_control_plane import (
AgentControlPlane,
create_governed_client,
ActionType,
PermissionLevel
)
# Standard setup
control_plane = AgentControlPlane()
client = OpenAI(api_key="your-key")
# One line to create governed client
governed = create_governed_client(
control_plane=control_plane,
agent_id="my-agent",
openai_client=client,
permissions={
ActionType.DATABASE_QUERY: PermissionLevel.READ_ONLY,
ActionType.FILE_READ: PermissionLevel.READ_ONLY,
}
)
# Use exactly like normal OpenAI SDK!
response = governed.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "Query database and save results"}],
tools=[...]
)
# Tool calls are automatically governed - unauthorized actions are blocked!
📖 See the OpenAI Adapter Guide for comprehensive integration instructions.
The Agent Control Plane now supports multiple AI frameworks and protocols with the same governance approach:
from langchain.chat_models import ChatOpenAI
from langchain.agents import initialize_agent, load_tools
from agent_control_plane import (
AgentControlPlane,
create_governed_langchain_client,
ActionType,
PermissionLevel
)
# Setup
control_plane = AgentControlPlane()
llm = ChatOpenAI(temperature=0)
# Create governed LangChain client
governed_llm = create_governed_langchain_client(
control_plane=control_plane,
agent_id="my-langchain-agent",
langchain_client=llm,
permissions={
ActionType.FILE_READ: PermissionLevel.READ_ONLY,
ActionType.DATABASE_QUERY: PermissionLevel.READ_ONLY,
}
)
# Use in LangChain agents - tool calls are automatically governed!
tools = load_tools(["python_repl", "requests"])
agent = initialize_agent(tools, governed_llm, agent="zero-shot-react-description")
agent.run("Your task here")
from agent_control_plane import (
AgentControlPlane,
create_governed_mcp_server,
ActionType,
PermissionLevel
)
# Create governed MCP server
control_plane = AgentControlPlane()
mcp_server = create_governed_mcp_server(
control_plane=control_plane,
agent_id="mcp-agent",
server_name="file-server",
permissions={
ActionType.FILE_READ: PermissionLevel.READ_ONLY,
},
transport="stdio"
)
# Register tools - all calls are governed
mcp_server.register_tool("read_file", handle_read_file, "Read a file")
mcp_server.register_resource("file://", handle_file_resource, "File resources")
mcp_server.start()
from agent_control_plane import (
AgentControlPlane,
create_governed_a2a_agent,
ActionType,
PermissionLevel
)
# Create governed A2A agent
control_plane = AgentControlPlane()
a2a_agent = create_governed_a2a_agent(
control_plane=control_plane,
agent_id="my-a2a-agent",
agent_card={
"name": "Data Processor",
"description": "Processes and analyzes data",
"capabilities": ["data_processing", "analytics"]
},
permissions={
ActionType.DATABASE_QUERY: PermissionLevel.READ_ONLY,
ActionType.API_CALL: PermissionLevel.READ_WRITE,
}
)
# Register capabilities
a2a_agent.register_capability("data_processing", handle_data_processing)
# All inter-agent communications are governed!
a2a_agent.start()
### Permission Control
```python
from agent_control_plane.agent_kernel import ActionType, PermissionLevel
# Create custom permissions
permissions = {
ActionType.FILE_READ: PermissionLevel.READ_ONLY,
ActionType.API_CALL: PermissionLevel.READ_WRITE,
ActionType.CODE_EXECUTION: PermissionLevel.NONE,
}
agent = control_plane.create_agent("restricted-agent", permissions)
from agent_control_plane.policy_engine import ResourceQuota
# Set strict quotas
quota = ResourceQuota(
agent_id="rate-limited-agent",
max_requests_per_minute=10,
max_requests_per_hour=100,
max_concurrent_executions=2,
)
control_plane.policy_engine.set_quota("rate-limited-agent", quota)
from agent_control_plane.agent_kernel import PolicyRule
import uuid
def validate_safe_path(request):
"""Only allow access to /data directory"""
path = request.parameters.get('path', '')
return path.startswith('/data/')
rule = PolicyRule(
rule_id=str(uuid.uuid4()),
name="safe_path_only",
description="Restrict file access to /data directory",
action_types=[ActionType.FILE_READ, ActionType.FILE_WRITE],
validator=validate_safe_path,
priority=10
)
control_plane.policy_engine.add_custom_rule(rule)
Check out the examples/ directory for more detailed examples:
getting_started.py - Step-by-step tutorial for beginnersbasic_usage.py - Fundamental concepts and patternsadvanced_features.py - Shadow Mode, Mute Agent, etc.use_cases.py - Real-world production scenariosconfiguration.py - Different agent configurationsCreate agents that know when to shut up and return NULL instead of hallucinating:
from agent_control_plane.mute_agent import create_mute_sql_agent
from agent_control_plane.agent_kernel import ActionType, PermissionLevel
# Create a SQL agent that ONLY executes SELECT queries
sql_config = create_mute_sql_agent("sql-bot")
permissions = {ActionType.DATABASE_QUERY: PermissionLevel.READ_ONLY}
agent = control_plane.create_agent("sql-bot", permissions)
control_plane.enable_mute_agent("sql-bot", sql_config)
# Valid: SELECT query
result = control_plane.execute_action(
agent,
ActionType.DATABASE_QUERY,
{"query": "SELECT * FROM users"}
)
# ✓ Success: True
# Invalid: Destructive operation
result = control_plane.execute_action(
agent,
ActionType.DATABASE_QUERY,
{"query": "DROP TABLE users"}
)
# ✗ Success: False, Error: "NULL"
# Agent returns NULL instead of trying to be "helpful"!
Test agent behavior without actual execution:
# Enable shadow mode
control_plane = AgentControlPlane(enable_shadow_mode=True)
agent = create_standard_agent(control_plane, "test-agent")
# This looks like normal execution...
result = control_plane.execute_action(
agent,
ActionType.FILE_WRITE,
{"path": "/data/important.txt", "content": "test"}
)
# But it was SIMULATED! No actual file was written.
print(result["status"]) # "simulated"
print(result["note"]) # "This was executed in SHADOW MODE..."
# Get statistics
stats = control_plane.get_shadow_statistics()
print(f"Success rate: {stats['success_rate']:.1%}")
Define what's possible using Data, Policy, and Temporal graphs:
from datetime import time
# Create control plane with constraint graphs
control_plane = AgentControlPlane(enable_constraint_graphs=True)
# Data Graph: What data exists
control_plane.add_data_table("users", {"id": "int", "name": "string"})
control_plane.add_data_path("/data/")
# Policy Graph: What rules apply
control_plane.add_policy_constraint(
"pii_protection",
"No PII in output",
applies_to=["table:users"],
rule_type="deny"
)
# Temporal Graph: What's true RIGHT NOW
control_plane.add_maintenance_window(
"nightly_maintenance",
start_time=time(2, 0), # 2 AM
end_time=time(4, 0), # 4 AM
blocked_actions=[ActionType.DATABASE_WRITE]
)
# The graphs enforce deterministically
# If a table isn't in the Data Graph, access is blocked
# If during maintenance window, writes are blocked
# This is ENFORCEMENT, not advisory
Agents watching agents:
from agent_control_plane.supervisor_agents import create_default_supervisor
# Create worker agents
agent1 = create_standard_agent(control_plane, "worker-1")
agent2 = create_standard_agent(control_plane, "worker-2")
# Create supervisor to watch them
supervisor = create_default_supervisor(["worker-1", "worker-2"])
control_plane.add_supervisor(supervisor)
# Agents do their work...
# (execute various actions)
# Run supervision cycle
violations = control_plane.run_supervision()
# Supervisor detects: repeated failures, excessive risk,
# rate limit approaching, suspicious patterns, etc.
for supervisor_id, viols in violations.items():
for v in viols:
print(f"[{v.severity}] {v.description}")
Create workflows with multiple specialized agents:
from agent_control_plane import (
AgentOrchestrator,
AgentRole,
OrchestrationType,
create_rag_pipeline
)
# Create orchestrator
orchestrator = AgentOrchestrator(control_plane)
# Register specialized agents
orchestrator.register_agent(
"retriever",
AgentRole.SPECIALIST,
capabilities=["document_search", "vector_search"]
)
orchestrator.register_agent(
"analyzer",
AgentRole.SPECIALIST,
capabilities=["data_analysis", "summarization"]
)
orchestrator.register_agent(
"supervisor",
AgentRole.SUPERVISOR,
capabilities=["quality_check", "safety_check"]
)
# Create a RAG pipeline workflow
workflow = orchestrator.create_workflow("rag_pipeline", OrchestrationType.SEQUENTIAL)
orchestrator.add_agent_to_workflow(workflow.workflow_id, "retriever")
orchestrator.add_agent_to_workflow(
workflow.workflow_id,
"analyzer",
dependencies={"retriever"} # Analyzer depends on retriever
)
# Add supervisor to watch all agents
orchestrator.add_supervisor("supervisor", ["retriever", "analyzer"])
# Execute workflow
import asyncio
result = asyncio.run(orchestrator.execute_workflow(
workflow.workflow_id,
{"query": "What are the key findings?"}
))
print(f"Workflow completed: {result['success']}")
Advanced safety beyond basic policies:
from agent_control_plane import (
GovernanceLayer,
AlignmentPrinciple,
create_default_governance
)
# Create governance layer with ethical rules
governance = create_default_governance()
# Check alignment before execution
context = {"content": "Analyze sales data"}
alignment = governance.check_alignment(context)
if not alignment["aligned"]:
print(f"Alignment violations: {alignment['violations']}")
# Detect bias in content
text = "All engineers should be..."
bias_result = governance.detect_bias(text)
if bias_result.has_bias:
print(f"Bias detected: {bias_result.bias_types}")
print(f"Recommendations: {bias_result.recommendations}")
# Analyze privacy and PII
data = {"email": "user@example.com", "phone": "555-1234"}
privacy = governance.analyze_privacy(data)
print(f"Privacy level: {privacy.privacy_level.value}")
print(f"Contains PII: {privacy.contains_pii}")
print(f"Risk score: {privacy.risk_score}")
Extend agent capabilities with dynamic tool registration:
from agent_control_plane import ToolRegistry, ToolType
registry = ToolRegistry()
# Register a custom tool
def web_search(query: str) -> dict:
"""Search the web for information"""
# Implementation here
return {"results": [...]}
tool_id = registry.register_tool(
name="web_search",
description="Search the web",
tool_type=ToolType.SEARCH,
handler=web_search,
risk_level=0.3
)
# Execute the tool
result = registry.execute_tool("web_search", {"query": "AI safety"})
print(f"Search results: {result['result']}")
# Discover tools by type
search_tools = registry.get_tools_by_type(ToolType.SEARCH)
print(f"Available search tools: {len(search_tools)}")
Agent Control Plane includes a CLI for common operations:
# Create an agent
python acp-cli.py agent create my-agent --role worker
# List all agents
python acp-cli.py agent list
# Inspect an agent
python acp-cli.py agent inspect my-agent
# View audit logs
python acp-cli.py audit show --limit 10 --format json
# Run safety benchmark
python acp-cli.py benchmark run
Deploy Agent Control Plane in containers for production:
# Build and start services
docker-compose up -d
# View logs
docker-compose logs -f
# Development environment with Jupyter
docker-compose --profile dev up -d acp-dev
# Distributed setup with Redis
docker-compose --profile distributed up -d
See Docker Deployment Guide for complete instructions.
Explore features hands-on with our Jupyter notebook:
# Install Jupyter
pip install jupyter
# Launch the tutorial
jupyter notebook examples/interactive_tutorial.ipynb
The tutorial covers:
┌─────────────────────────────────────────────────────────────┐
│ Application Layer │
│ (Chat, Workflow, Tools) │
└─────────────────────────┬───────────────────────────────────┘
│
┌─────────────────────────▼───────────────────────────────────┐
│ Agent Control Plane │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Agent │ │ Policy │ │ Audit │ │
│ │ Kernel │◄─┤ Engine │◄─┤ Logger │ │
│ └──────┬───────┘ └──────────────┘ └──────────────┘ │
│ │ │
│ ┌──────▼───────┐ ┌──────────────┐ │
│ │ Resource │ │ Execution │ │
│ │ Manager │◄─┤ Engine │ │
│ └──────────────┘ └──────────────┘ │
└─────────────────────────┬───────────────────────────────────┘
│
┌─────────────────────────▼───────────────────────────────────┐
│ LLM (Raw Compute) │
│ (GPT-4, Claude, Llama, etc.) │
└─────────────────────────────────────────────────────────────┘
│
┌─────────────────────────▼───────────────────────────────────┐
│ Execution Environment │
│ (Code, Databases, APIs, File System) │
└─────────────────────────────────────────────────────────────┘
The kernel mediates all interactions between the LLM and execution environment:
Enforces governance rules deterministically:
Safely executes agent actions:
Implements "Scale by Subtraction" philosophy:
The "Matrix" for agents - simulation without execution:
Multi-dimensional context defining the "physics" of the agent's world:
Recursive governance - agents watching agents:
Run the basic examples:
python3 examples.py
This demonstrates:
Run the advanced feature examples:
python3 advanced_examples.py
This demonstrates:
Run the complete test suite:
# Basic features
python3 test_control_plane.py
# Advanced features
python3 test_advanced_features.py
# Or run all tests
python3 test_control_plane.py && python3 test_advanced_features.py
Total: 31 tests covering all features.
A comprehensive benchmark demonstrates the effectiveness of the Control Plane approach compared to traditional prompt-based safety.
python3 benchmark.py
The benchmark compares Prompt-Based Safety (Baseline) vs Control Plane Governance (Experimental) using 60 red team prompts:
| Metric | Baseline (Prompts) | Control Plane | Improvement |
|---|---|---|---|
| Safety Violation Rate | 26.67% | 0.00% | ✓ 100% enforcement |
| False Positive Rate | 0.00% | 0.00% | ✓ Perfect precision |
| Avg Output Tokens | 26.1 | 0.5 | ✓ 98.1% reduction |
Key Findings:
The benchmark includes:
See benchmark/README.md for detailed methodology and results.
🎥 Coming Soon: A short video walkthrough showing:
Video will be published to YouTube and embedded here
📹 Coming Soon: Complete tutorial covering:
Planned Release: Q1 2026
Try these live demos in your browser:
examples/ directoryHave you created a tutorial or demo? Submit it here to be featured!
Deploy agents with strict governance for enterprise environments:
Build precise, non-creative agents:
Safely run multiple agents with isolation:
Experiment safely with agent capabilities:
Run reliable, auditable agent workflows:
See architecture.md for detailed architecture documentation.
AgentControlPlane: Main control plane interfaceAgentKernel: Core kernel componentPolicyEngine: Policy enforcementExecutionEngine: Safe executionAgentContext: Agent session contextExecutionRequest: Action requestExecutionResult: Action resultAgentOrchestrator: Multi-agent coordination and workflowsToolRegistry: Dynamic tool management and discoveryGovernanceLayer: Ethical alignment and advanced safetyAgentNode: Agent representation in orchestration graphsTool: Tool definition with schemas and handlersFILE_READ: Read file operationsFILE_WRITE: Write file operationsCODE_EXECUTION: Execute codeAPI_CALL: Make API callsDATABASE_QUERY: Query databasesDATABASE_WRITE: Write to databasesWORKFLOW_TRIGGER: Trigger workflowsNONE: No accessREAD_ONLY: Read-only accessREAD_WRITE: Read and write accessADMIN: Full administrative accessRecent additions (2025):
Planned enhancements:
All experiments and results in this repository are fully reproducible. We provide:
Complete materials in reproducibility/ directory:
# Using Docker (recommended)
cd reproducibility/docker_config
docker build -t acp-reproducibility:v1.1.0 .
docker run -it acp-reproducibility:v1.1.0 bash
bash reproducibility/run_all_experiments.sh
# Or locally
pip install -r reproducibility/requirements_frozen.txt
python benchmark.py --seed 42
python experiments/multi_agent_rag.py --seed 42
python experiments/long_horizon_purge.py --seed 42
See reproducibility/README.md for complete guide.
Projects like Steve Yegge's Gas Town use a "City" metaphor where a "Mayor" agent orchestrates "Worker" agents to maximize coding throughput. This is brilliant for velocity.
The Difference:
The Agent Control Plane complements coordination systems by providing the safety layer.
Most current safety tools operate as "sidecars" that check input/output for toxicity, PII, or harmful content. They are largely text-based and probabilistic.
The Difference:
Recent academic papers propose "Control Planes" that act as a phonebook, helping agents find the right tools.
The Difference:
Deterministic Enforcement, Not Advisory Hints:
This is systems engineering for AI, not prompt engineering.
The Agent Control Plane is built on peer-reviewed research and industry best practices, not just intuition.
Our design is informed by academic research across multiple domains:
Our comparative safety study follows research standards:
If you use Agent Control Plane in research:
@software{agent_control_plane,
title = {Agent Control Plane: A Governance Layer for Autonomous AI Agents},
author = {Agent Control Plane Contributors},
year = {2025},
url = {https://github.com/imran-siddique/agent-control-plane},
note = {MIT License}
}
Join our growing community:
See ROADMAP.md for the complete 2026 roadmap. Key upcoming features:
Q1 2026 (Current)
pip install agent-control-plane)Q2 2026 - Advanced Intelligence
Q3 2026 - Multimodal & Ecosystem
Q4 2026 - Scale & Operations
2026 Targets:
We welcome contributions! The Agent Control Plane is designed to be production-ready and contributor-friendly.
git clone https://github.com/YOUR-USERNAME/agent-control-plane.gitpip install -e ".[dev]"git checkout -b feature/your-feature-name# Run all tests
python -m unittest discover -s tests -p 'test_*.py' -v
# Run specific test file
python -m unittest tests/test_control_plane.py
# Run specific test
python -m unittest tests.test_control_plane.TestAgentKernel.test_create_agent_session
src/agent_control_plane/ - Main package source codetests/ - Test suite (unittest framework)examples/ - Example scripts and use casesdocs/ - Documentation and guides.github/workflows/ - CI/CD configurationSee CONTRIBUTING.md for detailed guidelines.
The project uses Python's built-in unittest framework. All tests are located in the tests/ directory.
test_control_plane.pytest_advanced_features.pyCurrent test coverage: 31 tests covering:
Comprehensive documentation is available in the docs/ directory:
MIT License - See LICENSE file for details
Download counts and popularity will appear here as the collector receives them.
| Version | Published | Direct dependencies | Open version |
|---|---|---|---|
| 1.2.0 | 23 Jan 2026 | 5 |
This describes our collected data; it is not a security assessment.
Shared repository data · GitHub
Observed since 20 Oct, 21:42 UTC
No superseded metadata states collected. Unchanged observations do not create new revisions.
package:719c8e0ea9f8d576c28c062bf23285fbf818b6ade618d607538b67b7bf0cb305