Strands Agents Quickstart

pip install visiq "strands-agents>=0.1"

import functools
import inspect

from visiq import Governor, ToolBlocked  # pip install visiq

gov = Governor(agent_id="support-bot")


def scenario_search(query: str) -> list:
    # A stand-in retrieval source — your real RAG store returns the same shape:
    # a list of {"page_content", "metadata"} dicts gate_documents can filter.
    return [
        {"page_content": f"Q3 revenue was $4.2M. (matched: {query})",
         "metadata": {"classification": "internal"}},
    ]


def governed(name, fn):
    # Route one tool body through the gate. gate_tool decides BEFORE the body
    # runs: a deny raises ToolBlocked; a mask verdict hands the callback ONLY the
    # redacted arguments, never the originals.
    sig = inspect.signature(fn)

    @functools.wraps(fn)
    def wrapper(*a, **kw):
        bound = sig.bind(*a, **kw)
        bound.apply_defaults()
        args = dict(bound.arguments)
        try:
            return gov.gate_tool(name, args, lambda effective: fn(**effective))
        except ToolBlocked as e:
            return f"[BLOCKED BY POLICY] {e.reason}"

    wrapper.__name__ = name
    return wrapper


def issue_refund(customer_id: str, amount: int) -> str:
    return f"Refunded ${amount} to {customer_id}"


def search_knowledge(query: str) -> str:
    # Retrieval governance: drop / redact documents before the model sees them.
    docs = gov.gate_documents(scenario_search(query), query=query)
    return "\n\n".join(d["page_content"] for d in docs)

from strands import Agent
from strands.hooks import HookProvider, HookRegistry
from strands.hooks.events import BeforeToolCallEvent


class VisiqHooks(HookProvider):
    def register_hooks(self, registry: HookRegistry) -> None:
        registry.add_callback(BeforeToolCallEvent, self.before_tool)

    def before_tool(self, event: BeforeToolCallEvent) -> None:
        gov.gate_tool(event.tool_use["name"],
                      dict(event.tool_use.get("input", {})),
                      lambda effective: None)


agent = Agent(tools=[issue_refund, search_knowledge], hooks=[VisiqHooks()])
gov.start(tools=[{"name": "issue_refund"}, {"name": "search_knowledge"}])
gov.flush()  # deliver buffered audit events before the process exits