🔌 Framework Integrations

from agent_os.integrations.base import BaseAdapter, AdapterConfig
from agent_os import KernelSpace
from typing import Any, Callable

class MyFrameworkAdapter(BaseAdapter):
    """Custom adapter for MyFramework agents."""
    
    def __init__(self, config: AdapterConfig = None):
        super().__init__(config or AdapterConfig())
        self.kernel = KernelSpace(policy=self.config.policy)
    
    def wrap(self, agent: Any, **kwargs) -> Any:
        """Wrap a MyFramework agent with kernel governance."""
        
        # Store original execution method
        original_execute = agent.execute
        
        # Create governed wrapper
        @self.kernel.register
        async def governed_execute(task: str):
            # Pre-execution hook
            self.kernel.emit("agent:execute:start", {"task": task})
            
            # Execute through kernel
            result = await original_execute(task)
            
            # Post-execution hook
            self.kernel.emit("agent:execute:complete", {"result": result})
            return result
        
        # Replace execution method
        agent.execute = governed_execute
        return agent

from agent_os.core import PolicyViolation

class MyFrameworkAdapter(BaseAdapter):
    
    def _intercept_tool_call(self, tool_name: str, args: dict) -> dict:
        """Intercept and validate tool calls before execution."""
        
        # Check against kernel policy
        decision = self.kernel.check_action({
            "type": "tool_call",
            "tool": tool_name,
            "arguments": args
        })
        
        if decision.blocked:
            raise PolicyViolation(
                f"Tool '{tool_name}' blocked by policy: {decision.reason}"
            )
        
        # Apply any argument transformations
        if decision.modified_args:
            args = decision.modified_args
            
        return args
    
    def _wrap_tools(self, agent: Any) -> None:
        """Wrap all agent tools with interception."""
        for tool in agent.tools:
            original_fn = tool.function
            
            def governed_fn(*args, **kwargs):
                kwargs = self._intercept_tool_call(tool.name, kwargs)
                return original_fn(*args, **kwargs)
            
            tool.function = governed_fn

from agent_os.integrations import register_adapter

# Register your custom adapter
register_adapter("myframework", MyFrameworkAdapter)

# Now use it like built-in integrations
from agent_os.integrations import myframework_kernel

# Wrap your agent
my_agent = MyFrameworkAgent(...)
governed = myframework_kernel.wrap(my_agent, policy="strict")

# Execute with full kernel protection
result = governed.execute("process sensitive data")