From zero to governed agent in 5 minutes.
pip install agent-os-kernelfrom agent_os import KernelSpace
kernel = KernelSpace(policy="strict")
@kernel.register
async def my_agent(task: str):
return f"Processed: {task}"
# Run with kernel governance
import asyncio
result = asyncio.run(kernel.execute(my_agent, "Hello, Agent OS!"))
print(result)That's it. Your agent now runs with kernel-level policy enforcement.
pip install agent-os-kernelOptional extras:
pip install agent-os-kernel[cmvk] # Verification
pip install agent-os-kernel[observability] # Prometheus/OpenTelemetry
pip install agent-os-kernel[full] # EverythingCreate a file called my_agent.py:
from agent_os import KernelSpace
# Initialize the kernel with strict policy
kernel = KernelSpace(policy="strict")
@kernel.register
async def analyze_data(task: str):
"""Your agent logic goes here."""
# This could be any LLM call, data processing, etc.
return f"Analysis complete: {task}"
# Execute with governance
if __name__ == "__main__":
import asyncio
result = asyncio.run(
kernel.execute(analyze_data, "Summarize Q4 sales data")
)
print(result)python my_agent.pyOutput:
Analysis complete: Summarize Q4 sales data
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
β USER SPACE (Your Code) β
β analyze_data() runs here β
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ€
β KERNEL SPACE (Agent OS) β
β Every action checked against policies β
β Violations β SIGKILL (non-catchable) β
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
@kernel.registerwraps your function with kernel governancekernel.execute()runs your agent through the policy engine- If policy violated β automatic SIGKILL before execution
from agent_os import KernelSpace
from openai import OpenAI
kernel = KernelSpace(policy="strict")
client = OpenAI()
@kernel.register
async def smart_agent(task: str):
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": task}]
)
return response.choices[0].message.content
if __name__ == "__main__":
import asyncio
result = asyncio.run(kernel.execute(smart_agent, "What is 2+2?"))
print(result)See what happens when your agent tries something blocked:
@kernel.register
async def dangerous_agent(task: str):
import os
os.remove("/etc/passwd") # β This will be blocked!
return "Done"Output:
β οΈ POLICY VIOLATION DETECTED
β οΈ Signal: SIGKILL
β οΈ Action: file_write
β οΈ Status: TERMINATED
The kernel blocked the action before it executed.
| Time | Tutorial | What You'll Learn |
|---|---|---|
| 10 min | 30-Minute Deep Dive | Policies, signals, VFS |
| 15 min | Episodic Memory | Agent memory that persists |
| 15 min | Verification | Detect hallucinations |
| 20 min | Time-Travel Debugging | Replay and debug decisions |
from agent_os.integrations import LangChainKernel
from langchain.agents import AgentExecutor
kernel = LangChainKernel()
governed_agent = kernel.wrap(my_langchain_agent)from agent_os.integrations import OpenAIKernel
kernel = OpenAIKernel()
governed = kernel.wrap_assistant(assistant, client)from agent_os.integrations import CrewAIKernel
kernel = CrewAIKernel()
governed = kernel.wrap(my_crew)- π Full Documentation
- π GitHub Issues
- π‘ Examples