API Reference

from avakill import Guard, protect, PolicyViolation, ConfigError, RateLimitExceeded

Guard(
    policy: str | Path | dict | PolicyConfig | None = None,
    logger: AuditLogger | None = None,
    agent_id: str | None = None,
    self_protection: bool = True,
    signing_key: bytes | None = None,
    verify_key: bytes | None = None,
    rate_limit_backend: RateLimitBackend | None = None,
    normalize_tools: bool = False,
    approval_store: ApprovalStore | None = None,
    cross_call_correlation: bool = False,
)

Guard.evaluate(
    tool: str,
    args: dict[str, Any] | None = None,
    *,
    agent_id: str | None = None,
    session_id: str | None = None,
    metadata: dict[str, Any] | None = None,
    override: bool = False,
) -> Decision

decision = guard.evaluate(tool="execute_sql", args={"query": "DROP TABLE users"})
if not decision.allowed:
    print(f"Blocked by {decision.policy_name}: {decision.reason}")

Guard.evaluate_or_raise(
    tool: str,
    args: dict[str, Any] | None = None,
    **kwargs,
) -> Decision

Guard.session(
    agent_id: str | None = None,
    session_id: str | None = None,
) -> GuardSession

with guard.session(agent_id="my-agent") as session:
    session.evaluate(tool="search", args={"q": "hello"})
    session.evaluate(tool="read_file", args={"path": "/tmp/data.txt"})
    print(session.call_count)  # → 2

Guard.reload_policy(path: str | Path | None = None) -> None

Guard.watch(**kwargs) -> PolicyWatcher

watcher = guard.watch()
await watcher.start()
# Policy reloads automatically on file change
await watcher.stop()

await Guard.unwatch() -> None

Guard.policy_status -> str

@protect
def my_tool(arg: str): ...

@protect(guard=guard, tool_name="custom_name", on_deny="return_none")
def my_tool(arg: str): ...

@protect(policy="strict.yaml", on_deny="callback", deny_callback=my_handler)
async def risky_tool(arg: str): ...

try:
    guard.evaluate_or_raise(tool="delete_user", args={"id": "123"})
except PolicyViolation as e:
    print(e.tool_name)     # "delete_user"
    print(e.decision)      # Decision object

try:
    guard.evaluate_or_raise(tool="web_search", args={"q": "test"})
except RateLimitExceeded as e:
    print(e.decision.reason)  # Rate limit details

from openai import OpenAI
from avakill import GuardedOpenAIClient

client = GuardedOpenAIClient(OpenAI(), policy="avakill.yaml")
response = client.chat.completions.create(model="gpt-4o", tools=[...], messages=[...])
# Denied tool_calls are removed from the response
# response.avakill_decisions contains all decisions

from anthropic import Anthropic
from avakill import GuardedAnthropicClient

client = GuardedAnthropicClient(Anthropic(), policy="avakill.yaml")
response = client.messages.create(model="claude-sonnet-4-5-20250514", tools=[...], messages=[...])
# Denied tool_use blocks are removed from response.content

from avakill import AvaKillCallbackHandler

handler = AvaKillCallbackHandler(policy="avakill.yaml")
agent.invoke({"input": "..."}, config={"callbacks": [handler]})
# Raises PolicyViolation before the tool executes

from avakill import MCPProxyServer

proxy = MCPProxyServer(
    upstream_cmd="python",
    upstream_args=["server.py"],
    policy="avakill.yaml",
)
await proxy.run()

from avakill import PolicyIntegrity

# Sign a file with HMAC
sig_path = PolicyIntegrity.sign_file("avakill.yaml", key_bytes)

# Sign with Ed25519
sig_path = PolicyIntegrity.sign_file_ed25519("avakill.yaml", private_key_bytes)

# Verify (auto-detects HMAC or Ed25519)
valid = PolicyIntegrity.verify_file("avakill.yaml", key_bytes)

from avakill import FileSnapshot

snap = FileSnapshot.from_path("avakill.yaml")
print(snap.sha256)  # SHA-256 hash
print(snap.size)    # File size in bytes
print(snap.mode)    # File permissions

ok, message = snap.verify("avakill.yaml")

from avakill import get_json_schema
schema = get_json_schema()  # Returns dict — JSON Schema for PolicyConfig

from avakill import generate_prompt
prompt = generate_prompt(
    tools_list=["file_read", "shell_exec", "db_query"],
    use_case="code assistant",
)
# Returns a self-contained LLM prompt for generating policies

from avakill import get_metrics_registry
from prometheus_client import start_http_server

registry = get_metrics_registry()
start_http_server(9090, registry=registry)

from avakill import PolicyWatcher

watcher = guard.watch()
await watcher.start()
# ... policy file changes are detected and reloaded ...
await watcher.stop()

async with guard.watch() as watcher:
    # Policy auto-reloads on file changes
    pass

DaemonServer(
    guard: Guard,
    socket_path: str | Path | None = None,
    pid_file: str | Path | None = None,
    normalizer: ToolNormalizer | None = None,
    max_connections: int = 100,
    transport: ServerTransport | None = None,
    tcp_port: int | None = None,
    os_enforce: bool = False,
    on_ready: Callable[[str], None] | None = None,
)

await DaemonServer.start() -> None

await DaemonServer.stop() -> None

await DaemonServer.serve_forever() -> None

@staticmethod
DaemonServer.is_running(pid_file: str | Path | None = None) -> tuple[bool, int | None]

DaemonClient(
    socket_path: str | Path | None = None,
    timeout: float = 5.0,
    transport: ClientTransport | None = None,
    tcp_port: int | None = None,
)

DaemonClient.evaluate(request: EvaluateRequest) -> EvaluateResponse

DaemonClient.ping() -> bool

DaemonClient.try_evaluate(request: EvaluateRequest) -> EvaluateResponse | None

from avakill.daemon.protocol import EvaluateRequest

request = EvaluateRequest(
    agent="claude-code",
    tool="Bash",
    args={"command": "rm -rf /"},
)

from avakill.daemon.protocol import EvaluateResponse

from avakill.daemon.protocol import (
    serialize_request, deserialize_request,
    serialize_response, deserialize_response,
)

# Newline-delimited JSON over Unix socket
data = serialize_request(request)    # -> bytes
req = deserialize_request(data)      # -> EvaluateRequest
data = serialize_response(response)  # -> bytes
resp = deserialize_response(data)    # -> EvaluateResponse

ToolNormalizer(custom_mappings: dict[str, dict[str, str]] | None = None)

ToolNormalizer.normalize(tool: str, agent: str | None = None) -> str

normalizer = ToolNormalizer()
normalizer.normalize("Bash", agent="claude-code")       # → "shell_execute"
normalizer.normalize("run_shell_command", agent="gemini-cli")  # → "shell_execute"
normalizer.normalize("unknown_tool", agent="claude-code")      # → "unknown_tool"

ToolNormalizer.denormalize(canonical: str, agent: str) -> str | None

PolicyCascade.discover(cwd: Path | None = None) -> list[tuple[PolicyLevel, Path]]

PolicyCascade.load(cwd: Path | None = None) -> PolicyConfig

@staticmethod
PolicyCascade.merge(configs: list[PolicyConfig]) -> PolicyConfig

from avakill.hooks.base import HookAdapter

from avakill.enforcement.landlock import LandlockEnforcer

# Check availability
LandlockEnforcer.available()  # → True on Linux 5.13+

enforcer = LandlockEnforcer()

# Dry run — see what would be restricted
ruleset = enforcer.generate_ruleset(policy_config)

# Apply — IRREVERSIBLE for the current process
enforcer.apply(policy_config)

from avakill.enforcement.sandbox_exec import SandboxExecEnforcer

# Check availability
SandboxExecEnforcer.available()  # → True on macOS

enforcer = SandboxExecEnforcer()

# Generate SBPL profile string
profile = enforcer.generate_profile(policy_config)

# Write to file
enforcer.write_profile(policy_config, Path("avakill.sb"))

from avakill.enforcement.tetragon import TetragonPolicyGenerator

generator = TetragonPolicyGenerator()

# Generate TracingPolicy YAML
yaml_str = generator.generate(policy_config)

# With optional parameters
yaml_str = generator.generate(policy_config, match_binaries=["/usr/bin/python3"], action="Sigkill")

# Write to file
generator.write(policy_config, Path("tetragon-policy.yaml"))

from avakill.compliance.assessor import ComplianceAssessor

assessor = ComplianceAssessor(guard=guard, logger=logger)

# Assess single framework
report = assessor.assess("soc2")

# Assess all frameworks
reports = assessor.assess_all()  # → dict[str, ComplianceReport]

from avakill.compliance.reporter import ComplianceReporter

reporter = ComplianceReporter()
table = reporter.to_rich_table(report)  # Rich Table for terminal
json_str = reporter.to_json(report)     # JSON string
md_str = reporter.to_markdown(report)   # Markdown string

from avakill.core.approval import ApprovalStore

async with ApprovalStore("approvals.db") as store:
    # Create approval request (default TTL: 1 hour)
    request = await store.create(tool_call, decision, agent="claude-code", ttl_seconds=3600)

    # List pending
    pending = await store.get_pending()

    # Approve or deny
    approved = await store.approve(request.id, approver="admin")
    denied = await store.deny(request.id, approver="admin")

    # Clean up expired
    count = await store.cleanup_expired()

    # Find approved request for tool+agent
    req = await store.get_approved_for_tool("shell_execute", agent="claude-code")

    # Resolve a prefix ID (12-char) to full UUID
    full_id = await store.resolve_id("abc123def456")

    # Get a single request by ID
    req = await store.get(request.id)

    # Close the database connection
    await store.close()

from avakill.analytics.engine import AuditAnalytics

analytics = AuditAnalytics(logger=sqlite_logger)

# Denial trends (time-bucketed)
trends = await analytics.denial_trend(hours=24, bucket_minutes=60)

# Per-tool usage summary
usage = await analytics.tool_usage_summary()  # → {"tool": {"allowed": N, "denied": N}}

# Per-agent risk scores (0.0 = safe, 1.0 = all denied)
scores = await analytics.agent_risk_scores()

# Per-rule effectiveness
effectiveness = await analytics.policy_effectiveness()

from avakill.launcher.core import ProcessLauncher