> ## 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 > Risk-score every tool call and agent message, enforce policy rules, and mask PII in Agency Swarm agencies. [Agency Swarm](https://github.com/VRSEN/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. Risk-score incoming messages and mask PII in agent responses using Agency Swarm's native guardrail decorators. Intercept individual tool calls before they execute by mixing Vaultak checks into any BaseTool subclass. ## Install ```bash theme={null} pip install vaultak agency-swarm ``` Sign up at [vaultak.com](https://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. ```python theme={null} 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. ```python theme={null} 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. ```python theme={null} 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: ```python theme={null} 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 | ## Links * [Agency Swarm on GitHub](https://github.com/VRSEN/agency-swarm) * [Vaultak dashboard](https://app.vaultak.com) * [Vaultak API reference](/api-reference)