alovia-watchdog

Runtime protection for AI agents. Guard your agent's inputs with your Alovia fleet key.

Install

pip install alovia-watchdog

Use

from alovia_watchdog import Watchdog, guard_input

wd = Watchdog(api_key="wd_live_...")

# Pass session_id: it groups a conversation's turns so the scope-expansion and
# cascade-anomaly detectors can see across them. Without it they have nothing to
# correlate and stay silent, so always send a stable per-conversation id.
@guard_input(wd, agent="support-bot", session_id=lambda: chat_id)
def handle(message):
    ...

guard_input raises WatchdogBlocked when verdict.enforce says to stop, so the wrapped function never runs on a malicious or unresolved input. On a network error it honors fail_mode ("closed" blocks, "open" allows).

You can also check directly:

verdict = wd.check("some user input", agent="support-bot", session_id=chat_id)
if verdict.enforce:
    ...

Use verdict.enforce, not verdict.blocked, to decide whether to stop. enforce answers "should I stop": it always stops on block, and it also honors fail_mode for challenge and unknown. blocked only answers "did the server say block", so code that checks blocked still lets challenge and unknown through even when fail_mode is "closed".

LangChain

Optional extra:

pip install alovia-watchdog[langchain]

WatchdogToolMiddleware gates every tool call through Watchdog before it runs, blocking off-mission or injected calls with a ToolMessage instead of letting the tool execute:

from langchain.agents import create_agent
from alovia_watchdog import Watchdog
from alovia_watchdog.integrations.langchain import WatchdogToolMiddleware

wd = Watchdog(api_key="wd_live_...")

agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[WatchdogToolMiddleware(wd, agent="support-bot", session_id=lambda: chat_id)],
)

For async agent graphs, the middleware also exposes awrap_tool_call, LangChain calls it automatically wherever the graph runs async.