Agency Swarm - Vaultak

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

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

Parameter	Type	Default	Description
`api_key`	`str`	—	Your Vaultak API key — required
`agent_name`	`str`	`"agency-swarm-agent"`	Label shown in the Vaultak dashboard
`RISK_THRESHOLD`	`float`	`7.0`	Score (0-10) at or above which actions are blocked

What gets monitored

Event	Vaultak action
Incoming message (input guardrail)	Risk-scored; blocked via `tripwire_triggered=True` if score >= threshold
Policy check	Validated 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

Input & Output Guardrails

BaseTool Mixin

​Install

​Guardrail Approach

​Tool-Level Approach

​Combined Example

​Stricter thresholds for sensitive agencies

​Configuration reference

​What gets monitored

​Links

Install

Guardrail Approach

Tool-Level Approach

Combined Example

Stricter thresholds for sensitive agencies

Configuration reference

What gets monitored

Links