AI Agent Safety for FastAPI Applications

SafeClaw by Authensor integrates into FastAPI applications as a dependency injection layer, providing deny-by-default action gating for AI agent tool calls. Every subprocess.run(), file operation, and httpx request is checked against your YAML policy before execution. Install with npx @authensor/safeclaw and wire it into your FastAPI dependency system.

Why FastAPI Agents Need Gating

FastAPI is the go-to Python framework for AI agent APIs. It powers LangServe endpoints, custom agent APIs, and streaming chat backends. These APIs execute agent tool calls on the server, giving agents access to subprocess, os, and httpx. A single prompt injection through a chat endpoint can escalate to shell execution.

SafeClaw gates every action with 446 tests and hash-chained audit logging.

Installation

npx @authensor/safeclaw
pip install httpx  # for async HTTP to SafeClaw

Policy

version: 1
defaultAction: deny

rules:
  - action: file.read
    path:
      glob: "./data/**"
    decision: allow

- action: file.write
    path:
      glob: "./output/**"
    decision: allow

- action: process.exec
    command:
      startsWith: "python"
    decision: allow

- action: network.request
    host:
      in: ["api.openai.com", "api.anthropic.com"]
    decision: allow

- action: file.read
    path:
      glob: "*/.env"
    decision: deny

- action: process.exec
    command:
      contains: "pip install"
    decision: prompt

FastAPI Dependency

# dependencies/safeclaw.py
import httpx
from fastapi import Depends, HTTPException

class SafeClawGate:
    def __init__(self, endpoint: str = "http://localhost:9800"):
        self.endpoint = endpoint
        self.client = httpx.AsyncClient(base_url=endpoint)

async def check(self, action_request: dict) -> dict:
        try:
            response = await self.client.post("/check", json=action_request)
            response.raise_for_status()
            return response.json()
        except httpx.HTTPError:
            # Fail closed
            return {"allowed": False, "reason": "SafeClaw unreachable"}

async def require(self, action_request: dict) -> None:
        """Check and raise HTTPException if denied."""
        decision = await self.check(action_request)
        if not decision["allowed"]:
            raise HTTPException(
                status_code=403,
                detail=f"SafeClaw denied: {decision['reason']}"
            )

gate = SafeClawGate()

async def get_gate() -> SafeClawGate:
    return gate

Route Integration

# routes/agent.py
from fastapi import APIRouter, Depends
from dependencies.safeclaw import SafeClawGate, get_gate
import subprocess
from pathlib import Path

router = APIRouter(prefix="/api/agent")

@router.post("/read-file")
async def read_file(
    path: str,
    gate: SafeClawGate = Depends(get_gate)
):
    await gate.require({"action": "file.read", "path": path})
    content = Path(path).read_text()
    return {"content": content}

@router.post("/write-file")
async def write_file(
    path: str,
    content: str,
    gate: SafeClawGate = Depends(get_gate)
):
    await gate.require({"action": "file.write", "path": path})
    Path(path).write_text(content)
    return {"status": "written"}

@router.post("/exec")
async def exec_command(
    command: str,
    gate: SafeClawGate = Depends(get_gate)
):
    await gate.require({"action": "process.exec", "command": command})
    result = subprocess.run(
        command.split(),
        capture_output=True,
        text=True,
        timeout=30
    )
    return {"stdout": result.stdout, "stderr": result.stderr}

@router.post("/fetch")
async def fetch_url(
    url: str,
    method: str = "GET",
    gate: SafeClawGate = Depends(get_gate)
):
    from urllib.parse import urlparse
    host = urlparse(url).hostname
    await gate.require({
        "action": "network.request",
        "host": host,
        "url": url,
        "method": method
    })
    async with httpx.AsyncClient() as client:
        response = await client.request(method, url)
        return {"status": response.status_code, "body": response.text}

LangServe Integration

If you use LangServe with FastAPI:

from langserve import add_routes
from langchain_core.tools import tool
from dependencies.safeclaw import gate

@tool
async def safe_read_file(path: str) -> str:
    """Read a file with SafeClaw gating."""
    await gate.require({"action": "file.read", "path": path})
    return Path(path).read_text()

Add to your LangServe chain
add_routes(app, chain_with_safe_tools, path="/agent")

Startup

# main.py
from fastapi import FastAPI
from contextlib import asynccontextmanager
from routes.agent import router

@asynccontextmanager
async def lifespan(app: FastAPI):
    # SafeClaw runs as sidecar, started via npx @authensor/safeclaw
    yield

app = FastAPI(lifespan=lifespan)
app.include_router(router)

Every decision is hash-chained. MIT licensed, works with Claude and OpenAI.

Cross-References

• Python Integration
• LangChain Tool Wrapping
• Django Integration
• Docker Compose Deployment
• Deny-by-Default Explained