Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.vaultak.com/llms.txt

Use this file to discover all available pages before exploring further.

Agency Swarm is a multi-agent orchestration framework built around role-based agents and tool-calling. Vaultak plugs into two integration points it exposes natively.

Input & Output Guardrails

Risk-score incoming messages and mask PII in agent responses using Agency Swarm’s native guardrail decorators.

BaseTool Mixin

Intercept individual tool calls before they execute by mixing Vaultak checks into any BaseTool subclass.

​
Install

pip install vaultak agency-swarm
Sign up at vaultak.com to get your API key (starts with vtk_).

​
Guardrail Approach

Agency Swarm’s @input_guardrail and @output_guardrail decorators let you intercept messages at the agency boundary without touching any tool code.
  • Input guardrail — risk-scores each incoming message before the agent processes it; blocks if the score meets or exceeds your threshold.
  • Output guardrail — masks PII in agent responses before they reach the user.
import asyncio
import os

from agency_swarm import (
    Agent,
    Agency,
    GuardrailFunctionOutput,
    RunContextWrapper,
    input_guardrail,
    output_guardrail,
)
from vaultak import Vaultak

vt = Vaultak(api_key=os.environ["VAULTAK_API_KEY"], agent_name="agency-swarm-agent")
RISK_THRESHOLD = 7.0


@input_guardrail
async def vaultak_input_guard(
    context: RunContextWrapper,
    agent: Agent,
    user_input: str | list[str],
) -> GuardrailFunctionOutput:
    text = user_input if isinstance(user_input, str) else " ".join(user_input)
    result = await asyncio.to_thread(vt.score_action, action="user-message", context={"input": text})
    if result.score >= RISK_THRESHOLD:
        return GuardrailFunctionOutput(
            output_info=f"[Vaultak] Blocked — risk score {result.score:.1f}/10. Review at app.vaultak.com",
            tripwire_triggered=True,
        )
    await asyncio.to_thread(vt.check_policy, tool_name="user-message", input_data=text)
    return GuardrailFunctionOutput(output_info="", tripwire_triggered=False)


@output_guardrail
async def vaultak_output_guard(
    context: RunContextWrapper,
    agent: Agent,
    response_text: str,
) -> GuardrailFunctionOutput:
    if isinstance(response_text, str):
        masked = await asyncio.to_thread(vt.mask_pii, response_text)
        if masked != response_text:
            return GuardrailFunctionOutput(output_info=masked, tripwire_triggered=True)
    return GuardrailFunctionOutput(output_info="", tripwire_triggered=False)


ceo = Agent(
    name="CEO",
    instructions="You orchestrate the agency and answer user questions.",
    input_guardrails=[vaultak_input_guard],
    output_guardrails=[vaultak_output_guard],
)

agency = Agency(agents=[ceo])
asyncio.to_thread() is required because Vaultak SDK calls are synchronous and guardrail functions are async. This keeps the event loop non-blocking.

​
Tool-Level Approach

For per-tool risk scoring, subclass BaseTool through the VaultakMixin. The mixin intercepts run(), calls Vaultak before execution, and masks PII in string outputs afterward. 
import os
from typing import Any

from agency_swarm.tools import BaseTool
from pydantic import Field
from vaultak import Vaultak

vt = Vaultak(api_key=os.environ["VAULTAK_API_KEY"], agent_name="agency-swarm-agent")
RISK_THRESHOLD = 7.0


class VaultakMixin:
    def run(self) -> Any:
        tool_name = self.__class__.__name__
        args = self.model_dump()  # type: ignore[attr-defined]

        result = vt.score_action(action=tool_name, context=args)
        if result.score >= RISK_THRESHOLD:
            raise RuntimeError(
                f"[Vaultak] '{tool_name}' blocked — risk score {result.score:.1f}/10. "
                "Review at app.vaultak.com"
            )

        vt.check_policy(tool_name=tool_name, input_data=str(args))

        output = super().run()  # type: ignore[misc]
        return vt.mask_pii(output) if isinstance(output, str) else output


class LookupCustomer(VaultakMixin, BaseTool):
    """Look up a customer record by ID."""

    customer_id: str = Field(..., description="The customer ID to look up.")

    def run(self) -> str:
        return f"Customer {self.customer_id}: Alice Smith, alice@example.com"
Inherit VaultakMixin before BaseTool so Python’s MRO calls VaultakMixin.run() first, then chains to your tool’s run() via super().

​
Combined Example

Use both approaches together for defence-in-depth: guardrails screen all messages, the mixin secures individual high-risk tools. 
import asyncio
import os

from agency_swarm import Agent, Agency, GuardrailFunctionOutput, RunContextWrapper, input_guardrail
from agency_swarm.tools import BaseTool
from pydantic import Field
from vaultak import Vaultak

vt = Vaultak(api_key=os.environ["VAULTAK_API_KEY"], agent_name="agency-swarm-agent")
RISK_THRESHOLD = 7.0


@input_guardrail
async def vaultak_input_guard(
    context: RunContextWrapper, agent: Agent, user_input: str | list[str]
) -> GuardrailFunctionOutput:
    text = user_input if isinstance(user_input, str) else " ".join(user_input)
    result = await asyncio.to_thread(vt.score_action, action="user-message", context={"input": text})
    if result.score >= RISK_THRESHOLD:
        return GuardrailFunctionOutput(
            output_info=f"[Vaultak] Blocked — risk score {result.score:.1f}/10. Review at app.vaultak.com",
            tripwire_triggered=True,
        )
    return GuardrailFunctionOutput(output_info="", tripwire_triggered=False)


class VaultakMixin:
    def run(self):
        tool_name = self.__class__.__name__
        args = self.model_dump()
        result = vt.score_action(action=tool_name, context=args)
        if result.score >= RISK_THRESHOLD:
            raise RuntimeError(
                f"[Vaultak] '{tool_name}' blocked — risk score {result.score:.1f}/10. "
                "Review at app.vaultak.com"
            )
        vt.check_policy(tool_name=tool_name, input_data=str(args))
        output = super().run()
        return vt.mask_pii(output) if isinstance(output, str) else output


class SendEmail(VaultakMixin, BaseTool):
    """Send an email to a recipient."""

    to: str = Field(..., description="Recipient email address.")
    subject: str = Field(..., description="Email subject.")
    body: str = Field(..., description="Email body.")

    def run(self) -> str:
        return f"Email sent to {self.to} — subject: '{self.subject}'"


ceo = Agent(
    name="CEO",
    instructions="You are an executive assistant. Use SendEmail for outbound communication.",
    tools=[SendEmail],
    input_guardrails=[vaultak_input_guard],
)

agency = Agency(agents=[ceo])

​
Stricter thresholds for sensitive agencies

For agencies with access to databases, payment systems, or external APIs, lower the threshold to block medium-risk actions: 
RISK_THRESHOLD = 5.0  # Block anything scoring above medium risk

​
Configuration reference

ParameterTypeDefaultDescription
api_keystr—Your Vaultak API key — required
agent_namestr"agency-swarm-agent"Label shown in the Vaultak dashboard
RISK_THRESHOLDfloat7.0Score (0-10) at or above which actions are blocked

​
What gets monitored

EventVaultak action
Incoming message (input guardrail)Risk-scored; blocked via tripwire_triggered=True if score >= threshold
Policy checkValidated against your dashboard-configured rules
Tool call (VaultakMixin)Risk-scored; blocked via RuntimeError if score >= threshold
Tool output (VaultakMixin)Scanned for PII and masked before returning to the agent
Agent response (output guardrail)PII masked before the response reaches the user