🚀 Quickstart Guide

pip install agent-os-kernel

python -c "import agent_os; print(agent_os.__version__)"

# Install with LangChain integration
pip install agent-os-kernel[langchain]

# Install with CrewAI integration
pip install agent-os-kernel[crewai]

# Install with all integrations
pip install agent-os-kernel[all]

# Install with Redis support for AMB
pip install agent-os-kernel[redis]

from agent_os import KernelSpace

# Initialize the kernel with default strict policy
kernel = KernelSpace(policy="strict")

@kernel.register
async def my_first_agent(task: str) -> str:
    """
    A simple agent that processes tasks.
    The kernel automatically enforces safety policies
    on every action this agent attempts.
    """
    # Simulate some agent work
    result = f"Processed task: {task}"
    
    # Any dangerous operations would be blocked
    # by the kernel before they execute
    return result

import asyncio

async def main():
    # Execute the governed agent
    result = await kernel.execute(my_first_agent, "analyze sales data")
    print(result)
    
    # View the execution trace
    trace = kernel.get_trace()
    print(f"Execution completed in {trace.duration_ms}ms")
    print(f"Policy violations: {trace.violations}")

# Run the example
asyncio.run(main())

"""
my_first_agent.py - A minimal governed agent example
"""
import asyncio
from agent_os import KernelSpace

# Initialize kernel with strict safety policy
kernel = KernelSpace(policy="strict")

@kernel.register
async def analyze_data(query: str) -> str:
    """Agent that analyzes data with kernel governance."""
    # In a real agent, this might call an LLM
    # The kernel intercepts and validates every action
    return f"Analysis complete for: {query}"

@kernel.register  
async def write_report(data: str) -> str:
    """Agent that writes reports with governance."""
    return f"Report generated: {len(data)} chars processed"

async def main():
    # Execute governed agents
    analysis = await kernel.execute(analyze_data, "Q4 revenue trends")
    report = await kernel.execute(write_report, analysis)
    
    print(report)
    
    # Check execution metrics
    metrics = kernel.metrics()
    print(f"\n--- Kernel Metrics ---")
    print(f"Total executions: {metrics.total_executions}")
    print(f"Policy violations blocked: {metrics.blocked_actions}")
    print(f"Average latency: {metrics.avg_latency_ms:.2f}ms")

if __name__ == "__main__":
    asyncio.run(main())

# policy.yaml
name: healthcare_policy
version: "1.0"

rules:
  # Block any PHI-related operations
  - action: "read_file"
    pattern: "*.phi"
    effect: deny
    signal: SIGKILL
    
  # Require approval for external API calls
  - action: "http_request"
    domains: ["*"]
    effect: require_approval
    timeout: 30s
    
  # Allow internal operations
  - action: "compute"
    effect: allow
    
  # Log all database operations
  - action: "database_*"
    effect: allow
    audit: true

from agent_os import KernelSpace, Policy

# Load from YAML file
kernel = KernelSpace(policy="./policy.yaml")

# Or define programmatically
policy = Policy(
    name="my_policy",
    rules=[
        {"action": "file_write", "pattern": "/tmp/*", "effect": "allow"},
        {"action": "file_write", "pattern": "*", "effect": "deny"},
        {"action": "http_request", "domains": ["api.safe.com"], "effect": "allow"},
    ]
)
kernel = KernelSpace(policy=policy)

from agent_os import KernelSpace
from agent_os.exceptions import PolicyViolation

kernel = KernelSpace(policy="strict")

@kernel.register
async def risky_agent(task: str):
    # This action will be blocked by the kernel
    # The agent code never actually executes the delete
    await kernel.action("file_delete", path="/etc/passwd")
    return "Done"

async def main():
    try:
        await kernel.execute(risky_agent, "cleanup")
    except PolicyViolation as e:
        print(f"Blocked: {e.action} - {e.reason}")
        print(f"Signal sent: {e.signal}")  # SIGKILL

asyncio.run(main())

from agent_os import KernelSpace, Signal

kernel = KernelSpace()

@kernel.register
async def long_running_agent(task: str):
    for i in range(100):
        await kernel.checkpoint()  # Check for signals
        await process_chunk(i)
    return "Complete"

async def main():
    # Start agent in background
    agent_id = await kernel.spawn(long_running_agent, "process all data")
    
    # Pause after 5 seconds
    await asyncio.sleep(5)
    await kernel.signal(agent_id, Signal.SIGSTOP)
    print("Agent paused")
    
    # Resume after user confirmation
    input("Press Enter to resume...")
    await kernel.signal(agent_id, Signal.SIGCONT)
    
    # Wait for completion
    result = await kernel.wait(agent_id)
    print(result)

# When an agent attempts a blocked action:
# 1. Kernel intercepts the action BEFORE execution
# 2. Policy engine evaluates the action
# 3. If denied: SIGKILL sent immediately
# 4. Agent terminates, action never executes

# Example policy rule that triggers SIGKILL:
rules:
  - action: "execute_code"
    pattern: "rm -rf *"
    effect: deny
    signal: SIGKILL
    reason: "Destructive command blocked"

from langchain.agents import create_openai_functions_agent
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from agent_os.integrations import langchain_kernel

# Create your LangChain agent as usual
llm = ChatOpenAI(model="gpt-4")
tools = [
    Tool(name="search", func=search_web, description="Search the web"),
    Tool(name="calculate", func=calculator, description="Do math"),
]
agent = create_openai_functions_agent(llm, tools, prompt)

# Wrap with Agent OS governance - one line!
governed_agent = langchain_kernel.wrap(
    agent,
    policy="strict",
    signals_enabled=True
)

# Execute with full kernel protection
result = await governed_agent.ainvoke({"input": "What is 2+2?"})

from crewai import Agent, Task, Crew
from agent_os.integrations import crewai_kernel

# Define your CrewAI agents
researcher = Agent(
    role="Researcher",
    goal="Find accurate information",
    backstory="Expert researcher with attention to detail"
)

writer = Agent(
    role="Writer", 
    goal="Create compelling content",
    backstory="Skilled technical writer"
)

# Create the crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task]
)

# Wrap entire crew with governance
governed_crew = crewai_kernel.wrap(
    crew,
    policy="moderate",
    inter_agent_trust=True  # Enable IATP for agent communication
)

# Execute with kernel protection
result = governed_crew.kickoff()

from autogen import AssistantAgent, UserProxyAgent
from agent_os.integrations import autogen_kernel

# Create AutoGen agents
assistant = AssistantAgent(
    name="assistant",
    llm_config={"model": "gpt-4"}
)

user_proxy = UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER"
)

# Wrap with governance
governed_assistant = autogen_kernel.wrap(assistant, policy="strict")
governed_proxy = autogen_kernel.wrap(user_proxy, policy="strict")

# Chat with kernel protection
governed_proxy.initiate_chat(
    governed_assistant,
    message="Write a Python function to sort a list"
)

from openai import OpenAI
from agent_os.integrations import openai_kernel

client = OpenAI()

# Create assistant
assistant = client.beta.assistants.create(
    name="Data Analyst",
    instructions="You analyze data and create visualizations",
    model="gpt-4-turbo",
    tools=[{"type": "code_interpreter"}]
)

# Wrap with governance
governed_assistant = openai_kernel.wrap(
    assistant,
    client=client,
    policy="strict",
    allowed_file_types=[".csv", ".json", ".xlsx"]
)

# Use governed assistant
thread = governed_assistant.create_thread()
governed_assistant.add_message(thread.id, "Analyze this sales data")
run = governed_assistant.run(thread.id)

from agent_os import KernelSpace, govern

# Use the @govern decorator for any async function
kernel = KernelSpace(policy="strict")

@govern(kernel)
async def my_custom_agent(input_data):
    """Works with any framework or custom code."""
    # Your agent logic here
    result = await some_framework.process(input_data)
    return result

# Or wrap at runtime
from some_framework import Agent
agent = Agent()
governed = kernel.wrap(agent.run)

# Clone the repository
git clone https://github.com/imran-siddique/agent-os.git
cd agent-os

# Install development dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Submit a pull request!