Documentation Index
Fetch the complete documentation index at: https://docs.definable.ai/llms.txt
Use this file to discover all available pages before exploring further.
Hooks let you inject custom logic at four points in the message processing pipeline. Use them for logging, access control, content filtering, analytics, or any cross-cutting concern.
Hook Points
The InterfaceHook Protocol
Implement any combination of these four methods. All are optional — only implement the ones you need:
from definable.agent.interface import InterfaceHook, InterfaceMessage, InterfaceResponse, InterfaceSession
class MyHook:
async def on_message_received(self, message: InterfaceMessage) -> bool | None:
"""Called when a message arrives. Return False to skip it."""
...
async def on_before_respond(
self, message: InterfaceMessage, session: InterfaceSession
) -> InterfaceMessage | None:
"""Called before agent processing. Return a new message to modify it."""
...
async def on_after_respond(
self, message: InterfaceMessage, response: InterfaceResponse, session: InterfaceSession
) -> InterfaceResponse | None:
"""Called after agent produces a response. Return a new response to modify it."""
...
async def on_error(self, error: Exception, message: InterfaceMessage | None) -> None:
"""Called when an error occurs."""
...
Adding Hooks
Use .add_hook() for fluent chaining:
from definable.agent.interface import TelegramInterface, LoggingHook, AllowlistHook
interface = (
TelegramInterface(agent=agent, bot_token="YOUR_BOT_TOKEN")
.add_hook(LoggingHook())
.add_hook(AllowlistHook(allowed_user_ids={"123", "456"}))
)
interface = TelegramInterface(
agent=agent,
bot_token="YOUR_BOT_TOKEN",
hooks=[LoggingHook(), AllowlistHook(allowed_user_ids={"123"})],
)
on_message_received, if any hook returns False, processing stops and subsequent hooks are skipped.
Built-in Hooks
LoggingHook
Logs incoming messages and errors for observability:
from definable.agent.interface import LoggingHook
interface.add_hook(LoggingHook())
on_message_received (user, chat, text preview) and on_error (exception details).
AllowlistHook
Restricts access to a set of user IDs:
from definable.agent.interface import AllowlistHook
interface.add_hook(AllowlistHook(allowed_user_ids={"user-123", "user-456"}))
False from on_message_received).
Writing Custom Hooks
Access Control
class BusinessHoursHook:
"""Only respond during business hours."""
async def on_message_received(self, message):
from datetime import datetime
hour = datetime.now().hour
if hour < 9 or hour >= 17:
return False # Silently drop messages outside 9am-5pm
return None # Continue processing
Content Filtering
class ContentFilterHook:
"""Filter inappropriate content from responses."""
async def on_after_respond(self, message, response, session):
if response.content and contains_inappropriate(response.content):
from definable.agent.interface import InterfaceResponse
return InterfaceResponse(
content="I'm sorry, I can't provide that information."
)
return None # Keep original response
Analytics
class AnalyticsHook:
"""Track message and response metrics."""
def __init__(self):
self.message_count = 0
self.error_count = 0
async def on_message_received(self, message):
self.message_count += 1
return None
async def on_error(self, error, message):
self.error_count += 1
await report_to_dashboard(error, message)
Message Enrichment
class ContextHook:
"""Add context to messages before agent processing."""
async def on_before_respond(self, message, session):
from definable.agent.interface import InterfaceMessage
user_tier = await lookup_user_tier(message.platform_user_id)
enriched_text = f"[User tier: {user_tier}] {message.text}"
return InterfaceMessage(
platform=message.platform,
platform_user_id=message.platform_user_id,
platform_chat_id=message.platform_chat_id,
platform_message_id=message.platform_message_id,
text=enriched_text,
username=message.username,
images=message.images,
audio=message.audio,
metadata=message.metadata,
)
Error Notification
class SlackAlertHook:
"""Send critical errors to Slack."""
def __init__(self, webhook_url: str):
self.webhook_url = webhook_url
async def on_error(self, error, message):
import httpx
user = message.platform_user_id if message else "unknown"
await httpx.AsyncClient().post(self.webhook_url, json={
"text": f"Interface error for user {user}: {error}"
})
Hook Execution Order
When multiple hooks are registered:
on_message_received: Runs in order. First False vetoes the message and stops the chain.on_before_respond: Runs in order. If a hook returns a modified message, subsequent hooks see the modified version.on_after_respond: Runs in order. If a hook returns a modified response, subsequent hooks see the modified version.on_error: All hooks run, regardless of whether earlier hooks raised exceptions.
