Agent Rails — agent-core to Exo Mapping

# agent-core pattern
class AgentRail(ABC):
    name: str
    priority: int = 50  # Lower = runs first

    @abstractmethod
    async def handle(self, event: AgentCallbackEvent, context: dict) -> RailResult:
        ...

# agent-core pattern
@rail(name="block_rm", events=[AgentCallbackEvent.ON_TOOL_CALL], priority=10)
async def block_rm(event, context):
    if context["tool_name"] == "rm_rf":
        return RailResult.ABORT
    return RailResult.CONTINUE

# agent-core — uses @rail decorator and untyped context dict
@rail(name="block_rm", events=[AgentCallbackEvent.ON_TOOL_CALL], priority=10)
async def block_rm(event, context):
    if context["tool_name"] == "rm_rf":
        return RailResult.ABORT
    return RailResult.CONTINUE

agent = Agent(name="assistant", rails=[block_rm])

from exo.hooks import HookPoint
from exo.rail import Rail, RailAction
from exo.rail_types import RailContext, ToolCallInputs


class BlockDangerousTool(Rail):
    """Block execution of a specific tool by name."""

    def __init__(self, blocked_tool: str) -> None:
        super().__init__("block_tool", priority=10)
        self.blocked_tool = blocked_tool

    async def handle(self, ctx: RailContext) -> RailAction | None:
        if (
            ctx.event == HookPoint.PRE_TOOL_CALL
            and isinstance(ctx.inputs, ToolCallInputs)
            and ctx.inputs.tool_name == self.blocked_tool
        ):
            return RailAction.ABORT
        return RailAction.CONTINUE

# agent-core — upstream rail writes to context["extra"]
@rail(name="rate_counter", priority=10)
async def rate_counter(event, context):
    context["extra"]["calls_remaining"] = 5
    return RailResult.CONTINUE

@rail(name="rate_logger", priority=80)
async def rate_logger(event, context):
    remaining = context["extra"].get("calls_remaining", "unknown")
    print(f"Rate limit remaining: {remaining}")
    return RailResult.CONTINUE

from exo.rail import Rail, RailAction
from exo.rail_types import RailContext


class RateCounterRail(Rail):
    """Write rate-limit state for downstream rails."""

    def __init__(self) -> None:
        super().__init__("rate_counter", priority=10)

    async def handle(self, ctx: RailContext) -> RailAction | None:
        ctx.extra["calls_remaining"] = 5
        return RailAction.CONTINUE


class RateLoggerRail(Rail):
    """Read rate-limit state from upstream rails."""

    def __init__(self) -> None:
        super().__init__("rate_logger", priority=80)

    async def handle(self, ctx: RailContext) -> RailAction | None:
        remaining = ctx.extra.get("calls_remaining", "unknown")
        print(f"Rate limit remaining: {remaining}")
        return RailAction.CONTINUE

# agent-core
agent = Agent(name="my_agent", rails=[block_rm, rate_counter, rate_logger])

from exo.agent import Agent
from exo.rail import Rail

# Rails are passed as a list — RailManager is created automatically
agent = Agent(
    name="my_agent",
    model="openai:gpt-4o",
    rails=[
        BlockDangerousTool("rm_rf"),
        RateCounterRail(),
        RateLoggerRail(),
    ],
)

# Equivalent manual setup (if you need fine-grained control):
from exo.hooks import HookPoint
from exo.rail import RailManager

manager = RailManager()
manager.add(BlockDangerousTool("rm_rf"))

agent2 = Agent(name="my_agent2", model="openai:gpt-4o")
for point in HookPoint:
    agent2.hook_manager.add(point, manager.hook_for(point))