TerminalRenderer (rich terminal UI) and a _DefaultRenderer (silent auto-approve for CI), but you can build custom renderers for any delivery channel — Slack, web dashboards, email, mobile push notifications, or even voice interfaces.
The Renderer Protocol
Every renderer must implement four async methods:| Method | Signature | When Called |
|---|---|---|
render_approval | async (ctx, risk) -> Verdict | MEDIUM-risk actions needing simple approve/deny |
render_challenge | async (ctx, risk, challenge_type) -> ChallengeResult | HIGH/CRITICAL-risk actions needing verification |
render_info | async (message) -> None | Informational messages (no decision required) |
render_auto_approved | async (ctx, risk) -> None | LOW-risk actions that were auto-approved |
All renderer methods are
async. Even if your implementation is synchronous under the hood (e.g., blocking HTTP call), you must declare the methods as async def. Use await asyncio.to_thread(...) to wrap blocking I/O.Step-by-Step: Slack Renderer
A Slack renderer sends approval requests as interactive messages with buttons and collects responses via a callback URL or polling.Implement the Renderer
import asyncio
import json
import aiohttp
from attesta import (
ActionContext,
RiskAssessment,
ChallengeType,
ChallengeResult,
Verdict,
)
class SlackRenderer:
"""Sends approval requests to Slack and waits for button responses."""
def __init__(
self,
webhook_url: str,
channel: str,
callback_url: str,
timeout_seconds: float = 300.0,
):
self.webhook_url = webhook_url
self.channel = channel
self.callback_url = callback_url
self.timeout = timeout_seconds
async def render_approval(
self, ctx: ActionContext, risk: RiskAssessment
) -> Verdict:
"""Post an interactive message and wait for a button click."""
blocks = self._build_approval_blocks(ctx, risk)
request_id = await self._post_message(blocks)
response = await self._wait_for_response(request_id)
if response == "approve":
return Verdict.APPROVED
return Verdict.DENIED
async def render_challenge(
self,
ctx: ActionContext,
risk: RiskAssessment,
challenge_type: ChallengeType,
) -> ChallengeResult:
"""Post a challenge prompt and collect the response."""
blocks = self._build_challenge_blocks(ctx, risk, challenge_type)
request_id = await self._post_message(blocks)
response = await self._wait_for_response(request_id)
return ChallengeResult(
passed=(response == "approve"),
challenge_type=challenge_type,
responder=self.channel,
)
async def render_info(self, message: str) -> None:
"""Post an informational message to the channel."""
blocks = [
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f":information_source: *Attesta* {message}",
},
}
]
await self._post_message(blocks)
async def render_auto_approved(
self, ctx: ActionContext, risk: RiskAssessment
) -> None:
"""Optionally notify on auto-approvals. Often a no-op."""
pass # Silent for low-risk actions
# -- internal helpers ------------------------------------------
def _build_approval_blocks(
self, ctx: ActionContext, risk: RiskAssessment
) -> list[dict]:
risk_emoji = {
"low": ":large_green_circle:",
"medium": ":large_yellow_circle:",
"high": ":red_circle:",
"critical": ":no_entry:",
}
emoji = risk_emoji.get(risk.level.value, ":grey_question:")
return [
{
"type": "header",
"text": {
"type": "plain_text",
"text": "Attesta Approval Request",
},
},
{
"type": "section",
"fields": [
{
"type": "mrkdwn",
"text": f"*Action:*\n`{ctx.function_name}`",
},
{
"type": "mrkdwn",
"text": (
f"*Risk:* {emoji} "
f"{risk.level.value.upper()} "
f"({risk.score:.2f})"
),
},
],
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*Call:*\n```{ctx.description}```",
},
},
{
"type": "actions",
"elements": [
{
"type": "button",
"text": {"type": "plain_text", "text": "Approve"},
"style": "primary",
"value": "approve",
},
{
"type": "button",
"text": {"type": "plain_text", "text": "Deny"},
"style": "danger",
"value": "deny",
},
],
},
]
def _build_challenge_blocks(
self,
ctx: ActionContext,
risk: RiskAssessment,
challenge_type: ChallengeType,
) -> list[dict]:
blocks = self._build_approval_blocks(ctx, risk)
# Insert a challenge question before the action buttons
challenge_block = {
"type": "section",
"text": {
"type": "mrkdwn",
"text": (
f":warning: *{challenge_type.value.upper()} Challenge*\n"
f"Please review the action details carefully "
f"before responding."
),
},
}
blocks.insert(-1, challenge_block)
return blocks
async def _post_message(self, blocks: list[dict]) -> str:
"""Post a block-kit message and return a request ID."""
import uuid
request_id = uuid.uuid4().hex
async with aiohttp.ClientSession() as session:
payload = {
"channel": self.channel,
"blocks": blocks,
"metadata": {
"event_type": "attesta_approval",
"event_payload": {"request_id": request_id},
},
}
await session.post(self.webhook_url, json=payload)
return request_id
async def _wait_for_response(self, request_id: str) -> str:
"""Poll for a response (replace with webhook in production)."""
# In production, use a webhook callback or WebSocket
# This is a simplified polling example
async with aiohttp.ClientSession() as session:
deadline = asyncio.get_event_loop().time() + self.timeout
while asyncio.get_event_loop().time() < deadline:
resp = await session.get(
f"{self.callback_url}/responses/{request_id}"
)
data = await resp.json()
if data.get("status") == "responded":
return data["value"]
await asyncio.sleep(2.0)
return "deny" # Timeout defaults to deny
Register with Attesta
from attesta import gate
slack = SlackRenderer(
webhook_url="https://hooks.slack.com/services/T.../B.../xxx",
channel="#ai-approvals",
callback_url="https://your-api.example.com/attesta",
)
@gate(renderer=slack)
def deploy_to_production(service: str) -> str:
return f"Deployed {service}"
Step-by-Step: Web UI Renderer
A web-based renderer sends approval requests to a dashboard and receives responses via WebSocket or server-sent events.Define the Renderer
import asyncio
import json
import uuid
import aiohttp
from attesta import (
ActionContext,
RiskAssessment,
ChallengeType,
ChallengeResult,
Verdict,
)
class WebUIRenderer:
"""Sends approval requests to a web dashboard via REST API."""
def __init__(
self,
api_base_url: str,
api_key: str,
timeout_seconds: float = 600.0,
):
self.api_base = api_base_url.rstrip("/")
self.api_key = api_key
self.timeout = timeout_seconds
async def render_approval(
self, ctx: ActionContext, risk: RiskAssessment
) -> Verdict:
request_id = await self._create_request(ctx, risk, "approval")
response = await self._poll_response(request_id)
return Verdict.APPROVED if response == "approved" else Verdict.DENIED
async def render_challenge(
self,
ctx: ActionContext,
risk: RiskAssessment,
challenge_type: ChallengeType,
) -> ChallengeResult:
request_id = await self._create_request(
ctx, risk, "challenge", challenge_type
)
response = await self._poll_response(request_id)
return ChallengeResult(
passed=(response == "approved"),
challenge_type=challenge_type,
responder="web-dashboard",
)
async def render_info(self, message: str) -> None:
async with aiohttp.ClientSession() as session:
await session.post(
f"{self.api_base}/notifications",
headers={"Authorization": f"Bearer {self.api_key}"},
json={"type": "info", "message": message},
)
async def render_auto_approved(
self, ctx: ActionContext, risk: RiskAssessment
) -> None:
# Log to dashboard activity feed
async with aiohttp.ClientSession() as session:
await session.post(
f"{self.api_base}/activity",
headers={"Authorization": f"Bearer {self.api_key}"},
json={
"type": "auto_approved",
"action": ctx.function_name,
"risk_score": risk.score,
},
)
async def _create_request(
self,
ctx: ActionContext,
risk: RiskAssessment,
request_type: str,
challenge_type: ChallengeType | None = None,
) -> str:
request_id = uuid.uuid4().hex
payload = {
"request_id": request_id,
"type": request_type,
"action": ctx.function_name,
"description": ctx.description,
"risk_score": risk.score,
"risk_level": risk.level.value,
"environment": ctx.environment,
"agent_id": ctx.agent_id,
}
if challenge_type:
payload["challenge_type"] = challenge_type.value
async with aiohttp.ClientSession() as session:
await session.post(
f"{self.api_base}/requests",
headers={"Authorization": f"Bearer {self.api_key}"},
json=payload,
)
return request_id
async def _poll_response(self, request_id: str) -> str:
async with aiohttp.ClientSession() as session:
deadline = asyncio.get_event_loop().time() + self.timeout
while asyncio.get_event_loop().time() < deadline:
resp = await session.get(
f"{self.api_base}/requests/{request_id}",
headers={"Authorization": f"Bearer {self.api_key}"},
)
data = await resp.json()
if data.get("status") in ("approved", "denied"):
return data["status"]
await asyncio.sleep(1.0)
return "denied" # Timeout = deny
Example: Multi-Channel Renderer
A renderer that fans out approval requests to multiple channels simultaneously and accepts the first response.import asyncio
from attesta import (
ActionContext,
RiskAssessment,
ChallengeType,
ChallengeResult,
Verdict,
)
class MultiChannelRenderer:
"""Sends to multiple renderers; first response wins."""
def __init__(self, renderers: list):
self.renderers = renderers
async def render_approval(
self, ctx: ActionContext, risk: RiskAssessment
) -> Verdict:
tasks = [
asyncio.create_task(r.render_approval(ctx, risk))
for r in self.renderers
]
done, pending = await asyncio.wait(
tasks, return_when=asyncio.FIRST_COMPLETED
)
# Cancel remaining tasks
for task in pending:
task.cancel()
return done.pop().result()
async def render_challenge(
self,
ctx: ActionContext,
risk: RiskAssessment,
challenge_type: ChallengeType,
) -> ChallengeResult:
tasks = [
asyncio.create_task(
r.render_challenge(ctx, risk, challenge_type)
)
for r in self.renderers
]
done, pending = await asyncio.wait(
tasks, return_when=asyncio.FIRST_COMPLETED
)
for task in pending:
task.cancel()
return done.pop().result()
async def render_info(self, message: str) -> None:
await asyncio.gather(
*(r.render_info(message) for r in self.renderers)
)
async def render_auto_approved(
self, ctx: ActionContext, risk: RiskAssessment
) -> None:
await asyncio.gather(
*(r.render_auto_approved(ctx, risk) for r in self.renderers)
)
from attesta import Attesta
attesta = Attesta(
renderer=MultiChannelRenderer([
SlackRenderer(webhook_url="...", channel="#approvals", callback_url="..."),
WebUIRenderer(api_base_url="https://...", api_key="..."),
])
)
Inheriting from BaseRenderer
If you prefer an abstract-base-class approach over the structural protocol, you can extendBaseRenderer:
from attesta.renderers.base import BaseRenderer
from attesta import (
ActionContext, RiskAssessment,
ChallengeType, ChallengeResult, Verdict,
)
class EmailRenderer(BaseRenderer):
"""Sends approval requests via email with magic-link responses."""
def __init__(self, smtp_host: str, from_addr: str, to_addr: str):
self.smtp_host = smtp_host
self.from_addr = from_addr
self.to_addr = to_addr
async def render_approval(
self, ctx: ActionContext, risk: RiskAssessment
) -> Verdict:
# Send email with approve/deny links
# Wait for callback
...
async def render_challenge(
self,
ctx: ActionContext,
risk: RiskAssessment,
challenge_type: ChallengeType,
) -> ChallengeResult:
# Send challenge email, collect response
...
async def render_auto_approved(
self, ctx: ActionContext, risk: RiskAssessment
) -> None:
pass # No notification for auto-approvals
async def render_info(self, message: str) -> None:
# Send informational email
...
BaseRenderer is an abstract class that will raise TypeError at instantiation if you forget to implement any required method. The Renderer protocol, by contrast, only catches missing methods at runtime isinstance() checks or through static type checkers.Renderer Selection at Runtime
Attesta auto-detects the best renderer when none is explicitly provided:1. If renderer= is passed to @gate or Attesta() → use that
2. If stdin is a TTY and `rich` is installed → TerminalRenderer
3. Otherwise → _DefaultRenderer (auto-approve, for CI/headless)
from attesta import Attesta
# Always use your custom renderer, never auto-detect
attesta = Attesta(renderer=SlackRenderer(...))
The
_DefaultRenderer auto-approves everything. In production, always specify an explicit renderer to ensure human oversight is not accidentally bypassed in non-TTY environments (containers, cron jobs, systemd services).Testing Custom Renderers
import pytest
from attesta import (
ActionContext, RiskAssessment, RiskLevel,
ChallengeType, Verdict, Renderer,
)
@pytest.mark.asyncio
async def test_slack_renderer_satisfies_protocol():
renderer = SlackRenderer(
webhook_url="https://example.com",
channel="#test",
callback_url="https://example.com",
)
assert isinstance(renderer, Renderer)
@pytest.mark.asyncio
async def test_render_info_does_not_raise():
renderer = SlackRenderer(
webhook_url="https://example.com",
channel="#test",
callback_url="https://example.com",
)
# Should not raise even if the webhook is unreachable in tests
# (wrap in try/except or mock the HTTP call)
...
Protocols
Full Renderer protocol specification
Renderers Concept
How Attesta selects and uses renderers